DistributedSingleThreadNetwork.cpp

Go to the documentation of this file.

#include "DistributedSingleThreadNetwork.h"
#include "MPIAllToAllCommunicatorFactory.h"

#include "SingleThreadNetwork.h"
#include "SpikeSender.h"

#include <boost/format.hpp>

#include <iostream>
#include <string>

using std::string;

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




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

DistributedSingleThreadNetwork::DistributedSingleThreadNetwork( SimParameter sp )
        : DistributedNetwork(1, MPI::COMM_WORLD, sp )
{
    init();
}

DistributedSingleThreadNetwork::DistributedSingleThreadNetwork( MPI::Intracomm &mpiCommunicator, SimParameter sp )
        : DistributedNetwork(1, mpiCommunicator, sp )
{
    init();
}

void DistributedSingleThreadNetwork::init()
{

    spikeBuffer = new PropagatedSpikeBuffer(simParam.minDelay.in_steps( get_dt() ), simParam.maxDelay.in_steps( get_dt() ) );
    localDelayMap = new LocalDelayMap;

    stgPool = new SpikeTargetGroupPool;

    localSpikeScheduler = new SingleThreadSpikeScheduler(*localDelayMap, *stgPool, *spikeBuffer, simParam) ;

    targetNodesMap = new TargetNodesMap();
    globalDelayMap = new GlobalSingleTargetDelayMap();

    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);

    localAnalogMsgDispatcher = new SingleThreadAnalogMsgDispatcher;

    // create incoming and outgoing dispatchers
    for (int i = 0; i < mpi_comm.Get_size(); ++i) {
        sources2BufPositionsMaps.push_back(new DistIncomingAnalogSources2BufPosMap);
        incomingDispatchers.push_back(new DistributedIncomingAnalogMsgDispatcher(
                                          (delay_t)simParam.minDelay.in_steps( get_dt() ),
                                          &(*mpiInBuffers)[i],
                                          sources2BufPositionsMaps[i]));

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

    distAnalogMsgDispatcher = new DistributedAnalogMessageDispatcher(mpi_comm.Get_size(),
                              simParam.minDelay.in_steps( get_dt() ),
                              incomingDispatchers,
                              outgoingDispatchers,
                              *localAnalogMsgDispatcher);



    SingleThreadSimEngine *ste = new SingleThreadSimEngine(0, *spikeScheduler, *distAnalogMsgDispatcher, *this );
    localSimEngine = ste;

    // setEnginePtr( ste );

    delayObjectMap = new AnalogDelayObjectMap;

    localAnalogMsgCreator = new STAnalogMessageCreator(
                                *static_cast<SingleThreadAnalogMsgDispatcher *>(localAnalogMsgDispatcher),
                                *ste,
                                *delayObjectMap,
                                (delay_t)simParam.minDelay.in_steps( get_dt() )) ;


    distAnalogMsgCreator = new DistributedAnalogMessageCreator(
                               *ste,
                               *distAnalogMsgDispatcher,
                               (delay_t)simParam.minDelay.in_steps( get_dt() ));

    distIncomingSpikeScheduler = new STDistributedIncomingSpikeScheduler(*mpiInBuffers,
                                 *globalDelayMap,
                                 *spikeBuffer,
                                 (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,
                                          *distAnalogMsgDispatcher,
                                          *mpiAllToAllComm,
                                          *this );

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

    // Make sure that all net works are trhough the init befor anything else can happen
    mpi_comm.Barrier();

}

DistributedSingleThreadNetwork::~DistributedSingleThreadNetwork()
{
    delete distEngine;
    delete mpiAllToAllComm;
    delete distIncomingSpikeScheduler;
    delete localAnalogMsgDispatcher;
    delete localSimEngine;
    delete spikeScheduler;
    delete mpiInBuffers;
    delete mpiOutBuffers;
    delete globalDelayMap;
    delete targetNodesMap;
    delete localSpikeScheduler;
    delete stgPool;
    delete localDelayMap;
    delete spikeBuffer;
    delete localAnalogMsgCreator;
    for (int i = 0; i < mpi_comm.Get_size() ; ++i) {
        delete incomingDispatchers[i];
        delete outgoingDispatchers[i];
        delete sources2BufPositionsMaps[i];
    }
    delete distAnalogMsgDispatcher;
    delete distAnalogMsgCreator;
    delete delayObjectMap;
}


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


void DistributedSingleThreadNetwork::_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 ( delay > simParam.maxDelay.in_steps( get_dt() ) ) {
            throw(
                PCSIM::ConstructionException(
                    "DistributedSingleThreadNetwork::addSpikeMessage",
                    str( boost::format("Specified delay (%1%) out of range (min=%2% ms, max=%3% ms).") % (delay * simParam.dt.in_ms()) % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() )
                )
            );
        }

        SingleThreadNetwork::addLocalSpikeMessage( localDelayMap, stgPool, localSimEngine, sender_port, receiver, port, delay );

    } else if (sender.node == mpi_comm.Get_rank()) {
        targetNodesMap->addTargetNode(sender_port, receiver.node );
        _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(
                    "DistributedSingleThreadNetwork::addSpikeMessage",
                    str( boost::format("Specified delay (%1%) 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->find(glengineids[sender.node][sender.eng], sender_port, (delaystep_t)delay);

        if( tg == no_spikegroup ) {
            tg = stgPool->addSpikeTarget( localSimEngine->getObject(receiver), port );
            globalDelayMap->insert( glengineids[sender.node][sender.eng], sender_port, (delaystep_t)delay, tg );
        } else {
            stgPool->addSpikeTarget( tg, localSimEngine->getObject(receiver), port);
        }
        _nSpikeMessages++;
    }

}

// ------------------------------------ analog messages -------------------------------------- //
template<typename analogSrcType, typename analogDestType>
inline void DistributedSingleThreadNetwork::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()) {

        localAnalogMsgCreator->addAnalogMessage( sender, srcPortOrField, receiver, destPortOrField, delay );

    } else if (sender.node == mpi_comm.Get_rank()) {
        distAnalogMsgCreator->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(
                    "DistributedSingleThreadNetwork::addAnalogMessage",
                    str( boost::format("Specified delay (%1%) out of range (min=%2% ms, max=%3% ms).") % (delay * simParam.dt.in_ms()) % simParam.minDelay.in_ms() % simParam.maxDelay.in_ms() )
                )
            );
        }
        distAnalogMsgCreator->addIncomingAnalogMessage( sender, srcPortOrField, receiver, destPortOrField, delay);
    }
}

void DistributedSingleThreadNetwork::_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( get_dt() ));
}

void DistributedSingleThreadNetwork::_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( get_dt() ));
}

void DistributedSingleThreadNetwork::_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( get_dt() ) );
}

void DistributedSingleThreadNetwork::_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( get_dt() ) );
}


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

void DistributedSingleThreadNetwork::_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);
        distAnalogMsgDispatcher->initialize();
        initialized = true;
    }
}

void DistributedSingleThreadNetwork::_reset_()
{
    if ( ! initialized )
        initialize();
    spikeScheduler->reset();
    distEngine->reset();
    distAnalogMsgDispatcher->reset(get_dt().in_sec());
    reseted = true;
}


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

---