---

PG_owner_computes_graph.hpp

#ifndef _PG_OWNER_COMPUTES_GRAPH_HPP_
#define _PG_OWNER_COMPUTES_GRAPH_HPP_

#include <map>
#include <vector>
#include <set>
#include <list>
#include <utility>
#include <numeric>
#include <algorithm>
#include <limits>
#include <assert.h>

#include "ParGraph.hpp"

namespace ParGraph
{
using std::pair;
using std::make_pair;

typedef std::map< const VertexID_t, pair<size_t, size_t> > KnownVerticesMap_type;
typedef std::map< const VertexID_t, pair<size_t, size_t> >::iterator known_vertex_type;
typedef std::map< const VertexID_t, pair<size_t, size_t> >::const_iterator const_known_vertex_type;
}

#include "PG_property.hpp"
#include "PG_edges.hpp"
#include "PG_property_map.hpp"
#include "PG_vertex_sets.hpp"
#include "PG_heap.hpp"

namespace std
{
inline bool operator<( const ParGraph::known_vertex_type & v1, const ParGraph::known_vertex_type & v2 )
{
    return v1->first < v2->first;
}
}

namespace ParGraph
{

inline VertexID_t get_id( known_vertex_type v )
{
  return v->first;
}
inline VertexID_t get_id( const_known_vertex_type v )
{
  return v->first;
}

template< class Graph, class Iter >
class LocalVertexIterator
{
  Iter iter;
  Graph * g;
  void find_next()
  {
    while( iter != g->known_vertices.end() )
    {
      if( (*iter).second.first % (_libbase.size() + 1) == _libbase.rank() ) return;
      iter++;
    }
  }
public:
  LocalVertexIterator() {}
  LocalVertexIterator( Graph & _g )
  {
    g = &_g;
    iter = g->known_vertices.begin();
    find_next();
  }
  void operator++(int)
  {
    if( iter != g->known_vertices.end() )
    {
      iter++;
      find_next();
    }
  }
  bool operator==( const LocalVertexIterator<Graph,Iter> & other )
  {
    return iter == other.iter;
  }
  bool operator!=( const LocalVertexIterator<Graph,Iter> & other )
  {
    return iter != other.iter;
  }
  Iter operator*()
  {
      assert(iter!= g->known_vertices.end());
    return iter;
  }
  void goto_end()
  {
    iter = g->known_vertices.end();
  }
  int get_index()
  {
    return (*iter).second.first / (_libbase.size() + 1);
  }
};

template< class Graph >
struct SimpleDistribution
{
  Graph & g;

  SimpleDistribution( Graph & _g ) : g(_g) {}

  void notify_id( VertexID_t ) {}
  void notify_about_vertices() { g.notify_about_vertices(); }
  void write_proc( typename Graph::KnownVertex, AllToAllBuffer & ) {}
  proc_t read_proc( AllToAllBuffer & ) { return _libbase.size(); }
};

template< class Graph >
struct StandardDistribution
{
  AllGatherBuffer notify_buffer;
  Graph & g;

  StandardDistribution( Graph & _g ) : notify_buffer(), g(_g) {}

  void notify_id( VertexID_t id ) { notify_buffer.put( id ); }
  void notify_about_vertices()
        {
            g.notify_about_vertices( notify_buffer );
        }
  void write_proc( typename Graph::KnownVertex id , AllToAllBuffer & buf ) { buf.put( g.owning_processor(id) ); }
  proc_t read_proc( AllToAllBuffer & buf )
        {
            proc_t id;
            buf.get( &id );
            return id;
        }
};

template < class VProperty >
class OwnerComputesGraph
{
public:
  typedef OwnerComputesGraph< VProperty > self_type;
  typedef ParGraph::known_vertex_type KnownVertex;
  typedef typename ParGraph::KnownVerticesMap_type::const_iterator const_KnownVertex;
  typedef ParGraph::KnownVerticesMap_type KnownVerticesMap_type;
  typedef LocalVertexIterator< 
      self_type, KnownVertex > VertexIterator;
  typedef LocalVertexIterator<
      const self_type,
      const_KnownVertex > const_VertexIterator;
  typedef VProperty VertexProperty;

