DistributedNetwork.cpp

Go to the documentation of this file.

#include "DistributedNetwork.h"
#include "PCSIMException.h"
#include "SpikeSender.h"
#include "SimTime.h"
#include "SimObjectRegistry.h"

#include <string>
using std::string;

#include <boost/format.hpp>

#include <mpi.h>

#include <iostream>
using std::cerr;
using std::endl;

DistributedNetwork::DistributedNetwork( int nThreads, SimParameter sp ) :
        SimNetwork(MPI::COMM_WORLD, sp, &globalRoundRobin ), mpi_comm(MPI::COMM_WORLD), _nThreads(nThreads)
{
    init();
}

DistributedNetwork::DistributedNetwork( int nThreads, MPI::Intracomm &mpiCommunicator, SimParameter sp ) :
        SimNetwork(mpiCommunicator, sp, &globalRoundRobin ), mpi_comm(mpiCommunicator), _nThreads(nThreads)
{
    init();
}



void DistributedNetwork::init()
{
    setupGlEngineIDs(_nThreads);
    incomingConnections.resize(mpi_comm.Get_size(), false);
    outgoingConnections.resize(mpi_comm.Get_size(), false);
    objectCounter.resize( maxGlobalEngineID()+1 );
    spikePortCounter.resize( maxGlobalEngineID()+1 );
    for( size_t i=0; i<spikePortCounter.size(); i++)
        spikePortCounter[0] = 0;
    // ghostPortInfo.resize( maxGlobalEngineID()+1 );
    ghostFirstPortInfo.resize( maxGlobalEngineID()+1 );
    _mpi_rank = mpi_comm.Get_rank();

}

DistributedNetwork::~DistributedNetwork()
{
    /* NOOP */
}

uint32 DistributedNetwork::getUniqueSeedOverMpi( uint32 seed )
{
    uint32 seed_buffer = seed;
    mpi_comm.Bcast((void *)&seed_buffer, (int)1, MPI::UNSIGNED, 0);
    return seed_buffer;
}

void DistributedNetwork::seed_noise_rng( uint32 noiseRNGseed )
{
    uint32 seed = getUniqueSeedOverMpi( noiseRNGseed );
    vector<uint32> global_sim_seeds( maxGlobalEngineID()+1 );
    fillSeedVector( seed, global_sim_seeds );

    vector<uint32> local_sim_seeds( _nThreads );
    for( int e = 0;  e < _nThreads; e++ ) {
        local_sim_seeds[e] = global_sim_seeds[ glengineids[ _mpi_rank ][ e ] ];
    }
    localSimEngine->seed( local_sim_seeds );


    seed = getUniqueSeedOverMpi( makeSeed( simParam.constructionRNGSeed ) );
    global_sim_seeds.resize( mpi_size() + 1 );
    fillSeedVector( seed, global_sim_seeds );
    objectVariationRNDEngine->seed( global_sim_seeds[ mpi_rank() ] );

}

void DistributedNetwork::setupGlEngineIDs(unsigned int numLocalEngines)
{
    uint32 *nEngineArray  = new uint32[mpi_comm.Get_size()];
    // setup all the global engine ids
    uint32 nEngines = numLocalEngines;
    mpi_comm.Allgather((void*)&nEngines, (int)1, MPI::UNSIGNED, (void*)nEngineArray, (int)1, MPI::UNSIGNED);
    int glEngineCounter = 0;
    glengineids.resize(mpi_comm.Get_size());
    for (int i=0 ; i < mpi_comm.Get_size(); ++i) {
        glengineids[i].resize(nEngineArray[i]);
        for (unsigned int j = 0 ; j < nEngineArray[i] ; ++j) {
            glengineids[i][j] = glEngineCounter;
            location.push_back( SimEngine::ID( i, j ) );
            glEngineCounter++;
        }
    }
    //cerr << "setupGlEngineIDs rank " << mpi_comm.Get_rank() << endl ;
    _max_global_engine_id = glEngineCounter-1;
    delete [] nEngineArray;
}

// ----------------------------- conversion between local and global engine IDs ---------------------------//

gl_engineid_t DistributedNetwork::getGlobalEngineID( const SimEngine::ID &eid ) const
{
    if( ( eid.node < glengineids.size() ) && ( eid.engine < glengineids[ eid.node ].size() ) ) {
        return glengineids[ eid.node ][ eid.engine ];
    } else {
        return INVALID_GLOBAL_ENGINE_ID;
    }
}

