DistributedMultiThreadNetwork.cpp

Go to the documentation of this file.

#include "DistributedMultiThreadNetwork.h"
#include "MPIAllToAllCommunicatorFactory.h"
#include "MultiThreadNetwork.h"
#include "SpikeSender.h"
#include "PCSIMException.h"

#include <iostream>

using std::cout;
using std::cerr;
using std::endl;

#include <string>
using std::string;

#include <boost/format.hpp>



DistributedMultiThreadNetwork::DistributedMultiThreadNetwork(int numThreads, SimParameter sp )
        : DistributedNetwork( numThreads, MPI::COMM_WORLD, sp ), nThreads(numThreads), thrPool(numThreads-1)
{
    init();
}

DistributedMultiThreadNetwork::DistributedMultiThreadNetwork(int numThreads, MPI::Intracomm &mpiCommunicator, SimParameter sp )
        : DistributedNetwork( numThreads, mpiCommunicator, sp ), nThreads(numThreads), thrPool(numThreads-1)
{
    init();
}


DistributedMultiThreadNetwork::~DistributedMultiThreadNetwork()
{
    delete distEngine;
    delete mpiAllToAllComm;
    delete distIncomingSpikeScheduler;
    delete localSimEngine;
    delete spikeScheduler;
    delete mpiInBuffers;
    delete mpiOutBuffers;
    delete globalDelayMap;
    delete targetNodesMap;
    delete localSpikeScheduler;
    for (int i = 0 ; i < nThreads ; ++i) {
        delete STBuffers[i];
    }
    delete mtTables;

    // delete dispatchers
    for (int i = 0; i < mpi_comm.Get_size(); ++i)
        delete sources2BufPositionsMaps[i];

    for (int thr = 0; thr < nThreads; ++thr) {
        delete mTanalogMsgDispatchers[thr];
        delete distAnalogMsgDispatchers[thr];
        for (int i = 0; i < mpi_comm.Get_size(); ++i) {
            delete incomingDispatchers[thr][i];
            delete outgoingDispatchers[thr][i];
        }
    }

    delete mtDistCycledAnalogMsgDispatcher;

    // delete creators
    delete mtAnalogMsgCreator;
    delete analogDelayObjectsMap;
    for (int thr = 0; thr < nThreads; ++thr) {
        delete stAnalogMsgCreators[thr];
        delete distAnalogMessageCreators[thr];
    }

}

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

    mtTables = new MTSpikeRoutingTables(nThreads);

    STBuffers.resize(nThreads);
    for (int i = 0 ; i < nThreads ; ++i) {
        STBuffers[i] = new PropagatedSpikeBuffer(simParam.minDelay.in_steps( get_dt() ), simParam.maxDelay.in_steps( get_dt() ));
    }

    localSpikeScheduler = new MultiThreadSpikeScheduler(nThreads, *mtTables, STBuffers, simParam ) ;

    targetNodesMap = new TargetNodesMap();
    globalDelayMap = new GlobalMultiTargetDelayMap(nThreads);

    mpiOutBuffers = new MPIOutputBufferVector(mpi_comm.Get_size(), glengineids[mpi_comm.Get_rank()]);

    mpiInBuffers = new MPIInputBufferVector(glengineids, mpi_comm.Get_size());

    spikeScheduler = new DistributedSpikeScheduler(*localSpikeScheduler,
                     *targetNodesMap,
                     *mpiOutBuffers,
                     nThreads);


    // ********* setup mt local dispatchers
    for (int thr = 0 ; thr < nThreads ; ++thr)
        mTanalogMsgDispatchers.push_back(new MultiThreadAnalogMsgDispatcher( (delay_t)simParam.minDelay.in_steps( get_dt() ) ) );

    // ********** setup distributed dispatchers
    incomingDispatchers.resize(nThreads);
    outgoingDispatchers.resize(nThreads);

    for (int i = 0; i < mpi_comm.Get_size(); ++i)
        sources2BufPositionsMaps.push_back(new DistIncomingAnalogSources2BufPosMap);

    for (int thr = 0 ; thr < nThreads; ++thr) {
        for (int i = 0; i < mpi_comm.Get_size(); ++i) {
            incomingDispatchers[thr].push_back(new DistributedIncomingAnalogMsgDispatcher(
                                                   (delay_t)simParam.minDelay.in_steps( get_dt() ),
                                                   &(*mpiInBuffers)[i],
                                                   sources2BufPositionsMaps[i]));

            outgoingDispatchers[thr].push_back(new DistributedOutgoingAnalogMsgDispatcher(
                                                   0,
                                                   (delay_t)simParam.minDelay.in_steps( get_dt() ),
                                                   &(*mpiOutBuffers)[i]));
        }

        distAnalogMsgDispatchers.push_back(new DistributedAnalogMessageDispatcher(mpi_comm.Get_size(),
                                           (delay_t)simParam.minDelay.in_steps( get_dt() ),
                                           incomingDispatchers[thr],
                                           outgoingDispatchers[thr],
                                           *mTanalogMsgDispatchers[thr]));
    }

    mtDistCycledAnalogMsgDispatcher = new MTDistributedCycledAnalogMsgDispatcher(thrPool,
                                      distAnalogMsgDispatchers);

    // local engine
    AnalogMessageDispatcherVectorImpl<DistributedAnalogMessageDispatcher> dispatcherVector(distAnalogMsgDispatchers);

    localSimEngine = new MultiThreadSimEngine(0, nThreads, thrPool, *spikeScheduler, dispatcherVector, *this );

    // ****** setup creators
    // local creators
    analogDelayObjectsMap = new AnalogDelayObjectMap;

    for (int i = 0; i < nThreads ; ++i) {
        stAnalogMsgCreators.push_back(new STAnalogMessageCreator(
                                          mTanalogMsgDispatchers[i]->sTLocalDispatcher(),
                                          (dynamic_cast<MultiThreadSimEngine *>(localSimEngine))->getSTEngine(i),
                                          *analogDelayObjectsMap,
                                          (delay_t)simParam.minDelay.in_steps( simParam.dt) ) );
    }

    mtAnalogMsgCreator = new MTAnalogMessageCreator(mTanalogMsgDispatchers,
                         *dynamic_cast<MultiThreadSimEngine *>(localSimEngine),
                         *analogDelayObjectsMap,
                         (delay_t)simParam.minDelay.in_steps( simParam.dt));
    // distributed creators
    for (int i = 0; i < nThreads ; ++i) {
        distAnalogMessageCreators.push_back( new DistributedAnalogMessageCreator(
                                                 (dynamic_cast<MultiThreadSimEngine *>(localSimEngine))->getSTEngine(i),
                                                 *distAnalogMsgDispatchers[i],
                                                 (delay_t)simParam.minDelay.in_steps( get_dt() )) );

    }



    distIncomingSpikeScheduler = new MTDistributedIncomingSpikeScheduler(nThreads,
                                 thrPool,
                                 *mpiInBuffers,
                                 *globalDelayMap,
                                 STBuffers,
                                 (delay_t)simParam.minDelay.in_steps( get_dt() ));

    MPIAllToAllCommunicatorFactory mpi_comm_factory(MPIAllToAllCommunicatorFactory::Default);

    mpiAllToAllComm = mpi_comm_factory.getCommunicator(*mpiInBuffers,
                      *mpiOutBuffers,
                      mpi_comm,
                      incomingConnections,
                      outgoingConnections);


    distEngine = new DistributedSimEngine(*localSimEngine,
                                          *distIncomingSpikeScheduler,
                                          *spikeScheduler,
                                          *mtDistCycledAnalogMsgDispatcher,
                                          *mpiAllToAllComm,
                                          *this );

    setupConstructRNGEngines();
    DistributedNetwork::seed_noise_rng( makeSeed( simParam.simulationRNGSeed ) );
    
    // Make sure that all net works are through the init befor anything else can happen
    mpi_comm.Barrier();

}


