/* Partial MPI library based on Sockets in Linux,

   using TCP/IP protocol.

   No local buffering of messages is implemented, so that all messages

   must be received in the order sent, and receives with wildcard

   sources are not supported. the following subroutines are implemented:

   MPI_Init, MPI_Finalize, MPI_Send, MPI_Recv, MPI_Isend, MPI_Irecv

   MPI_Test, MPI_Wait, MPI_Sendrecv, MPI_Ssend, MPI_Issend, MPI_Waitall

   MPI_Waitany, MPI_Get_count, MPI_Initialized, MPI_Comm_size

   MPI_Comm_rank, MPI_Comm_dup, MPI_Comm_split, MPI_Comm_free

   MPI_Cart_create, MPI_Cart_coords, MPI_Cart_get, MPI_Cart_shift

   MPI_Cart_rank, MPI_Cart_sub, MPI_Dims_create

   MPI_Bcast, MPI_Barrier, MPI_Reduce, MPI_Scan

   MPI_Allreduce, MPI_Gather, MPI_Allgather, MPI_Scatter, MPI_Alltoall

   MPI_Gatherv, MPI_Allgatherv, MPI_Scatterv, MPI_Alltoallv

   MPI_Reduce_scatter, MPI_Abort, MPI_Wtime, MPI_Wtick, MPI_Type_extent

   MPI_Request_free, MPI_Get_processor_name

   Sockets are described in Unix: Network Programming, vol. 1, by

   W. Richards Stevens [Prentice Hall PTR, Upper Saddle River, NJ, 1998].

   The Message Passing Interface (MPI) is described in the reference,

   M. Snir, S. Otto, S. Huss-Lederman, D. Walker, and J. Dongarra,

   MPI: The Complete Reference [MIT Press, Cambridge, MA,1996].

   The file MPIerrs is used throughout for error messages

   written by viktor k. decyk, ucla

   copyright 2002, regents of the university of california.

   all rights reserved.

   no warranty for proper operation of this software is given or implied.

   software or information may be copied, distributed, and used at own

   risk; it may not be distributed without this notice included verbatim

   with each file. 

   update: january 26, 2004                                              */

#include <stdlib.h>

#include <stdio.h>

#include <string.h>

#include <math.h>

#include "mpi.h"

#include <sys/types.h>

#include <sys/errno.h>

#include <sys/socket.h>

#include <netinet/in.h>

#include <netinet/tcp.h>

#include <arpa/inet.h>

#include <sys/utsname.h>

#include <sys/fcntl.h>

#include <sys/ioctl.h>

#include <sys/uio.h>

#include <net/if.h>

#include <netdb.h>

#include <unistd.h>

#include <signal.h>

/* MAXS = maximum number of nodes connected */

#define MAXS                 64

/* MAXM = maximum number of outstanding messages on a node */

#define MAXM                 (2*MAXS)

/* MAXC = maximum number of communicators */

#define MAXC                 10

/* MAXD = maximum number of topology dimensions */

#define MAXD                 6

/* internal mpi common block

   nproc = number of real or virtual processors obtained

   idproc = processor id

   epref = array of socket endpoints for each participating node

   ioc = array of context pointers for notifier function

   stime = first time stamp if MPI_Init successful

   mapcomm = communicator map

   maxfds = maximum descriptor value

   fdsset = fd_set reprsenting all endpoints

   sset = signal mask                                               */

   static int nproc = -1, idproc;

   static int epref[MAXS+2];

   static int ioc[MAXS+2][4], mapcomm[MAXC][MAXS+MAXD+3];

   static struct timeval stime;

   static int maxfds;

   static fd_set fdsset;

   static sigset_t sset;

/* internal common block for non-blocking messages

   monitor = (0,1,2) = (suppress,display,display & log) monitor messages

   curreq = request record for transmission parameters

   header = message envelope

   rwrec = read/write record for asynchronous messages

   trash = trash bin for unwanted data

   mqueue = message request queue                                     */

   struct RWRec {

      int ref;

      int ioflag;

      void *buf;

      size_t nbytes;

      int flags;

      void *sbuf;

      int sln;

      int len;

      struct timeval ts[2];

      int nextm;

      int nfatal;

   };

   struct ipdata {

      struct iovec iov[2];

      int hdata[4];

   };

   static int monitor = 1, curreq[MAXM][5], mqueue[MAXS+2][2];

   static struct ipdata header[MAXM];

   static struct RWRec rwrec[MAXM];

   static unsigned char trash[1024];

static FILE *unit2;

/* prototypes for internal procedures */

static void alarmf(int signo);

static void notifier(int signo);

long otpinit(int np, struct sockaddr_in *addrn, int alen);

long etmsec(struct timeval *ptime);

long sbtusec(struct timeval *pstime, struct timeval *crtime);

int ioresult(int *pblock);

int checkesc(long stk);

void sndmsgf(MPI_Request request, int dest);

void rcvmsgf(MPI_Request request, int source);

static int imax(int val1, int val2);

static int imin(int val1, int val2);

static float flmax(float val1, float val2);

static float flmin(float val1, float val2);

static double dmax(double val1, double val2);

static double dmin(double val1, double val2);

static void iredux(int *recvbuf, int *sendbuf, int offset, int count, MPI_Op op);

static void fredux(float *recvbuf, float *sendbuf, int offset, int count, MPI_Op op);  

static void dredux(double *recvbuf, double *sendbuf, int offset, int count, MPI_Op op);

void writerrs(char *source, int ierror);

void rwstat(int request, FILE *unit);

void wqueue(FILE *unit);

void messwin(int nvp);

void logmess(int idp, int lstat, int lsize, int mticks, int tag);

void showmess(int idp, int istat, int istyle);

void showdism(int ibin, int nbin, int mbin, int lmax, int istyle);

void shospeed(float atime, float ctime, float arate, float crate);

void Logname(char *name);

void Set_Mon(int monval);

int Get_Mon();

void delmess();

/* function definitions */

