MPIOutputSpikeBuffer.h

Go to the documentation of this file.

#ifndef MPIOutputSpikeBuffer_H_
#define MPIOutputSpikeBuffer_H_

#include "globaldefinitions.h"
#include <vector>
#include <algorithm>
#include <cstddef>

using std::vector;
using std::copy;

#include <iostream>
using std::cout;
using std::endl;


template<typename T>
class MPIOutputSpikeBufferVector;


template<typename T = unsigned short>
class MPIOutputSpikeBuffer
{
public:

    MPIOutputSpikeBuffer(vector<gl_engineid_t> &engIDs);



    typedef T coding_element_type ;

    T guard_value;

    virtual ~MPIOutputSpikeBuffer();
    
    void initialize(void * spikeBuffer, size_t spikeBufferSize);

    void appendSpike(local_objectid_t oid, int timestamp, engineid_t eng = 0);

    void nextCycle();


    int prepareNextSerializedMPIBuffer(size_t currentBufferSize);
    
    int prepareNextSerializedMPIBuffer()
    {
        return prepareNextSerializedMPIBuffer(buf_size);        
    };

    bool hasNextSerializedMPIBuffer();

    const T* const getBuffer();

    void setFinishedFlag(bool flag);

protected:
    

    int nLocalEngines;

    size_t buf_size;


    vector<vector<T> > _buffers;

    vector<gl_engineid_t> *gl_engineids;

    T* serialBuffer;


    vector<int> timestamps;

    bool isInit;

    void initSerialization();

    gl_engineid_t s_curr_eng;

    int s_curr_timestamp;

    typedef typename vector<T>::const_iterator vector_iterator;

    vector_iterator s_curr_src_pos;

    vector_iterator s_src_end_pos;

    bool serializationFinished;

    friend class MPIOutputSpikeBufferVector<T>;

};


template<typename T = unsigned short>
class MPIOutputSpikeBufferVector
{
public:

    MPIOutputSpikeBufferVector(int numBuffers, vector<gl_engineid_t> &engIDs,
                          size_t buffer_size = MPIBUFFER_BLOCK_SIZE);

    ~MPIOutputSpikeBufferVector();

    T *getBuffersPool()
    {
        return serialBuffersPool;
    }

    void nextCycle()
    {
        for (int i = 0; i < nNodes; ++i) {
            _outputbuffers[i]->nextCycle();
        }
    }

    void setFinishedFlag(bool flag)
    {
        for (int i = 0; i < nNodes; ++i) {
            _outputbuffers[i]->setFinishedFlag(flag);
        }
    }

    MPIOutputSpikeBuffer<T> & operator[] (int idx)
    {
        return *_outputbuffers[idx];
    }

private:

    int nNodes;

    size_t buf_size;

    vector<MPIOutputSpikeBuffer<T>*> _outputbuffers;


    T *serialBuffersPool;
};

template<typename T>
MPIOutputSpikeBuffer<T>::MPIOutputSpikeBuffer(vector<gl_engineid_t> &engIDs) :
                                                          gl_engineids(&engIDs)
{
    guard_value = 0xFF;
    for (unsigned int i = 1; i < sizeof(T); ++i)
        guard_value = (guard_value << 8) + 0xFF;

    nLocalEngines = gl_engineids->size();

    _buffers.resize(nLocalEngines);
    timestamps.resize(nLocalEngines, int(-1));
    for (int j = 0 ; j < nLocalEngines ; ++j) {
        _buffers[j].push_back((*gl_engineids)[j]);
    }

    serializationFinished = false;
}

template<typename T>
void MPIOutputSpikeBuffer<T>::initialize(void * spikeBuffer, 
                                                                        size_t spikeBufferSize)
{
        serialBuffer = (T *)spikeBuffer;
        buf_size = spikeBufferSize;     
}

template<typename T>
MPIOutputSpikeBuffer<T>::~MPIOutputSpikeBuffer()
{}

template<typename T>
void MPIOutputSpikeBuffer<T>::appendSpike(local_objectid_t oid,
                                     int timestamp, engineid_t eng)
{
    if (timestamps[eng] < timestamp) {
        timestamps[eng] = timestamp;
        _buffers[eng].push_back(guard_value);
        _buffers[eng].push_back(timestamp);
    }
    _buffers[eng].push_back(oid);
}

template<typename T>
void MPIOutputSpikeBuffer<T>::nextCycle()
{
    for (int i  = 0 ; i < nLocalEngines ; ++i) {
        _buffers[i].resize(1);
        timestamps[i] = -1;
    }
    isInit = false;
}

template<typename T>
void MPIOutputSpikeBuffer<T>::initSerialization()
{

    serializationFinished = false;
    engineid_t first_non_empty = 0;
    for (int j = 0 ; j < nLocalEngines; ++j) {
        _buffers[j].push_back(guard_value);
    }
    while (first_non_empty < nLocalEngines && _buffers[first_non_empty].size() == 2)
        first_non_empty++;
    if (first_non_empty == nLocalEngines) {
        serializationFinished = true;
        s_curr_timestamp = -1;
    } else {
        s_curr_eng = first_non_empty;
        s_curr_timestamp = -1;
        s_curr_src_pos = _buffers[first_non_empty].begin();
        s_src_end_pos = _buffers[first_non_empty].end();
    }
}

