MPIInputBuffer.cpp

Go to the documentation of this file.

#include "MPIInputBuffer.h"

#include "PCSIMException.h"

#include <cassert>

#include <algorithm>

#include <cstring> 

using std::min;


MPIInputBuffer::MPIInputBuffer(int nEngines)
        : slicer(MPIBufferSlicer::tpInputBuffer)
{
    // Initialize analog msg counters (each for every engine)
    analogMsgCounters.resize(nEngines, 0);
    initialized = false;
}

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

void MPIInputBuffer::initialize(MPIMessageSpec msgSpec, size_t spikeBufferSize, size_t maxMPIMsgSize,
                                void *baseBufferPtr, void *analogBuffer, void *spikeBuffer)
{
    this->baseBufferPtr = baseBufferPtr;
    currentMsgInfo = msgSpec;
    mpiInputSpikeBuffer.initialize((MPIInputSpikeBuffer<>::coding_element_type *)spikeBuffer);

    analog_buf = analogBuffer;
    spike_buf = spikeBuffer;

    size_t analog_buffer_size_elements = (double *)spikeBuffer - (double *)analogBuffer;

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

    slicer.initialize(analog_buffer_size_elements,
                      maxMPIMsgSize, spikeBufferSize);

    analogMPIDatatype = MPI::DOUBLE;
    if (sizeof(MPIInputSpikeBuffer<>::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
        MPI::Datatype mixed_data_types[2];
        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(MPIInputSpikeBuffer<>::coding_element_type);

        mixedDisplacements[0] = 0;
        mixedDisplacements[1] = mixedCounts[0] * sizeof(double);

        mixedMPIDataType = MPI::Datatype::Create_struct(2, mixedCounts, mixedDisplacements, mixed_data_types);

        mixedMPIDataType.Commit();
        mixedMsgAbsoluteDisplacement = (char *)( (double *)spike_buf - mixedCounts[0] ) - (char *)baseBufferPtr;
    }
    currentMsgInfo.hasContent = true;
    initialized = true;
}


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

bool MPIInputBuffer::hasNextBufferSlice()
{
    if (slicer.currentSliceType == MPIBufferSlicer::sliceAnalog ||
            slicer.currentSliceType == MPIBufferSlicer::sliceUndefined ) {
        if (totalAnalogMsgCounter)
            return true;
        else
            return !mpiInputSpikeBuffer.isLastReceivedBuffer();
    }
    return !mpiInputSpikeBuffer.isLastReceivedBuffer();
}

MPIMessageSpec & MPIInputBuffer::prepareNextBufferSlice()
{
    slicer.calcNextBufferSliceDimensions();

    // setup currentMsgInfo
    switch (slicer.currentSliceType) {
    case MPIBufferSlicer::sliceAnalog :
        *currentMsgInfo.displacement = ((char *)analog_buf + slicer.currentSlicePos) - (char *)baseBufferPtr;
        *currentMsgInfo.buffer = (char *)analog_buf + slicer.currentSlicePos;
        *currentMsgInfo.count = slicer.currentAnalogSliceSize / sizeof(double);
        *currentMsgInfo.datatype = analogMPIDatatype;
        currentMsgInfo.content_type = MPIMessageSpec::contentAnalog;
        break;
    case MPIBufferSlicer::sliceMixed :
        *currentMsgInfo.displacement = mixedMsgAbsoluteDisplacement;
        *currentMsgInfo.buffer = (char *)baseBufferPtr + mixedMsgAbsoluteDisplacement;
        *currentMsgInfo.count = 1;
        *currentMsgInfo.datatype = mixedMPIDataType;
        currentMsgInfo.content_type = MPIMessageSpec::contentMixed;
        break;
    case MPIBufferSlicer::sliceSpiking :
        *currentMsgInfo.displacement = (char *)spike_buf - (char *)baseBufferPtr;
        *currentMsgInfo.buffer = spike_buf;
        *currentMsgInfo.count = spike_buffer_size_elements;
        *currentMsgInfo.datatype = spikingMPIDatatype;
        currentMsgInfo.content_type = MPIMessageSpec::contentSpiking;
        break;
    case MPIBufferSlicer::sliceUndefined :
        assert( 0 );
        break;
    }
    return currentMsgInfo;
}


MPIInputBufferVector::MPIInputBufferVector(vector< vector< gl_engineid_t> > & glengineids, int numBuffers)
        : initialized(false), nNodes(numBuffers), mpiExchBlocksInfo(numBuffers)
{
        for (int i = 0; i < nNodes ; ++i) {             
        _buffers.push_back(MPIInputBuffer(glengineids[i].size()));
        }
}


void MPIInputBufferVector::initialize(int minDelay, size_t maxMPIMessageSize, size_t spikeBufferSize)
{
    // Calculate the size of the memory pool
    max_mpi_msg_size = maxMPIMessageSize;
    spike_buffer_size = spikeBufferSize;

    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 + spike_buffer_size;
    }

        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, max_mpi_msg_size, memoryPool, analogBufPtr, spikeBufPtr);
        analogBufPtr = (char *)analogBufPtr + _buffers[i].totalAnalogMsgCounter * sizeof(double) * minDelay + spike_buffer_size;
    }

    initialized = true;
}


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

---