gl_engineid_t DistributedNetwork::getGlobalEngineID( engineid_t eng ) const
{
    return getGlobalEngineID( SimEngine::ID( mpi_comm.Get_rank(), eng) );
}


gl_engineid_t DistributedNetwork::maxGlobalEngineID(void) const
{
    return _max_global_engine_id;
}

gl_engineid_t DistributedNetwork::maxLocalEngineID(void) const
{
    return _nThreads-1;
}

const SimEngine::ID DistributedNetwork::getLocation( gl_engineid_t gEID ) const
{
    if( gEID < location.size() ) {
        return location[gEID];
    } else {
        return SimEngine::ID::Invalid;
    }
}

local_objectid_t DistributedNetwork::getGhostID( gl_engineid_t gEID, SimObjectFactory const& objFactory  )
{
    object_type_t tid = objFactory.getObjectTypeID();
    if( gEID >= objectCounter.size() ) {
        return INVALID_LOCAL_OBJECT_ID;
    }

    // resize if necessary
    if( tid >= objectCounter[gEID].size() ) {
        objectCounter[gEID].resize( tid + 1 );
        objectCounter[gEID][tid] = 0;
        //ghostPortInfo[gEID].resize( tid + 1 );
        ghostFirstPortInfo[gEID].resize( tid + 1 );
    }

    // cerr << "spc.sz=" << spikePortCounter.size() << endl;
    local_objectid_t oid  = objectCounter[gEID][tid]++;

    int nsop = theSimObjectRegistry->getObject( objFactory.getObjectTypeID() ).nSpikeOutputPorts();
    if( nsop  > 0 ) {
        if( oid >= ghostFirstPortInfo[gEID][tid].size() ) {
            ghostFirstPortInfo[gEID][tid].resize( oid + 1 );
        }
        ghostFirstPortInfo[gEID][tid][oid] = spikePortCounter[gEID];
    } else {
        // pi.first_spike_output_port_id = INVALID_SPIKE_PORT_ID;
    }
    //cerr << "-- getGhostID node " << mpi_comm.Get_rank() << ": pi(" << gEID << "," << (int)tid << ")=" << pi << endl;
    spikePortCounter[gEID] += nsop;

    return oid;

}


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

void DistributedNetwork::_addObject_( const SimObjectFactory &objFactory, const SimEngine::ID &loc, SimObject::ID &id )
{
    gl_engineid_t gEID = getGlobalEngineID( loc );
    // cerr << "gEID =" << gEID << endl;
    if( gEID != INVALID_GLOBAL_ENGINE_ID ) {
        id.node = loc.node;
        id.eng  = loc.engine;
        if( loc.node == mpi_comm.Get_rank() ) {
            localSimEngine->addObject( objFactory, id );
        } else {
            id.type = objFactory.getObjectTypeID();
            id.localid = getGhostID( gEID, objFactory );
        }
    } else {
        id = SimObject::ID::Invalid;
        throw( PCSIM::ConstructionException( "DistributedSingleThreadNetwork::addObject", "Invalid sim engine id " + loc.toString() + " specified." ) );
    }
}

void DistributedNetwork::_mount_( const SimObjectFactory &objFactory, const SimObject::ID &mountpoint, SimObject::ID &id )
{
    SimEngine::ID loc( mountpoint.node, mountpoint.eng );
    gl_engineid_t gEID = getGlobalEngineID( loc );
    if( gEID != INVALID_GLOBAL_ENGINE_ID ) {
        id.node = loc.node;
        id.eng  = loc.engine;
        if( loc.node == mpi_comm.Get_rank() ) {
            localSimEngine->mount( objFactory, mountpoint, id );
        } else {
            id.type = objFactory.getObjectTypeID();
            id.localid = getGhostID( gEID, objFactory );
        }
    } else {
        id = SimObject::ID::Invalid;
        throw( PCSIM::ConstructionException( "DistributedSingleThreadNetwork::mount", "Invalid mountpoint id " + mountpoint.toString() + " specified." ) );
    }
}

void DistributedNetwork::_insert_( const SimObjectFactory &objFactory, const SimObject::ID &container, SimObject::ID &id )
{
    SimEngine::ID loc( container.node, container.eng );
    gl_engineid_t gEID = getGlobalEngineID( loc );
    if( gEID != INVALID_GLOBAL_ENGINE_ID ) {
        id.node = loc.node;
        id.eng  = loc.engine;
        if( loc.node == mpi_comm.Get_rank() ) {
            localSimEngine->insert( objFactory, container, id );
        } else {
            id.type = objFactory.getObjectTypeID();
            id.localid = getGhostID( gEID, objFactory );
        }
    } else {
        id = SimObject::ID::Invalid;
        throw( PCSIM::ConstructionException( "DistributedSingleThreadNetwork::insert", "Invalid container id " + container.toString() + " specified." ) );
    }
}

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

