MultiThreadNetwork.cpp

Go to the documentation of this file.

#include "MultiThreadNetwork.h"
#include "SpikeSender.h"
#include "SingleThreadNetwork.h"
#include "PCSIMException.h"

#include <mpi.h>

#include <string>
using std::string;

#include <boost/format.hpp>

// ------------------------------ constructor  / destructor --------------------------------------- //

MultiThreadNetwork::MultiThreadNetwork(int nT, SimParameter sp ) :
//        SimNetwork( MPI::COMM_WORLD, sp, localRoundRobin ), nThreads(nT), thrPool( nThreads-1 )
        SimNetwork( MPI::COMM_WORLD, sp, &localRoundRobin ), nThreads(nT), thrPool( nThreads-1 )
{
    init();
}

MultiThreadNetwork::MultiThreadNetwork(int nT, MPI::Intracomm &comm, SimParameter sp ) :
//        SimNetwork( comm, sp, localRoundRobin ), nThreads(nT), thrPool( nThreads-1 )
        SimNetwork( comm, sp, &localRoundRobin ), nThreads(nT), thrPool( nThreads-1 )
{
    init();
}

void MultiThreadNetwork::init()
{
        if (simParam.minDelay.in_sec() < simParam.dt.in_sec())
                throw(
                    PCSIM::ConstructionException("MultiThreadNetwork::init",
                str(boost::format("minDelay (%1%) smaller than dt (%2%)") 
                    % simParam.minDelay.in_sec() % simParam.dt.in_sec() )
            ) 
        );

    tables = new MTSpikeRoutingTables(nThreads);

    for (int i = 0; i < nThreads; ++i) {
        PropagatedSpikeBuffer *buf = new PropagatedSpikeBuffer(simParam.minDelay.in_steps( simParam.dt ),
                                     simParam.maxDelay.in_steps( simParam.dt ));
        STBuffers.push_back(buf);
    }

    spikeScheduler = new MultiThreadSpikeScheduler(nThreads, *tables, STBuffers, simParam );

    analogDelayObjectsMap = new AnalogDelayObjectMap;

    for (int i = 0 ; i < nThreads; ++i)
        analogMsgDispatchers.push_back(new MultiThreadAnalogMsgDispatcher( (delay_t)simParam.minDelay.in_steps( simParam.dt) ) );



    AnalogMessageDispatcherVectorImpl<MultiThreadAnalogMsgDispatcher>
    dispatchersVector(analogMsgDispatchers);

    simEngine = new MultiThreadSimEngine(0, nThreads, thrPool, *spikeScheduler,
                                         dispatchersVector,
                                         *this );

    // ----- create the analog message creators ----------//
    for (int i = 0; i < nThreads ; ++i) {
        stAnalogMsgCreators.push_back(new STAnalogMessageCreator(analogMsgDispatchers[i]->sTLocalDispatcher(),
                                      simEngine->getSTEngine(i), *analogDelayObjectsMap,
                                      (delay_t)simParam.minDelay.in_steps( simParam.dt) ) );
    }

    mtAnalogMsgCreator = new MTAnalogMessageCreator(analogMsgDispatchers, *simEngine, *analogDelayObjectsMap,
                         (delay_t)simParam.minDelay.in_steps( simParam.dt));

    initialized = false;

    setupConstructRNGEngines();
    MultiThreadNetwork::seed_noise_rng( makeSeed( simParam.simulationRNGSeed ) );

}

MultiThreadNetwork::~MultiThreadNetwork()
{
    delete simEngine;
    delete spikeScheduler;
    delete tables;
    delete analogDelayObjectsMap;
    for (int i = 0; i < nThreads; ++i) {
        delete STBuffers[i];
        delete stAnalogMsgCreators[i];
        delete analogMsgDispatchers[i];
    }
    delete mtAnalogMsgCreator;
}

void MultiThreadNetwork::seed_noise_rng( uint32 noiseRNGseed )
{
    vector<uint32> sim_seeds(nThreads);
    fillSeedVector( noiseRNGseed, sim_seeds );
    simEngine->seed( sim_seeds );
    objectVariationRNDEngine->seed( makeSeed( simParam.constructionRNGSeed ) ) ;
}

// ------------------------- adding / getting objects ----------------------- //

void MultiThreadNetwork::_addObject_( const SimObjectFactory &objFactory, const SimEngine::ID &loc, SimObject::ID &id )
{
    if( loc.node == _mpi_rank && loc.engine < this->nThreads ) {
        id.node = _mpi_rank;
        id.eng  = loc.engine;
        simEngine->addObject( objFactory, id );
    } else {
        id = SimObject::ID::Invalid;
        if( loc.node != _mpi_rank )
            throw(
                PCSIM::ConstructionException(
                    "MultiThreadNetwork::addObject",
                    str( boost::format("Specified node (%1%) not equal to mpi rank (%2%)" ) % loc.node % _mpi_rank )
                )
            );
        else
            throw(
                PCSIM::ConstructionException(
                    "MultiThreadNetwork::addObject",
                    str( boost::format("Specified engine (%1%) out of range (0,%2%)" ) % 0 % this->nThreads ) 
                )
            );
    }
}