int MPI_Init(int *argc, char ***argv) {

/* initialize the MPI execution environment

   input: argc, argv, output: none

local data                                  */

   int ierror, nerr, nv, i, alen;

   int portnum = 0, pshift = 0, pnumid[3], epnum[8];

   long oss, response;

   char *cnerr = 0;

   struct utsname uinfo;

   struct hostent *info;

   struct in_addr address, oldadd;

   char ename[36], myself[36], location[18];

   struct sockaddr_in addrn, addrb, addrl;

   struct timeval ptime;

   struct sigaction act;

   MPI_Status stat;

   FILE *unit3;

/* internal mpi common block

   nproc = number of real or virtual processors obtained

   idproc = processor id

   epref = array of socket endpoints for each participating node

   ioc = array of context pointers for notifier function

   ioc[i][0] = endpoint reference for notifier for endpoint i

   ioc[i][1] = processor id for listener for endpoint i

   ioc[i][2] = handle for current receive from endpoint i

   ioc[i][3] = handle for current send to endpoint i

   stime = first time stamp if MPI_Init successful

   maxfds = maximum descriptor value

   fdsset = fd_set reprsenting all endpoints

   sset = signal mask

   mapcomm = communicator map

   mapcomm[i][0:nproc-1] = actual proc id for given rank in communicator i

   mapcomm[i][nproc:MAXS-1] = MPI_UNDEFINED

   mapcomm[i][MAXS] = number of processes in comm i

   mapcomm[i][MAXS+1] = rank for this node in comm i

   mapcomm[i][MAXS+2] = ndims = number of dimensions in topology in comm i

   mapcomm[i][iMAXS+3:MAXS+2+ndims] = size of dimension in comm i,

   negative if non-periodic

   mapcomm[i][MAXS+3+ndims:MAXS+2+MAXD] = 0                             */  

/* internal common block for non-blocking messages

   monitor = (0,1,2) = (suppress,display,display & log) monitor messages

   curreq = request record for transmission parameters, see rwstat

   header = message envelope

   hdata[i][0] = communicator, hdata[i][1] = tag, hdata[i][2] = datatype

   hdata[i][3] = length of data (in bytes) for message handle i

   rwrec = read/write record for asynchronous messages, see rwstat

   mqueue = message request queue

   mqueue[i][0] = end of message queue for receives from endpoint i

   mqueue[i][1] = end of message queue for sends to endpoint i          */

/* internal common block for adsp

   epref0 = endpoint reference used by tcp and adsp providers */

/* Initialize common block data */

   for (i = 0; i < MAXS+2; i++) {

      epref[i] = 0;

      for (nv = 0; nv < 2; nv++)

         mqueue[i][nv] = 0;

   }

   for (i = 0; i < MAXC; i++) {

      for (nv = 0; nv < MAXS+MAXD+3; nv++)

         mapcomm[i][nv] = 0;

   }

   for (i = 0; i < MAXM; i++) {

      for (nv = 0; nv < 5; nv++)

         curreq[i][nv] = 0;

   }

/* Set MPI_COMM_WORLD and MPI_COMM_SELF mapping */

   for (i = 0; i < MAXS; i++) {

      mapcomm[0][i] = i;

      mapcomm[1][i] = MPI_UNDEFINED;

   }

/* Zero out descriptor array */

   FD_ZERO(&fdsset);

/* Open error file */

   unit2 = fopen("MPIerrs","w");

   if (monitor==2)

      fprintf(unit2,"MPI_Init started\n");

/* Obtain the current time stamp */

   oss = gettimeofday(&stime,NULL);

   if (oss < 0) {

      oss = errno;

      fprintf(unit2,"Gettimeofday Error = %ld, %s\n",oss,strerror(oss));

   }

/* Obtain information about the Internet environment */

   oss = uname(&uinfo);

   if (oss < 0) {

      ierror = errno;

      fprintf(unit2,"Uname Error = %d, %s\n",ierror,strerror(ierror));

      return ierror;

   }

   strcpy(myself,uinfo.nodename);

/* debug */

   if (monitor==2) {

      fprintf(unit2,"local host name=%s\n",myself);

   }

   info = gethostbyname(uinfo.nodename);

   if (info==NULL) {

      oss = h_errno;

      fprintf(unit2,"Gethostbyname Error = %ld for %s\n",oss,myself);

      fprintf(unit2,"%s, Trying INADDR_ANY\n",hstrerror(oss));

      address.s_addr = INADDR_ANY;

   }

   else

/* debug */

      if (monitor==2) {

         i = 1;

         cnerr = ((info->h_addr_list)[i]);

         while (cnerr) {

            address.s_addr = *(in_addr_t *) cnerr;

            cnerr = inet_ntoa(address);

            strcpy(myself,cnerr);

            fprintf(unit2,"other local addresses=%d,%s\n",i,myself);

            i =+ 1;

            cnerr = ((info->h_addr_list)[i]);

         }

      }

      address.s_addr = *(in_addr_t *) ((info->h_addr_list)[0]);

/* debug */

   if (monitor==2) {

      cnerr = inet_ntoa(address);

      strcpy(myself,cnerr);

      fprintf(unit2,"local address=%s\n",myself);

   }

   oldadd = address;

/*

   Everyone opens a port

*/

/* Open file containing portnum (and possibly participating nodes)

   first line in nodelist file on all nodes contains common portnum

   if the file is missing or empty, a default number of 5013 is used */

   unit3 = fopen("nodelist_ip","r");

   if (!unit3)

      unit3 = fopen("nodelist","r");

   if (unit3) {

      cnerr = fgets(location,11,unit3);

      if (!cnerr)

         strcpy(location,"5013");

/* Replace trailing newlines with nulls */

      cnerr = strchr(location,'\n');

      if (cnerr)

         cnerr[0] = '\0';

      if (location[0]=='\0')

         strcpy(location,"5013");

   }

   else

      strcpy(location,"5013");

/* Convert string to integer */

   nerr = sscanf(location,"%d",&portnum);

   if ((!nerr) || (portnum < 5000) || (portnum > 49152))

      portnum = 5013;

/* Construct sockaddr_in of listener */

   addrn.sin_family = AF_INET;

   addrn.sin_port = portnum;

   addrn.sin_addr = address;

   alen = sizeof(addrn);

   for (i = 0; i < 8; i++)

      addrn.sin_zero[i] = '\0';

   strcpy(ename,myself);

/* Save copy of portnum */

   pnumid[0] = portnum;

   nproc = 0;

/* Initialize synchronous listener endpoint provider */

   oss = otpinit(MAXS+1,&addrn,alen);

   if (oss) {

      ierror = oss;

      nerr = MPI_Finalize();

      return ierror;

   }

/* Set struct sockaddr_in for address which is returned */

   nerr = alen;

   oss = getsockname(epref[MAXS+1],(struct sockaddr *)&addrb,&nerr);

   if (oss < 0) {

      ierror = errno;

      fprintf(unit2,"Getsockname Error = %d, %s\n",ierror,strerror(ierror));

      nerr = MPI_Finalize();

      return ierror;

   }

/* Pass processor id to listener notifier */

   ioc[MAXS+1][1] = nproc + 1;

/* debug */

   if (monitor==2)

      fprintf(unit2,"listener=%s,port=%d,socket=%d\n",myself,portnum,

              epref[MAXS+1]);

/*

   Determine if node is master (idproc=0) or slave (idproc>0).

   on the master node, the second and subsequent lines of nodelist file

   contain IP addresses of the nodes participating, in dotted-decimal

   format (for example, "12.13.14.15").

   if this list of nodes is missing, then the node is a slave.

   every node also makes a connection to itself.

*/

   if (unit3) {

      cnerr = fgets(location,17,unit3);

   }

/* Must be slave */

   if (!unit3 || !cnerr)

      idproc = 1;

   else {

/* Replace trailing newlines with nulls */

      cnerr = strchr(location,'\n');

      if (cnerr)

         cnerr[0] = '\0';

/* Must be slave */

      if (location[0]=='\0')

         idproc = 1;

/* Must be master */

      else

         idproc = 0;

   }

/* 

   * * * begin main iteration loop * * *

   Prepare to accept connection

*/

/* Establish connection end */

/* Listen for an incoming connection request */

L10: oss = listen(epref[MAXS+1],1);

   if (oss < 0) {

      ierror = errno;

      fprintf(unit2,"Listen Error = %d, %s\n",ierror,strerror(ierror));

      nerr = MPI_Finalize();

      return ierror;

   }

/* Prepare alarm */

   act.sa_handler = alarmf;

   sigemptyset(&act.sa_mask);

   act.sa_flags = 0;

   oss = sigaction(SIGALRM,&act,NULL);

   if (oss < 0) {

      ierror = errno;

      fprintf(unit2,"Sigaction SIGALRM Error = %d, %s\n",ierror,

              strerror(ierror));

      nerr = MPI_Finalize();

      return ierror;

   }

/* For connections to oneself, jump to connect */

   if (idproc==nproc) {

      ioc[MAXS+1][1] = MAXS+2;

      goto L70;

   }

/* Set alarm */

   alarm(60);

/* Accept an incoming connection request */

   nerr = alen;

   epref[nproc] = accept(epref[MAXS+1],(struct sockaddr *)&addrl,&nerr);

/* Clear alarm */

   alarm(0);

   if (epref[nproc] < 0) {

      ierror = errno;

      if (ierror==EINTR)

         ierror = ETIMEDOUT;

      fprintf(unit2,"Accept Error = %d, %s\n",ierror,strerror(ierror));

      return ierror;

   }

/* debug */

   if (monitor==2) {

      cnerr = inet_ntoa(addrl.sin_addr);

      fprintf(unit2,"socket=%d accepted connection with=%s,%d\n",

              epref[nproc],cnerr,addrl.sin_port);

   }

   ioc[MAXS+1][1] = 0;

/* Receive portnum for verification and processor id from remote node */

   nproc += 1;

   mapcomm[0][MAXS] = nproc;

/* Reset iocompletion flag to notifier */

   ioc[MAXS+1][1] = nproc + 1;

   ierror = MPI_Recv(epnum,3,MPI_INT,nproc-1,3,0,&stat);

/* Extract processor id on first connection */

   if (nproc==1) {

      idproc = epnum[1];

      pshift = portnum - epnum[2];

      portnum = epnum[2];

      mapcomm[0][MAXS+1] = idproc;

   }

/* Check if remote portnum agrees with local portnum */

   nerr = pnumid[0] - epnum[0];

/* Send reject flag to remote node */

   ierror = MPI_Send(&nerr,1,MPI_INT,nproc-1,4,0);

/* Reject if portnums disagree */

   if (nerr) {

      fprintf(unit2,"Session rejected, idproc = %d\n",idproc);

      fprintf(unit2,"Portnums do not agree\n");

      ierror = 5;

      fprintf(unit2,"remote portnum = %d\n",epnum[0]);

      nerr = MPI_Finalize();

      return ierror;

   }

/* Check for processor number overflow */

   if (nproc > MAXS) {

      fprintf(unit2,"processor number overflow, nproc = %d\n",nproc);

      ierror = 6;

      nerr = MPI_Finalize();

      return ierror;

   }

/* debug */

   if (monitor==2)

      fprintf(unit2,"connection accepted with idproc=%d\n",nproc-1);

/* Accept more connections */

   if (idproc >= nproc)

      goto L10;

/*

   Master prepares to start connection

*/

/* Find internet address of requested node */

/* Convert a character string into an address */

L40: oss = inet_aton(location,&address);

   if (!oss) {

      fprintf(unit2,"inet_aton failed, location = %s\n",location);

      fprintf(unit2,"Invalid nodelist file possibly being used\n");

      ierror = oss;

      nerr = MPI_Finalize();

      return ierror;

   }

/* Second cpu on a node uses incremented port number */

   if (address.s_addr==oldadd.s_addr)

      pshift = pshift + 1;

   else {

      pshift = 0;

      oldadd = address;

   }

   pnumid[0] = portnum + pshift;

   addrn.sin_family = AF_INET;

   addrn.sin_port = pnumid[0];

/* addrn.sin_addr.s_addr = INADDR_ANY; */

   addrn.sin_addr = address;

   for (i = 0; i < 8; i++)

      addrn.sin_zero[i] = '\0';

/* Convert an address into a character string */

   cnerr = inet_ntoa(address);

   strcpy(ename,cnerr);

/* Establish connection end */

/* Create a new socket for this endpoint */

L70: addrb.sin_port = 0;

/* Initialize synchronous endpoint provider */

   oss = otpinit(nproc,&addrb,alen);

   if (oss) {

      ierror = oss;

      nerr = MPI_Finalize();

      return ierror;

   }

/* debug */

   if (monitor==2)

      fprintf(unit2,"socket=%d requesting connection with=%s,%d\n",

              epref[nproc],ename,pnumid[0]);

/* Pause to let OS get some time */

   if (checkesc(1)) {

      ierror = -9;

      writerrs("MPI_Init: ",ierror);

      return ierror;

   }

/* Obtain the current time stamp */

   oss = gettimeofday(&ptime,NULL);

/* Set alarm */

   alarm(60);

/* Request a connection to a remote peer */

   oss = connect(epref[nproc],(struct sockaddr *)&addrn,alen);

/* Clear alarm */

   alarm(0);

   if (oss < 0) {

      ierror = errno;

      if (ierror==EINTR)

         ierror = ETIMEDOUT;

      fprintf(unit2,"Connect Error = %d, %s\n",ierror,strerror(ierror));

      return ierror;

   }

/* Accept an incoming connection request to oneself */

   if (idproc==nproc) {

      nerr = alen;

      epref[MAXS] = accept(epref[MAXS+1],(struct sockaddr *)&addrl,&nerr);

      if (epref[MAXS] < 0) {

         ierror = errno;

         fprintf(unit2,"Accept Error = %d, %s\n",ierror,strerror(ierror));

         return ierror;

      }

/* debug */

      if (monitor==2) {

         cnerr = inet_ntoa(addrl.sin_addr);

         fprintf(unit2,"self socket=%d accepted connection with=%s,%d\n",

                 epref[MAXS],cnerr,addrl.sin_port);

      }

   }

   ioc[nproc][1] = 0;

/* debug */

   if (monitor==2) {

      nerr = alen;

      oss = getpeername(epref[nproc],(struct sockaddr *)&addrl,&nerr);

      if (oss < 0) {

         oss = errno;

         fprintf(unit2,"Getpeername Error = %ld, %s\n",oss,strerror(oss));

         addrl.sin_addr.s_addr = 0;

      }

      cnerr = inet_ntoa(addrl.sin_addr);

      fprintf(unit2,"tentative connection started with=%s,%d\n",cnerr,

              addrl.sin_port);

   }

/* Send portnum for verification and processor id to remote node */

   nerr = nproc;

   nproc += 1;

   mapcomm[0][MAXS] = nproc;

   if (nerr > idproc) {

/* Set processor id */

      pnumid[1] = nerr;

      pnumid[2] = portnum;

      ierror = MPI_Send(pnumid,3,MPI_INT,nproc-1,3,0);

/* Read and check reject flag */

      ierror = MPI_Recv(&nerr,1,MPI_INT,nproc-1,4,0,&stat);

      if (nerr) {

         fprintf(unit2,"Connection rejected, reject info = %d\n",nerr);

         fprintf(unit2,"Portnums do not agree, idproc = %d\n",nproc-1);

         ierror = 12;

         nerr = MPI_Finalize();

         return ierror;

      }

   }

/* Check for processor number overflow */

   if (nproc > MAXS) {

      fprintf(unit2,"processor number overflow, nproc = %d\n",nproc);

      ierror = 6;

      nerr = MPI_Finalize();

      return ierror;

   }

/* debug */

   if (monitor==2)

      fprintf(unit2,"connection confirmed with idproc=%d\n",nproc-1);

/* Pass current location to next node */

   if (nproc > (idproc+2))

      nerr = MPI_Send(location,4,MPI_INT,idproc+1,1,0);

/* Read location of next node from file */

   if (idproc==0) {

      if ((nproc >= 2) || (!strcmp(location,"self")) || (!strcmp(location,myself))) {

         if (!unit3)

            goto L90;

         cnerr = fgets(location,17,unit3);

         if (!cnerr)

            goto L90;

/* Replace trailing newlines with nulls */

         cnerr = strchr(location,'\n');

         if (cnerr)

            cnerr[0] = '\0';

         if (location[0]=='\0')

            goto L90;

      }

   }

/* Receive location of next node from another processor */

   else {

      nerr = MPI_Recv(location,4,MPI_INT,idproc-1,1,0,&stat);

/* End of file marker received */

      if (stat.len==0)

         goto L90;

   }

/* Start another connection */

   goto L40;

/* 

   * * * end main iteration loop * * *

*/

/* All expected nodes activated */

L90: nv = nproc - 1;

/* debug */

   if (monitor==2)

      fprintf(unit2,"all nodes activated, idproc, nproc=%d,%d\n",

              idproc,nproc);

/* Send null record to next processor */

   if (idproc < nv)

      nerr = MPI_Send(location,0,MPI_INT,idproc+1,1,0);

   if (unit3)

      fclose(unit3);

/* Check number of processors */

   if (idproc==nv) {

      for (i = 1; i <= nv; i++) {

         nerr = MPI_Send(&nproc,1,MPI_INT,nv-i,2,0);

      }

   }

   else {

      nerr = MPI_Recv(&response,1,MPI_INT,nv,2,0,&stat);

/* Local processor does not agree with last processor on total number */

      if (response != nproc) {

         fprintf(unit2,"processor number error, local/remote nproc = %d,%ld\n",

                 nproc,response);

         ierror = 7;

         nerr = MPI_Finalize();

         return ierror;

      }

   }

/* Clear unused MPI_COMM_WORLD mapping */

   for (i = nproc; i < MAXS; i++) {

      mapcomm[0][i] = MPI_UNDEFINED;

   }

   mapcomm[0][MAXS+2] = 0;

/* Set MPI_COMM_SELF */

   mapcomm[1][0] = idproc;

   mapcomm[1][MAXS] = 1;

   mapcomm[1][MAXS+1] = 0;

   mapcomm[1][MAXS+2] = 0;

/* Set descriptor array for notifier */

   maxfds = -1;

   for (i = 0; i < MAXS+1; i++) {

      if (epref[i]) {

         maxfds = imax(epref[i],maxfds);

         FD_SET(epref[i],&fdsset);

      }

   }

   maxfds = maxfds + 1;

/* Prepare signal mask */

   oss = sigemptyset(&sset);

   oss = sigaddset(&sset,SIGIO);

/* Prepare signal action */

   act.sa_handler = notifier;

   sigemptyset(&act.sa_mask);

   act.sa_flags = 0;

   oss = sigaction(SIGIO,&act,NULL);

   if (oss < 0) {

      ierror = errno;

      fprintf(unit2,"Sigaction SIGIO Error = %d, %s\n",ierror,

              strerror(ierror));

      nerr = MPI_Finalize();

      return ierror;

   }

/* Set providers to asynchronous mode */

   nv = 1;

   nerr = getpid();

   for (i = 0; i < MAXS+1; i++) {

      if (epref[i]) {

         oss = fcntl(epref[i],F_SETOWN,nerr);

         if (oss < 0) {

            oss = errno;

            fprintf(unit2,"F_SETOWN fcntl Error, oss = %ld, %s\n",oss,

                    strerror(oss));

            return oss;

         }

         oss = ioctl(epref[i],FIOASYNC,&nv);

         if (oss < 0) {

            oss = errno;

            fprintf(unit2,"FIOASYNC ioctl Error, oss = %ld, %s\n",oss,

                    strerror(oss));

            return oss;

         }

         oss = ioctl(epref[i],FIONBIO,&nv);

         if (oss < 0) {

            oss = errno;

            fprintf(unit2,"FIONBIO ioctl Error, oss = %ld, %s\n",oss,

                    strerror(oss));

            return oss;

         }

      }

   }

/* Create window for showing MPI message status */

   if (monitor > 0) {

      messwin(nproc);

      checkesc(1);

      if (monitor==2)

         fprintf(unit2,"MPI_Init complete\n\n");

   }

/* Set error code to success */

   ierror = 0;

   return ierror;

}

