communicator_mpi.cpp

Go to the documentation of this file.

 
#include <unistd.h> // for usleep
#include "layout.h"
 
// exception handler for uncaught throw.
static std::terminate_handler default_handler = std::set_terminate(Communicator::abort);
 
// define static members
#ifdef ENABLE_MULTI_INSTANCE
int Communicator_impl::m_Ninstance   = 1; // number of instances
int Communicator_impl::m_instance_id = 0; // id of present instance
#endif
 
int Communicator_impl::m_global_rank = 0;
int Communicator_impl::m_global_size = 1;
 
int Communicator_impl::m_grid_rank = 0;
int Communicator_impl::m_grid_size = 1;
 
MPI_Comm Communicator_impl::m_comm;
 
char Communicator_impl::default_grid_map[] = "xyzt";
 
//====================================================================
// class methods
int Communicator_impl::init(int *pargc, char ***pargv)
{
  LOG;
 
  int is_initialized = 0;
  MPI_Initialized(&is_initialized);
 
  if (is_initialized) {
    fprintf(stderr, "Communicator: MPI already initialized. skip.\n");
  } else {
#ifdef USE_OPENMP
    int required = MPI_THREAD_FUNNELED;
    int provided;
 
    MPI_Init_thread(pargc, pargv, required, &provided);
 
    if (provided < required) {
      fprintf(stderr, "Communicator: MPI not supporting sufficient thread level. exiting.\n");
      exit(EXIT_FAILURE);
    }
#else
    MPI_Init(pargc, pargv);
#endif
  }
 
  MPI_Comm_size(MPI_COMM_WORLD, &m_global_size);
  MPI_Comm_rank(MPI_COMM_WORLD, &m_global_rank);
 
  // grid size and rank equal to global ones for a moment until layout is set.
  m_grid_size = m_global_size;
  m_grid_rank = m_global_rank;
 
  //- initialize m_comm, thanks to Aoyama-san.
  MPI_Comm_dup(MPI_COMM_WORLD, &m_comm);
 
  return EXIT_SUCCESS;
}
 
 
//====================================================================
int Communicator_impl::finalize()
{
  LOG;
  return MPI_Finalize();
}
 
 
//====================================================================
int Communicator_impl::setup(int Ninstance)
{
  LOG;
 
#ifdef ENABLE_MULTI_INSTANCE
  if ((Ninstance == 0) || (m_global_size % Ninstance != 0)) {
    printf("%s: invalid number of instance: %d\n", "Communicator::init", Ninstance);
    abort();
  }
 
  m_Ninstance = Ninstance;
 
  int gsize = m_global_size / Ninstance;
  m_instance_id = m_global_rank / gsize;
 
  MPI_Comm_split(MPI_COMM_WORLD, m_instance_id, 0 /* key */, &m_comm);
#else
//  m_Ninstance = 1;
//  m_instance_id = 0;
 
  MPI_Comm_dup(MPI_COMM_WORLD, &m_comm);
#endif
 
  MPI_Comm_size(m_comm, &m_grid_size);
  MPI_Comm_rank(m_comm, &m_grid_rank);
 
  Communicator_impl::Layout::layout_setup();
 
  status();
 
  return EXIT_SUCCESS;
}
 
 
//====================================================================
int Communicator_impl::setup(
  const std::vector<int>& lattice_size,
  std::vector<int>& grid_size,
  int Ninstance)
{
  LOG;
 
#ifdef ENABLE_MULTI_INSTANCE
  if ((Ninstance == 0) || (m_global_size % Ninstance != 0)) {
    printf("%s: invalid number of instance: %d\n", "Communicator::init", Ninstance);
    abort();
  }
 
  m_Ninstance = Ninstance;
 
  int gsize = m_global_size / Ninstance;
  m_instance_id = m_global_rank / gsize;
 
  MPI_Comm_split(MPI_COMM_WORLD, m_instance_id, 0 /* key */, &m_comm);
#else
//  m_Ninstance = 1;
//  m_instance_id = 0;
 
  MPI_Comm_dup(MPI_COMM_WORLD, &m_comm);
#endif
 
  MPI_Comm_size(m_comm, &m_grid_size);
  MPI_Comm_rank(m_comm, &m_grid_rank);
 
  Communicator_impl::Layout::layout_setup(lattice_size, grid_size);
 
  status();
 
  return EXIT_SUCCESS;
}
 
 
//====================================================================
void Communicator_impl::abort()
{
  LOG;
  MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
  // unreached.
}
 
 
//====================================================================
// information
bool Communicator_impl::is_primary()
{
  return m_grid_rank == 0;
}
 
 
//====================================================================
int Communicator_impl::self()
{
  return m_grid_rank;
}
 
 
//====================================================================
int Communicator_impl::size()
{
  return m_grid_size;
}
 
 
//====================================================================
bool Communicator_impl::is_primary_master()
{
  return m_global_rank == 0;
}
 
 
//====================================================================
#ifdef ENABLE_MULTI_INSTANCE
 