template<typename T>
const T* const MPIOutputSpikeBuffer<T>::getBuffer()
{
    return serialBuffer;
}

template<typename T>
void MPIOutputSpikeBuffer<T>::setFinishedFlag(bool flag)
{
    *(serialBuffer+1) = flag;
}


template<typename T>
int MPIOutputSpikeBuffer<T>::prepareNextSerializedMPIBuffer(size_t currentBufferSize)
{
    if (!isInit) {
        initSerialization();
        isInit = true;
    }

    T *dest_pos = serialBuffer;
    T *dest_end_pos = serialBuffer + currentBufferSize;
    bool withoutHeader = true;

    // jump two positions ahead, to leave space for the "finished" indicator
    // and the length of the buffer
    dest_pos+=2;

    if (serializationFinished) {
        int length = 2;
        *serialBuffer = length ;
        *(serialBuffer + 1) = (T)serializationFinished;
        return length;
    }


    if (s_curr_timestamp != -1) { // we are not at the beginning of a buffer, so fill up the header
        *(dest_pos++) = (*gl_engineids)[s_curr_eng];
        *(dest_pos++) = guard_value;
        *(dest_pos++) = s_curr_timestamp;
        withoutHeader = false;
    }
    // transfer whole engine-buffers in the serialBuffer while there is space
    while (!serializationFinished &&
            (s_src_end_pos - s_curr_src_pos) <= ((dest_end_pos - dest_pos)-1) ) {
        copy(s_curr_src_pos, s_src_end_pos, dest_pos);
        dest_pos += s_src_end_pos - s_curr_src_pos;
        // go to the next src buffer (skip the empty ones)
        s_curr_eng++;
        while (s_curr_eng < nLocalEngines && _buffers[s_curr_eng].size() == 2)
            s_curr_eng++;

        // check if we are finished. If we are not, then setup the pointers
        if (s_curr_eng == nLocalEngines) {
            serializationFinished = true;
        } else {
            s_curr_timestamp = -1;
            s_curr_src_pos = _buffers[s_curr_eng].begin();
            s_src_end_pos = _buffers[s_curr_eng].end();
        }
        *(dest_pos++) = guard_value;
        withoutHeader = true;
    }

    if (!serializationFinished) {

        // there is only space to copy a part of the current buffer, so try to copy as much as possible
        typename vector<T>::const_iterator last_possible_pos = s_curr_src_pos + ((dest_end_pos - dest_pos) -2);
        // adjust the last_possible_pos since we may be on a guard or a timestamp
        if (last_possible_pos - s_curr_src_pos >= 2) {
            if (*(last_possible_pos-2) == guard_value )
                last_possible_pos-=2;
            else
                if (*(last_possible_pos-1) == guard_value )
                    last_possible_pos-- ;
        }

        if (withoutHeader) {
            // have we enough space to put the header and at least one oid?
            if (last_possible_pos - s_curr_src_pos >= 4) {
                *(dest_pos++) = (*gl_engineids)[s_curr_eng];
                *(dest_pos++) = guard_value;
                *(dest_pos++) = *(s_curr_src_pos + 2);
                s_curr_src_pos += 3;
            }
        }

        if (last_possible_pos - s_curr_src_pos > 0) {
            copy(s_curr_src_pos, last_possible_pos, dest_pos);
            dest_pos += last_possible_pos - s_curr_src_pos;
            s_curr_src_pos = last_possible_pos;
            if (*s_curr_src_pos == guard_value)
                s_curr_src_pos+=2;
            typename vector<T>::const_iterator timestamp_pos = last_possible_pos;
            while (*timestamp_pos != guard_value)
                timestamp_pos--;
            s_curr_timestamp = *(timestamp_pos+1);
            *(dest_pos++) = guard_value;
            *(dest_pos++) = guard_value;
        }
    }


    int length = dest_pos - serialBuffer;
    *serialBuffer = length ;     
    *(serialBuffer+1) = (T)serializationFinished;

    return length;
}

template<typename T>
bool MPIOutputSpikeBuffer<T>::hasNextSerializedMPIBuffer()
{
    return !serializationFinished;
}


template<typename T>
MPIOutputSpikeBufferVector<T>::MPIOutputSpikeBufferVector(int numBuffers,
        vector<gl_engineid_t> &engIDs,
        size_t buffer_size) :
        nNodes(numBuffers), buf_size(buffer_size)
{

    serialBuffersPool = new T[nNodes*buf_size];

    _outputbuffers.resize(nNodes);

    for (int i = 0; i < nNodes; ++i) {
        _outputbuffers[i] = new MPIOutputSpikeBuffer<T>(engIDs);
        _outputbuffers[i]->initialize(serialBuffersPool + i*buf_size, buf_size);        
    }

    nextCycle();
}


template<typename T>
MPIOutputSpikeBufferVector<T>::~MPIOutputSpikeBufferVector()
{
    for (int i = 0; i < nNodes; ++i) {
        delete _outputbuffers[i];
    }

    delete [] serialBuffersPool;
}


#endif /*MPIOutputSpikeBuffer_H_*/

---