static void alarmf(int signo) {

/* callback function to handle connect/accept alarms */

   return;

}

static void notifier(int signo) {

/* notifier function for asynchronous and completion events

local data */

   int i, j, l, n, m;

   long num, oss;

   struct timeval timeout = {0,0};

   fd_set lfdsset;

/* internal mpi common block

   epref = array of endpoint references for each participating node

   maxfds = maximum descriptor value

   fdsset = fd_set reprsenting all endpoints                         */

/* internal common block for non-blocking messages

   rwrec = read/write record for asynchronous messages

   trash = trash bin for unwanted data

   mqueue = message request queue                                    */

/* Check if any descriptors are ready for reading */

   lfdsset = fdsset;

   num = select(maxfds,&lfdsset,NULL,NULL,&timeout);

/* Check for error in select */

   if (num <= 0) {

      num = 0;

      goto L20;

   }

   l = 0;

/* Find which descriptor has a read event */

   for (i = 0; i < nproc; i++) {

      if (FD_ISSET(epref[i],&lfdsset)) {

/* Get reference to endpoint */

         n = ioc[i][0];

/* Read pointer to current read record */

         m = ioc[i][2] - 1;

/* Read data sent from a remote peer */

         if ((m >= 0) && (m < MAXM)) {

/* Obtain the current time stamp */

            oss = gettimeofday(&rwrec[m].ts[1],NULL);

L10:        oss = recv(rwrec[m].ref,rwrec[m].buf,rwrec[m].nbytes,

                       rwrec[m].flags);

            if (oss < 0)

               oss = errno - 1024;

/* Process data which arrived */

            if (oss > 0) {

/* Clear non-fatal error code */

               rwrec[m].nfatal = 0;

/* Set actual length received */

               rwrec[m].len += oss;

/* Check if all the data has arrived */

               if (rwrec[m].len < header[m].hdata[3]) {

/* Incomplete message */

                  if (rwrec[m].nbytes > oss) {

/* Readjust buffer pointer */

                     rwrec[m].buf = (void *)(((unsigned char *)rwrec[m].buf) + oss);

                     rwrec[m].nbytes -= oss;

                     if (rwrec[m].len >= 0)

                        rwrec[m].ioflag += 1;

                  }

/* Header is received, readjust parameters to receive data */

                  else if (rwrec[m].ioflag==1) {

                     rwrec[m].buf = rwrec[m].sbuf;

                     rwrec[m].nbytes = imin(header[m].hdata[3],rwrec[m].sln);

                     rwrec[m].ioflag = 2;

                  }

/* Data is received and buffer is full */

                  else {

                     rwrec[m].buf = &trash;

                     rwrec[m].nbytes = 1024;

                  }

                  goto L10;

               }

/* Message complete */

               else {

/* Obtain the current time stamp */

                  oss = gettimeofday(&rwrec[m].ts[1],NULL);

/* Set iocompletion flag */

                  rwrec[m].ioflag = 0;

/* Get next message if messages are queued */

                  if (rwrec[m].nextm > 0) {

                     m = rwrec[m].nextm;

                     if (m==mqueue[n][0])

                        mqueue[n][0] = 0;

                     ioc[i][2] = m;

                     m -= 1;

                     goto L10;

                  }

/* Clear pointer to current send record */

                  else

                     ioc[i][2] = 0;

               }

            }

/* Check for errors */

            else {

/* Check for EOF */

               if (oss==0)

                  oss = -1;

/* Quit if no data is available */

               if ((oss+1024) != EWOULDBLOCK) {

/* Set iocompletion flag to error */

                  rwrec[m].ioflag = oss;

                  ioc[i][2] = 0;

               }

            }

         }

/* All receive events processed */

         l += 1;

         if (l==num)

            goto L20;

      }

   }

/* Check if any descriptors are ready for writing */

L20: lfdsset = fdsset;

   oss = select(maxfds,NULL,&lfdsset,NULL,&timeout);

   if (oss > 0)

      num += oss;

/* Check for error in select */

   if (oss <= 0)

      return;

/* Find which descriptor has a write event */

   for (j = 0; j < nproc+1; j++) {

      i = j;

      if (j==nproc)

         i = MAXS;

      if (FD_ISSET(epref[i],&lfdsset)) {

/* Get reference to endpoint */

         n = ioc[i][0];

/* Read pointer to current send record */

         m = ioc[i][3] - 1;

/* Send data to a remote peer */

         if ((m >= 0) && (m < MAXM)) {

/* Obtain the current time stamp */

            oss = gettimeofday(&rwrec[m].ts[1],NULL);

L30:        if (rwrec[m].ioflag==1)

               oss = writev(rwrec[m].ref,rwrec[m].buf,rwrec[m].nbytes);

            else

               oss = send(rwrec[m].ref,rwrec[m].buf,rwrec[m].nbytes,

                      rwrec[m].flags);

            if (oss < 0)

               oss = errno - 1024;

/* Process data which has been sent */

            if (oss > 0) {

/* Clear non-fatal error code */

               rwrec[m].nfatal = 0;

/* Set actual length sent */

               rwrec[m].len += oss;

/* Check for incomplete header */

               if (rwrec[m].len < 0) {

                  header[m].iov[0].iov_base = 

                     (void *)(((unsigned char *)header[m].iov[0].iov_base) + oss);

                  header[m].iov[0].iov_len -= oss;

                  goto L30;

               }

/* Check for incomplete data */

               else if (rwrec[m].sln > rwrec[m].len) {

/* Header is sent, readjust parameters to send data */

                  if (rwrec[m].ioflag==1) {

                     rwrec[m].buf = rwrec[m].sbuf;

                     rwrec[m].nbytes = rwrec[m].sln;

                     oss -= header[m].iov[0].iov_len;

                  }

/* Readjust buffer pointer */

                  rwrec[m].buf = (void *)(((unsigned char *)rwrec[m].buf) + oss);

                  rwrec[m].nbytes -= oss;

                  rwrec[m].ioflag += 1;

                  goto L30;

               }

/* Data is sent */

               else {

/* Obtain the current time stamp */

                  oss = gettimeofday(&rwrec[m].ts[1],NULL);

/* Set iocompletion flag */

                  rwrec[m].ioflag = 0;

/* Get next message if messages are queued */

                  if (rwrec[m].nextm > 0) {

                     m = rwrec[m].nextm;

                     if (m==mqueue[n][1])

                        mqueue[n][1] = 0;

                     ioc[i][3] = m;

                     m -= 1;

                     goto L30;

                  }

/* Clear pointer to current send record */

                  else

                     ioc[i][3] = 0;

               }

            }

/* Check for errors */

            else {

/* Check for EOF */

               if (oss==0)

                  oss = -1;

/* Quit if no data can be sent */

               if ((oss+1024) != EWOULDBLOCK) {

/* Set iocompletion flag to error */

                  rwrec[m].ioflag = oss;

                  ioc[i][3] = 0;

               }

            }

         }

/* All send events processed */

         l += 1;

         if (l==num)

            return;

      }

   }

   return;

}