  typedef SetWrapper< self_type > VertexSet;
  typedef std::list< VertexSet > VertexSets;
  typedef typename VertexSets::iterator VertexSetHandle;

private:
  VertexID_t lower_free_ID;
  EdgeID_t lower_free_edge_ID;
  PG_heap< VProperty > local_vertices;
  KnownVerticesMap_type known_vertices;
  VertexSets registered_vertex_sets;
  SimpleDistribution< self_type > simple_helper;
  StandardDistribution< self_type > standard_helper;

public:
  OwnerComputesGraph();

  void set_user_vertex_id_range( VertexID_t id ) { lower_free_ID = id + _libbase.rank(); }
  void set_user_edge_id_range( EdgeID_t id ) { lower_free_edge_ID = id + _libbase.rank(); }
  EdgeID_t get_free_edge_id() { EdgeID_t tmp = lower_free_edge_ID;
                                lower_free_edge_ID += _libbase.size(); return tmp; }

  pair< KnownVertex, bool > insert();
  pair< KnownVertex, bool > insert( const VertexProperty & );
  pair< KnownVertex, bool > insert( VertexID_t );
  pair< KnownVertex, bool > insert( VertexID_t, const VertexProperty & );
  void erase( KnownVertex );

  inline VProperty & get_property_reference( VertexID_t v );
  inline const VProperty & get_property_reference( VertexID_t v ) const;
  inline VProperty & get_property_reference( KnownVertex );
  inline const VProperty & get_property_reference( typename KnownVerticesMap_type::const_iterator ) const;
  inline KnownVertex get_known_vertex( VertexID_t v );
  inline KnownVertex get_known_vertex( VertexID_t v ) const;
  inline KnownVertex get_known_vertex( KnownVertex v );

  inline bool is_vertex_local( VertexID_t ) const;
  inline bool is_vertex_local( KnownVertex ) const;
  inline bool is_vertex_local( const_KnownVertex ) const;
  inline bool is_vertex_known( VertexID_t ) const;
  inline size_t owning_processor( VertexID_t ) const;
  inline size_t owning_processor( KnownVertex ) const;
  inline size_t owning_processor( const_KnownVertex ) const;
  inline KnownVertex ink_known_vertex_counter( VertexID_t );
  inline KnownVertex ink_known_vertex_counter_and_set_processor( VertexID_t, proc_t );
  inline void ink_known_vertex_counter( KnownVertex );
  inline void dek_known_vertex_counter( VertexID_t );
  inline void dek_known_vertex_counter( KnownVertex );
  inline KnownVertex make_vertex_known( VertexID_t, proc_t );

  pair< VertexIterator, VertexIterator > get_local_vertices();
  pair< const_VertexIterator, const_VertexIterator > get_local_vertices() const;

  template< class DistHelper >
  void redistribute( DistHelper & );
  template< class PMap, class DistHelper >
  void redistribute( PMap, DistHelper & );

  template< class DistHelper >
  void do_distribution( std::map< KnownVertex, proc_t > &, DistHelper & );
  void notify_about_vertices();
  void notify_about_vertices_check();
  void notify_about_vertices(AllGatherBuffer &);

  SimpleDistribution< self_type > & use_simple_distribution()
        { return simple_helper; }
  StandardDistribution< self_type > & use_standard_distribution()
        { return standard_helper; }

  friend VertexSetHandle add_vertex_set<>( self_type & );
  friend void remove_vertex_set<>( self_type &, VertexSetHandle );

  friend class LocalVertexIterator< self_type, KnownVertex >;
  friend class LocalVertexIterator< const self_type,
           typename KnownVerticesMap_type::const_iterator >;