SimObject * DistributedMultiThreadNetwork::_getObject_(const SimObject::ID &id)
{
    if (id.node == mpi_comm.Get_rank())
        return localSimEngine->getObject(id);
    return NULL;
}

void DistributedMultiThreadNetwork::_addSpikeMessage_(const SimObject::ID &sender, const port_t, const spike_port_id_t sender_port, const SimObject::ID &receiver, const port_t port, const step_t delay)
{

    if (sender.node != receiver.node) {
        if (mpi_comm.Get_rank() == sender.node) {
            outgoingConnections[receiver.node] = true;
        } else if (mpi_comm.Get_rank() == receiver.node) {
            incomingConnections[sender.node] = true;
        }
    }

    if (sender.node == mpi_comm.Get_rank() && receiver.node == mpi_comm.Get_rank()) {

        if( (!sender.isInTheSameSTEngineWith(receiver) && delay < simParam.minDelay.in_steps( get_dt() ) ) || delay > simParam.maxDelay.in_steps( get_dt() ) ) {
            throw(
                PCSIM::ConstructionException(
                    "DistributedMultiThreadNetwork::addSpikeMessage",
                    str( boost::format("Specified delay (%1% ms) out of range (min=%2% ms, max=%3% ms).") 
                                          % (delay * simParam.dt.in_ms()) % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() ) 
                )
            );
        }

        MultiThreadNetwork::addLocalSpikeMessage( mtTables, localSimEngine, simParam, sender, sender_port, receiver, port, delay );

    } else if (sender.node == mpi_comm.Get_rank()) {

        targetNodesMap->addTargetNode(sender_port, receiver.node, sender.eng);
        _nSpikeMessages++;

    } else if (receiver.node == mpi_comm.Get_rank()) {
        if( delay < simParam.minDelay.in_steps( get_dt() ) || delay > simParam.maxDelay.in_steps( get_dt() ) ) {
            throw(
                PCSIM::ConstructionException(
                    "DistributedMultiThreadNetwork::addSpikeMessage",
                    str( boost::format("Specified delay (%1% ms) out of range (min=%2% ms, max=%3% ms)." ) % ( delay * simParam.dt.in_ms() ) % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() )
                )
            );
        }

        spikegroupid_t tg = (*globalDelayMap)[receiver.eng].find(glengineids[sender.node][sender.eng], sender_port, (delaystep_t)delay);

        if( tg == no_spikegroup ) {
            tg = mtTables->stgPool.addSpikeTarget( localSimEngine->getObject(receiver), port );
            // cerr << "NODE " << mpi_comm.Get_rank() << " insertdelay(srcnode=" << sender.node << " sed=" << sender_port << ", del=" << delay << ",tg=" << tg <<") type=" << (int)receiver.type << endl;
            (*globalDelayMap)[receiver.eng].insert( glengineids[sender.node][sender.eng], sender_port, (delaystep_t)delay, tg );
        } else {
            mtTables->stgPool.addSpikeTarget( tg, localSimEngine->getObject(receiver), port);
        }
        _nSpikeMessages++;

    }

}

