mpi_manager.h

Go to the documentation of this file.

/*
 *  mpi_manager.h
 *
 *  This file is part of NEST.
 *
 *  Copyright (C) 2004 The NEST Initiative
 *
 *  NEST is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  NEST is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with NEST.  If not, see <http://www.gnu.org/licenses/>.
 *
 */
 
#ifndef MPI_MANAGER_H
#define MPI_MANAGER_H
 
// Generated includes:
#include "config.h"
 
// C includes:
#include <unistd.h>
#ifdef HAVE_MPI
#include <mpi.h>
#endif
 
// C++ includes:
#include <cassert>
#include <cmath>
#include <iostream>
#include <limits>
#include <numeric>
#include <vector>
 
// Includes from libnestutil:
#include "manager_interface.h"
#include "stopwatch.h"
 
// Includes from nestkernel:
#include "nest_types.h"
#include "spike_data.h"
#include "target_data.h"
 
 
namespace nest
{
 
class MPIManager : public ManagerInterface
{
public:
  // forward declaration of internal classes
  class OffGridSpike;
 
  MPIManager();
  ~MPIManager() override
  {
  }
 
  void initialize( const bool ) override;
  void finalize( const bool ) override;
  void set_status( const Dictionary& ) override;
  void get_status( Dictionary& ) override;
 
  void init_mpi( int* argc, char** argv[] );
#ifdef HAVE_MPI
  void set_communicator( MPI_Comm );
 
  MPI_Comm
  get_communicator()
  {
    return comm;
  };
#endif
 
  size_t get_num_processes() const;
 
  size_t get_rank() const;
 
  size_t get_process_id_of_vp( const size_t vp ) const;
 
  /*
   * Return the process id of the node with the specified node ID.
   */
  size_t get_process_id_of_node_id( const size_t node_id ) const;
 
  void mpi_finalize( int exitcode );
 
  void mpi_abort( int exitcode );
 
  // gather all send_buffer vectors on other mpi process to recv_buffer
  // vector
  void communicate( std::vector< size_t >& send_buffer, std::vector< size_t >& recv_buffer );
 
  void communicate( std::vector< unsigned int >& send_buffer,
    std::vector< unsigned int >& recv_buffer,
    std::vector< int >& displacements );
 
  void communicate( std::vector< OffGridSpike >& send_buffer,
    std::vector< OffGridSpike >& recv_buffer,
    std::vector< int >& displacements );
 
  void communicate( std::vector< double >& send_buffer,
    std::vector< double >& recv_buffer,
    std::vector< int >& displacements );
 
  void communicate( std::vector< unsigned long >& send_buffer,
    std::vector< unsigned long >& recv_buffer,
    std::vector< int >& displacements );
 
  void
  communicate( std::vector< int >& send_buffer, std::vector< int >& recv_buffer, std::vector< int >& displacements );
 
  void communicate( double, std::vector< double >& );
  void communicate( std::vector< int >& );
  void communicate( std::vector< long >& );
 
  void communicate_Allreduce_sum_in_place( double buffer );
  void communicate_Allreduce_sum_in_place( std::vector< double >& buffer );
  void communicate_Allreduce_sum_in_place( std::vector< int >& buffer );
  void communicate_Allreduce_sum( std::vector< double >& send_buffer, std::vector< double >& recv_buffer );
 
  bool equal_cross_ranks( const double value );
 
  std::string get_processor_name();
 
  bool is_mpi_used();
 
  size_t get_buffer_size_target_data() const;
 
  unsigned int get_send_recv_count_target_data_per_rank() const;
 
  size_t get_buffer_size_spike_data() const;
 
  unsigned int get_send_recv_count_spike_data_per_rank() const;
 
  size_t get_send_buffer_size_secondary_events_in_int() const;
 
  size_t get_recv_buffer_size_secondary_events_in_int() const;
 
#ifdef HAVE_MPI
 
  void communicate_Alltoall_( void* send_buffer, void* recv_buffer, const unsigned int send_recv_count );
 
