MPIOutputBuffer.cpp

Go to the documentation of this file.

#include "MPIOutputBuffer.h"

#include <cassert>

#include <cstring>



MPIOutputBuffer::MPIOutputBuffer(vector<gl_engineid_t> &engIDs)
        : initialized(false), mpiOutputSpikeBuffer(engIDs), slicer(MPIBufferSlicer::tpOutputBuffer)
{
    analogMsgCounters.resize(engIDs.size(), 0);
}


MPIOutputBuffer::~MPIOutputBuffer()
{
        if (initialized)
        if (slicer.thereIsMixedDataType)
                mixedMPIDataType.Free();
}

void MPIOutputBuffer::startNewMPIExchange()
{
    slicer.reset();
}

void MPIOutputBuffer::initialize(MPIMessageSpec msgSpec, size_t spikeBufferSize, size_t maxMPIMsgSize,
                                 void *baseBufferPtr, void *analogBuffer, void *spikeBuffer)
{
    this->baseBufferPtr = baseBufferPtr;
    currentMsgInfo = msgSpec;
    analog_buf = analogBuffer;
    spike_buf = spikeBuffer;

    slicer.initialize((char *)spikeBuffer - (char *)analogBuffer,
                      maxMPIMsgSize, spikeBufferSize);

    // prepare mixed
    size_t analog_buffer_size_elements = (double *)spikeBuffer - (double *)analogBuffer;

    spike_buffer_size_elements = spikeBufferSize / sizeof(MPIOutputSpikeBuffer<>::coding_element_type);

    mpiOutputSpikeBuffer.initialize((MPIOutputSpikeBuffer<>::coding_element_type *)spikeBuffer, spike_buffer_size_elements);

    slicer.initialize(analog_buffer_size_elements,
                      maxMPIMsgSize, spikeBufferSize);

    analogMPIDatatype = MPI::DOUBLE;
    if (sizeof(MPIOutputSpikeBuffer<>::coding_element_type) == 4)
        spikingMPIDatatype = MPI::LONG;
    else
        spikingMPIDatatype = MPI::SHORT;

    // initialize the mixed mpi data type
    if (slicer.thereIsMixedDataType) {
        // set the  data types comprising the mixed data type        
        mixed_data_types[0] = analogMPIDatatype;
        mixed_data_types[1] = spikingMPIDatatype;
        // set the mixed counts and displacements
        if (maxMPIMsgSize == 0)
            mixedCounts[0] = analog_buffer_size_elements;
        else
            mixedCounts[0] = analog_buffer_size_elements % (maxMPIMsgSize / sizeof(double));
        mixedCounts[1] = spikeBufferSize / sizeof(MPIOutputSpikeBuffer<>::coding_element_type);

        mixedDisplacements[0] = 0;
        mixedDisplacements[1] = mixedCounts[0] * sizeof(double);
        
        // initialize mixed data type to some dummy initial value, so that free will work afterwards
        mixedMPIDataType = MPI::Datatype::Create_struct(2, mixedCounts, mixedDisplacements, mixed_data_types);
    }

    initialized = true;
}

bool MPIOutputBuffer::hasNextBufferSlice()
{
    if (slicer.currentSliceType == MPIBufferSlicer::sliceAnalog ||
            slicer.currentSliceType == MPIBufferSlicer::sliceUndefined ) {
        if (totalAnalogMsgCounter)
            return true;
        else {     
            return mpiOutputSpikeBuffer.hasNextSerializedMPIBuffer();
        }
    }    
    return mpiOutputSpikeBuffer.hasNextSerializedMPIBuffer();
}