// ------------------------------------ analog messages -------------------------------------- //
template<typename analogSrcType, typename analogDestType>
void DistributedMultiThreadNetwork::addGenericAnalogMessage(const SimObject::ID &sender, analogSrcType srcPortOrField, const SimObject::ID &receiver, analogDestType destPortOrField, delay_t delay)
{

    if (sender.node != receiver.node) {
        if (mpi_comm.Get_rank() == sender.node) {
            outgoingConnections[receiver.node] = true;
        } else if (mpi_comm.Get_rank() == receiver.node) {
            incomingConnections[sender.node] = true;
        }
    }

    if (sender.node == mpi_comm.Get_rank() && receiver.node == mpi_comm.Get_rank()) {
        if (sender.eng == receiver.eng) {
            stAnalogMsgCreators[sender.eng]->addAnalogMessage(sender,
                    srcPortOrField,
                    receiver,
                    destPortOrField,
                    delay);
        } else {
                if(  delay < simParam.minDelay.in_steps( simParam.dt ) || delay > simParam.maxDelay.in_steps( simParam.dt ) ) {                 
                throw(
                    PCSIM::ConstructionException(
                        "DistributedMultiThreadNetwork::addGenericAnalogMessage",
                        str( boost::format("Specified delay (%1% ms) out of range (min=%2% ms, max=%3% ms)." ) % (delay * simParam.dt.in_ms()) % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() )
                    )
                );
            }
            mtAnalogMsgCreator->addAnalogMessage(sender,
                                                 srcPortOrField,
                                                 receiver,
                                                 destPortOrField,
                                                 delay);
        }
    } else if (sender.node == mpi_comm.Get_rank()) {
        distAnalogMessageCreators[sender.eng]->addOutgoingAnalogMessage( sender, srcPortOrField, receiver.node);

    } else if (receiver.node == mpi_comm.Get_rank()) {
        if ( delay < simParam.minDelay.in_steps( get_dt() ) || delay > simParam.maxDelay.in_steps( get_dt() ) ) {
            throw(
                PCSIM::ConstructionException(
                    "DistributedMultiThreadNetwork::addGenericAnalogMessage",
                    str( boost::format("Specified delay (%1% ms) out of range (min=%2% ms, max=%3% ms).") % ( delay * simParam.dt.in_ms() ) % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() )
                )
            );
        }

        distAnalogMessageCreators[receiver.eng]->addIncomingAnalogMessage( sender, srcPortOrField, receiver, destPortOrField, delay);

    }

}



void DistributedMultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, int sender_port, const SimObject::ID &receiver, int recv_port, const Time &delay)
{
    addGenericAnalogMessage(sender, (analog_port_id_t)sender_port, receiver, (analog_port_id_t)recv_port, (delay_t)delay.in_steps( simParam.dt));
}


void DistributedMultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, int sender_port, const SimObject::ID &receiver, string destfield, const Time &delay)
{
    addGenericAnalogMessage(sender, (analog_port_id_t)sender_port, receiver, destfield, (delay_t)delay.in_steps( simParam.dt));
}


void DistributedMultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, string srcfield, const SimObject::ID &receiver, int recv_port, const Time &delay)
{
    addGenericAnalogMessage(sender, srcfield, receiver, (analog_port_id_t)recv_port, (delay_t)delay.in_steps( simParam.dt));
}


void DistributedMultiThreadNetwork::_addAnalogMessage_(const SimObject::ID &sender, string srcfield, const SimObject::ID &receiver, string destfield, const Time &delay)
{
    addGenericAnalogMessage(sender, srcfield, receiver, destfield, (delay_t)delay.in_steps( simParam.dt));
}

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

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

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


void DistributedMultiThreadNetwork::_initialize_()
{
    if (!initialized) {
        localSimEngine->initialize();
        mpiInBuffers->initialize(simParam.minDelay.in_steps( get_dt() ), 0, MPIBUFFER_BLOCK_SIZE);
        mpiOutBuffers->initialize(simParam.minDelay.in_steps( get_dt() ), 0, MPIBUFFER_BLOCK_SIZE);
        for (int i = 0 ; i < nThreads; ++i)
            distAnalogMsgDispatchers[i]->initialize();
        initialized = true;
    }
}

void DistributedMultiThreadNetwork::_reset_()
{
    if( ! initialized )
        initialize();
    spikeScheduler->reset();
    distEngine->reset();
    for (int i = 0 ; i < nThreads; ++i) {
        distAnalogMsgDispatchers[i]->reset(get_dt().in_sec());
        mpiOutBuffers->nextCycle();
    }
    reseted = true;
}

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


void DistributedMultiThreadNetwork::noiseRandEngineOutput( vector<uint32> & R )
{
    localSimEngine->noiseRandEngineOutput( R );
}

---