---

ParGraph.hpp

#ifndef _PAR_GRAPH_HPP_
#define _PAR_GRAPH_HPP_

#include <stddef.h>
#include "PG_mpi_traits.hpp"

/* content: communication layer
            abstract communication buffers -> class AllToAllBuffer
*/

namespace ParGraph
{
//typedef unsigned int size_t;
typedef unsigned int proc_t;

typedef unsigned int VertexID_t;
typedef unsigned int EdgeID_t;

void ParGraph_init( int& argc, char**& argv );
void ParGraph_init(int& argc, char**& argv, MPI_Comm& user_comm);
void ParGraph_finalize();

bool PG_ReduceOr( bool );
bool PG_ReduceAnd( bool );
size_t PG_ReduceAdd( size_t );
template< class A, class Function >
A PG_Reduce( A a, Function f )
{
  A* tmp = new A[_libbase.size()];
  MPI_Allgather( &a, 1, mpi_traits<A>::mpi_type,
                 tmp, 1, mpi_traits<A>::mpi_type, _libbase.comm() );
  return std::accumulate( tmp, tmp+_libbase.size(), A(), f );
}

class comm_layer_t
{
    MPI_Comm _comm;
    proc_t   _rank;
    proc_t   _size;
    bool     _mpi_was_init;
    int     *_sendrecv_buffer, *scounts, *sdispls, *rcounts, *rdispls;
public:
    inline MPI_Comm comm() const
    {
        return _comm;
    }

    inline size_t rank() const
    {
        return _rank;
    }

    inline size_t size() const
    {
        return _size;
    }

    friend void ParGraph_init( int&, char**& );
    friend void ParGraph_init( int&, char**&, MPI_Comm& );
    friend void ParGraph_common_init();
    friend void ParGraph_finalize();
    friend void _pg_mpi_reduce_op_wrapper( void*, void*, int*, MPI_Datatype* );
    template< class A >
    friend A PG_Reduce( A, void (*op)(void*,void*) );
};

extern comm_layer_t _libbase;

inline size_t rank()
{
  return _libbase.rank();
}

inline size_t size()
{
  return _libbase.size();
}

const int COMMUNICATION_BUFFER_INITIAL_SIZE = 10000;

class AllGatherBuffer
{
    char *send_buffer;
    int send_size;
    int send_maxsize;
    int send_position;

    char *recv_buffer;
    int *recv_buffer_parts;
    int *recv_size;
    int cur_recv_position;
    int whole_recv_size;
    proc_t last_recv_proc;

public:
    AllGatherBuffer();
    ~AllGatherBuffer();

    template< class A >
    void put( A );
    template< class A >
    void put( const A * element, size_t count );

    void communicate();

    inline bool is_empty() { return cur_recv_position >= whole_recv_size; }
    template< class A >
    proc_t get( A * element, size_t count = 1 );
};

template< class A >
inline void
AllGatherBuffer::put( A element )
{
  put( & element, 1 );
}

template< class A >
void
AllGatherBuffer::put( const A * element, size_t count )
{
  if( send_buffer == 0 )
  {
    send_buffer = (char*) malloc( COMMUNICATION_BUFFER_INITIAL_SIZE );
    send_maxsize = COMMUNICATION_BUFFER_INITIAL_SIZE;
  }

  int diff;
  diff = sizeof(int)*count;     // WORKAROUND
//  MPI_Pack_size( count, mpi_traits<A>::mpi_type, _libbase.comm(), &diff );

  if( send_size + diff > send_maxsize )
  {
    send_maxsize = 2 * (send_size + diff);
    if( (send_buffer = (char*) realloc( (void*) send_buffer, send_maxsize )) == 0 )
    {
      std::cerr<<"Not enough Memory in AllGatherBuffer"<<std::endl; 
      exit( 1 );
    }
  }
  send_size += diff;

  for( size_t i = 0 ; i < count ; i++ ) // WORKAROUND Begin
  {
    *(reinterpret_cast<A *> (send_buffer + send_position)) = *element;
    element++;
    send_position+=sizeof(int);
  } // WORKAROUND End
//  MPI_Pack( const_cast<A*>(element), count, mpi_traits<A>::mpi_type,
//            send_buffer, send_maxsize, &send_position,
//            _libbase.comm() );
}

template< class A >
proc_t
AllGatherBuffer::get( A * element, size_t count )
{
  for(; last_recv_proc < _libbase.size() - 1 && 
        recv_buffer_parts[last_recv_proc+1]<=cur_recv_position ; last_recv_proc++ );
  for( size_t i = 0 ; i < count ; i++ ) // WORKAROUND Begin
  {
    *element = *(reinterpret_cast<A *> (recv_buffer + cur_recv_position));
    element++;
    cur_recv_position+=sizeof(int);
  } // WORKAROUND End
//  MPI_Unpack( recv_buffer, whole_recv_size, &cur_recv_position,
//            element, count, mpi_traits<A>::mpi_type,
//            _libbase.comm() );

  return last_recv_proc;
}



class AllToAllBuffer
{
    char **send_buffer;
    int *send_size;
    int *send_maxsize;
    int *send_position;
    proc_t cur_proc;