  void communicate_Alltoallv_( void* send_buffer,
    const int* send_counts,
    const int* send_displacements,
    void* recv_buffer,
    const int* recv_counts,
    const int* recv_displacements );
 
#endif /* HAVE_MPI */
 
  template < class D >
  void communicate_Alltoall( std::vector< D >& send_buffer,
    std::vector< D >& recv_buffer,
    const unsigned int send_recv_count );
  template < class D >
  void communicate_target_data_Alltoall( std::vector< D >& send_buffer, std::vector< D >& recv_buffer );
  template < class D >
  void communicate_spike_data_Alltoall( std::vector< D >& send_buffer, std::vector< D >& recv_buffer );
  template < class D >
  void communicate_off_grid_spike_data_Alltoall( std::vector< D >& send_buffer, std::vector< D >& recv_buffer );
  template < class D >
  void communicate_secondary_events_Alltoallv( std::vector< D >& send_buffer, std::vector< D >& recv_buffer );
 
  void synchronize();
 
  bool any_true( const bool );
 
  double time_communicate( int num_bytes, int samples = 1000 );
  double time_communicatev( int num_bytes, int samples = 1000 );
  double time_communicate_offgrid( int num_bytes, int samples = 1000 );
  double time_communicate_alltoall( int num_bytes, int samples = 1000 );
  double time_communicate_alltoallv( int num_bytes, int samples = 1000 );
 
  void set_buffer_size_target_data( size_t buffer_size );
  void set_buffer_size_spike_data( size_t buffer_size );
 
  bool increase_buffer_size_target_data();
 
  bool adaptive_target_buffers() const;
 
  void set_recv_counts_secondary_events_in_int_per_rank( const std::vector< int >& recv_counts_in_int_per_rank );
 
  size_t get_recv_count_secondary_events_in_int( const size_t source_rank ) const;
 
  size_t get_recv_displacement_secondary_events_in_int( const size_t source_rank ) const;
 
  size_t get_send_count_secondary_events_in_int( const size_t target_rank ) const;
 
  size_t get_send_displacement_secondary_events_in_int( const size_t target_rank ) const;
 
  size_t get_done_marker_position_in_secondary_events_send_buffer( const size_t target_rank ) const;
 
  size_t get_done_marker_position_in_secondary_events_recv_buffer( const size_t source_rank ) const;
 
  void communicate_recv_counts_secondary_events();
 
private:
  int num_processes_;               
  int rank_;                        
  int send_buffer_size_;            
  int recv_buffer_size_;            
  bool use_mpi_;                    
  size_t buffer_size_target_data_;  
  // communication of connections
 
  size_t buffer_size_spike_data_;  
  // communication of spikes
 
  size_t max_buffer_size_target_data_;  
  // communication of connections
 
  bool adaptive_target_buffers_;  
  // connections resize on the fly
 
  double growth_factor_buffer_spike_data_;
  double growth_factor_buffer_target_data_;
 
  double shrink_factor_buffer_spike_data_;
 
  unsigned int send_recv_count_spike_data_per_rank_;
  unsigned int send_recv_count_target_data_per_rank_;
 
  std::vector< int > recv_counts_secondary_events_in_int_per_rank_;
 
  std::vector< int >
    send_counts_secondary_events_in_int_per_rank_;  
 
  std::vector< int > recv_displacements_secondary_events_in_int_per_rank_;
 
  std::vector< int > send_displacements_secondary_events_in_int_per_rank_;
 
#ifdef HAVE_MPI
 
  std::vector< int > comm_step_;
 
  unsigned int COMM_OVERFLOW_ERROR;  //<! array containing communication partner for each step.
 
 
  MPI_Comm comm;
  MPI_Datatype MPI_OFFGRID_SPIKE;
 
  void communicate_Allgather( std::vector< unsigned int >& send_buffer,
    std::vector< unsigned int >& recv_buffer,
    std::vector< int >& displacements );
 