  friend void do_asking_for_position<>( self_type &, AllToAllBuffer & );
private:
  KnownVertex insert_from_remote( VertexID_t );
  void do_distribution_on_vertex_sets( std::map< KnownVertex, proc_t > & );
  void compose_distribution( int *, int, int *, int * );
  inline size_t get_vertex_index( VertexID_t ) const;
  inline size_t get_vertex_index( KnownVertex ) const;
  inline size_t get_vertex_position( VertexID_t ) const;
  inline size_t get_vertex_position( KnownVertex ) const;
  inline void dek_known_vertex_counter_and_unmark_position( VertexID_t );
  inline void dek_known_vertex_counter_and_unmark_position( KnownVertex );
  inline void ink_known_vertex_counter_and_set_position( VertexID_t, size_t );
};

template< class Property >
OwnerComputesGraph<Property>::OwnerComputesGraph() : simple_helper(*this), standard_helper(*this)
{
    lower_free_edge_ID = _libbase.rank();
    lower_free_ID = _libbase.rank();
}

template< class Property >
pair< typename OwnerComputesGraph<Property>::KnownVertex, bool >
OwnerComputesGraph<Property>::insert()
{
    VertexID_t new_ID = lower_free_ID;
    lower_free_ID += _libbase.size();
    insert( new_ID );
    return new_ID;
}

template< class Property >
pair< typename OwnerComputesGraph<Property>::KnownVertex, bool >
OwnerComputesGraph<Property>::insert( const Property & prop )
{
    VertexID_t new_ID = lower_free_ID;
    lower_free_ID += _libbase.size();
    insert( new_ID, prop );
    return new_ID;
}

template< class Property >
pair< typename OwnerComputesGraph<Property>::KnownVertex, bool >
OwnerComputesGraph<Property>::insert( VertexID_t vid )
{
  KnownVertex v = known_vertices.find(vid);
  if( v == known_vertices.end() )
  {
    size_t vpos = local_vertices.insert() * (_libbase.size() + 1) + _libbase.rank();
    return known_vertices.insert( make_pair( vid, make_pair( vpos, 1 ) ) );
  }
  else
  {
    if( is_vertex_local( v ) )
    {
      get_property_reference( v ) = Property();
    }
  }
  return make_pair( v, false );
}

template< class Property >
typename OwnerComputesGraph<Property>::KnownVertex
OwnerComputesGraph<Property>::insert_from_remote( VertexID_t vid )
{
  KnownVertex v = known_vertices.find(vid);
  if( v == known_vertices.end() )
  {
    size_t vpos = local_vertices.insert() * (_libbase.size() + 1) + _libbase.rank();
    pair<KnownVertex,bool> tmp = known_vertices.insert( make_pair( vid, make_pair( vpos, 1 ) ) );
    return tmp.first;
  }
  else
  {
    size_t vpos = local_vertices.insert() * (_libbase.size() + 1) + _libbase.rank();
    v->second.first = vpos;
    v->second.second++;
  }
  return v;
}

template< class Property >
pair< typename OwnerComputesGraph<Property>::KnownVertex, bool >
OwnerComputesGraph<Property>::insert( VertexID_t vid, const Property & prop )
{
  KnownVertex v = known_vertices.find(vid);
  if( v == known_vertices.end() )
  {
    size_t vpos = local_vertices.insert(prop) * (_libbase.size() + 1) + _libbase.rank();
    return known_vertices.insert( make_pair( vid, make_pair( vpos, 1 ) ) );
  }
  else
  {
   if( is_vertex_local( v ) )
    {
      get_property_reference( v ) = prop;
    }
  }
  return make_pair( v, false );
}

template< class Property >
void
OwnerComputesGraph<Property>::erase( KnownVertex v )
{
  local_vertices.erase( get_vertex_index( v ) );
  dek_known_vertex_counter_and_unmark_position( v );
}

template< class VProperty >
inline VProperty &
OwnerComputesGraph<VProperty>::get_property_reference( VertexID_t v )
{ 
    return local_vertices[ get_vertex_index(v) ];
}

template< class VProperty >
inline const VProperty &
OwnerComputesGraph<VProperty>::get_property_reference( VertexID_t v ) const
{ 
    return local_vertices[ get_vertex_index(v) ];
}

template< class VProperty >
inline VProperty &
OwnerComputesGraph<VProperty>::get_property_reference( KnownVertex v )
{ 
    return local_vertices[ v->second.first / (_libbase.size() + 1) ];
}

template< class VProperty >
inline const VProperty &
OwnerComputesGraph<VProperty>::get_property_reference( typename KnownVerticesMap_type::const_iterator v ) const
{ 
    return local_vertices[ v->second.first / (_libbase.size() + 1) ];
}

template< class VProperty >
inline typename OwnerComputesGraph<VProperty>::KnownVertex
OwnerComputesGraph<VProperty>::get_known_vertex( VertexID_t v )
{
    assert( known_vertices.find(v) != known_vertices.end() );
  return known_vertices.find(v);
}

template< class VProperty >
inline typename OwnerComputesGraph<VProperty>::KnownVertex
OwnerComputesGraph<VProperty>::get_known_vertex( VertexID_t v ) const
{ 
    assert( known_vertices.find(v) != known_vertices.end() );
  return known_vertices.find(v);
}

template< class VProperty >
inline typename OwnerComputesGraph<VProperty>::KnownVertex
OwnerComputesGraph<VProperty>::get_known_vertex( KnownVertex v )
{
  return v;
}

template< class Property >
inline bool
OwnerComputesGraph<Property>::is_vertex_local( VertexID_t v ) const
{
    const_KnownVertex it = known_vertices.find(v);
    if( it == known_vertices.end() ) return false;
    return it->second.first % (_libbase.size() + 1) == _libbase.rank();
}

template< class Property >
inline bool
OwnerComputesGraph<Property>::is_vertex_local( KnownVertex v ) const
{
//    assert( owning_processor(v) != _libbase.size() );
  return owning_processor(v) == _libbase.rank();
}

template< class Property >
inline bool
OwnerComputesGraph<Property>::is_vertex_local( const_KnownVertex v ) const
{
//    assert( owning_processor(v) != _libbase.size() );
  return owning_processor(v) == _libbase.rank();
}

template< class Property >
inline bool
OwnerComputesGraph<Property>::is_vertex_known( VertexID_t v ) const
{
    return known_vertices.find(v) != known_vertices.end();
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::get_vertex_position( VertexID_t v ) const
{
    assert( known_vertices.find(v) != known_vertices.end() );
  return (*known_vertices.find(v)).second.first;
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::get_vertex_position( KnownVertex v ) const
{
  return v->second.first;
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::get_vertex_index( VertexID_t v ) const
{
  return get_vertex_position(v) / (_libbase.size() + 1 );
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::get_vertex_index( KnownVertex v ) const
{
  return get_vertex_position(v) / (_libbase.size() + 1 );
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::owning_processor( VertexID_t v ) const
{
    assert( known_vertices.find(v) != known_vertices.end() );
  return owning_processor( known_vertices.find( v ) );
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::owning_processor( KnownVertex v ) const
{
  return v->second.first % (_libbase.size() + 1 );
}

template< class Property >
inline size_t
OwnerComputesGraph<Property>::owning_processor( const_KnownVertex v ) const
{
  return v->second.first % (_libbase.size() + 1 );
}

template< class Property >
inline typename OwnerComputesGraph<Property>::KnownVertex
OwnerComputesGraph<Property>::ink_known_vertex_counter( VertexID_t v )
{
  KnownVertex it = known_vertices.find(v);
  if( it == known_vertices.end() )
  {
    bool tmp;
    boost::tie(it,tmp) = known_vertices.insert( make_pair( v, make_pair( _libbase.size(), 1 ) ) );
  }
  else
  {
    it->second.second++;
  }
  return it;
}

template< class Property >
inline void
OwnerComputesGraph<Property>::ink_known_vertex_counter( KnownVertex v )
{
  v->second.second++;
}

template< class Property >
inline typename OwnerComputesGraph<Property>::KnownVertex
OwnerComputesGraph<Property>::ink_known_vertex_counter_and_set_processor( VertexID_t v, proc_t p )
{
    assert( p<=_libbase.size() && p != _libbase.rank() );
  KnownVertex it = known_vertices.find(v);
  if( it == known_vertices.end() )
  {
    bool tmp;
    boost::tie(it,tmp) = known_vertices.insert( make_pair( v, make_pair( p, 1 ) ) );
  }
  else
  {
    it->second.second++;
    if( it->second.first % (_libbase.size() + 1) == _libbase.size() )
        it->second.first = (it->second.first / (_libbase.size() + 1)) * (_libbase.size() + 1) + p;
  }
  return it;
}

template< class Property >
inline typename OwnerComputesGraph<Property>::KnownVertex
OwnerComputesGraph<Property>::make_vertex_known( VertexID_t v, proc_t p )
{
    assert( p<=_libbase.size() );
  KnownVertex it = known_vertices.find(v);
  if( it == known_vertices.end() )
  {
    bool tmp;
    boost::tie(it,tmp) = known_vertices.insert( make_pair( v, make_pair( p, 0 ) ) );
  }
  else
  {
    if( it->second.first % (_libbase.size() + 1) == _libbase.size() )
        it->second.first = (it->second.first / (_libbase.size() + 1)) * (_libbase.size() + 1) + p;
  }
  return it;
}

template< class Property >
inline void
OwnerComputesGraph<Property>::ink_known_vertex_counter_and_set_position( VertexID_t v, size_t pos )
{
  KnownVertex it = known_vertices.find(v);
  if( it == known_vertices.end() )
  {
    known_vertices.insert( make_pair( v, make_pair( pos, 1 ) ) );
  }
  else
  {
    it->second.second++;
    it->second.first = pos;
  }
}

template< class Property >
inline void
OwnerComputesGraph<Property>::dek_known_vertex_counter( VertexID_t v )
{
  KnownVertex it = known_vertices.find(v);
  if( ( --(it->second.second ) ) == 0 )
      known_vertices.erase(it);
}

template< class Property >
inline void
OwnerComputesGraph<Property>::dek_known_vertex_counter( KnownVertex v )
{
  if( ( --(v->second.second) ) == 0 )
      known_vertices.erase(v);
}

template< class Property >
inline void
OwnerComputesGraph<Property>::dek_known_vertex_counter_and_unmark_position( VertexID_t v )
{
  KnownVertex it = known_vertices.find(v);
  (*it).second.first = _libbase.size();
  if( ( --(it->second.second) ) == 0 )
      known_vertices.erase(it);
}

template< class Property >
inline void
OwnerComputesGraph<Property>::dek_known_vertex_counter_and_unmark_position( KnownVertex v )
{
  v->second.first = _libbase.size();
  if( ( --(v->second.second) ) == 0 )
      known_vertices.erase(v);
}

template< class Property >
pair< typename OwnerComputesGraph<Property>::VertexIterator, typename OwnerComputesGraph<Property>::VertexIterator >
OwnerComputesGraph<Property>::get_local_vertices()
{
  VertexIterator tmp(*this), tmp2(*this);
  tmp2.goto_end();
  return make_pair( tmp, tmp2 );
}

template< class Property >
pair< typename OwnerComputesGraph<Property>::const_VertexIterator, typename OwnerComputesGraph<Property>::const_VertexIterator >
OwnerComputesGraph<Property>::get_local_vertices() const
{
  const_VertexIterator tmp(*this), tmp2(*this);
  tmp2.goto_end();
  return make_pair( tmp, tmp2 );
}

inline int pg_max( int a, int b ) { return std::max(a,b); }

template< class Property >
void
OwnerComputesGraph<Property>::compose_distribution( int *sizes, int best_size, int *send_count_buf,
                                                    int *send_count )
{
  proc_t sendproc = 0, recvproc = 0;  
  for(; sendproc < _libbase.size() ; sendproc++ )
  {
    for(; sizes[sendproc] > best_size + 1 && recvproc < _libbase.size() ; recvproc++ )
    {
      if( best_size > sizes[recvproc] )
      {
        size_t tmp = std::min( sizes[sendproc] - best_size - 1,
                               best_size - sizes[recvproc] );
        sizes[sendproc]                 -= tmp;
        sizes[recvproc]                 += tmp;

        if( sendproc == _libbase.rank() )
        {
          (*send_count) += tmp; send_count_buf[recvproc] = tmp;
        }
      }
    }
  }
}

template< class Property >
template< class DistHelper >
void
OwnerComputesGraph<Property>::redistribute( DistHelper & dh )
{
  int *sizes = new int[_libbase.size()];
  int local_size = local_vertices.size();

  MPI_Allgather( &local_size, 1, MPI_INT, sizes, 1, MPI_INT, _libbase.comm() );
  int all_size = std::accumulate( sizes, sizes+_libbase.size(),0 );
  int max_local_size = std::accumulate( sizes, sizes+_libbase.size(),0, pg_max );
  int best_size = all_size/_libbase.size();

  if( max_local_size >= best_size + best_size / 2 )
  {
    int *send_count_buf =  new int[_libbase.size()], send_count;
    for( proc_t i = 0; i < _libbase.size() ; i++ ) send_count_buf[i] = 0;

    compose_distribution( sizes, best_size, send_count_buf, &send_count );

    std::map< KnownVertex, proc_t > movements;
    VertexIterator vit(*this);

    for( proc_t i = 0 ; i < _libbase.size() ; i++ )
    {
      for( int j = 0 ; j < send_count_buf[i] ; j++ )
      {
        movements.insert( make_pair( *vit, i ) );
        vit++;
      }
    }
    delete[] send_count_buf;
    do_distribution( movements, dh );
  }
  delete[] sizes;
}

template< class Property >
template< class PropertyMap, class DistHelper >
void
OwnerComputesGraph<Property>::redistribute( PropertyMap procmap, DistHelper & dh )
{
  VertexIterator vit, vend;
  boost::tie( vit, vend ) = get_local_vertices();
  std::map< KnownVertex, proc_t > movements;
  for(; vit != vend ; vit++ )
  {
    proc_t target = get( procmap, get_vertex_reference( vit ) );
    if( target != _libbase.rank() && target < _libbase.size() )
    {
      movements.insert( make_pair( *vit, target ) );
    }
  }
  do_distribution( movements, dh );
}

template< class Property >
template< class DistHelper >
void
OwnerComputesGraph<Property>::do_distribution( std::map< KnownVertex, proc_t > & movements, DistHelper & dh )
{
  AllToAllBuffer buffer;
  std::map< KnownVertex, proc_t >::iterator vit = movements.begin(),
                                           vend = movements.end();

  for(; vit != vend ; vit++ )
  {
    buffer.set_proc( vit->second );
    buffer.put( get_id( vit->first ) );
    put_property_into_buffer( *this, dh, get_property_reference( vit->first ), buffer );
    erase( vit->first );
  }

  buffer.communicate();

  while( ! buffer.is_empty() )
  {
    VertexID_t id;
    buffer.get( &id, 1 );
    KnownVertex v = insert_from_remote( id );
    get_property_from_buffer( *this, dh, get_property_reference( v ), buffer );
    dh.notify_id( id );
  }

  do_distribution_on_vertex_sets( movements );
  dh.notify_about_vertices();
}

template< class Property >
void
OwnerComputesGraph<Property>::notify_about_vertices( )
{
  AllGatherBuffer notify_buffer;
  pair< VertexIterator, VertexIterator > lv = get_local_vertices();
  VertexIterator it = lv.first;

  for(; it != lv.second ; it++ )
  {
    VertexID_t id = get_id(*it);
    int pos = get_vertex_position( *it );
    notify_buffer.put( &id, 1 );
    notify_buffer.put( &pos, 1 );
  }

  notify_buffer.communicate();

  while( ! notify_buffer.is_empty() )
  {
    VertexID_t id;
    int pos;
    notify_buffer.get( &id, 1 );
    notify_buffer.get( &pos, 1 );
    KnownVertex it = known_vertices.find( id );
    if( it != known_vertices.end() )
    {
      (*it).second.first = pos;
    }
  }
}

template< class Property >
void
OwnerComputesGraph<Property>::notify_about_vertices_check( )
{
  AllGatherBuffer notify_buffer;
  pair< VertexIterator, VertexIterator > lv = get_local_vertices();
  VertexIterator it = lv.first;
  if(rank()==0)std::cout<<"check"<<std::endl;
  for(; it != lv.second ; it++ )
  {
    VertexID_t id = get_id(*it);
    int pos = get_vertex_position( *it );
    notify_buffer.put( &id, 1 );
    notify_buffer.put( &pos, 1 );
  }

  notify_buffer.communicate();

  while( ! notify_buffer.is_empty() )
  {
    VertexID_t id;
    int pos;
    notify_buffer.get( &id, 1 );
    notify_buffer.get( &pos, 1 );
    KnownVertex it = known_vertices.find( id );
    if( it != known_vertices.end() )
    {
        if( (*it).second.first % (_libbase.size() + 1) != pos % (_libbase.size() + 1) )
            std::cout<<"P"<<rank()<<": v"<<id<<" hier "<< (*it).second.first % (_libbase.size() + 1)<<
                " soll sein "<<pos % (_libbase.size() + 1)<<std::endl;
    }
  }
}

template< class Property >
void
OwnerComputesGraph<Property>::notify_about_vertices( AllGatherBuffer & notify_buffer  )
{
  notify_buffer.communicate();

  while( ! notify_buffer.is_empty() )
  {
    VertexID_t id;
    proc_t p = notify_buffer.get( &id );
    if( p != _libbase.rank() )
    {
      KnownVertex it = known_vertices.find( id );
      if( it != known_vertices.end() )
      {
        size_t pos = (*it).second.first / (_libbase.size() + 1);
        (*it).second.first = p + pos * (_libbase.size() + 1);
      }
    }
  }
}

template< class Property >
void
OwnerComputesGraph<Property>::do_distribution_on_vertex_sets( std::map< KnownVertex, proc_t > & movements )
{
  AllToAllBuffer buffer;
  VertexID_t * tmp = new VertexID_t[ movements.size() * _libbase.size() ];
  int * size = new int[ _libbase.size() ];

  typename VertexSets::iterator sets = registered_vertex_sets.begin();

  for(; sets != registered_vertex_sets.end() ; sets++ )
  {
    for( proc_t i = 0 ; i < _libbase.size() ; i++ ) size[ i ] = 0;

    std::map< KnownVertex, proc_t >::iterator movementsit = movements.begin();
    std::set< KnownVertex >::iterator setit = (*sets).begin(), tmpsetit;
    while( setit != (*sets).end() && movementsit != movements.end() )
    {
      if( get_id(*setit) == get_id(movementsit->first) )
      {
        tmp[ size[ (*movementsit).second ] + movements.size() * (*movementsit).second ] = get_id(*setit);
        size[ (*movementsit).second ]++;
        tmpsetit = setit;
        tmpsetit++;
        (*sets).erase( setit );
        setit = tmpsetit;
        movementsit++;
      }
      else if( get_id(*setit) < get_id(movementsit->first) )
      {
        setit++;
      }
      else
      {
        movementsit++;
      }
    }
    for( proc_t i = 0 ; i < _libbase.size() ; i++ )
    {
      buffer.set_proc( i );
      buffer.put( size[i] );
      if( size[i] > 0 )
          buffer.put( &tmp[ i * movements.size() ], size[i] );
      size[i] = 0;
    }
  }
  delete[] tmp;
  delete[] size;
  buffer.communicate();

  for( proc_t i = 0 ; i < _libbase.size() ; i++ )
  {
    sets = registered_vertex_sets.begin();
    for(; sets != registered_vertex_sets.end(); sets++ )
    {
      int readsize;
      buffer.get( &readsize, 1 );
      for(; readsize > 0 ; readsize-- )
      {
        VertexID_t v;
        buffer.get( &v, 1 );
        (*sets).insert( v );
      }
    }
  }
}

inline void
ask_for_position( proc_t p, VertexID_t v, AllToAllBuffer & buf )
{
  buf.set_proc( p );
  buf.put( v );
}

template< class Graph >
void
do_asking_for_position( Graph & g, AllToAllBuffer & buf )
{
  buf.communicate();

  while( ! buf.is_empty() )
  {
    VertexID_t v;
    proc_t p = buf.get( &v, 1 );
    buf.set_proc( p );
    buf.put( v );
    buf.put( g.get_vertex_position( v ) );
  }

  buf.communicate();

  while( ! buf.is_empty() )
  {
    VertexID_t v;
    int pos;
    buf.get( &v, 1 );
    buf.get( &pos, 1 );
    g.ink_known_vertex_counter_and_set_position( v, pos );
  }
}

} // namespace ParGraph

#endif // _PG_OWNER_COMPUTES_GRAPH_HPP_

---