int Communicator_impl::self_global()
{
  return m_global_rank;
}
 
 
//====================================================================
int Communicator_impl::world_id()
{
  return m_instance_id;
}
 
 
#endif
 
//====================================================================
// synchronize
int Communicator_impl::sync()
{
  LOG;
  return MPI_Barrier(m_comm);
}
 
 
//====================================================================
// sync w/o (possible) busy wait
int Communicator_impl::sync_usleep()
{
  LOG;
  char        dummy    = '\0';
  const int   interval = 100000; // 100ms
  MPI_Request req;
  MPI_Ibcast(&dummy, 1, MPI_BYTE, 0, m_comm, &req);
 
  MPI_Status status;
  int        finished = 0;
 
  while (!finished)
  {
    usleep(interval);
    MPI_Test(&req, &finished, &status);
  }
  return finished;
}
 
 
//====================================================================
#ifdef ENABLE_MULTI_INSTANCE
 
int Communicator_impl::sync_global()
{
  LOG;
  return MPI_Barrier(MPI_COMM_WORLD);
}
 
 
#endif
 
//====================================================================
// data transfer: base cases
int Communicator_impl::Base::broadcast(size_t size, void *data, int sender)
{
  LOG;
  return MPI_Bcast(data, size, MPI_BYTE, sender, m_comm);
}
 
 
//====================================================================
int Communicator_impl::Base::exchange(size_t size, void *recv_buf, void *send_buf, int idir, int ipm, int itag)
{
  LOG;
 
  MPI_Status status;
  int        p_send, p_recv;
 
  assert(ipm == 1 || ipm == -1);
 
  if (Layout::m_grid_dims[idir] == 1) {  // no need to transfer
    memcpy(recv_buf, send_buf, size);
    return EXIT_SUCCESS;
  }
 
  if (ipm == 1) {  // downward shift
    p_send = Layout::m_ipe_dn[idir];
    p_recv = Layout::m_ipe_up[idir];
  } else {  // upward shift
    p_send = Layout::m_ipe_up[idir];
    p_recv = Layout::m_ipe_dn[idir];
  }
 
  int tag_send = Layout::tag(self(), idir, -ipm);
  int tag_recv = Layout::tag(p_recv, idir, -ipm);
 
  return MPI_Sendrecv(
    send_buf, size, MPI_BYTE, p_send, tag_send,
    recv_buf, size, MPI_BYTE, p_recv, tag_recv,
    m_comm, &status);
}
 
 
//====================================================================
int Communicator_impl::Base::send_1to1(size_t size, void *recv_buf, void *send_buf, int send_to, int recv_from, int tag)
{
  LOG;
 
  MPI_Status status;
 
  if (send_to == recv_from) {
    memcpy(recv_buf, send_buf, size);
  } else {
    if (self() == recv_from)
      MPI_Send(send_buf, size, MPI_BYTE, send_to, tag, m_comm);
 
    if (self() == send_to)
      MPI_Recv(recv_buf, size, MPI_BYTE, recv_from, tag, m_comm, &status);
  }
 
  // sync should be taken outside.
 
  return EXIT_SUCCESS;
}
 
 
//====================================================================
int Communicator_impl::Base::reduce(int count, void *recv_buf, void *send_buf, MPI_Datatype type, MPI_Op op, int pattern)
{
  LOG;
  return MPI_Allreduce((void *)send_buf, (void *)recv_buf, count, type, op, Layout::m_sub_comm[pattern]);
}
 
 
//====================================================================
// data transfer for specific datatypes
int Communicator_impl::broadcast_string(int count, string& data, int sender)
{
  LOG;
  assert(count == 1);
 
  size_t size = 0;
 
  // broadcast string length.
  if (Communicator::self() == sender) {
    size = data.size();
  }
  MPI_Bcast((void *)&size, sizeof(size_t), MPI_BYTE, sender, m_comm);
 
  // allocate buffer. pack data at sender.
  char *buf = new char[size + 1];
  memset(buf, '\0', size + 1);
 
  if (Communicator::self() == sender) {
    data.copy(buf, size, 0);
  }
 
  // do broadcast.
  int retv = MPI_Bcast((void *)buf, size, MPI_BYTE, sender, m_comm);
 
  if (Communicator::self() != sender) {
    data = string(buf);
  }
 
  delete [] buf;
 
  return retv;
}
 
 
//====================================================================
// info
double Communicator_impl::get_time()
{
  return MPI_Wtime();
}
 
 
//====================================================================
// debug
int Communicator_impl::status()
{
#ifdef DEBUG
#ifdef ENABLE_MULTI_INSTANCE
  printf("global_rank=%2d/%2d: ngrid=%d, grid_id=%d: grid_rank=%2d/%2d\n",
         m_global_rank, m_global_size,
         m_Ninstance, m_instance_id,
         m_grid_rank, m_grid_size);
#else
  printf("grid_rank=%2d/%2d\n",
         m_grid_rank, m_grid_size);
#endif
#endif
 
 
  return EXIT_SUCCESS;
}
 
 
//====================================================================
//============================================================END=====