MultiThreadSimEngine.cpp

Go to the documentation of this file.

#include "MultiThreadSimEngine.h"
#include "SimNetwork.h"

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

SimEngineAdvanceJob::SimEngineAdvanceJob(SingleThreadSimEngine *eng, int numSteps) :
        engine(eng), nSteps(numSteps)
{}

void SimEngineAdvanceJob::setNumSteps(int nSteps)
{
    this->nSteps = nSteps;
}

void SimEngineAdvanceJob::start()
{
    engine->advance(nSteps);
}

SimEngineResetJob::SimEngineResetJob(SingleThreadSimEngine *eng)
{
    this->eng = eng;
}

void SimEngineResetJob::start()
{
    eng->reset();
}

SimEngineInitializeJob::SimEngineInitializeJob(SingleThreadSimEngine *eng)
{
    this->eng = eng;
}

void SimEngineInitializeJob::start()
{
    eng->initialize();
}



void CycleAdvanceJob::start()
{
    vector< AnalogDelayObject *>::const_iterator it;
    for (it = analog_delay_objects->begin(); it != analog_delay_objects->end(); ++it)
        (*it)->advanceCycle();
}

MultiThreadSimEngine::MultiThreadSimEngine(int ID,
        int numThreads,
        ThreadPool &thrPool,
        SpikeScheduler &scheduler,
        AnalogMessageDispatcherVector &analogDispatchers,
        SimNetwork &net)
        : SimEngine(ID, scheduler, net), _numThreads(numThreads),
        thr_pool(thrPool)
{
    advance_jobs.reserve(_numThreads - 1);
    reset_jobs.reserve(_numThreads - 1);
    initialize_jobs.reserve(_numThreads - 1);
    mtCycleAdvanceJobs.reserve(_numThreads - 1);
    engines.reserve(_numThreads);
    analogDelayObjects.resize(_numThreads);
    thr_barrier = new boost::barrier(_numThreads);
    for (int i = 0; i < _numThreads; ++i) {
        analogDispatchers[i].setBarrier(thr_barrier);
        engines.push_back(new SingleThreadSimEngine(i, spikeScheduler, analogDispatchers[i], net));
        if (i < _numThreads - 1) {
            SimEngineAdvanceJob the_advance_job(engines[i], net.simParameter().minDelay.in_steps( net.get_dt() ));
            advance_jobs.push_back(the_advance_job);
            SimEngineResetJob the_reset_job(engines[i]);
            reset_jobs.push_back(the_reset_job);
            SimEngineInitializeJob the_initialize_job(engines[i]);
            initialize_jobs.push_back(the_initialize_job);
        }
        CycleAdvanceJob the_cycle_advance_job(&analogDelayObjects[i]);
        mtCycleAdvanceJobs.push_back(the_cycle_advance_job);
    }
}

MultiThreadSimEngine::~MultiThreadSimEngine()
{
    for (int i = 0 ; i < _numThreads ; ++i) {
        delete engines[i];
    }
    delete thr_barrier;
}

void MultiThreadSimEngine::seed( uint32 noiseSeed )
{
    for (int i = 0 ; i < _numThreads ; ++i) {
        engines[i]->seed( noiseSeed );
    }
}

void MultiThreadSimEngine::seed( vector<uint32> const& sim_seeds )
{
    for (int i = 0 ; i < _numThreads ; ++i) {
        engines[i]->seed( sim_seeds[i] );
    }
}

void MultiThreadSimEngine::noiseRandEngineOutput( vector<uint32> & r )
{
    r.resize( _numThreads );
    for (int i = 0 ; i < _numThreads ; ++i) {
        unsigned v = (unsigned)((*(engines[i]->noiseRandomEngine()))() * 200000);
        // cerr << "r=" << v << endl;
        r[i] = v;
    }
}

void MultiThreadSimEngine::mount( const SimObjectFactory &objFactory, const SimObject::ID &mountpoint, SimObject::ID &id )
{
    addObject( objFactory, id );
    one_way_link( id, mountpoint );
}