long otpinit(int np, struct sockaddr_in *addrn, int alen) {

/* this subroutine initializes a Socket provider for

   index np, using address specified in addrn.

   provider is left in asynchronous, blocking mode

   np = index to endpoint reference array

   addrn = address to which endpoint is to be bound

   alen = length of addrn structure

   returns error indicator

   input: np, addrn, alen

local data                                                       */

   long oss, value;

/* internal mpi common block

   epref = array of endpoint references for each participating node

   ioc = array of context pointers for notifier function                  */

   oss = 0;

/* Create a synchronous endpoint provider */

   epref[np] = socket(AF_INET,SOCK_STREAM,0);

   if (epref[np] < 0) {

      oss = errno;

      fprintf(unit2,"Socket Error = %ld, %s\n",oss,strerror(oss));

      return oss;

   }

/* Set TCP_NODELAY option */

   value = 1;

   oss = setsockopt(epref[np],IPPROTO_TCP,TCP_NODELAY,&value,4);

   if (oss < 0) {

      oss = errno;

      fprintf(unit2,"TCP_NODELAY setsockopt Error %ld, %s\n",oss,strerror(oss));

      return oss;

   }

/* Set SO_RCVBUF option */

   value = 65536;

/* oss = setsockopt(epref[np],SOL_SOCKET,SO_RCVBUF,&value,4); */

   if (oss < 0) {

      oss = errno;

      fprintf(unit2,"SO_RCVBUF setsockopt Error %ld, %s\n",oss,strerror(oss));

      return oss;

   }

/* Set SO_SNDBUF option */

   value = 65536;

/* oss = setsockopt(epref[np],SOL_SOCKET,SO_SNDBUF,&value,4); */

   if (oss < 0) {

      oss = errno;

      fprintf(unit2,"SO_SNDBUF setsockopt Error %ld, %s\n",oss,strerror(oss));

      return oss;

   }

/* Set SO_DEBUG option for trpt program */

   value = 1;

/* oss = setsockopt(epref[np],SOL_SOCKET,SO_DEBUG,&value,4); */

   if (oss < 0) {

      oss = errno;

      fprintf(unit2,"SO_DEBUG setsockopt Error %ld, %s\n",oss,strerror(oss));

      return oss;

   }

/* Assign an address to an endpoint */

   oss = bind(epref[np],(struct sockaddr *)addrn,alen);

   if (oss < 0) {

      oss = errno;

      fprintf(unit2,"Bind Error = %ld, %s\n",oss,strerror(oss));

      return oss;

   }

/* Pass processor epref index and iocompletion to notifier */

   ioc[np][0] = np;

   ioc[np][1] = 1;

/* Clear read and send record pointers */

   ioc[np][2] = 0;

   ioc[np][3] = 0;

   return oss;

}

