pcu.c

Go to the documentation of this file.

/****************************************************************************** 

  Copyright 2011 Scientific Computation Research Center, 
      Rensselaer Polytechnic Institute. All rights reserved.
  
  This work is open source software, licensed under the terms of the
  BSD license as described in the LICENSE file in the top-level directory.

*******************************************************************************/
#include <string.h>
#include <stdarg.h>
#include "PCU.h"
#include "pcu_msg.h"
#include "pcu_pmpi.h"
#include "pcu_order.h"
#include "noto_malloc.h"
#include "reel.h"
#include <sys/types.h> /*required for mode_t for mkdir on some systems*/
#include <sys/stat.h> /*using POSIX mkdir call for SMB "foo/" path*/
#include <errno.h> /* for checking the error from mkdir */

enum state { uninit, init };
static enum state global_state = uninit;
static pcu_msg global_pmsg;

static pcu_msg* get_msg()
{
  return &global_pmsg;
}

int PCU_Comm_Init(void)
{
  if (global_state != uninit)
    reel_fail("nested calls to Comm_Init");
  pcu_pmpi_init(MPI_COMM_WORLD);
  pcu_set_mpi(&pcu_pmpi);
  pcu_make_msg(&global_pmsg);
  global_state = init;
  /* turn ordering on by default, call
     PCU_Comm_Order(false) after PCU_Comm_Init
     to disable this */
  PCU_Comm_Order(true);
  return PCU_SUCCESS;
}

int PCU_Comm_Free(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Free called before Comm_Init");
  if (global_pmsg.order)
    pcu_order_free(global_pmsg.order);
  pcu_free_msg(&global_pmsg);
  pcu_pmpi_finalize();
  global_state = uninit;
  return PCU_SUCCESS;
}

int PCU_Comm_Self(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Self called before Comm_Init");
  return pcu_mpi_rank();
}

int PCU_Comm_Peers(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Peers called before Comm_Init");
  return pcu_mpi_size();
}

void PCU_Comm_Begin(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Begin called before Comm_Init");
  pcu_msg_start(get_msg());
}

int PCU_Comm_Pack(int to_rank, const void* data, size_t size)
{
  if (global_state == uninit)
    reel_fail("Comm_Pack called before Comm_Init");
  if ((to_rank < 0)||(to_rank >= pcu_mpi_size()))
    reel_fail("Invalid rank in Comm_Pack");
  memcpy(pcu_msg_pack(get_msg(),to_rank,size),data,size);
  return PCU_SUCCESS;
}

int PCU_Comm_Send(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Send called before Comm_Init");
  pcu_msg_send(get_msg());
  return PCU_SUCCESS;
}

bool PCU_Comm_Listen(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Listen called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    return pcu_order_receive(m->order, m);
  return pcu_msg_receive(m);
}

int PCU_Comm_Sender(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Sender called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    return pcu_order_received_from(m->order);
  return pcu_msg_received_from(m);
}

bool PCU_Comm_Unpacked(void)
{
  if (global_state == uninit)
    reel_fail("Comm_Unpacked called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    return pcu_order_unpacked(m->order);
  return pcu_msg_unpacked(m);
}

int PCU_Comm_Unpack(void* data, size_t size)
{
  if (global_state == uninit)
    reel_fail("Comm_Unpack called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    memcpy(data,pcu_order_unpack(m->order,size),size);
  else
    memcpy(data,pcu_msg_unpack(m,size),size);
  return PCU_SUCCESS;
}

void PCU_Comm_Order(bool on)
{
  if (global_state == uninit)
    reel_fail("Comm_Order called before Comm_Init");
  pcu_msg* m = get_msg();
  if (on && (!m->order))
    m->order = pcu_order_new();
  if ((!on) && m->order) {
    pcu_order_free(m->order);
    m->order = NULL;
  }
}

void PCU_Barrier(void)
{
  if (global_state == uninit)
    reel_fail("Barrier called before Comm_Init");
  pcu_barrier(&(get_msg()->coll));
}

void PCU_Add_Doubles(double* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Add_Doubles called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_add_doubles,p,n*sizeof(double));
}

double PCU_Add_Double(double x)
{
  double a[1];
  a[0] = x;
  PCU_Add_Doubles(a, 1);
  return a[0];
}

void PCU_Min_Doubles(double* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Min_Doubles called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_min_doubles,p,n*sizeof(double));
}

double PCU_Min_Double(double x)
{
  double a[1];
  a[0] = x;
  PCU_Min_Doubles(a, 1);
  return a[0];
}

void PCU_Max_Doubles(double* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Max_Doubles called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_max_doubles,p,n*sizeof(double));
}

double PCU_Max_Double(double x)
{
  double a[1];
  a[0] = x;
  PCU_Max_Doubles(a, 1);
  return a[0];
}

