---

ipc.c

Go to the documentation of this file.

#include <stdio.h>
#include <stdarg.h>
#include <time.h>
#include <string.h>

#ifdef _CRAYMPP
#define IPC_USE_SHMEM
#include <intrinsics.h>
#include <mpp/shmem.h>


/* Wrapper to avoid problems with multiple inclusion due to Cray bug
   reported to them 7/27/98 */
#ifdef _CRAYT3E
#undef _CRAYT3E
#include <mpp/sync_proto.h>
#define _CRAYT3E
#else
#include <mpp/sync_proto.h>
#endif
#endif

/* Needed for memmove command */
#ifndef IPC_USE_SHMEM
#include <string.h>
#endif

#include "ipc.h"


/******************************************************************************/
/* Defines and typedefs                                                       */
/******************************************************************************/

#define IPC_PE_MSG_DELAY 0.1

#define IPC_MAXFILENAMELENGTH 1024



/******************************************************************************/
/* Global parameters                                                          */
/******************************************************************************/

int         ipc_msg_level = IPC_STD; 
const char* ipc_msg_level_docstring = 
"Level of notification messages to display.  Standard levels:\n\n"
          
"  IPC_None     \n"
"  IPC_Error    \n"
"  IPC_Warning  \n"
"  IPC_Caution  \n"
"  IPC_Alert    \n"
"  IPC_Summary  \n"
"  IPC_Std      \n"
"  IPC_Verbose  \n"
"  IPC_Overwhelm\n\n"

"The last one is special in that it overrides the values of\n"
"ipc_msg_level_forceall and ipc_msg_synch_level, forcing\n"
"all PEs to present all messages and to synchronize while doing\n"
"so.";


int         ipc_msg_forceall_level = IPC_NONE;
const char* ipc_msg_forceall_level_docstring =
"Messages with a debug level higher than this level (see ipc_msg_level\n"
"for definitions) will be printed by all PEs who make the call, regardless\n"
"of whether the call declares that only one should print. Useful for\n"
"debugging on multiple PEs when some PEs are reporting errors or warnings\n"
"in the totals yet no messages were displayed.";


int         ipc_msg_synch_level = IPC_ERROR;
const char* ipc_msg_synch_level_docstring =
"Attempt to print messages with a msg level higher than this level (see\n"
"ipc_msg_level for definitions) in order of PE number, rather than\n"
"interleaving the output from different PEs.  The synchronization is\n"
"currently accomplished only by having each PE delay for a short\n"
"time proportional to its PE number.  As a consequence, it slows\n"
"down execution while not guaranteeing perfect order of output.\n"
"However, it has the advantage of not failing catastrophically in\n"
"error conditions where PEs become out of synch due to missed\n"
"barriers, which is important for an error message handler.";


int         ipc_exit_on_error_num = 0;
const char* ipc_exit_on_error_num_docstring =
"If nonzero, when this many errors have been reached, the program\n"  
"will exit automatically.";
  

int         ipc_max_warnings = 100;
const char* ipc_max_warnings_docstring =
"Maximum number of warnings to be printed, per PE.  If this limit is\n"
"reached, further warnings will still increment the warning counter\n"
"but no messages will be printed.";


int         ipc_max_errors = 100;
const char* ipc_max_errors_docstring =
"Maximum number of error messages to be printed, per PE.  If this limit\n"
"is reached, further errors will still increment the error counter\n"
"but no messages will be printed.";



/******************************************************************************/
/* Global variables                                                           */
/******************************************************************************/

FILE *ipc_logfile     = NULL;   
int   ipc_warnings    = 0;      
int   ipc_errors      = 0;      



/******************************************************************************/
/* Prototypes for private functions                                           */
/******************************************************************************/

ipc_status ipc_put_base(void *target_data, void *source_data,
                        ipc_datatype datatype, size_t count, int process);

ipc_status ipc_get_base(void *target_data, void *source_data,
                        ipc_datatype datatype, size_t count, int process);

void       ipc_notify_base(int terminateline, int print_pe, int message_level, const char *format, va_list args);


/******************************************************************************/
/* Public and private routines                                                */
/******************************************************************************/


