MPIOutputBuffer.h

Go to the documentation of this file.

#ifndef MPIOUTPUTBUFFER_H_
#define MPIOUTPUTBUFFER_H_


#include <numeric>

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

#include <mpi.h>

#include "globaldefinitions.h"
#include "MPIBufferSlicer.h"
#include "MPIOutputSpikeBuffer.h"
#include "MPIExchangeBlocksInfo.h"

using std::cerr;

class MPIOutputBufferVector;

class MPIOutputBuffer
{
public:
    MPIOutputBuffer(vector<gl_engineid_t> &engIDs);
                                
    virtual ~MPIOutputBuffer();

    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 *getBuffer();

    void setFinishedFlag(bool finished);

    void nextCycle();

    MPIOutputSpikeBuffer<> &spikeBuf()
    {
        return mpiOutputSpikeBuffer;
    };

                    
        MPIOutputSpikeBuffer<> & mpiOutputSpikingBuffer();
        
        MPIMessageSpec &getCurrentMPIMsgSpecHolder();        
        
        // for testing
        int getMixedCounts(int idx)
        {
                return mixedCounts[idx];        
        }            
        
        // for testing
        int getMixedDispl(int idx)
        {
                return mixedDisplacements[idx]; 
        }
        
        void getMPIExchangeBlocksInfo();                                

protected:
        bool initialized;
        
        void *analog_buf;
        
        void *spike_buf;
        
        void *baseBufferPtr;
        
    MPI::Datatype mixedMPIDataType;

    MPI::Datatype spikingMPIDatatype;

    MPI::Datatype analogMPIDatatype;
    
    MPIMessageSpec currentMsgInfo;
    
    MPI::Datatype mixed_data_types[2];
    
    void initialize(MPIMessageSpec msgSpec, size_t spikeBufferSize, size_t maxMPIMsgSize,
                    void *baseBufferPtr, void *analogBuffer, void *spikeBuffer);
    
    // 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];
    
    size_t spike_buffer_size_elements;
        
    vector<unsigned> analogMsgCounters;
    
    unsigned totalAnalogMsgCounter;
    
    MPIOutputSpikeBuffer<> mpiOutputSpikeBuffer;
    
    MPIBufferSlicer slicer;
    
    friend class MPIOutputBufferVector;
};



class MPIOutputBufferVector
{
public:

    MPIOutputBufferVector(int numBuffers, vector<gl_engineid_t> &engIDs);

    virtual ~MPIOutputBufferVector();
    
    
    void initialize(int minDelay, size_t maxMPIMessageSize = 0, 
                    size_t spikeBufferSize = MPIBUFFER_BLOCK_SIZE);

    void *getBaseBufferPtr()
    {
        return (void *)memoryPool;
    }

    void nextCycle()
    {
        for (int i = 0; i < nNodes; ++i) {
            _buffers[i].nextCycle();
        }
    }

    void setFinishedFlag(bool flag)
    {
        for (int i = 0; i < nNodes; ++i) {
            _buffers[i].setFinishedFlag(flag);
        }
    }

    MPIOutputBuffer & operator[] (int idx)
    {
        return _buffers[idx];
    }
    
    void startNewMPIExchange();
    
    void prepareNextBufferSlices();
    
    MPIExchangeBlocksInfo & getMPIExchangeBlocksInfo();

private:
        bool initialized;
        
    int nNodes;
    
    size_t spike_buffer_size;

    vector<MPIOutputBuffer> _buffers;


    char *memoryPool;
    
    MPIExchangeBlocksInfo mpiExchBlocksInfo;
};

#endif /*MPIOUTPUTBUFFER_H_*/

---