void MultiThreadNetwork::_addObject_( const SimObjectFactory &objFactory, SimObject::ID &id )
{
//    SimEngine::ID loc = distributionStrategy( this );
    SimEngine::ID loc = (*distributionStrategy)( this );
    return addObject( objFactory, loc, id );
}

void MultiThreadNetwork::_mount_( const SimObjectFactory &objFactory, const SimObject::ID &mountpoint, SimObject::ID &id )
{
    if( mountpoint.node == _mpi_rank && mountpoint.eng < this->nThreads ) {
        id.node = _mpi_rank;
        id.eng  = mountpoint.eng;
        simEngine->mount( objFactory, mountpoint, id );
    } else {
        id = SimObject::ID::Invalid;
        if( mountpoint.node != _mpi_rank )
            throw(
                PCSIM::ConstructionException(
                    "MultiThreadNetwork::mount",
                    str( boost::format("Specified node (%1%) not equal to mpi rank (%2%)" ) % mountpoint.node % _mpi_rank )
                )
            );
        else
            throw(
                PCSIM::ConstructionException(
                    "MultiThreadNetwork::mount",
                    str( boost::format("Specified engine (%1%) out of range (0,%2%)" ) % 0 % this->nThreads )
                )
            );
    }
}

void MultiThreadNetwork::_insert_( const SimObjectFactory &objFactory, const SimObject::ID &container, SimObject::ID &id )
{
    if( container.node == _mpi_rank && container.eng < this->nThreads ) {
        id.node = _mpi_rank;
        id.eng  = container.eng;
        simEngine->insert( objFactory, container, id );
    } else {
        id = SimObject::ID::Invalid;
        if( container.node != _mpi_rank )
            throw(
                PCSIM::ConstructionException(
                    "MultiThreadNetwork::mount",
                    str( boost::format("Specified node (%1%) not equal to mpi rank (%2%)") % container.node % _mpi_rank ) 
                )
            );
        else
            throw(
                PCSIM::ConstructionException(
                    "MultiThreadNetwork::mount",
                    str( boost::format("Specified engine (%1%) out of range (0,%2%)")  % 0 % this->nThreads ) 
                )
            );
    }
}


SimObject * MultiThreadNetwork::_getObject_(const SimObject::ID &id)
{
    return simEngine->getObject(id);
}

// -------------------------------------- connections ------------------------------------------ //

void MultiThreadNetwork::_connect_( SimObject::ID const& src, port_t out, const SimObject::ID &dst, port_t in, int delay )
{
    SimObject *src_obj = simEngine->getObject(src);
    SimObject *dst_obj = simEngine->getObject(dst);

    Time delay_to_use;
    if( delay < 0 ) {
        delay_to_use = Time::sec( dst_obj->getManagedDelay() );
    } else {
        delay_to_use = Time::steps( delay, get_dt() );
    }

    if( src_obj->outputPortType(out) == SimObject::spiking && dst_obj->inputPortType( in ) == SimObject::spiking ) {
        addSpikeMessage( src, out, dst, in, delay_to_use );
    } else if ( src_obj->outputPortType(out) == SimObject::analog && dst_obj->inputPortType( in ) == SimObject::analog ) {
        addAnalogMessage( src, out, dst, in, delay_to_use );
    } else {
        throw(
            PCSIM::ConstructionException(
                "MultiThreadNetwork::_connect_",
                str( boost::format("Can not connect specified source (%1%) and destination (%2%) object: no matching out (%3%) and in (%4%) ports.") % src.toString() % dst.toString() % out % in )
            )
        );
    }

}

// ---------------------------------- spike messages -------------------------------------- //

void MultiThreadNetwork::_addSpikeMessage_(const SimObject::ID &sender, const port_t out, const SimObject::ID &receiver, const port_t in, const Time &delay)
{               
    if (  ( sender.eng != receiver.eng && delay < simParam.minDelay) || delay > simParam.maxDelay ) {
        throw(
            PCSIM::ConstructionException(
                "MultiThreadNetwork::addSpikeMessage",
                str( boost::format("Specified delay (%1%) out of range (min=%2% ms, max=%3% ms)." ) % delay.in_ms() % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() ) 
            )
        );
    }
    SimObject *src_obj = simEngine->getObject(sender);
    SpikeSender *ss = dynamic_cast<SpikeSender*>( src_obj );
    if( ss == NULL ) {
        throw( PCSIM::ConstructionException( "MultiThreadNetwork::addSpikeMessage", 
                str( boost::format("Specified simulation object (%1%, %2%) is not a spike sender.") % typeid(*src_obj).name() % sender.toString() ) ) ) ;
    }
    if( ss->getSpikePort( out ) != NULL ) {
        MultiThreadNetwork::addLocalSpikeMessage( tables, simEngine, simParam, sender, ss->getSpikePort( out )->ID(), receiver, in, delay.in_steps( get_dt() ) );
        _nSpikeMessages++;
    } else {
        throw( PCSIM::ConstructionException( "MultiThreadNetwork::addSpikeMessage", str( boost::format("Specified spike output port (%1%) does not exist") % out ) ) );
    }
}