    char *recv_buffer;
    int *recv_buffer_parts;
    int *recv_size;
    int cur_recv_position;
    int whole_recv_size;
    proc_t last_recv_proc;

public:
    AllToAllBuffer();
    ~AllToAllBuffer();

    inline void set_proc( proc_t proc ) { cur_proc = proc; }
    template< class A >
    inline void put( A element );
    template< class A >
    inline void put( const A * element, size_t count );
    template< class A >
    void put( const A * element, size_t count, proc_t proc );

    void communicate();

    inline bool is_empty() { return cur_recv_position >= whole_recv_size; }
    template< class A >
    proc_t get( A * element, size_t count = 1 );
};

template< class A >
inline void
AllToAllBuffer::put( A element )
{
  put( & element, 1, cur_proc );
}

template< class A >
inline void
AllToAllBuffer::put( const A * element, size_t count )
{
  put( element, count, cur_proc );
}

template<>
inline void
AllToAllBuffer::put( const bool * element, size_t count )
{
// TODO: greater count than 1
  int value = (*element) ? -1 : 0;
  put( & value, 1, cur_proc );
}

template< class A >
void
AllToAllBuffer::put( const A * element, size_t count, proc_t proc )
{
  if( send_buffer[proc] == 0 )
  {
    send_buffer[proc] = (char*) malloc( COMMUNICATION_BUFFER_INITIAL_SIZE );
    send_maxsize[proc] = COMMUNICATION_BUFFER_INITIAL_SIZE;
  }

  int diff;
//  MPI_Pack_size( count, mpi_traits<A>::mpi_type, _libbase.comm(), &diff );
  diff=sizeof(int)*count; // WORKAROUND
  if( send_size[proc] + diff > send_maxsize[proc] )
  {
    send_maxsize[proc] = 2 * (send_size[proc] + diff);
    if( (send_buffer[proc] = (char*) realloc( (void*) send_buffer[proc], send_maxsize[proc] ))
        == 0 )
    {
      std::cerr<<"Not enough Memory in AllToAllBuffer"<<std::endl; 
      exit( 1 );
    }
  }
  send_size[proc] += diff;

  for( size_t i = 0 ; i < count ; i++ ) // WORKAROUND Begin
  {
    *(reinterpret_cast<A *> (send_buffer[proc] + send_position[proc])) = *element;
    element++;
    send_position[proc]+=sizeof(int);
  } // WORKAROUND End
  //MPI_Pack( const_cast<A*>(element), count, mpi_traits<A>::mpi_type,
  //          send_buffer[proc], send_maxsize[proc], &send_position[proc],
  //          _libbase.comm() );
}

template< class A >
proc_t
AllToAllBuffer::get( A * element, size_t count )
{
  for(; last_recv_proc < _libbase.size() - 1 && 
        recv_buffer_parts[last_recv_proc+1]<=cur_recv_position ; last_recv_proc++ );
  for( size_t i = 0 ; i < count ; i++ ) // WORKAROUND Begin
  {
    *element = *(reinterpret_cast<A *> (recv_buffer + cur_recv_position));
    element++;
    cur_recv_position+=sizeof(int);
  } // WORKAROUND End

//  MPI_Unpack( recv_buffer, whole_recv_size, &cur_recv_position,
//            element, count, mpi_traits<A>::mpi_type,
//            _libbase.comm() );
  return last_recv_proc;
}

template<>
inline proc_t
AllToAllBuffer::get( bool * element, size_t count )
{
// TODO: greater count than 1
  int value;
  proc_t p = get( & value, 1 );
  *element = (value==0) ? false : true;
  return p;
}
} // namespace ParGraph

#endif // _PAR_GRAPH_HPP_

---