void MultiThreadSimEngine::insert( const SimObjectFactory &objFactory, const SimObject::ID &mountpoint, SimObject::ID &id )
{
    addObject( objFactory, id );
    two_way_link( id, mountpoint );
}

void MultiThreadSimEngine::two_way_link( const SimObject::ID &id1, const SimObject::ID &id2 )
{
    SimObject *obj1 = engines[id1.eng]->getObject( id1 );
    SimObject *obj2 = engines[id2.eng]->getObject( id2 );

    obj1->incoming( obj2, id2, id1, network );
    obj2->outgoing( obj1, id1, id2, network );

    obj2->incoming( obj1, id1, id2, network );
    obj1->outgoing( obj2, id2, id1, network );
}

void MultiThreadSimEngine::one_way_link( const SimObject::ID &src, const SimObject::ID &dst )
{
    SimObject *s = engines[src.eng]->getObject( src );
    SimObject *d = engines[dst.eng]->getObject( dst );

    d->incoming( s, src, dst, network );
    s->outgoing( d, dst, src, network );
}

void MultiThreadSimEngine::reset()
{
    for (int thr = 0 ; thr < _numThreads-1 ; thr++)
        thr_pool.dispatch(thr, reset_jobs[thr]);
    engines[_numThreads-1]->reset();
    thr_pool.waitAll();
    currentStepWithinCycle = 0;
}

void MultiThreadSimEngine::initialize()
{
    for (int thr = 0 ; thr < _numThreads-1 ; thr++) {
        thr_pool.dispatch(thr, initialize_jobs[thr]);
    }
    engines[_numThreads-1]->initialize();
    thr_pool.waitAll();
}


void MultiThreadSimEngine::advance(int nSteps)
{
    int nCycles;
    int reminderSteps;
    int interval = default_steps_per_cycle;
    int leftStepsToFinishPreviousCycle;

    if (currentStepWithinCycle) {
        if (nSteps + currentStepWithinCycle >= interval) {
            leftStepsToFinishPreviousCycle = default_steps_per_cycle - currentStepWithinCycle;
            advanceSeveralStepsWithinACycle(leftStepsToFinishPreviousCycle);
            finalizeCycle();
        } else {
            leftStepsToFinishPreviousCycle = nSteps;
            advanceSeveralStepsWithinACycle(leftStepsToFinishPreviousCycle);
        }
        nSteps -=leftStepsToFinishPreviousCycle;
    }


    reminderSteps = nSteps % interval;
    nCycles = (nSteps - reminderSteps) / interval ;
    for (int c = 0 ; c < nCycles ; ++c ) {
        advanceOneFullCycle();
    }

    if (reminderSteps > 0 )
        advanceSeveralStepsWithinACycle(reminderSteps);

    currentStepWithinCycle = reminderSteps;
}


void MultiThreadSimEngine::advanceSeveralStepsWithinACycle(int steps)
{
    for (int i = 0 ; i < _numThreads-1 ; i++ )
        advance_jobs[i].setNumSteps(steps);

    for (int thr = 0 ; thr < _numThreads-1; ++thr)
        thr_pool.dispatch(thr, advance_jobs[thr]);

    engines[_numThreads-1]->advance(steps);
    thr_pool.waitAll();
}


void MultiThreadSimEngine::finalizeCycle()
{
    spikeScheduler.nextCycle();

    for (int thr = 0; thr < _numThreads-1; ++thr) {
        if (analogDelayObjects[thr].size()) {
            thr_pool.dispatch(thr, mtCycleAdvanceJobs[thr]);
        }
    }
    if (analogDelayObjects[_numThreads-1].size()) {
        mtCycleAdvanceJobs[_numThreads-1].start();
    }
    thr_pool.waitAll();
}

void MultiThreadSimEngine::advanceOneFullCycle()
{
    advanceSeveralStepsWithinACycle(default_steps_per_cycle);
    finalizeCycle();
}

SingleThreadSimEngine & MultiThreadSimEngine::getSTEngine(engineid_t eng)
{
    return *engines[eng];
}

---