auryn::SyncBuffer Class Reference

Buffer object to capsulate native MPI_Allgather for SpikingGroups. More...

#include <SyncBuffer.h>

Public Member Functions
	SyncBuffer (mpi::communicator *com)
	The default contructor. More...
virtual	~SyncBuffer ()
	The default destructor. More...
void	sync ()
	Synchronize spikes and additional information across ranks. More...
void	push (SpikeDelay *delay, const NeuronID size)
	Pushes a spike delay with all its spikes to the SyncBuffer. More...
void	null_terminate_send_buffer ()
	Terminate send buffer. More...
void	pop (SpikeDelay *delay, const NeuronID size)
	Rerieves a spike delay with all its spikes from the SyncBuffer. More...
int	get_max_send_buffer_size ()
	Return max_send_size value which determines the size of the MPI AllGather operation. More...
unsigned int	get_sync_count ()
	Return sync count since object instantiation. More...
unsigned int	get_overflow_count ()
	Return overflow count since object instantiation. More...

Detailed Description

Buffer object to capsulate native MPI_Allgather for SpikingGroups.

The class stores the recent history of transmission buffer sizes and tries to determine an optimal size on-line.

Constructor & Destructor Documentation

SyncBuffer()

SyncBuffer::SyncBuffer

(

mpi::communicator *

com

)

The default contructor.

{
        mpicom = com;
        init();
}

~SyncBuffer()

SyncBuffer::~SyncBuffer ( )

virtual

The default destructor.

{
        delete [] pop_offsets;
        delete [] pop_delta_spikes;
        delete [] last_spike_pos;

        delete [] rank_send_sum;
        delete [] rank_send_sum2;
        delete [] rank_recv_count;
        delete [] rank_displs;
}

Member Function Documentation

get_max_send_buffer_size()

int SyncBuffer::get_max_send_buffer_size

(

)

Return max_send_size value which determines the size of the MPI AllGather operation.

{
        return max_send_size;
}

get_overflow_count()

unsigned int SyncBuffer::get_overflow_count

(

)

Return overflow count since object instantiation.

{
        return overflow_counter;
}

get_sync_count()

unsigned int SyncBuffer::get_sync_count

(

)

Return sync count since object instantiation.

{
        return sync_counter;
}

null_terminate_send_buffer()

void SyncBuffer::null_terminate_send_buffer

(

)

Terminate send buffer.

{
        // puts a "delta spike" just behind the last unrolled delay of the last group
        // std::cout << " term " << carry_offset << std::endl;
        send_buf.push_back(carry_offset);
}

pop()

void SyncBuffer::pop	(	SpikeDelay *	delay,
		const NeuronID	size
	)

Rerieves a spike delay with all its spikes from the SyncBuffer.

{
        // TODO consider passing the current rank, because in principle it should not require the sync
        
        const SYNCBUFFER_DELTA_DATATYPE grid_size = (SYNCBUFFER_DELTA_DATATYPE)size*MINDELAY;

        // clear all receiving buffers in the relevant time range
        for (NeuronID i = 1 ; i < MINDELAY+1 ; ++i ) {
                delay->get_spikes(i)->clear();
                delay->get_attributes(i)->clear();
        }

        // loop over different rank input segments in recv_buf
        for (int r = 0 ; r < mpicom->size() ; ++r ) {

                // when we enter this function we know this is a new group
                last_spike_pos[r] = 0;

                //read current difference element from buffer and interpret as unrolled
                NeuronID * iter = &recv_buf[r*max_send_size+pop_offsets[r]]; // first spike

                while ( true ) {

                        // std::cout << "have " << pop_delta_spikes[r] << std::endl;
                        
                        if ( pop_delta_spikes[r] == undefined_delta_size ) {
                                // read delta spike value from buffer and update iterator
                                iter = read_delta_spike_from_buffer(iter, pop_delta_spikes[r]);
                        }

                        SYNCBUFFER_DELTA_DATATYPE unrolled_spike = last_spike_pos[r] + pop_delta_spikes[r];
                        if ( unrolled_spike < grid_size ) {

                                // decode spike positon on time slice grid
                                const int slice = unrolled_spike/size;
                                const NeuronID spike = unrolled_spike%size;
                                
                                // push spike to appropriate time slice
                                delay->get_spikes(slice+1)->push_back(spike);
                                // std::cout << "slice " << slice << " spike " << spike << std::endl;

                                // save last position
                                last_spike_pos[r] = unrolled_spike;
                                pop_delta_spikes[r] = undefined_delta_size;

                                // now that we know where the spike belongs we read the spike attributes from the buffer
                                for ( int k = 0 ; k < delay->get_num_attributes() ; ++k ) { // loop over attributes
                                        AurynFloat attrib;
                                        iter = read_attribute_from_buffer(iter, attrib);
                                        delay->get_attributes(slice+1)->push_back(attrib);
                                        // std::cout << "read " << std::scientific << attrib << std::endl;
                                }

                        } else {
                                pop_delta_spikes[r] -= (grid_size-last_spike_pos[r]);
                                // std::cout << "keep " << pop_delta_spikes[r] << std::endl;
                                break;
                        }
                }
                pop_offsets[r] = iter - &recv_buf[r*max_send_size]; // save offset in recv_buf section
        }

        // // TEST TODO comment after testing
        // for (NeuronID i = 1 ; i < MINDELAY+1 ; ++i ) {
        //      SpikeContainer * myvector = delay->get_spikes(i);
        //      std::sort (myvector->begin(), myvector->end());
        // }

#ifdef DEBUG
        if ( mpicom->rank() == 0 ) {
                for ( NeuronID slice = 0 ; slice < MINDELAY ; ++slice ) {
                        if ( delay->get_attributes(slice+1)->size() != delay->get_num_attributes()*delay->get_spikes(slice+1)->size() ) {
                                std::cout << "   " << delay->get_spikes(slice+1)->size() << " spikes extracted in time slice " << slice+1 << std::endl
                                        << "   " << delay->get_attributes(slice+1)->size() << " attributes extracted in time slice " << slice+1
                                        << std::endl;
                        }
                }
        }
#endif // DEBUG
}