int ipc_datatype_size(ipc_datatype datatype)
{
  switch(datatype){
  case IPC_RAW8:                return sizeof(i8              ); 
  case IPC_RAW32:               return sizeof(i32             ); 
#ifndef NO_I64
  case IPC_RAW64:               return sizeof(i64             ); 
#endif
  case IPC_CHAR:                return sizeof(signed char     ); 
  case IPC_SHORT:               return sizeof(signed short    ); 
  case IPC_INT:                 return sizeof(signed int      ); 
  case IPC_LONG:                return sizeof(signed long     ); 
#ifndef LONG_LONG_UNAVAILABLE
  case IPC_LONG_LONG:           return sizeof(signed long long);
#endif    
  case IPC_UNSIGNED_CHAR:       return sizeof(unsigned char   ); 
  case IPC_UNSIGNED_SHORT:      return sizeof(unsigned short  ); 
  case IPC_UNSIGNED:            return sizeof(unsigned int    ); 
  case IPC_UNSIGNED_LONG:       return sizeof(unsigned long   ); 
  case IPC_FLOAT:               return sizeof(float           ); 
  case IPC_DOUBLE:              return sizeof(double          ); 
#ifndef LONG_DOUBLE_UNAVAILABLE
  case IPC_LONG_DOUBLE:         return sizeof(long double     );
#endif
  default:
    ipc_notify(IPC_ALL,IPC_ERROR,"ipc_datatype %d unknown",datatype);
    return(sizeof(int));
  }
}



int ipc_num_processes(void) /* inline this */
{
#ifdef IPC_USE_SHMEM
  return _num_pes();
#else
  return 1; /* Assume single-process execution for now */
#endif
}


  
int ipc_my_process(void) /* inline this */
{
#ifdef IPC_USE_SHMEM
  return _my_pe();
#else
  return 0; /* Assume single-process execution for now */
#endif
}



void ipc_barrier(void)
{ 
#ifdef IPC_USE_SHMEM
  /* barrier(); */
  shmem_barrier_all();
#else
  if (ipc_num_processes() != 1) {
    ipc_notify(IPC_ONE,IPC_ERROR,"ipc_barrier with multiple processes (%d) unimplemented",ipc_num_processes());
    exit(-1);  
  }
#endif
}             



void ipc_set_barrier(void)
{ 
#ifdef IPC_USE_SHMEM

#ifdef __cplusplus
  ipc_notify(IPC_ONE,IPC_ERROR,"ipc_set_barrier not implemented");
#else
  set_barrier();
#endif

#else

  if (ipc_num_processes() != 1) {
    ipc_notify(IPC_ONE,IPC_ERROR,"ipc_set_barrier with multiple processes (%d) unimplemented",ipc_num_processes());
    exit(-1);  
  }

#endif
}             



/*****************************************************************************/
/* Implementation of PUT, GET, PUT_TO, and GET_TO */