void PCU_Add_Ints(int* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Add_Ints called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_add_ints,p,n*sizeof(int));
}

int PCU_Add_Int(int x)
{
  int a[1];
  a[0] = x;
  PCU_Add_Ints(a, 1);
  return a[0];
}

void PCU_Add_Longs(long* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Add_Longs called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_add_longs,p,n*sizeof(long));
}

long PCU_Add_Long(long x)
{
  long a[1];
  a[0] = x;
  PCU_Add_Longs(a, 1);
  return a[0];
}

void PCU_Add_SizeTs(size_t* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Add_SizeTs called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_add_sizets,p,n*sizeof(size_t));
}

size_t PCU_Add_SizeT(size_t x)
{
  size_t a[1];
  a[0] = x;
  PCU_Add_SizeTs(a, 1);
  return a[0];
}

void PCU_Min_SizeTs(size_t* p, size_t n) {
  if (global_state == uninit)
    reel_fail("Min_SizeTs called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_min_sizets,p,n*sizeof(size_t));
}

size_t PCU_Min_SizeT(size_t x) {
  size_t a[1];
  a[0] = x;
  PCU_Min_SizeTs(a, 1);
  return a[0];
}

void PCU_Max_SizeTs(size_t* p, size_t n) {
  if (global_state == uninit)
    reel_fail("Max_SizeTs called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_max_sizets,p,n*sizeof(size_t));
}

size_t PCU_Max_SizeT(size_t x) {
  size_t a[1];
  a[0] = x;
  PCU_Max_SizeTs(a, 1);
  return a[0];
}

void PCU_Exscan_Ints(int* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Exscan_Ints called before Comm_Init");
  int* originals;
  NOTO_MALLOC(originals,n);
  for (size_t i=0; i < n; ++i)
    originals[i] = p[i];
  pcu_scan(&(get_msg()->coll),pcu_add_ints,p,n*sizeof(int));
  //convert inclusive scan to exclusive
  for (size_t i=0; i < n; ++i)
    p[i] -= originals[i];
  noto_free(originals);
}

int PCU_Exscan_Int(int x)
{
  int a[1];
  a[0] = x;
  PCU_Exscan_Ints(a, 1);
  return a[0];
}

void PCU_Exscan_Longs(long* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Exscan_Longs called before Comm_Init");
  long* originals;
  NOTO_MALLOC(originals,n);
  for (size_t i=0; i < n; ++i)
    originals[i] = p[i];
  pcu_scan(&(get_msg()->coll),pcu_add_longs,p,n*sizeof(long));
  //convert inclusive scan to exclusive
  for (size_t i=0; i < n; ++i)
    p[i] -= originals[i];
  noto_free(originals);
}

long PCU_Exscan_Long(long x)
{
  long a[1];
  a[0] = x;
  PCU_Exscan_Longs(a, 1);
  return a[0];
}

void PCU_Min_Ints(int* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Min_Ints called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_min_ints,p,n*sizeof(int));
}

int PCU_Min_Int(int x)
{
  int a[1];
  a[0] = x;
  PCU_Min_Ints(a, 1);
  return a[0];
}

void PCU_Max_Ints(int* p, size_t n)
{
  if (global_state == uninit)
    reel_fail("Max_Ints called before Comm_Init");
  pcu_allreduce(&(get_msg()->coll),pcu_max_ints,p,n*sizeof(int));
}

int PCU_Max_Int(int x)
{
  int a[1];
  a[0] = x;
  PCU_Max_Ints(a, 1);
  return a[0];
}

int PCU_Or(int c)
{
  return PCU_Max_Int(c);
}

int PCU_And(int c)
{
  return PCU_Min_Int(c);
}

int PCU_Proc_Self(void)
{
  if (global_state == uninit)
    reel_fail("Proc_Self called before Comm_Init");
  return pcu_pmpi_rank();
}

int PCU_Proc_Peers(void)
{
  if (global_state == uninit)
    reel_fail("Proc_Peers called before Comm_Init");
  return pcu_pmpi_size();
}

int PCU_Comm_Rank(int* rank)
{
  if (global_state == uninit)
    reel_fail("Comm_Rank called before Comm_Init");
  *rank = pcu_mpi_rank();
  return PCU_SUCCESS;
}

int PCU_Comm_Size(int* size)
{
  if (global_state == uninit)
    reel_fail("Comm_Size called before Comm_Init");
  *size = pcu_mpi_size();
  return PCU_SUCCESS;
}

bool PCU_Comm_Initialized(void)
{
  return global_state == init;
}

int PCU_Comm_Start(PCU_Method method)
{
  (void)method; //warning silencer
  if (global_state == uninit)
    reel_fail("Comm_Start called before Comm_Init");
  pcu_msg_start(get_msg());
  return PCU_SUCCESS;
}

int PCU_Comm_Packed(int to_rank, size_t* size)
{
  if (global_state == uninit)
    reel_fail("Comm_Packed called before Comm_Init");
  if ((to_rank < 0)||(to_rank >= pcu_mpi_size()))
    reel_fail("Invalid rank in Comm_Packed");
  *size = pcu_msg_packed(get_msg(),to_rank);
  return PCU_SUCCESS;
}

int PCU_Comm_Write(int to_rank, const void* data, size_t size)
{
  if (global_state == uninit)
    reel_fail("Comm_Write called before Comm_Init");
  if ((to_rank < 0)||(to_rank >= pcu_mpi_size()))
    reel_fail("Invalid rank in Comm_Write");
  pcu_msg* msg = get_msg();
  PCU_MSG_PACK(msg,to_rank,size);
  memcpy(pcu_msg_pack(msg,to_rank,size),data,size);
  return PCU_SUCCESS;
}

bool PCU_Comm_Receive(void)
{
  while (PCU_Comm_Unpacked())
    if (!PCU_Comm_Listen())
      return false;
  return true;
}

bool PCU_Comm_Read(int* from_rank, void** data, size_t* size)
{
  if (global_state == uninit)
    reel_fail("Comm_Read called before Comm_Init");
  if (!PCU_Comm_Receive())
    return false;
  *from_rank = PCU_Comm_Sender();
  PCU_COMM_UNPACK(*size);
  *data = PCU_Comm_Extract(*size);
  return true;
}

static void safe_mkdir(const char* path, mode_t mode)
{
  int err;
  errno = 0;
  err = mkdir(path, mode);
  if (err != 0 && errno != EEXIST)
    reel_fail("PCU: could not create directory \"%s\"\n", path);
}

static void append(char* s, size_t size, const char* format, ...)
{
  int len = strlen(s);
  va_list ap;
  va_start(ap, format);
  vsnprintf(s + len, size - len, format, ap);
  va_end(ap);
}


void PCU_Debug_Open(void)
{
  if (global_state == uninit)
    reel_fail("Debug_Open called before Comm_Init");

  const int fanout = 2048;
  const int bufsize = 1024;
  char* path = noto_malloc(bufsize);
  path[0] = '\0';
  if (PCU_Comm_Peers() > fanout) {
    mode_t const dir_perm = S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH;
    strcpy(path, "debug/");
    safe_mkdir(path, dir_perm);
    int self = PCU_Comm_Self();
    append(path, bufsize, "%d/", self / fanout);
    if (self % fanout == 0)
      safe_mkdir(path, dir_perm);
    PCU_Barrier();
  }

  append(path,bufsize, "%s", "debug");
  pcu_msg* msg = get_msg();
  if ( ! msg->file)
    msg->file = pcu_open_parallel(path,"txt");
  noto_free(path);
}

void PCU_Debug_Print(const char* format, ...)
{
  if (global_state == uninit)
    reel_fail("Debug_Print called before Comm_Init");
  pcu_msg* msg = get_msg();
  if ( ! msg->file)
    return; //Print is a no-op if no file is open
  va_list ap;
  va_start(ap,format);
  vfprintf(msg->file,format,ap);
  va_end(ap);
  fflush(msg->file);
}

int PCU_Comm_From(int* from_rank)
{
  if (global_state == uninit)
    reel_fail("Comm_From called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    *from_rank = pcu_order_received_from(m->order);
  else
    *from_rank = pcu_msg_received_from(m);
  return PCU_SUCCESS;
}

int PCU_Comm_Received(size_t* size)
{
  if (global_state == uninit)
    reel_fail("Comm_Received called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    *size = pcu_order_received_size(m->order);
  else
    *size = pcu_msg_received_size(m);
  return PCU_SUCCESS;
}

void* PCU_Comm_Extract(size_t size)
{
  if (global_state == uninit)
    reel_fail("Comm_Extract called before Comm_Init");
  pcu_msg* m = get_msg();
  if (m->order)
    return pcu_order_unpack(m->order,size);
  return pcu_msg_unpack(m,size);
}

void PCU_Switch_Comm(MPI_Comm new_comm)
{
  if (global_state == uninit)
    reel_fail("Switch_Comm called before Comm_Init");
  pcu_pmpi_switch(new_comm);
}

MPI_Comm PCU_Get_Comm(void)
{
  if (global_state == uninit)
    reel_fail("Get_Comm called before Comm_Init");
  return pcu_pmpi_comm();
}

double PCU_Time(void)
{
  return MPI_Wtime();
}

void PCU_Protect(void)
{
  reel_protect();
}