DistributedIncomingAnalogMsgDispatcher.cpp

Go to the documentation of this file.

#include "DistributedIncomingAnalogMsgDispatcher.h"

#include <algorithm>

using std::max;

DistributedIncomingAnalogMsgDispatcher::DistributedIncomingAnalogMsgDispatcher(delay_t minimumDelay, MPIInputBuffer *mpiInputBuffer,
        DistIncomingAnalogSources2BufPosMap *bufferPositionsOfIncomingSources)
        : minDelay(minimumDelay), bufPosOfIncomingSources(bufferPositionsOfIncomingSources), inputBuffer(mpiInputBuffer)
{
    initialized = false;
    cycled_buf_pos_source_eng_ids = new vector< engineid_t >;

    //port_msgs_positions = new buff_pos2msglist_pos_type;
    //field_msgs_positions = new buff_pos2msglist_pos_type;
    // this should be shared among every dispatcher belonging to the same incoming buffer (from different engines)
    // buffPosOfIncomingSources = new DistIncomingAnalogSources2BufPosMap();
}

DistributedIncomingAnalogMsgDispatcher::~DistributedIncomingAnalogMsgDispatcher()
{
    if (!initialized) {
        // delete field_msgs_positions;
        // delete port_msgs_positions;
        delete cycled_buf_pos_source_eng_ids;
    }

    // delete analog delay objects that are created inside the bufPosOfIncomingSources?
    vector< IncomingAnalogDelayObject *>::iterator delayer_it;
    for (delayer_it = cycled_inc_msgs_delayer_destinations.begin();
            delayer_it != cycled_inc_msgs_delayer_destinations.end();
            ++delayer_it) {
        delete (*delayer_it);
    }
    
}

/*
void DistributedIncomingAnalogMsgDispatcher::dispatchMPIIncomingAnalogMsgs(int step)
{
    // dispatch first port messages
    port_incoming_analog_msgs_type::const_iterator port_msg_it;
    vector< unsigned >::const_iterator port_buff_pos_iter = port_msgs_buff_pos.begin();
    for (port_msg_it = port_analog_msgs.begin() ; port_msg_it != port_analog_msgs.end() ; ++port_msg_it) {
        double v = *(mpiBufferPtr + minDelay*(*port_buff_pos_iter) + step);
        vector< pair<SimObject *,analog_port_id_t> >::const_iterator port_dest_it;        
        for (port_dest_it = port_msg_it->begin(); port_dest_it != port_msg_it->end() ; ++port_dest_it)            
            port_dest_it->first->setAnalogInput(v, port_dest_it->second);        
        ++port_buff_pos_iter;
    }
 
    // then dispatch field messages
    field_incoming_analog_msgs_type::iterator field_msg_it;
    vector< unsigned >::const_iterator field_buff_pos_iter = field_msgs_buff_pos.begin();
    for (field_msg_it = field_analog_msgs.begin() ; field_msg_it != field_analog_msgs.end() ; ++field_msg_it) {
        double v = *(mpiBufferPtr+ minDelay*(*field_buff_pos_iter) + step);
        vector< double *>::iterator field_dest_it ;
        for (field_dest_it = field_msg_it->begin() ; field_dest_it != field_msg_it->end() ; ++field_dest_it)            
            *(*field_dest_it) = v;
        ++field_buff_pos_iter;
    }
}
*/

void DistributedIncomingAnalogMsgDispatcher::dispatchDelayerCycledIncomingAnalogMsgs()
{
    vector< unsigned >::const_iterator src_it;
    vector< IncomingAnalogDelayObject * >::iterator dest_it;
    for (src_it = cycled_inc_msgs_src_buff_pos.begin(),
            dest_it = cycled_inc_msgs_delayer_destinations.begin();
            src_it != cycled_inc_msgs_src_buff_pos.end();
            ++src_it, ++dest_it) {                              
        (*dest_it)->putNewCycleValues(mpiBufferPtr + minDelay*(*src_it));
    }
}


void DistributedIncomingAnalogMsgDispatcher::initialize()
{
    if (!initialized) {

        mpiBufferPtr = inputBuffer->getAnalogBuffer();

        //      adjust buffer positions
        vector< unsigned >::iterator it;
        vector< engineid_t >::iterator src_eng_it = cycled_buf_pos_source_eng_ids->begin();
        for ( it = cycled_inc_msgs_src_buff_pos.begin() ; it != cycled_inc_msgs_src_buff_pos.end(); ++it, ++src_eng_it)
            *it += ( *(src_eng_it) ? inputBuffer->getAnalogMsgCounter(*(src_eng_it) - 1) : 0);


        delete cycled_buf_pos_source_eng_ids;

        // initialize delayers
        vector< IncomingAnalogDelayObject *>::iterator delayer_it;
        for (delayer_it = cycled_inc_msgs_delayer_destinations.begin();
                delayer_it != cycled_inc_msgs_delayer_destinations.end();
                ++delayer_it) {
            (*delayer_it)->initDistributedMode();
        }

        //delete field_msgs_positions;
        //delete port_msgs_positions;
        initialized = true;
    }
}



void DistributedIncomingAnalogMsgDispatcher::reset(double dt)
{
    // initialize delayers
    vector< IncomingAnalogDelayObject *>::iterator delayer_it;
    for (delayer_it = cycled_inc_msgs_delayer_destinations.begin();
            delayer_it != cycled_inc_msgs_delayer_destinations.end();
            ++delayer_it) {
        (*delayer_it)->reset(dt);
    }
}

---