  void communicate_Allgather( std::vector< OffGridSpike >& send_buffer,
    std::vector< OffGridSpike >& recv_buffer,
    std::vector< int >& displacements );
 
  void communicate_Allgather( std::vector< int >& );
  void communicate_Allgather( std::vector< long >& );
 
  template < typename T >
  void communicate_Allgatherv( std::vector< T >& send_buffer,
    std::vector< T >& recv_buffer,
    std::vector< int >& displacements,
    std::vector< int >& recv_counts );
 
  template < typename T >
  void communicate_Allgather( std::vector< T >& send_buffer,
    std::vector< T >& recv_buffer,
    std::vector< int >& displacements );
 
#endif /* #ifdef HAVE_MPI */
 
public:
  class OffGridSpike
  {
    friend void MPIManager::init_mpi( int*, char*** );
 
  public:
    typedef unsigned int node_id_external_type;
 
    OffGridSpike()
      : node_id_( 0 )
      , offset_( 0.0 )
    {
    }
    OffGridSpike( node_id_external_type node_idv, double offsetv )
      : node_id_( node_idv )
      , offset_( offsetv )
    {
    }
 
    unsigned int
    get_node_id() const
    {
      return static_cast< node_id_external_type >( node_id_ );
    }
    void
    set_node_id( node_id_external_type node_id )
    {
      node_id_ = static_cast< double >( node_id );
    }
    double
    get_offset() const
    {
      return offset_;
    }
 
  private:
    double node_id_;  
    double offset_;   
 