long etmsec(struct timeval *ptime) {

/* returns elapsed time since in milliseconds since ptime */

   const long tscale=1000;

   long oss, msec;

   struct timeval pstime, crtime;

/* calculate time elapsed in microseconds */

   oss = gettimeofday(&crtime,NULL);

   if (oss < 0)

      return -1;

   pstime = *ptime;

   msec = (crtime.tv_usec - pstime.tv_usec)/tscale;

   msec = msec + (crtime.tv_sec - pstime.tv_sec)*tscale;

   return msec;

}

long sbtusec(struct timeval *pstime, struct timeval *crtime) {

/* returns difference time in microseconds between crtime and pstime */

   const long tscale=1000000;

   long usec;

   usec = (crtime->tv_usec - pstime->tv_usec);

   usec = usec + (crtime->tv_sec - pstime->tv_sec)*tscale;

   return usec;

}

int ioresult(int *pblock) {

/* this function returns ioResult for asynchronous procedures

   input: pblock                                              */

   return pblock[1];

}

int checkesc(long stk) {

/* this procedure allows user to abort a procedure by checking for

   escape, Cmd-. or Ctrl-C keystrokes.  Calling EventAvail also

   permits an idle procedure to time-share and checks for Quit Events

   returns true if an escape event occurred.

   recent keyboard events not processed are not removed from event queue

   stk = maximum number of sleepTicks (sixtieths of a second) that

   application agrees to relinquish the processor if no events are 

   pending for it.

   input: stk                                                */

   return 0;

}

int MPI_Finalize(void) {

/* terminate MPI execution environment

local data                             */

   int ierror, i;

   long oss, state;

   struct sigaction act;

/* internal mpi common block

   nproc = number of real or virtual processors obtained

   epref = array of endpoint references for each participating node

   mapcomm = communicator map                                            */

/* internal common block for non-blocking messages

   monitor = (0,1,2) = (suppress,display,display & log) monitor messages

   curreq = request record for transmission parameters                   */

   ierror = 0;

/* MPI already finalized */

   if (nproc < 0) {

      ierror = 1;

      return ierror;

   }

   if (monitor==2)

      fprintf(unit2,"MPI_Finalize started\n");

/* Close providers */

   for (i = 0; i <= MAXS+1; i++) {

      if (epref[i]) {

/* Close a provider of any type */

         oss = close(epref[i]);

         if (oss < 0) {

            oss = errno;

            fprintf(unit2,"Close Error, i, oss = %d, %ld, %s\n",i,oss,

                    strerror(oss));

            ierror = oss;

         }

      }

   }

/* Revert signal action to default */

   act.sa_handler = SIG_DFL;

   sigemptyset(&act.sa_mask);

   act.sa_flags = 0;

   oss = sigaction(SIGIO,&act,NULL);

/* Close window for showing MPI message status */

   if (monitor > 0) {

      logmess(0,0,-1,0,0);

      delmess();

   }

/* Nullify nproc */

   nproc = -1;

/* Nullify communicator mappings */

   for (i = 0; i < MAXC; i++) {

      mapcomm[i][MAXS] = 0;

   }

/* Nullify endpoint references */

   for (i = 0; i <= MAXS+1; i++) {

      epref[i] = 0;

   }

/* check if any messages remain outstanding */

   state = 0;

   for (i = 0; i < MAXM; i++) {

      if (curreq[i][0])

         state += 1;

   }

   if (state > 0) {

      fprintf(unit2,"%ld message(s) never cleared\n",state);

      for (i = 0; i < MAXM; i++) {

         if (curreq[i][0]) {

            if (curreq[i][0]==(-1))

              fprintf(unit2," transmission mode = send\n");

            else if (curreq[i][0]==1)

               fprintf(unit2," transmission mode = receive\n");

            fprintf(unit2," destination/source = %d\n",curreq[i][1]);

            fprintf(unit2," communicator = %d\n",curreq[i][2]);

            fprintf(unit2," tag = %d\n",curreq[i][3]);

            fprintf(unit2," datatype = %d\n",curreq[i][4]);

            fprintf(unit2,"\n");

         }

      }

   }

   if (monitor==2)

      fprintf(unit2,"MPI_Finalize complete\n");

/* Delete file if empty */

   if (!fseek(unit2,0,SEEK_END)) {

      i = ftell(unit2);

      fclose(unit2);

      if (!i)

         remove("MPIerrs");

   }

   return ierror;

}

int MPI_Send(void* buf, int count, MPI_Datatype datatype, int dest,

             int tag, MPI_Comm comm) {

/* blocking standard mode send

   buf = initial address of send buffer

   count = number of entries to send

   datatype = datatype of each entry

   dest = rank of destination

   tag = message tag

   comm = communicator

   input: buf, count, datatype, dest, tag, comm

local data                                                       */

   int ierror;

   MPI_Request request;

   MPI_Status status;

   ierror = MPI_Isend(buf,count,datatype,dest,tag,comm,&request);

   ierror = MPI_Wait(&request,&status);

   return ierror;

}

int MPI_Recv(void* buf, int count, MPI_Datatype datatype, int source,

             int tag, MPI_Comm comm, MPI_Status *status) {

/* blocking receive

   buf = initial address of receive buffer

   count = maximum number of entries to receive

   datatype = datatype of each entry

   source = rank of source

   tag = message tag

   comm = communicator

   status = return status

   input: count, datatype, source, tag, comm

   output: buf, status

local data                                                       */

   int ierror;

   MPI_Request request;

   ierror = MPI_Irecv(buf,count,datatype,source,tag,comm,&request);

   ierror = MPI_Wait(&request,status);

   return ierror;

}