void DistributedNetwork::_connect_( SimObject::ID const& sender, port_t out, const SimObject::ID &receiver, port_t in, int delay )
{
    int delay_to_use = 0 ;
    if( mpi_comm.Get_rank() == receiver.node ) {
        SimObject *rec_obj = localSimEngine->getObject(receiver);
        delay_to_use = delay < 0 ? (int)( rec_obj->getManagedDelay() / get_dt().in_sec() + 0.5 ) : delay ;
    }

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

        SimObject::PortType sender_port_type = (theSimObjectRegistry->getObject( sender.type )).outputPortType( out );
        SimObject::PortType receiver_port_type = (theSimObjectRegistry->getObject( receiver.type )).inputPortType( in );

        if( sender_port_type == SimObject::spiking && receiver_port_type == SimObject::spiking ) {
            _addSpikeMessage_( sender, out, getFirstSenderSpikePort( sender ) + out, receiver, in, delay_to_use );
        } else if( sender_port_type == SimObject::analog && receiver_port_type == SimObject::analog ) {
            addAnalogMessage( sender, out, receiver, in, Time::steps( delay_to_use, get_dt() ) );
        } else  {
            throw(
                PCSIM::ConstructionException(
                    "DistributedNetwork::_connect_",
                    str( boost::format("Can not connect specified source (%1%) and destination (%2%) object: no matching output (%3%) and input (%4%) ports.") % sender.toString() % receiver.toString() % out % in)
                )
            );
        }
    }
}

// ------------------------------------- messages ----------------------------------------- //

void DistributedNetwork::_addSpikeMessage_(const SimObject::ID &sender, const port_t out, const SimObject::ID &receiver, const port_t in, const Time &delay)
{
    SimObject::PortType sender_port_type = (theSimObjectRegistry->getObject( sender.type )).outputPortType( out );
    SimObject::PortType receiver_port_type = (theSimObjectRegistry->getObject( receiver.type )).inputPortType( in );
    if( sender_port_type == SimObject::spiking && receiver_port_type == SimObject::spiking ) {
        _addSpikeMessage_( sender, out, getFirstSenderSpikePort(sender)+ out, receiver, in, delay.in_steps( get_dt() ) );
    } else {
        throw(
            PCSIM::ConstructionException(
                "DistributedSingleThreadNetwork::_connect_",
                str( boost::format("Can not add spike message from source (%1%) to destination (%2%) object: no matching out (%3%) and in (%4%) ports.") % sender.toString() % receiver.toString() % out % in)
            )
        );
    }
}


ostream& operator<<(ostream &s, const DistributedNetwork::ConnectInfo &ci)
{
    return s << "("
           // << "md=" << (int)ci.managed_delay << ", "
           << "sp=" << (int)ci.first_spike_output_port_id  << ", "
           << "si=" << (int)ci.nSpikeInputPorts << ", "
           << "so=" << (int)ci.nSpikeOutputPorts << ", "
           << "ai=" << (int)ci.nAnalogInputPorts << ", "
           << "ao=" << (int)ci.nAnalogOutputPorts << ")";
}

spike_port_id_t DistributedNetwork::getFirstSenderSpikePort( const SimObject::ID &oid )
{
    if ( oid.node == mpi_comm.Get_rank() ) {
        SimObject *src_obj = localSimEngine->getObject(oid);
        //ci.managed_delay = (int)( src_obj->getManagedDelay() / get_dt().in_sec() + 0.5 );
        //if (ci.managed_delay == 0) {
        //cerr << " get_dt().in_sec() = " << get_dt().in_sec() << " getManagedDelay = " << src_obj->getManagedDelay() << endl;
        //}

        SpikeSender *ss = dynamic_cast<SpikeSender*>( src_obj );
        if( ss != NULL && ss->getSpikePort( 0 ) != NULL ) {
            return ss->getSpikePort( 0 )->ID();
        } else {
            return INVALID_SPIKE_PORT_ID;
        }
    } else {
        SimEngine::ID loc( oid.node, oid.eng );
        gl_engineid_t gEID = getGlobalEngineID( loc );
        return ghostFirstPortInfo[gEID][oid.type][oid.localid];
    }
}

---