    static void
    assert_datatype_compatibility_()
    {
      assert( std::numeric_limits< double >::digits > std::numeric_limits< node_id_external_type >::digits );
 
      // the next one is doubling up, better be safe than sorry
      const node_id_external_type maxnode_id = std::numeric_limits< node_id_external_type >::max();
      OffGridSpike ogs( maxnode_id, 0.0 );
      assert( maxnode_id == ogs.get_node_id() );
    }
  };
};
 
inline void
MPIManager::set_recv_counts_secondary_events_in_int_per_rank( const std::vector< int >& recv_counts_in_int_per_rank )
{
  recv_counts_secondary_events_in_int_per_rank_ = recv_counts_in_int_per_rank;
 
  std::partial_sum( recv_counts_secondary_events_in_int_per_rank_.begin(),
    recv_counts_secondary_events_in_int_per_rank_.end() - 1,
    recv_displacements_secondary_events_in_int_per_rank_.begin() + 1 );
}
 
inline size_t
MPIManager::get_recv_count_secondary_events_in_int( const size_t source_rank ) const
{
  return recv_counts_secondary_events_in_int_per_rank_[ source_rank ];
}
 
inline size_t
MPIManager::get_recv_displacement_secondary_events_in_int( const size_t source_rank ) const
{
  return recv_displacements_secondary_events_in_int_per_rank_[ source_rank ];
}
 
inline size_t
MPIManager::get_send_count_secondary_events_in_int( const size_t target_rank ) const
{
  return send_counts_secondary_events_in_int_per_rank_[ target_rank ];
}
 
inline size_t
MPIManager::get_send_displacement_secondary_events_in_int( const size_t target_rank ) const
{
  return send_displacements_secondary_events_in_int_per_rank_[ target_rank ];
}
 
inline size_t
MPIManager::get_done_marker_position_in_secondary_events_send_buffer( const size_t target_rank ) const
{
  return get_send_displacement_secondary_events_in_int( target_rank )
    + get_send_count_secondary_events_in_int( target_rank ) - 1;
}
 
inline size_t
MPIManager::get_done_marker_position_in_secondary_events_recv_buffer( const size_t source_rank ) const
{
  return get_recv_displacement_secondary_events_in_int( source_rank )
    + get_recv_count_secondary_events_in_int( source_rank ) - 1;
}
 
inline size_t
MPIManager::get_num_processes() const
{
  return num_processes_;
}
 
inline size_t
MPIManager::get_rank() const
{
  return rank_;
}
 
inline bool
MPIManager::is_mpi_used()
{
  return use_mpi_;
}
 
inline size_t
MPIManager::get_buffer_size_target_data() const
{
  return buffer_size_target_data_;
}
 
inline unsigned int
MPIManager::get_send_recv_count_target_data_per_rank() const
{
  return send_recv_count_target_data_per_rank_;
}
 
inline size_t
MPIManager::get_buffer_size_spike_data() const
{
  return buffer_size_spike_data_;
}
 
inline unsigned int
MPIManager::get_send_recv_count_spike_data_per_rank() const
{
  return send_recv_count_spike_data_per_rank_;
}
 
inline size_t
MPIManager::get_send_buffer_size_secondary_events_in_int() const
{
  return send_displacements_secondary_events_in_int_per_rank_
           [ send_displacements_secondary_events_in_int_per_rank_.size() - 1 ]
    + send_counts_secondary_events_in_int_per_rank_[ send_counts_secondary_events_in_int_per_rank_.size() - 1 ];
}
 
inline size_t
MPIManager::get_recv_buffer_size_secondary_events_in_int() const
{
  return recv_displacements_secondary_events_in_int_per_rank_
           [ recv_displacements_secondary_events_in_int_per_rank_.size() - 1 ]
    + recv_counts_secondary_events_in_int_per_rank_[ recv_counts_secondary_events_in_int_per_rank_.size() - 1 ];
}
 
inline void
MPIManager::set_buffer_size_target_data( const size_t buffer_size )
{
  assert( buffer_size >= static_cast< size_t >( 2 * get_num_processes() ) );
  if ( buffer_size <= max_buffer_size_target_data_ )
  {
    buffer_size_target_data_ = buffer_size;
  }
  else
  {
    buffer_size_target_data_ = max_buffer_size_target_data_;
  }
  send_recv_count_target_data_per_rank_ = static_cast< size_t >(
    floor( static_cast< double >( get_buffer_size_target_data() ) / static_cast< double >( get_num_processes() ) ) );
 
  assert( send_recv_count_target_data_per_rank_ * get_num_processes() <= get_buffer_size_target_data() );
}
 
inline void
MPIManager::set_buffer_size_spike_data( const size_t buffer_size )
{
  assert( buffer_size >= static_cast< size_t >( 2 * get_num_processes() ) );
  buffer_size_spike_data_ = buffer_size;
 
  send_recv_count_spike_data_per_rank_ = floor( get_buffer_size_spike_data() / get_num_processes() );
 
  assert( send_recv_count_spike_data_per_rank_ * get_num_processes() <= get_buffer_size_spike_data() );
}
 
inline bool
MPIManager::increase_buffer_size_target_data()
{
  assert( adaptive_target_buffers_ );
  if ( buffer_size_target_data_ >= max_buffer_size_target_data_ )
  {
    return false;
  }
  else
  {
    if ( buffer_size_target_data_ * growth_factor_buffer_target_data_ < max_buffer_size_target_data_ )
    {
      // this also adjusts send_recv_count_target_data_per_rank_
      set_buffer_size_target_data(
        static_cast< size_t >( floor( buffer_size_target_data_ * growth_factor_buffer_target_data_ ) ) );
    }
    else
    {
      // this also adjusts send_recv_count_target_data_per_rank_
      set_buffer_size_target_data( max_buffer_size_target_data_ );
    }
    return true;
  }
}
 
inline bool
MPIManager::adaptive_target_buffers() const
{
  return adaptive_target_buffers_;
}
 
#ifndef HAVE_MPI
inline std::string
MPIManager::get_processor_name()
{
  char name[ 1024 ];
  name[ 1023 ] = '\0';
  gethostname( name, 1023 );
  return name;
}
 
inline void
MPIManager::mpi_abort( int )
{
}
 
inline void
MPIManager::communicate( std::vector< int >& )
{
}
 
inline void
MPIManager::communicate( std::vector< long >& )
{
}
 
inline void
MPIManager::synchronize()
{
}
 
inline void
test_link( int, int )
{
}
 
inline void
test_links()
{
}
 
inline bool
MPIManager::any_true( const bool my_bool )
{
  return my_bool;
}
 
inline double
MPIManager::time_communicate( int, int )
{
  return 0.0;
}
 
inline double
MPIManager::time_communicatev( int, int )
{
  return 0.0;
}
 
inline double
MPIManager::time_communicate_offgrid( int, int )
{
  return 0.0;
}
 
inline double
MPIManager::time_communicate_alltoall( int, int )
{
  return 0.0;
}
 
inline double
MPIManager::time_communicate_alltoallv( int, int )
{
  return 0.0;
}
 
#endif /* HAVE_MPI */
 
#ifdef HAVE_MPI
template < class D >
void
MPIManager::communicate_Alltoall( std::vector< D >& send_buffer,
  std::vector< D >& recv_buffer,
  const unsigned int send_recv_count )
{
  void* send_buffer_int = static_cast< void* >( &send_buffer[ 0 ] );
  void* recv_buffer_int = static_cast< void* >( &recv_buffer[ 0 ] );
 
  communicate_Alltoall_( send_buffer_int, recv_buffer_int, send_recv_count );
}
 
template < class D >
void
MPIManager::communicate_secondary_events_Alltoallv( std::vector< D >& send_buffer, std::vector< D >& recv_buffer )
{
  void* send_buffer_int = static_cast< void* >( &send_buffer[ 0 ] );
  void* recv_buffer_int = static_cast< void* >( &recv_buffer[ 0 ] );
 
  communicate_Alltoallv_( send_buffer_int,
    &send_counts_secondary_events_in_int_per_rank_[ 0 ],
    &send_displacements_secondary_events_in_int_per_rank_[ 0 ],
    recv_buffer_int,
    &recv_counts_secondary_events_in_int_per_rank_[ 0 ],
    &recv_displacements_secondary_events_in_int_per_rank_[ 0 ] );
}
 
#else  // HAVE_MPI
template < class D >
void
MPIManager::MPIManager::communicate_Alltoall( std::vector< D >& send_buffer,
  std::vector< D >& recv_buffer,
  const unsigned int )
{
  recv_buffer.swap( send_buffer );
}
 
template < class D >
void
MPIManager::communicate_secondary_events_Alltoallv( std::vector< D >& send_buffer, std::vector< D >& recv_buffer )
{
  recv_buffer.swap( send_buffer );
}
 
#endif /* HAVE_MPI */
 
template < class D >
void
MPIManager::communicate_target_data_Alltoall( std::vector< D >& send_buffer, std::vector< D >& recv_buffer )
{
  const size_t send_recv_count_target_data_in_int_per_rank =
    sizeof( TargetData ) / sizeof( unsigned int ) * send_recv_count_target_data_per_rank_;
 
  communicate_Alltoall( send_buffer, recv_buffer, send_recv_count_target_data_in_int_per_rank );
}
 
template < class D >
void
MPIManager::communicate_spike_data_Alltoall( std::vector< D >& send_buffer, std::vector< D >& recv_buffer )
{
  const size_t send_recv_count_spike_data_in_int_per_rank =
    sizeof( SpikeData ) / sizeof( unsigned int ) * send_recv_count_spike_data_per_rank_;
 
  communicate_Alltoall( send_buffer, recv_buffer, send_recv_count_spike_data_in_int_per_rank );
}
 
template < class D >
void
MPIManager::communicate_off_grid_spike_data_Alltoall( std::vector< D >& send_buffer, std::vector< D >& recv_buffer )
{
  const size_t send_recv_count_off_grid_spike_data_in_int_per_rank =
    sizeof( OffGridSpikeData ) / sizeof( unsigned int ) * send_recv_count_spike_data_per_rank_;
 
  communicate_Alltoall( send_buffer, recv_buffer, send_recv_count_off_grid_spike_data_in_int_per_rank );
}
}
 
#endif /* MPI_MANAGER_H */