int MPI_Isend(void* buf, int count, MPI_Datatype datatype, int dest,

              int tag, MPI_Comm comm, MPI_Request *request) {

/* start a non-blocking send

   buf = initial address of send buffer

   count = number of entries to send

   datatype = datatype of each entry

   dest = rank of destination

   tag = message tag

   comm = communicator

   request = request handle

   input: buf, count, datatype, dest, tag, comm

   output: request

local data                                                           */

   int ierror, np, longw, i;

   long response;

/* internal mpi common block

   nproc = number of real or virtual processors obtained

   idproc = processor id

   epref = array of endpoint references for each participating node

   ioc = array of context pointers for notifier function

   sset = signal mask

   mapcomm = communicator map                                       */

/* internal common block for non-blocking messages

   monitor = (0,1,2) = (suppress,display,display & log) monitor messages

   curreq = request record for transmission parameters

   header = message envelope

   rwrec = read/write record for asynchronous messages

   mqueue = message request queue                                     */

   ierror = 0;

   if (dest==MPI_PROC_NULL) {

      *request = MPI_REQUEST_NULL;

      return ierror;

   }

/* Find space for record */

   i = -1;

L10: i += 1;

   if (i >= MAXM) {

      fprintf(unit2,"too many sends waiting, dest, tag = %d,%d,\n",

              dest,tag);

      *request = MPI_REQUEST_NULL;

      ierror = 14;

      writerrs("MPI_Isend: ",ierror);

      return ierror;

   }

   else if (curreq[i][0])

      goto L10;

/* Check for error conditions */

/* MPI not initialized        */

   if (nproc <= 0)

      ierror = 1;

/* Invalid comm */

   else if ((comm < 0) || (comm >= MAXC))

      ierror = 2;

/* Invalid count */

   else if (count < 0)

      ierror = 3;

/* Invalid destination */

   else if ((dest < 0) || (dest >= nproc)) {

      fprintf(unit2,"destination = %d\n",dest);

      ierror = 4;

   }

/* Invalid tag */

   else if (tag < (-1))

      ierror = 6;

/* Communicator errors */

   else {

      longw = mapcomm[comm][MAXS];

      np = mapcomm[comm][dest];

/* Communicator not mapped */

      if ((longw <= 0) || (longw > nproc))

         ierror = 2;

/* Invalid destination */

      else if ((dest < 0) || (dest >= longw)) {

         fprintf(unit2,"destination = %d\n",dest);

         ierror = 4;

      }

/* Invalid mapping */

      else if ((np < 0) || (np >= nproc)) {

         fprintf(unit2,"Invalid mapping, destination, node = %d,%d\n",dest,np);

         ierror = 2;

      }

   }

/* Handle errors */

   if (ierror) {

      writerrs("MPI_Isend: ",ierror);

      return ierror;

   }

/* Create header */

/* Iovec structure for header */

   header[i].iov[0].iov_base = (void *)&header[i].hdata[0];

   header[i].iov[0].iov_len = 16;

/* Save communicator */

   header[i].hdata[0] = comm;

/* Save tag */

   header[i].hdata[1] = tag;

/* Save datatype */

   header[i].hdata[2] = datatype;

/* Set destination id for selfsends */

   if (idproc==np)

      np = MAXS;

/* Set endpoint reference pointer */

   rwrec[i].ref = epref[np];

/* Reset iocompletion flag */

   rwrec[i].ioflag = 1;

/* Set pointer to header */

   rwrec[i].buf = (void *)&header[i].iov;

/* Set buffer length for header */

   rwrec[i].nbytes = 2;

/* Find buffer length for data */

   if ((datatype==MPI_INT) || (datatype==MPI_FLOAT))

      longw = 4*count;

   else if (datatype==MPI_DOUBLE)

      longw = 8*count;

   else if (datatype==MPI_BYTE)

      longw = count;

   else if ((datatype==MPI_2INT) || (datatype==MPI_FLOAT_INT)) 

      longw = 8*count;

   else if (datatype==MPI_DOUBLE_INT)

      longw = 16*count;

/* Invalid datatype */

   else {

      ierror = 7;

      writerrs("MPI_Isend: ",ierror);

      return ierror;

   }

/* Set pointer to data buffer */

   rwrec[i].sbuf = buf;

/* Set buffer lengths for data */

   rwrec[i].sln = longw;

   rwrec[i].len = -header[i].iov[0].iov_len;

/* Clear more flag */

   rwrec[i].flags = 0;

/* Clear next message flag */

   rwrec[i].nextm = 0;

/* Clear non-fatal error code */

   rwrec[i].nfatal = 0;

/* OTData structure for data */;

   header[i].iov[1].iov_base = rwrec[i].sbuf;

   header[i].iov[1].iov_len = longw;

/* Save length */

   header[i].hdata[3] = longw;

/* Obtain the current time stamp */

   response = gettimeofday(&rwrec[i].ts[0],NULL);

/* Limit notifications that can be sent to notifier */

   sigprocmask(SIG_BLOCK,&sset,NULL);

/* Append this message to send queue if necessary */

   if (ioc[np][3] > 0) {

      if (mqueue[np][1] > 0)

         rwrec[mqueue[np][1]-1].nextm = i + 1;

      else

         rwrec[ioc[np][3]-1].nextm = i + 1;

      mqueue[np][1] = i + 1;

/* Obtain the current time stamp */

      response = gettimeofday(&rwrec[i].ts[1],NULL);

      goto L30;

   }

/* First send 4 word header, then data */

      sndmsgf(i,np);

      response = ioresult((int *)&rwrec[i]);

/* Set pointer to current send record */

      if (response > 0)

         ioc[np][3] = i + 1;

/* Set iocompletion flag to error */

      else if (response < 0)

         ierror = response + 1024;

/* Allow Open Transport to resume sending events */

L30: sigprocmask(SIG_UNBLOCK,&sset,NULL);

/* Handle read and write errors */

   if (ierror) {

/* Write out readwrite record */

      rwstat(i,unit2);

      wqueue(unit2);

      for (i = 0; i < MAXM; i++) {

         if (curreq[i][0] != 0)

            rwstat(i,unit2);

      }

      writerrs("MPI_Send: ",ierror);

   }

/* Find actual destination */

   if (np==MAXS)

      np = idproc;

/* Log MPI message state change and display status */

   if (monitor > 0) 

      logmess(np,1,rwrec[i].sln,0,tag);

/* Save transmission mode as send */

   curreq[i][0] = -1;

/* Save destination/source id */

   curreq[i][1] = np;

/* Save communicator */

   curreq[i][2] = comm;

/* Save tag */

   curreq[i][3] = tag;

/* Save datatype */

   curreq[i][4] = datatype;

/* Assign request handle */

   *request = i;

   return ierror;

}

int MPI_Irecv(void* buf, int count, MPI_Datatype datatype, int source,

              int tag, MPI_Comm comm, MPI_Request *request) {

/* begin a non-blocking receive

   buf = initial address of receive buffer

   count = maximum number of entries to receive

   datatype = datatype of each entry

   source = rank of source

   tag = message tag

   comm = communicator

   request = request handle

   input: count, datatype, source, tag, comm

   output: buf, request

local data                                                            */

   int ierror, np, longw, i;

   long response;

/* internal mpi common block

   nproc = number of real or virtual processors obtained

   epref = array of endpoint references for each participating node

   ioc = array of context pointers for notifier function

   sset = signal mask

   mapcomm = communicator map                                         */

/* internal common block for non-blocking messages

   monitor = (0,1,2) = (suppress,display,display & log) monitor messages

   curreq = request record for transmission parameters

   header = message envelope

   rwrec = read/write record for asynchronous messages

   mqueue = message request queue                                     */

   ierror = 0;

   if (source==MPI_PROC_NULL) {

      *request = MPI_REQUEST_NULL;

      return ierror;

   }

/* Find space for record */

   i = -1;

L10: i += 1;

   if (i >= MAXM) {

      fprintf(unit2,"too many receives waiting, source, tag = %d,%d,\n",

              source,tag);

      *request = MPI_REQUEST_NULL;;

      ierror = 15;

      writerrs("MPI_Irecv: ",ierror);

      return ierror;

   }

   else if (curreq[i][0])

      goto L10;

/* Check for error conditions */

/* MPI not initialized        */

   if (nproc <= 0)

      ierror = 1;

/* Invalid comm */

   else if ((comm < 0) || (comm >= MAXC))

      ierror = 2;

/* Invalid count */

   else if (count < 0)

      ierror = 3;

/* Invalid source */

   else if ((source < 0) || (source >= nproc)) {

      if (source==MPI_ANY_SOURCE)

         fprintf(unit2,"MPI_ANY_SOURCE not supported\n");

      else

         fprintf(unit2,"source = %d\n",source);

      ierror = 5;

   }

/* Invalid tag */

   else if (tag < (-1))

      ierror = 6;

/* Communicator errors */

   else {

      longw = mapcomm[comm][MAXS];

      np = mapcomm[comm][source];

/* Communicator not mapped */

      if ((longw <= 0) || (longw > nproc))

         ierror = 2;

/* Invalid source */

      else if ((source < 0) || (source >= longw)) {

         fprintf(unit2,"source = %d\n",source);

         ierror = 4;

      }

/* Invalid mapping */

      else if ((np < 0) || (np >= nproc)) {

         fprintf(unit2,"Invalid mapping, source, node = %d,%d\n",source,np);

         ierror = 2;

      }

   }

/* Handle errors */

   if (ierror) {

      writerrs("MPI_Irecv: ",ierror);

      return ierror;

   }

/* Set header to undefined state */

   header[i].hdata[0] = -1;

   header[i].hdata[1] = -1;

   header[i].hdata[2] = -1;

/* Set endpoint reference pointer */

   rwrec[i].ref = epref[np];

/* Reset iocompletion flag for receive */

   rwrec[i].ioflag = 1;

/* Set pointer to header */

   rwrec[i].buf = (void *)&header[i].hdata[0];

/* Set buffer length for header */

   rwrec[i].nbytes = 16;

/* Set buffer length for data */

   if ((datatype==MPI_INT) || (datatype==MPI_FLOAT))

      longw = 4*count;

   else if (datatype==MPI_DOUBLE)

      longw = 8*count;

   else if (datatype==MPI_BYTE)

      longw = count;

   else if ((datatype==MPI_2INT) || (datatype==MPI_FLOAT_INT)) 

      longw = 8*count;

   else if (datatype==MPI_DOUBLE_INT)

      longw = 16*count;

/* Invalid datatype */

   else {

      ierror = 7;

      writerrs("MPI_Irecv: ",ierror);

      return ierror;

   }  

/* Set pointer to data buffer */

   rwrec[i].sbuf = buf;

/* Set buffer lengths */

   rwrec[i].sln = longw;

   rwrec[i].len = -rwrec[i].nbytes;

/* Clear more flag */

   rwrec[i].flags = 0;

/* Clear next message flag */

   rwrec[i].nextm = 0;

/* Clear non-fatal error code */

   rwrec[i].nfatal = 0;

/* Clear length */

   header[i].hdata[3] = 0;

/* Obtain the current time stamp */

   response = gettimeofday(&rwrec[i].ts[0],NULL);

/* Limit notifications that can be sent to notifier */

   sigprocmask(SIG_BLOCK,&sset,NULL);

/* Append this message to receive queue if necessary */

   if (ioc[np][2] > 0) {

      if (mqueue[np][0] > 0)

         rwrec[mqueue[np][0]-1].nextm = i + 1;

      else

         rwrec[ioc[np][2]-1].nextm = i + 1;

      mqueue[np][0] = i + 1;

/* Obtain the current time stamp */

      response = gettimeofday(&rwrec[i].ts[1],NULL);

      goto L40;

   }

/* First receive 4 word header, then data */

      rcvmsgf(i,np);

      response = ioresult((int *)&rwrec[i]);

/* Set pointer to current receive record */

      if (response > 0)

         ioc[np][2] = i + 1;

/* Set iocompletion flag to error */

      else if (response < 0)

         ierror = response + 1024;

/* Allow Open Transport to resume sending events */

L40: sigprocmask(SIG_UNBLOCK,&sset,NULL);

/* Handle read and write errors */

   if (ierror) {

/* Write out readwrite record */

      rwstat(i,unit2);

      wqueue(unit2);

      for (i = 0; i < MAXM; i++) {

         if (curreq[i][0] != 0)

            rwstat(i,unit2);

      }

      writerrs("MPI_Recv: ",ierror);

   }

/* Log MPI message state change and display status */

   if (monitor > 0) 

      logmess(np,2,rwrec[i].sln,0,tag);

/* Save transmission mode as receive */

   curreq[i][0] = 1;

/* Save destination/source id */

   curreq[i][1] = np;

/* Save communicator */

   curreq[i][2] = comm;

/* Save tag */

   curreq[i][3] = tag;

/* Save datatype */

   curreq[i][4] = datatype;

/* Assign request handle */

   *request = i;

   return ierror;

}