#define IPC_PUT_RAW(bits)                                                 \
ipc_status ipc_put ## bits(void *data, size_t count, int process) \
{  return ipc_put_base(data,data,IPC_RAW ## bits,count,process); }
IPC_PUT_RAW(8)
IPC_PUT_RAW(32)
IPC_PUT_RAW(64)
#undef IPC_PUT_RAW

#define IPC_GET_RAW(bits)                                                 \
ipc_status ipc_get ## bits(void *data, size_t count, int process) \
{  return ipc_get_base(data,data,IPC_RAW ## bits,count,process); }
IPC_GET_RAW(8)
IPC_GET_RAW(32)
IPC_GET_RAW(64)
#undef IPC_GET_RAW


#ifndef __cplusplus
/* \name    C user interface
    No inherent type-checking         */
ipc_status ipc_put(void *data, ipc_datatype datatype, size_t count, int process) /* inline this */
{  return ipc_put_base(data,data,datatype,count,process); }

ipc_status ipc_get(void *data, ipc_datatype datatype, size_t count, int process) /* inline this */
{  return ipc_get_base(data,data,datatype,count,process); }

ipc_status ipc_put_to(void *target_data, void *source_data, ipc_datatype datatype, size_t count, int process)
{  return ipc_put_base(target_data,source_data,datatype,count,process); }

ipc_status ipc_get_to(void *target_data, void *source_data, ipc_datatype datatype, size_t count, int process)
{  return ipc_get_base(target_data,source_data,datatype,count,process); }
#endif


#ifdef __cplusplus
/* \name    C++ user interface         

  One overloaded version per ipc_datatype; allows weak form of runtime
  type-checking (when NDEBUG is not set) while retaining source-code
  compatibility with the C version.

  For full compile-time type checking, C++-only versions are also
  provided, which are the same except that they have no ipc_datatype
  parameter.
*/

#ifdef NDEBUG
#define IPC_CHECK_DATATYPE(name,ipc_type,data) (void)datatype
#else
#define IPC_CHECK_DATATYPE(name,ipc_type,data)                                          \
  if (datatype != ipc_type &&                                                           \
        !((datatype==IPC_RAW8  && sizeof(data)==1) ||                                   \
          (datatype==IPC_RAW32 && sizeof(data)==4) ||                                   \
          (datatype==IPC_RAW64 && sizeof(data)==8)))                                    \
    ipc_notify(3,IPC_WARNING,"ipc_" #name " called with incorrect datatype (%d != %d, sizeof(data)=%d != %d)",\
             ipc_type,datatype,sizeof(data),ipc_datatype_size(ipc_type))
#endif

#define IPC_CALL(name, c_type,ipc_type)                                                 \
ipc_status ipc_ ## name         (c_type *data, size_t count, int process)               \
{   return ipc_ ## name ## _base(data,data,ipc_type,count,process); }                   \
                                                                                        \
ipc_status ipc_ ## name ## _to  (c_type *target, c_type *source, size_t count, int process) \
{   return ipc_ ## name ## _base(target,source,ipc_type,count,process); }               \
                                                                                        \
ipc_status ipc_ ## name                                                                 \
(c_type *data, ipc_datatype datatype, size_t count, int process)                        \
{                                                                                       \
  IPC_CHECK_DATATYPE(name,ipc_type,data);                                               \
  return ipc_ ## name ## _base(data,data,ipc_type,count,process);                       \
}                                                                                       \
                                                                                        \
ipc_status ipc_ ## name ## _to                                                          \
(c_type *target, c_type *source, ipc_datatype datatype, size_t count, int process)      \
{                                                                                       \
  IPC_CHECK_DATATYPE(name,ipc_type,source);                                             \
  return ipc_ ## name ## _base(target,source,ipc_type,count,process);                   \
}

IPC_CALL(put, signed short     ,IPC_SHORT           )
IPC_CALL(put, signed int       ,IPC_INT             )
IPC_CALL(put, signed long      ,IPC_LONG            )
#ifndef LONG_LONG_UNAVAILABLE
IPC_CALL(put, signed long long ,IPC_LONG_LONG       )
#endif
IPC_CALL(put, unsigned char    ,IPC_UNSIGNED_CHAR   )
IPC_CALL(put, unsigned short   ,IPC_UNSIGNED_SHORT  )
IPC_CALL(put, unsigned int     ,IPC_UNSIGNED        )
IPC_CALL(put, unsigned long    ,IPC_UNSIGNED_LONG   )
IPC_CALL(put, float            ,IPC_FLOAT           )
IPC_CALL(put, double           ,IPC_DOUBLE          )
#ifndef LONG_DOUBLE_UNAVAILABLE
IPC_CALL(put, long double      ,IPC_LONG_DOUBLE     )
#endif
 
IPC_CALL(get, signed short     ,IPC_SHORT           )
IPC_CALL(get, signed int       ,IPC_INT             )
IPC_CALL(get, signed long      ,IPC_LONG            )
#ifndef LONG_LONG_UNAVAILABLE
IPC_CALL(get, signed long long ,IPC_LONG_LONG       )
#endif
IPC_CALL(get, unsigned char    ,IPC_UNSIGNED_CHAR   )
IPC_CALL(get, unsigned short   ,IPC_UNSIGNED_SHORT  )
IPC_CALL(get, unsigned int     ,IPC_UNSIGNED        )
IPC_CALL(get, unsigned long    ,IPC_UNSIGNED_LONG   )
IPC_CALL(get, float            ,IPC_FLOAT           )
IPC_CALL(get, double           ,IPC_DOUBLE          )
#ifndef LONG_DOUBLE_UNAVAILABLE
IPC_CALL(get, long double      ,IPC_LONG_DOUBLE     )
#endif
#undef IPC_CALL


#endif /* __cplusplus */



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


/* Implementation of put */
ipc_status ipc_put_base(void *target_data, void *source_data,
                        ipc_datatype datatype, size_t count, int process)
{

#ifdef IPC_USE_SHMEM

  switch(ipc_datatype_size(datatype)){
  case 4:
    shmem_put32(target_data,source_data,count,process);
    break;
  case 8:
    shmem_put64(target_data,source_data,count,process);
    break;
  default:
    ipc_notify(IPC_ALL,IPC_ERROR,"ipc_put called with unknown datatype: %d",datatype);
    /* exit(-1);  */
  }
#else
  
  if (process != ipc_my_process()) {
    ipc_notify(IPC_ALL,IPC_ERROR,"ipc_put to remote process unimplemented");          
    exit(-1);  
  }
  else if (target_data != source_data)
    memmove(target_data, source_data, count*ipc_datatype_size(datatype));
#endif

  return IPC_NO_ERROR;
}



ipc_status ipc_get_base(void *target_data, void *source_data,
                        ipc_datatype datatype, size_t count, int process)
{

#ifdef IPC_USE_SHMEM

  switch(ipc_datatype_size(datatype)){
  case 4:
    shmem_get32(target_data,source_data,count,process);
    break;
  case 8:
    shmem_get64(target_data,source_data,count,process);
    break;
  default:
    ipc_notify(IPC_ALL,IPC_ERROR,"ipc_get called with unknown datatype: %d",datatype);
    /* exit(-1); */
  }

#else
  
  if (process != ipc_my_process()) {
    ipc_notify(IPC_ALL,IPC_ERROR,"ipc_get from remote process unimplemented");          
    exit(-1);  
  }

  if (target_data != source_data)
    memmove(target_data, source_data, count*ipc_datatype_size(datatype));
#endif

  return IPC_NO_ERROR;
}



/******************************************************************************/
/*
  Console message procedures, in increasing order of severity 
  The most severe ones exit with a given exit code; see ipc.h for
  IPC_EXIT_*, which are the available exit codes.  
*/


void ipc_pe_msg_delay( double scale )
{
  clock_t start=clock();
  while ((clock()-start)/(double)(CLOCKS_PER_SEC) < IPC_PE_MSG_DELAY*ipc_my_process()*scale); 
}



void ipc_init(void)
{
#ifdef _CRAYT3D
  shmem_set_cache_inv();  /* Set auto invalidate for remote writes */
#endif
}



void ipc_init_logfile(const char *basefilename)
{
  static char oldname[IPC_MAXFILENAMELENGTH];
  char buf[IPC_MAXFILENAMELENGTH];
  
  if (basefilename) {
    SNPRINTF(buf,IPC_MAXFILENAMELENGTH,"%s.log",basefilename);
    
    if (ipc_logfile && strncmp(basefilename,oldname,IPC_MAXFILENAMELENGTH)) {
      fclose(ipc_logfile);
      ipc_logfile=NULL;
    }
    
    if (!ipc_logfile)
      ipc_logfile=fopen(buf,"w");
    
    if (!ipc_logfile)
      ipc_notify(IPC_ALL,IPC_WARNING,"Runtime file %s could not be opened; log messages will be lost",buf);

    SNPRINTF(oldname,IPC_MAXFILENAMELENGTH,"%s",basefilename);
  }
}



void ipc_log(int print_pe, const char *format, ...)
{
  va_list args;
  va_start (args, format);
  if ((print_pe==IPC_ALL || print_pe==ipc_my_process()) && ipc_logfile)
    vfprintf (ipc_logfile,format, args);
  va_end (args);
}



void ipc_error(   void )  {}
void ipc_warning( void )  {}



int ipc_notify(int print_pe, int message_level, const char *format, ...)
{
  va_list args;
  va_start (args, format);
  ipc_notify_base(1,print_pe,message_level,format,args);
  va_end (args);

  /* Return a value just so this can be called within an expression */
  return 0;
}



void ipc_notify2(int terminateline, int print_pe, int message_level, const char *format, ...)
{
  va_list args;
  va_start (args, format);
  ipc_notify_base(terminateline,print_pe,message_level,format,args);
  va_end (args);
}



void ipc_notify_base(int terminateline, int print_pe, int message_level, const char *format, va_list args)
{
  /* Figure out whether and how to print the item */
  const int forcealltoprint = ipc_msg_forceall_level >= message_level || ipc_msg_level >=IPC_OVERWHELM;
  const int allpesprint     = forcealltoprint  ||  print_pe==IPC_ALL;
  const int thispeprints    = allpesprint      ||  print_pe==ipc_my_process();

  const int iserror         = IPC_ERROR   >= message_level;
  const int iswarning       = IPC_WARNING >= message_level && !iserror;

  const int pasterrorlimit  = iserror   && ipc_errors++   >= ipc_max_errors;
  const int pastwarnlimit   = iswarning && ipc_warnings++ >= ipc_max_warnings;
  const int pastlimit       = pasterrorlimit || pastwarnlimit;

  const int aterrorlimit    = iserror   && ipc_errors     == ipc_max_errors;
  const int atwarnlimit     = iswarning && ipc_warnings   == ipc_max_warnings;
  const int atlimit         = aterrorlimit || atwarnlimit;
  
  const int printthismsg    = thispeprints &&  ipc_msg_level >= message_level && !pastlimit;
  const int usedelay        = ipc_num_processes() > 1 && (ipc_msg_synch_level >= message_level || ipc_msg_level>=IPC_OVERWHELM);
  const int labelpes        = ipc_num_processes() > 1;

  // Temporarily modified to put all messages on stdout so that the GUI
  // won't have to interleave stderr and stdout.
  //  FILE* outputstream        = (message_level <= IPC_ALERT ? stderr : stdout);
  FILE* outputstream        = stdout;
  
  /* Print it, if needed */
  if (printthismsg) {
    if (usedelay)   ipc_pe_msg_delay( 1.0 );
    if (labelpes)   fprintf(outputstream,"PE = %3d:  ",ipc_my_process());
    if (iserror)  { fprintf(outputstream,"Error -- "  ); ipc_error(); }
    if (iswarning){ fprintf(outputstream,"Warning -- "); ipc_warning(); }
    
    vfprintf(outputstream, format, args);
    if (terminateline) fprintf( outputstream,".\n");

    if (atlimit)
      ipc_notify(print_pe,IPC_CAUTION,"Maximum number of %s reached (%d); the rest will be discarded",
                 (iserror ? "errors"   : "warnings"),
                 (iserror ? ipc_errors : ipc_warnings));
  }

  /* Abort if too many */
  if (iserror && ipc_exit_on_error_num && (ipc_errors >= ipc_exit_on_error_num))
    ipc_abort(IPC_EXIT_TOO_MANY_ERRORS,"Error limit reached",ipc_errors);
}



void ipc_exit(int status,const char *format, ...)
{
  va_list args;
  va_start (args, format);

  /* Only print PE info if it won't be redundant */
  if (ipc_num_processes() > 1) {
    ipc_pe_msg_delay( 1.0 );
    fprintf(stderr,"PE = %3d:  ",ipc_my_process());
  }
  
  fprintf(stderr,"Exiting -- ");
  
  /* Print format string */
  vfprintf(stderr, format, args);
  va_end (args);

  fprintf(stderr,".  There were %d error(s) and %d warning(s).\n",ipc_errors,ipc_warnings);
  
  /* Global cleanup */
  if (ipc_my_process()==0 && ipc_logfile)
    fclose(ipc_logfile);

  /* Syncronize everyone before exiting */
  ipc_barrier();
  exit(status);
}



void ipc_abort(int status,const char *format, ...)
{
  va_list args;
  va_start (args, format);

  /* Try not to lose any messages, at least on this processor */
  fflush(NULL);

  /* Only print PE info if it won't be redundant */
  if (ipc_num_processes() > 1) {

    /* Wait for a few seconds, hoping that other processors will flush
       their output too */
    clock_t start=clock();
    while ((clock()-start)/(double)(CLOCKS_PER_SEC) < 5); 

    ipc_pe_msg_delay( 1.0 );
    fprintf(stderr,"PE = %3d:  ",ipc_my_process());
  }
  
  fprintf(stderr,"Aborting -- ");
  
  /* Print format string */
  vfprintf(stderr, format, args);
  va_end (args);

  fprintf(stderr,".  There were %d error(s) and %d warning(s).\n",ipc_errors,ipc_warnings);

  /* Global cleanup */
  if (ipc_my_process()==0 && ipc_logfile)
    fclose(ipc_logfile);

#ifdef _CRAYT3E
  /* Abort all processors simultaneously */
  globalexit(status);
#else
  /* Assume single-processor */
  exit(status);
#endif
}


#ifdef __cplusplus
#ifndef NO_STRINGS
namespace Msg {
void ipc_verbose   (const string& s, const bool terminate)
  {  ::ipc_notify2(terminate,IPC_ONE,IPC_VERBOSE,   s.c_str());  }
void ipc_notify    (const string& s, const bool terminate)
  {  ::ipc_notify2(terminate,IPC_ONE,IPC_STD,       s.c_str());  }
void ipc_requested (const string& s, const bool terminate)
  {  ::ipc_notify2(terminate,IPC_ONE,IPC_REQUESTED, s.c_str());  }
void ipc_error     (const string& s, const bool terminate)
  {  ::ipc_notify2(terminate,IPC_ONE,IPC_ERROR,     s.c_str());  }
void ipc_warning   (const string& s, const bool terminate)
  {  ::ipc_notify2(terminate,IPC_ONE,IPC_WARNING,   s.c_str());  }
}
#endif
#endif

---