void MultiThreadNetwork::addLocalSpikeMessage(
    MTSpikeRoutingTables *arg_tables, SimEngine *arg_simEngine, const  SimParameter &SP,
    const SimObject::ID &sender, spike_port_id_t sender_port, const SimObject::ID &receiver, const port_t in_port, step_t delay )
{
    if ( sender.eng == receiver.eng ) {
        SingleThreadNetwork::addLocalSpikeMessage( &(arg_tables->localDelayMaps[sender.eng]), &(arg_tables->stgPool), arg_simEngine, sender_port, receiver, in_port, delay );
    } else {
        if( (step_t)delay < SP.minDelay.in_steps( SP.dt ) ) {
            throw( PCSIM::ConstructionException( "MultiThreadNetwork::addSpikeMessage", 
                        str( boost::format("Specified delay (%1%) smaller than minDelay (%2% ms).") % delay % SP.minDelay.in_ms() ) ) );
        }
        spikegroupid_t tg = arg_tables->mTDelayMap.find(sender.eng, sender_port, receiver.eng, (delaystep_t)delay);
        if( tg == no_spikegroup ) {
            tg = arg_tables->stgPool.addSpikeTarget( arg_simEngine->getObject( receiver ), in_port); // receiver, port
            arg_tables->mTDelayMap.insert( sender.eng, sender_port, receiver.eng, (delaystep_t)delay, tg ); // sender_type, sender, delay, targetgroup
        } else {
            arg_tables->stgPool.addSpikeTarget( tg, arg_simEngine->getObject( receiver ), in_port);
        }

    }
}

// ------------------------------------ analog messages -------------------------------------- //

void MultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, int sender_port, const SimObject::ID &receiver, int recv_port, const Time &delay)
{
    _nAnalogMessages++;
    addGenericAnalogMessage(sender, sender_port, receiver, recv_port, (delay_t)delay.in_steps( get_dt() ) );
}

void MultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, int sender_port, const SimObject::ID &receiver, string destfield, const Time &delay)
{
    _nAnalogMessages++;
    addGenericAnalogMessage(sender, sender_port, receiver, destfield, (delay_t)delay.in_steps( get_dt() ) );
}

void MultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, string srcfield, const SimObject::ID &receiver, int recv_port, const Time &delay)
{
    _nAnalogMessages++;
    addGenericAnalogMessage(sender, srcfield, receiver, recv_port, (delay_t)delay.in_steps( get_dt() ) );
}

void MultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, string srcfield, const SimObject::ID &receiver, string destfield, const Time &delay)
{
    _nAnalogMessages++;
    addGenericAnalogMessage(sender, srcfield, receiver, destfield, (delay_t)delay.in_steps( get_dt() ) );
}

// ------------------------------------------------------------------------------------------- //

gl_engineid_t MultiThreadNetwork::maxLocalEngineID(void) const
{
    return nThreads-1;
}

// -------------------------- running the simulation messages -------------------------------- //

void MultiThreadNetwork::_initialize_()
{
    if (!initialized) {
        simEngine->initialize();
        vector<MultiThreadAnalogMsgDispatcher *>::const_iterator it;
        for ( it = analogMsgDispatchers.begin(); it != analogMsgDispatchers.end(); ++it) {
            if (simParam.minDelay.in_steps( simParam.dt ) == 1)
                (*it)->setMinDelay(0);
            else
                (*it)->setMinDelay((delay_t)simParam.minDelay.in_steps(simParam.dt));
        }
        initialized = true;
    }
}

void MultiThreadNetwork::_reset_()
{
    if( ! initialized )
        initialize();
    spikeScheduler->reset();
    simEngine->reset();
    vector<MultiThreadAnalogMsgDispatcher *>::const_iterator it;
    for ( it = analogMsgDispatchers.begin(); it != analogMsgDispatchers.end(); ++it)
        (*it)->reset(get_dt().in_sec());
    reseted = true;
}

void MultiThreadNetwork::_advance_( int nSteps )
{
    if( ! reseted )
        reset();
    simEngine->advance( nSteps );
}

---