int MPI_Test(MPI_Request *request, int *flag, MPI_Status *status) {

/* check to see if a nonblocking send or receive operation has completed

   request = request handle

   flag = true if operation completed

   status = status object

   input: request

   output: request, flag, status

local data                                                            */

   int ierror, i, dest, source, slen, tag, rlen, rtag, nerr;

/* int j; */

   float ts;

   MPI_Comm comm, rcomm;

   MPI_Datatype datatype, rdatat;

/* mstime = maximum time (msec) to wait for message to start arriving

   mptime = maximum time (msec) to wait for next part of message      */

   static int mstime = 300000, mptime = 10000;

/* internal mpi common block

   nproc = number of real or virtual processors obtained

   ioc = array of context pointers for notifier function

   sset = signal mask                                               */

/* internal common block for non-blocking messages

   monitor = (0,1,2) = (suppress,display,display & log) monitor messages

   curreq = request record for transmission parameters

   header = message envelope

   rwrec = read/write record for asynchronous messages                   */

   ierror = 0;

/* Check for error conditions */

   i = *request;

/* MPI not initialized */

   if (nproc <= 0)

      ierror = 1;

/* Null request */

   else if (*request < 0) {

      *flag = 1;

      return 0;

   }

/* Invalid request handle */

   else if (i >= MAXM)

      ierror = 16;

   else if (curreq[i][0]==0)

      ierror = 16;

/* Handle errors */

   if (ierror) {

      status->error = ierror;

      writerrs("MPI_Test: ",ierror);

      return ierror;

   }

/* Set status to empty */

   status->source = -1;

   status->tag = -1;

   status->error = ierror;

   status->len = 0;

   status->type = 0;

   *flag = 0;

/* Check if data read or write has completed */

   if (ioresult((int *)&rwrec[i]) > 0) {

      if (checkesc(0)) {

         if (curreq[i][0] < 0)

            fprintf(unit2,"Send killed,dest,tag=%d,%d\n",

                    curreq[i][1],curreq[i][3]);

         else

            fprintf(unit2,"Receive killed,source,tag=%d,%d\n",

                    curreq[i][1],curreq[i][3]);

         ierror = -9;

         writerrs("MPI_Test: ",ierror);

         return ierror;

      }

/* Check Timeout */

      else {

/* Limit notifications that can be sent to notifier */

         sigprocmask(SIG_BLOCK,&sset,NULL);

/* Calculate time elapsed in milliseconds */

         nerr = etmsec(&rwrec[i].ts[1]);

/* Retry Send or Receive */

         if (nerr < mptime) {

/* Allow Open Transport to resume sending events */

            sigprocmask(SIG_UNBLOCK,&sset,NULL);

            return ierror;

         }

/* Check if message arrived during checkesc */

         nerr = ioresult((int *)&rwrec[i]);

         if (nerr <= 0) {

            fprintf(unit2,"Info: message arrived during checkesc\n");

            *flag = 1;

            goto L20;

         }

/* Try to determine cause of Timeout in Send */

         if (curreq[i][0] < 0) {

            dest = curreq[i][1];

            ts = .001*(float) etmsec(&rwrec[i].ts[0]);

            fprintf(unit2,"Send Timeout, %f sec, Retrying...\n",ts);

/* Debug information */

            fprintf(unit2,"destination=%d,size=%d,tag=%d\n",

                       dest,rwrec[i].sln,curreq[i][3]);

            if (idproc==dest)

               dest = MAXS;

/* Attempt to send next block of data */

            sndmsgf(i,dest);

            nerr = ioresult((int *)&rwrec[i]);

/* Non-fatal errors */

            if (rwrec[i].nfatal) {

               if ((rwrec[i].nfatal+1024)==EWOULDBLOCK)

                  fprintf(unit2,"Flow Control prevents sending data\n");

/* Do not wait more than 5 minutes for message to start sending */

               if (etmsec(&rwrec[i].ts[0]) >= mstime) {

                  nerr = rwrec[i].nfatal;

                  fprintf(unit2,"Send Retry failed\n");

                  ierror = nerr;

                  ioc[dest][3] = 0;

                  *flag = 1;

               }

            }

/* Fatal errors */

            else if (nerr < 0) {

               ierror = nerr + 1024;

               fprintf(unit2,"Send Error, oss = %d, %s\n",ierror,

                    strerror(ierror));

               ierror = nerr;

               *flag = 1;

            }

/* Data successfully sent */

            else {

/* Debug information */

               fprintf(unit2,"data block sent, current total = %d\n",

                       rwrec[i].len);

               if (!nerr)

                  *flag = 1;

            }

         }

/* Try to determine cause of Timeout in Receive */

         else {

            source = curreq[i][1];

            ts = .001*(float) etmsec(&rwrec[i].ts[0]);

            fprintf(unit2,"Receive Timeout, %f sec, Retrying...\n",ts);

/* Debug information */

            fprintf(unit2,"source=%d,size=%d,tag=%d\n",

                       source,rwrec[i].sln,curreq[i][3]);

/* Attempt to send next block of data */

            rcvmsgf(i,source);

            nerr = ioresult((int *)&rwrec[i]);

/* Non-fatal errors */

            if (rwrec[i].nfatal) {

               if ((rwrec[i].nfatal+1024)==EWOULDBLOCK)

                  fprintf(unit2,"Flow Control prevents accepting data\n");

/* Do not wait more than 5 minutes for message to start sending */

               if (etmsec(&rwrec[i].ts[0]) >= mstime) {

                  nerr = rwrec[i].nfatal;

                  fprintf(unit2,"Receive Retry failed\n");

                  ierror = nerr;

                  ioc[source][2] = 0;

                  *flag = 1;

               }

            }

/* Fatal errors */

            else if (nerr < 0) {

               ierror = nerr + 1024;

               fprintf(unit2,"Receive Error, oss = %d, %s\n",ierror,

                    strerror(ierror));

               ierror = nerr;

               *flag = 1;

            }

/* Data successfully received */

            else {

/* Debug information */

               fprintf(unit2,"data block received, current total = %d\n",

                       rwrec[i].len);

               if (!nerr)

                  *flag = 1;

            }

         }

/* Allow Open Transport to resume sending events */

L20:     sigprocmask(SIG_UNBLOCK,&sset,NULL);

         if (!(*flag))

            return ierror;

      }

   }

/* Data read or write has completed */

   else {

      nerr = ioresult((int *)&rwrec[i]);

      *flag = 1;

   }

/* Get requested length */

   slen = rwrec[i].sln;

/* Get actual length */

   rlen = rwrec[i].len;

/* Read current request record */

   tag = curreq[i][3];

/* Define length for MPI_Get_count */

   status->len = rlen;

/* Check for send errors */

   if (curreq[i][0] < 0) {

      dest = curreq[i][1];

/* Define type for MPI_Get_count */

      status->type = curreq[i][4];

/* Check for write errors */

      if (nerr < 0) {

         if (!ierror) {

            ierror = nerr + 1024;

            fprintf(unit2,"Send Error, oss = %d, %s\n",ierror,

                    strerror(ierror));

            fprintf(unit2,"dest, tag = %d,%d\n",dest,tag);

            ierror = nerr;

         }

      }

      else if (rlen != slen) {

         fprintf(unit2,"Send Length Error,dest,tag,requested/actual length=%d,%d,%d,%d\n",

                 dest,tag,slen,rlen);

         ierror = 8;

      }

/* Log MPI message state change and display status */

      if (monitor > 0) {

         if (checkesc(0)) {

            ierror = -9;

            writerrs("MPI_Test: ",ierror);

            return ierror;

         }

         else if (!ierror) {

/* Convert difference between time steps into microseconds */

            nerr = sbtusec(&rwrec[i].ts[0],&rwrec[i].ts[1]);

            logmess(dest,-1,rlen,nerr,tag);

         }

      }

      goto L30;

   }

/* Read current request record */

   source = curreq[i][1];

   comm = curreq[i][2];

   datatype = curreq[i][4];

/* Read header */

/* Get received tag from header */

   rtag = header[i].hdata[1];

/* Get received comm from header */

   rcomm = header[i].hdata[0];

/* Get received datatype from header */

   rdatat = header[i].hdata[2];

/* Define source, tag and type for MPI_Get_count */

   status->source = source;

   status->tag = rtag;

   status->type = rdatat;

/* Check for receive errors */

   if (nerr < 0) {

      if (!ierror) {

         ierror = nerr + 1024;

         fprintf(unit2,"Receive Error, oss = %d, %s\n",ierror,

                 strerror(ierror));

         fprintf(unit2,"source, tag = %d,%d\n",source,tag);

         ierror = nerr;

      }

   }

/* Length error */

   else if (rlen > slen) {

      fprintf(unit2,"Read Length Error, source, tag, requested/actual = %d,%d,%d,%d\n",

              source,tag,slen,rlen);

      fprintf(unit2,"Possibly attempting to receive data out of order\n");

      ierror = 13;

   }

/* Check for incomplete message, this should never be able to happen */

   else if (rlen < header[i].hdata[3]) {

      fprintf(unit2,"Incomplete Read, source, tag, requested/actual = %d,%d,%d,%d\n",    

              source,tag,header[i].hdata[3],rlen);

      ierror = 12;

   }

/* Comm error */

   else if (rcomm != comm) {

      fprintf(unit2,"Read Comm Error, source, tag, expected/received comm = %d,%d,%d,%d\n",

              source,tag,comm,rcomm);

      ierror = 9;

   }

/* Tag error */

   else if ((tag >= 0) && (rtag != tag)) {

      fprintf(unit2,"Read Tag Error, source, expected/received tag = %d,%d,%d\n",

              source,tag,rtag);

      fprintf(unit2,"Possibly attempting to receive data out of order\n");

      ierror = 10;

   }

/* Type error */

   else if (rdatat != datatype) {

      fprintf(unit2,"Read Type Error, source, tag, expected/received type = %d,%d,%d,%d\n",

              source,tag,datatype,rdatat);

      ierror = 11;

   }

/* Log MPI message state change and display status */

   if (monitor > 0) {

      if (checkesc(0)) {

         ierror = -9;

         writerrs("MPI_Test: ",ierror);

         return ierror;

      }

      else if (!ierror) {

/* Convert difference between time steps into microseconds */

         nerr = sbtusec(&rwrec[i].ts[0],&rwrec[i].ts[1]);

         logmess(source,-2,rlen,nerr,tag);

      }

   }

/* Store error code */

L30: status->error = ierror;

/* Nullify transmission mode */

   curreq[i][0] = 0;

/* Nullify request handle */

   *request = MPI_REQUEST_NULL;;

/* Handle read and write errors */

   if (ierror) {

/* Write out readwrite record */

      rwstat(i,unit2);

      wqueue(unit2);

      for (i = 0; i < MAXM; i++) {

         if (curreq[i][0] != 0)

            rwstat(i,unit2);

      }

      writerrs("MPI_Test: ",ierror);

   }

   return ierror;

}

