MPIInputBuffer.h

Go to the documentation of this file.

#ifndef MPIINPUTBUFFER_H_
#define MPIINPUTBUFFER_H_

#include <mpi.h>

#include <vector>
#include <numeric>

using std::vector;
using std::accumulate;
using std::partial_sum;

#include <iostream>

using std::cerr;
using std::cout;
using std::endl;


#include "MPIInputSpikeBuffer.h"
#include "MPIBufferSlicer.h"
#include "MPIExchangeBlocksInfo.h"

class MPIInputBufferVector;

class MPIInputBuffer
{
public:
    MPIInputBuffer(int nEngines = 1);
    virtual ~MPIInputBuffer();

    double *getAnalogBuffer()
    {
        return (double *)analog_buf;
    }

    unsigned & getAnalogMsgCounter(engineid_t eng = 0)
    {
        return analogMsgCounters[eng];
    }

    void calculateTotalAnalogMsgCounter()
    {
        totalAnalogMsgCounter = accumulate(analogMsgCounters.begin(), analogMsgCounters.end(), 0);
        partial_sum(analogMsgCounters.begin(), analogMsgCounters.end(), analogMsgCounters.begin());
    }

    void startNewMPIExchange();

    bool hasNextBufferSlice();

    MPIMessageSpec & prepareNextBufferSlice();

    void setHasNewContent(bool contentIndicator)
    {
        mpiInputSpikeBuffer.setHasNewContent(contentIndicator);
    };

    // These methods are for testing purposes only
    int getMixedCounts(int idx) const
    {
        return mixedCounts[idx];
    }

    int getMixedDispl(int idx) const
    {
        return mixedDisplacements[idx];
    }
    
    MPIInputSpikeBuffer<> & spikeBuf()
    {
        return mpiInputSpikeBuffer;     
    }
    
    MPIMessageSpec::ContentType_t & getContentType()
    {
        return currentMsgInfo.content_type;     
    }

protected:
    bool initialized;

    void *spike_buf;

    void *analog_buf;
    
    MPI::Datatype mixedMPIDataType;

    MPI::Datatype spikingMPIDatatype;

    MPI::Datatype analogMPIDatatype;

    MPIMessageSpec currentMsgInfo;
    
    void *baseBufferPtr;
    
    size_t spike_buffer_size_elements;

    vector<unsigned> analogMsgCounters;

    unsigned totalAnalogMsgCounter;
    
    int mixedMsgAbsoluteDisplacement;

    void initialize(MPIMessageSpec msgSpec, size_t spikeBufferSize, size_t maxMPIMsgSize,
                    void *baseBufferPtr, void *analogBuffer, void *spikeBuffer);

    MPIInputSpikeBuffer<> mpiInputSpikeBuffer;

    MPIBufferSlicer slicer;
    
    // counts for the mixed type, used only for testing.
    int mixedCounts[2];

    // displacements for the mixed type, used only for testing.
    MPI::Aint mixedDisplacements[2];
    
    friend class MPIInputBufferVector;

};


/*
 * Vector of all mpi input buffers. 
 * Simplifies the initialization procedure 
 * by performing memory allocation and setup of buffers. 
 * 
 */

class MPIInputBufferVector
{
public:


    MPIInputBufferVector(vector< vector< gl_engineid_t > > &glengineids, int numBuffers) ;

    virtual ~MPIInputBufferVector()
    {
        if (initialized)
                delete [] memoryPool;
    }

    MPIInputBuffer & operator[](int idx)
    {
        return _buffers[idx];
    }

    void *getBaseBufferPtr()
    {
        return memoryPool;
    }

    unsigned int size()
    {
        return nNodes;
    }

    void initialize(int minDelay, size_t maxMPIMessageSize = 0, size_t spikeBufferSize = MPIBUFFER_BLOCK_SIZE);
    
    
    void startNewMPIExchange()
    {
        vector<MPIInputBuffer>::iterator it;
        for (it = _buffers.begin(); it != _buffers.end(); ++it)
                it->startNewMPIExchange();              
    }
    
    void prepareNextBufferSlices()
    {
        vector<MPIInputBuffer>::iterator it;
        for (it = _buffers.begin(); it != _buffers.end(); ++it)
                it->prepareNextBufferSlice();   
    }
    
        MPIExchangeBlocksInfo & getMPIExchangeBlocksInfo();    

protected:
        bool initialized ;
        
    int nNodes;

    size_t spike_buffer_size;


    size_t max_mpi_msg_size;

    vector<MPIInputBuffer> _buffers;

    char * memoryPool;
    
    MPIExchangeBlocksInfo mpiExchBlocksInfo;

};



#endif /*MPIINPUTBUFFER_H_*/

---