Here is the call graph for this function:

push()

void SyncBuffer::push	(	SpikeDelay *	delay,
		const NeuronID	size
	)

Pushes a spike delay with all its spikes to the SyncBuffer.

{
        // DEBUG
        // std::cout << "Rank " << mpicom->rank() << "push\n";
        // delay->print();
        
        const SYNCBUFFER_DELTA_DATATYPE grid_size = (SYNCBUFFER_DELTA_DATATYPE)size*MINDELAY;

        SYNCBUFFER_DELTA_DATATYPE unrolled_last_pos = 0;
        // circular loop over different delay bins
        for (int slice = 0 ; slice < MINDELAY ; ++slice ) {
                SpikeContainer * sc = delay->get_spikes(slice+1);
                AttributeContainer * ac = delay->get_attributes(slice+1);

                // loop over all spikes in current delay time slice
                for (int i = 0 ; 
                                i < sc->size() ; 
                                ++i ) {
                        NeuronID spike = sc->at(i);
                        // compute unrolled position in current delay
                        SYNCBUFFER_DELTA_DATATYPE unrolled_pos = (SYNCBUFFER_DELTA_DATATYPE)(spike) + (SYNCBUFFER_DELTA_DATATYPE)size*slice; 
                        // compute vertical unrolled difference from last spike
                        SYNCBUFFER_DELTA_DATATYPE spike_delta = unrolled_pos + carry_offset - unrolled_last_pos;
                        // memorize current position in slice
                        unrolled_last_pos = unrolled_pos; 
                        // discard carry_offset since its only added to the first spike_delta
                        carry_offset = 0;

                        // overflow managment -- should only ever kick in for very very large SpikingGroups and very very sparse activity
                        while ( spike_delta >= max_delta_size ) {
                                send_buf.push_back( max_delta_size );
                                spike_delta -= max_delta_size;
                        }
                
                        // storing the spike delta (or its remainder) to buffer
                        send_buf.push_back(spike_delta);

                        // append spike attributes here in buffer
                        for ( int k = 0 ; k < delay->get_num_attributes() ; ++k ) { // loop over attributes
                                NeuronID cast_attrib = *(NeuronID*)(&(ac->at(i*delay->get_num_attributes()+k)));
                                send_buf.push_back(cast_attrib);
                                // std::cout << "store " << std::scientific << ac->at(i*delay->get_num_attributes()+k) << " int " << cast_attrib << std::endl;
                        }
                }
        }

        // set save carry_offset which is the remaining difference from the present group
        // plus because there might be more than one group without a spike ...
        carry_offset += grid_size-unrolled_last_pos;

}

Here is the call graph for this function:

sync()

void SyncBuffer::sync

(

)

Synchronize spikes and additional information across ranks.

{
        // allgather seems to be faster on the standard sim_background benchmark
        // however, replacing the following line by synchronize will run the new
        // code using Allgatherv in which each rank can send different amounts of data.
        sync_allgather();
}

---

The documentation for this class was generated from the following files:

• /home/zenke/auryn/src/auryn/SyncBuffer.h
• /home/zenke/auryn/src/auryn/SyncBuffer.cpp