void sndmsgf(MPI_Request request, int dest) {

/* send a message fragment

   request = request handle

   dest = rank of destination

local data                          */

   int i, np, nps;

   long response, oss;

/* internal mpi common block

   ioc = array of context pointers for notifier function            */

/* internal common block for non-blocking messages

   curreq = request record for transmission parameters

   rwrec = read/write record for asynchronous messages

   mqueue = message request queue                                     */

   i = request;

   np = dest;

/* Obtain the current time stamp */

   response = gettimeofday(&rwrec[i].ts[1],NULL);

/* Send data to a remote peer */

L10: if (rwrec[i].ioflag==1)

      response = writev(rwrec[i].ref,rwrec[i].buf,rwrec[i].nbytes);

   else

      response = send(rwrec[i].ref,rwrec[i].buf,rwrec[i].nbytes,

                      rwrec[i].flags);

   if (response < 0)

      response = errno - 1024;

/* Process data which has been sent */

   if (response > 0) {

/* Clear non-fatal error code */

      rwrec[i].nfatal = 0;

/* Set actual length sent */

      rwrec[i].len += response;

/* Check for incomplete header */

      if (rwrec[i].len < 0) {

         header[i].iov[0].iov_base = 

                     (void *)(((unsigned char *)header[i].iov[0].iov_base) + response);

         header[i].iov[0].iov_len -= response;

         goto L10;

      }

/* Check for incomplete data */

      else if (rwrec[i].sln > rwrec[i].len) {

/* Header is sent, readjust parameters to send data */

         if (rwrec[i].ioflag==1) {

            rwrec[i].buf = rwrec[i].sbuf;

            rwrec[i].nbytes = rwrec[i].sln;

            response -= header[i].iov[0].iov_len;

         }

/* Readjust buffer pointer */

         rwrec[i].buf = (void *)(((unsigned char *)rwrec[i].buf) + response);

         rwrec[i].nbytes -= response;

         rwrec[i].ioflag += 1;

      } 

/* Data is sent */

      else {

/* Obtain the current time stamp */

         response = gettimeofday(&rwrec[i].ts[1],NULL);

         rwrec[i].ioflag = 0;

/* Get next message if messages are queued */

         if (rwrec[i].nextm > 0) {

            i = rwrec[i].nextm;

            if (i==mqueue[np][1])

               mqueue[np][1] = 0;

            ioc[np][3] = i;

            i -= 1;

            goto L10;

         }

/* Clear pointer to current send record */

         else

            ioc[np][3] = 0;

      }

   }

/* Check for errors */

   else {

/* Check for EOF */

      if (response==0)

         response = -1;

      oss = response + 1024;

/* Process non-fatal errors */

      if (oss==EWOULDBLOCK)

         rwrec[i].nfatal = response;

/* Process fatal errors */

      else {

/* Find actual destination */

         nps = np;

         if (nps==MAXS)

            nps = idproc;

         if (response==(-1))

            fprintf(unit2,"Send Error, oss = %ld, EOF\n",response);

         else

            fprintf(unit2,"Send Error, oss = %ld, %s\n",oss,

                    strerror(oss));

         fprintf(unit2,"dest, tag = %d,%d\n",nps,header[i].hdata[1]);

/* Set iocompletion flag to error */

         rwrec[i].ioflag = response;

/* Clear pointer to current send record */

         ioc[np][3] = 0;

      }

   }

   return;

}

void rcvmsgf(MPI_Request request, int source) {

/* receive a message fragment

   request = request handle

   source = rank of source

local data                          */

   int i, np;

/* int j; */

   long response, oss;

/* internal mpi common block

   ioc = array of context pointers for notifier function            */

/* internal common block for non-blocking messages

   curreq = request record for transmission parameters

   rwrec = read/write record for asynchronous messages

   trash = trash bin for unwanted data

   mqueue = message request queue                                     */

   i = request;

   np = source;

/* Obtain the current time stamp */

   response = gettimeofday(&rwrec[i].ts[1],NULL);

/* Read data sent from a remote peer */

L10: response = recv(rwrec[i].ref,rwrec[i].buf,rwrec[i].nbytes,

                     rwrec[i].flags);

   if (response < 0)

      response = errno - 1024;

/* Process data which has arrived */

   if (response > 0) {

/* Clear non-fatal error code */

      rwrec[i].nfatal = 0;

/* Set actual length received */

      rwrec[i].len += response;

/* Check if all the data has arrived */

      if (rwrec[i].len < header[i].hdata[3]) {

/* Incomplete data */

         if (rwrec[i].nbytes > response) {

/* Readjust buffer pointer */

            rwrec[i].buf = (void *)(((unsigned char *)rwrec[i].buf) + response);

            rwrec[i].nbytes -= response;

            if (rwrec[i].len >= 0)

               rwrec[i].ioflag += 1;

         }

/* Header is received, readjust parameters to receive data */

         else if (rwrec[i].ioflag==1) {

            rwrec[i].buf = rwrec[i].sbuf;