MPIMessageSpec & MPIOutputBuffer::prepareNextBufferSlice()
{
    currentMsgInfo.hasContent = hasNextBufferSlice();

    slicer.calcNextBufferSliceDimensions();

    // setup currentMsgInfo
    switch (slicer.currentSliceType) {
    case MPIBufferSlicer::sliceAnalog :
        *currentMsgInfo.displacement = ((char *)analog_buf + slicer.currentSlicePos) - (char *)baseBufferPtr;
        *currentMsgInfo.buffer = (void *)((char *)analog_buf + slicer.currentSlicePos);
        *currentMsgInfo.count = slicer.currentAnalogSliceSize / sizeof(double);
        *currentMsgInfo.datatype = analogMPIDatatype;
        currentMsgInfo.content_type = MPIMessageSpec::contentAnalog;
        break;
    case MPIBufferSlicer::sliceMixed :
        mixedCounts[1] = mpiOutputSpikeBuffer.prepareNextSerializedMPIBuffer(slicer.allowedSpikeSliceSize/sizeof(MPIOutputSpikeBuffer<>::coding_element_type));
        // free previous mixed data type before creating new one
        mixedMPIDataType.Free();
        mixedMPIDataType = MPI::Datatype::Create_struct(2, mixedCounts, mixedDisplacements, mixed_data_types);
        mixedMPIDataType.Commit();
        *currentMsgInfo.displacement = (char *)( (double *)spike_buf - mixedCounts[0] ) - (char *)baseBufferPtr;
        *currentMsgInfo.buffer = (void *)( (double *)spike_buf - mixedCounts[0] ) ;
        *currentMsgInfo.datatype = mixedMPIDataType;
        *currentMsgInfo.count = 1;
        currentMsgInfo.content_type = MPIMessageSpec::contentMixed;        
        break;
    case MPIBufferSlicer::sliceSpiking :        
        *currentMsgInfo.displacement = (char *)spike_buf - (char *)baseBufferPtr;
        *currentMsgInfo.buffer = spike_buf;
        *currentMsgInfo.count = mpiOutputSpikeBuffer.prepareNextSerializedMPIBuffer(slicer.allowedSpikeSliceSize/sizeof(MPIOutputSpikeBuffer<>::coding_element_type));;
        *currentMsgInfo.datatype = spikingMPIDatatype;
        currentMsgInfo.content_type = MPIMessageSpec::contentSpiking;
        break;
    case MPIBufferSlicer::sliceUndefined :
        assert( 0 );
    }   
    return currentMsgInfo;
}

MPIMessageSpec& MPIOutputBuffer::getCurrentMPIMsgSpecHolder()
{
    return currentMsgInfo;
}

void *MPIOutputBuffer::getBuffer()
{
    return (void *)analog_buf;
}

void MPIOutputBuffer::setFinishedFlag(bool finished)
{
    mpiOutputSpikeBuffer.setFinishedFlag(finished);
}

void MPIOutputBuffer::nextCycle()
{
    mpiOutputSpikeBuffer.nextCycle();
}

MPIOutputSpikeBuffer<> & MPIOutputBuffer::mpiOutputSpikingBuffer()
{
    return mpiOutputSpikeBuffer;
}


MPIOutputBufferVector::MPIOutputBufferVector(int numBuffers, vector<gl_engineid_t> &engIDs)
        : initialized(false), nNodes(numBuffers), mpiExchBlocksInfo(numBuffers)
{
    _buffers.resize(nNodes, MPIOutputBuffer(engIDs));
}

MPIOutputBufferVector::~MPIOutputBufferVector()
{
        if (initialized)
        delete [] memoryPool;
}


void MPIOutputBufferVector::initialize(int minDelay, size_t maxMPIMessageSize,
                                       size_t spikeBufferSize)
{
    // Calculate the size of the memory pool
    spike_buffer_size = spikeBufferSize; // in bytes
    int pool_size = 0;
    for (int i = 0; i < nNodes; ++i) {
        _buffers[i].calculateTotalAnalogMsgCounter();
        // for each buffer allocate memory for the analog messages and one slice for the spiking messages
        pool_size += _buffers[i].totalAnalogMsgCounter * sizeof(double) * minDelay + spikeBufferSize;
    }

    memoryPool = new char[pool_size];
    
    memset(memoryPool, 0, pool_size);

    //  Setup buffer segment pointers for the individual input buffers
    void * analogBufPtr = (void *)memoryPool;
    for (int i = 0; i < nNodes; ++i) {
        void *spikeBufPtr = (void *)((char *)analogBufPtr + _buffers[i].totalAnalogMsgCounter * sizeof(double) * minDelay);
        _buffers[i].initialize(mpiExchBlocksInfo.getMsgSpec(i), spike_buffer_size, maxMPIMessageSize,
                               memoryPool, analogBufPtr, spikeBufPtr);
        analogBufPtr = (char *)analogBufPtr + _buffers[i].totalAnalogMsgCounter * sizeof(double) * minDelay + spike_buffer_size;
    }

    nextCycle();
    
    initialized = true;
}

void MPIOutputBufferVector::startNewMPIExchange()
{
    vector<MPIOutputBuffer>::iterator buf_it;
    for (buf_it = _buffers.begin() ; buf_it != _buffers.end() ; ++buf_it)
        buf_it->startNewMPIExchange();

}

void MPIOutputBufferVector::prepareNextBufferSlices()
{
    vector<MPIOutputBuffer>::iterator buf_it;
    for (buf_it = _buffers.begin() ; buf_it != _buffers.end() ; ++buf_it)
        buf_it->prepareNextBufferSlice();
}


MPIExchangeBlocksInfo & MPIOutputBufferVector::getMPIExchangeBlocksInfo()
{
        return mpiExchBlocksInfo;       
}

---