src/Message/CommMPI.cpp

Go to the documentation of this file.

// -*- C++ -*-
/***************************************************************************
 *
 * The IPPL Framework
 * 
 * This program was prepared by PSI. 
 * All rights in the program are reserved by PSI.
 * Neither PSI nor the author(s)
 * makes any warranty, express or implied, or assumes any liability or
 * responsibility for the use of this software
 *
 * Visit http://www.acl.lanl.gov/POOMS for more details
 *
 ***************************************************************************/

// -*- C++ -*-
/***************************************************************************
 *
 * The IPPL Framework
 * 
 *
 * Visit http://people.web.psi.ch/adelmann/ for more details
 *
 ***************************************************************************/

// include files
#include "Message/CommMPI.h"
#include "Message/Message.h"
#include "Utility/IpplInfo.h"
#include "Utility/PAssert.h"
#include "Profile/Profiler.h"

// include mpi header file
#include <mpi.h>

#include <string.h>
#include <stdlib.h>
#include <unistd.h>


// if an error occurs during myreceive more times than this, CommMPI
// will just exit.  Make it negative to totally disable checking for a
// maximum number of errors
#define MAX_MPI_ERRS    500


// static data to keep track of errors
static int numErrors = 0;
#if defined(IPPL_MPIXX)
static MPI_Aint size_of_MPI_INT; /* needed for tracing */
#else
static int size_of_MPI_INT; /* needed for tracing */
#endif

// temporary buffer used for speed
#define PSIZE 1024*16
#define PACKSIZE ((PSIZE)*sizeof(long))
static long mpipackbuf[PSIZE];



// constructor.  arguments: command-line args, and number of processes
// to start (if < 0, start the 'default' number, i.e. the number of
// hosts in a MPI virtual machine, the number of nodes in an O2K, etc)
// Note: The base-class constructor does not need the argument info or
// the number of nodes, it just by default sets the number of nodes=1
// The final argument indicates whether to run MPI_Init or not; IPPL
// may be run as another user of MPI, in a context where MPI_Init has
// already been called, in which case it can skip that step.
CommMPI::CommMPI(int& argc , char**& argv, int procs, bool mpiinit)
  : Communicate(argc, argv, procs), weInitialized(mpiinit)
{
  TAU_PROFILE("CommMPI::CommMPI()", "void (int, char **, int)", TAU_MESSAGE);
  int i, reported, rep_host, ierror, result_len;
  MPI_Status stat;
  char *currtok, *nexttok, *execname;

  // a little "string magic" to strip the absolute pathname off the executable
  currtok = strstr(argv[0],"/");
  if (!currtok) {
    execname = strdup(argv[0]);
  } else {
    currtok++;
    nexttok = strstr(currtok,"/");
    while (nexttok) {
      currtok = nexttok+1;
      nexttok = strstr(currtok,"/");
    }
    execname = strdup(currtok);
  }

  // initialize mpi
  if (weInitialized)
    MPI_Init(&argc, &argv);

  // restore original executable name without absolute path
  strcpy(argv[0],execname);

  // duplicate the MPI_COMM_WORLD communicator, so that we can use
  // a communicator that will not conflict with other users of MPI_COMM_WORLD
  MPI_Comm_dup(MPI_COMM_WORLD, &communicator);

  // determine the number of nodes running and my node number
  MPI_Comm_size(communicator,&TotalNodes);
  MPI_Comm_rank(communicator,&myHost);

  // make sure we do not have too many processes running
  if (procs > 0 && procs < TotalNodes) {
    // if this is a process that is beyond what we had requested, just exit
    if (myHost >= procs)
      Ippl::abort();
    TotalNodes = procs;
  }

  MPI_Type_size ( MPI_INT, &size_of_MPI_INT );
  if (myHost == 0) {    // this code is run by the master process
    // send a messages to each child node
    for (i = 1; i < TotalNodes; i++)
    {
      MPI_Send(&myHost, 1, MPI_INT, i, COMM_HOSTS_TAG, communicator);
      TAU_TRACE_SENDMSG(COMM_HOSTS_TAG, i, 1*size_of_MPI_INT);
    }

    // wait for the spawned processes to report back that they're ready
    int *child_ready = new int[TotalNodes];
    for (i = 0; i < TotalNodes; child_ready[i++] = 0);
    INFOMSG("CommMPI: Parent process waiting for children ..." << endl);
    reported = 1;               // since the parent is already ready
    while (reported < TotalNodes) {
      ierror = MPI_Recv(&rep_host, 1, MPI_INT, MPI_ANY_SOURCE,
          COMM_HOSTS_TAG, communicator, &stat);
      TAU_TRACE_RECVMSG(COMM_HOSTS_TAG, stat.MPI_SOURCE, 1*size_of_MPI_INT);
      if (rep_host >= 0 && rep_host < TotalNodes && !(child_ready[rep_host])) {
         child_ready[rep_host] = 1;
         reported++;
         INFOMSG("CommMPI: Child " << rep_host << " ready." << endl);
       }
       else {
         ERRORMSG("CommMPI: Error with child reporting to parent.  ");
         ERRORMSG("rep_host = " << rep_host);
         ERRORMSG(", child_ready[] = " << child_ready[rep_host] << endl);
       }
    }

    delete [] child_ready;
    INFOMSG("CommMPI: Initialization complete." << endl);

  }
  else {                        // this is a child process; get data from pops
    char host_name[MPI_MAX_PROCESSOR_NAME];
    ierror = MPI_Get_processor_name(host_name, &result_len);
    if (ierror >= 0) {
      INFOMSG("CommMPI: Started job " << myHost << " on host `");
      INFOMSG(host_name <<  "'." << endl);
    }
    else {
      ERRORMSG("CommMPI: failed" << endl);
    }

    // receive message from the master node
    int checknode;
    MPI_Recv(&checknode, 1, MPI_INT, 0, COMM_HOSTS_TAG, communicator,
             &stat);
    TAU_TRACE_RECVMSG(COMM_HOSTS_TAG, 0, 1*size_of_MPI_INT);
    if (checknode != 0)
      WARNMSG("CommMPI: Child received bad message during startup." << endl);

    // send back an acknowledgement
    MPI_Send(&myHost, 1, MPI_INT, 0, COMM_HOSTS_TAG, communicator);
    TAU_TRACE_SENDMSG(COMM_HOSTS_TAG, 0, 1*size_of_MPI_INT);
  }

  // set up the contexts and processes arrays properly
  if (TotalNodes > 1) {
    vector<int> proccount;
    proccount.push_back(1);
    for (i = 1; i < TotalNodes; i++) {
      Contexts.push_back(1);
      Processes.push_back(proccount);
    }
  }

}


// class destructor
CommMPI::~CommMPI(void)
{
  TAU_PROFILE("CommMPI::~CommMPI()", "void()", TAU_MESSAGE);
  int i, dieCode = 0;
  MPI_Status stat;

  // on all nodes, when running in parallel, get any extra messages not
  // yet received
  if (TotalNodes > 1) {
    int trial, node, tag;
    Message *msg;
    for (trial = 0; trial < 50000; ++trial) {
      do {
        node = COMM_ANY_NODE;
        tag = COMM_ANY_TAG;
        msg = myreceive(node, tag, COMM_SEND_TAG);
        if (msg != 0 && tag != IPPL_ABORT_TAG && tag != IPPL_EXIT_TAG) {
          WARNMSG("CommMPI: Found extra message from node " << node);
          WARNMSG(", tag " << tag << ": msg = " << *msg << endl);
        }
      } while (msg != 0);
    }
  }

  // broadcast a message to all other nodes to tell them to quit
  if (myNode() == 0) {
    // on master node, send out messages
    for (i = 1; i < TotalNodes; i++)
    {
      MPI_Send(&dieCode, 1, MPI_INT, i, COMM_DIE_TAG, communicator);
      TAU_TRACE_SENDMSG(COMM_DIE_TAG, i, 1*size_of_MPI_INT);
    }
  } else {
    // on client nodes, receive message
    MPI_Recv(&dieCode, 1, MPI_INT, 0, COMM_DIE_TAG, communicator, &stat);
    TAU_TRACE_RECVMSG(COMM_DIE_TAG, 0, 1*size_of_MPI_INT);
  }

  // delete the communicator we used
  MPI_Comm_free(&communicator);
  // if we did our own initialization, also do finalize operation.  But
  // if we did not initialize, skip the finalize as well
  
  /* 
     ada: it make no sense to call finalize from tha application,
     because the IPPL destructor is calles AFTER that, which causes:
     "0032-151 MPI is already finalized in string, task number" 
  */
  // if (weInitialized)
  MPI_Finalize();
}


// take the data from a Message object and pack it into the current send buf.
// each message is packed in this order:
//      tag, sending node, number of items             (3-int array)
//              type of item 1  (short)
//              size of item 1, in number of elements   (int)
//              item 1 data     (various)
//              ...
//              type of item N  (short)
//              size of item N, in number of elements   (int)
//              item N data     (various)
void *CommMPI::pack_message(Message *msg, int tag, int &buffsize, int node) {
  TAU_PROFILE("CommMPI::pack_message()", "(Message *, int, int, int)",
              TAU_MESSAGE);
  // calculate size of buffer
  buffsize = find_msg_length(*msg);

  // allocate storage for buffer
  void *pos = (buffsize > PACKSIZE) ? makebuffer(buffsize) : mpipackbuf;

  // pack message data and return the necessary pointer
  fill_msg_buffer(pos, *msg, tag, buffsize, node);
  return pos;
}


// send a message ... arguments are the Message itself, the
// destination node, the 'user' tag, and the 'encoding' tag.
// Messages should be sent via the underlying mechanism by using the
// encoding tag (one of the COMM_ tags),
// and should embed the information about what the user
// tag is in the data sent between nodes.  Return success.
bool CommMPI::mysend(Message *msg, int node, int tag, int etag) {
  TAU_PROFILE("CommMPI::mysend()", "bool (Message *, int, int, int)",
              TAU_MESSAGE);

  int nitems = msg->size(); 
  int errstat = (-1);
  int flag = false;
  MPI_Request request;
  MPI_Status status;

  MPI_Status rec_status;
  int src_node, rec_node, rec_tag, rec_size, rec_utag, bufid, rec_flag = 0;
  Message* newmsg = NULL;

  // pack the message data into the buffer
  int size;
  void *outbuffer = pack_message(msg, tag, size, node);

  // send the message (non-blocking)
  Inform dbgmsg("CommMPI", INFORM_ALL_NODES);
  //dbgmsg << "Sending MPI message of size " << size << " to node " << node;
  //dbgmsg << " with tag " << tag << "." << endl;
  errstat = MPI_Isend(outbuffer, size, MPI_BYTE, node, etag,
                      communicator, &request);
  TAU_TRACE_SENDMSG(etag, node, size);

  while (!flag) {
    if (!Ippl::retransmit()) {
      // get info about messages to be received
      bufid = MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, communicator,
                         &rec_flag, &rec_status);
      if ((bufid >= 0) && (rec_flag != 0) ) {
        // a message is available to be received
        src_node = rec_status.MPI_SOURCE;
        rec_tag = rec_status.MPI_TAG;
        MPI_Get_count(&rec_status, MPI_BYTE, &rec_size);
        //dbgmsg<<"Receiving MPI message of size " << rec_size << " from node ";
        //dbgmsg << src_node << "." << endl;
        if ( (rec_size >= 0) && (rec_tag >= 0) && (src_node >= 0) ) {
          // message is a valid one, so malloc the output buffer
          void *rec_buff = makebuffer(rec_size);

          // blocking receive, unpack message
          MPI_Recv(rec_buff, rec_size, MPI_BYTE, src_node, rec_tag,
                   communicator, &rec_status);
          TAU_TRACE_RECVMSG(rec_tag, src_node, rec_size);
          newmsg = unpack_message(rec_node, rec_utag, rec_buff);

          // if there was an error unpacking, then the message had a problem
          // and is invalid, so throw this one away
          if (newmsg == 0) {
            // free up the buffer
            cleanupMessage(rec_buff);

          } else {
            // tell the message to inform us when the buffer is finished
            newmsg->useCommunicate(this, rec_buff);

            // put message in my message queue
            add_msg(newmsg, rec_node, rec_utag);
          }

          // reset other receive information
          newmsg = NULL; // reset message pointer
          rec_flag = 0; // reset receive flag
        }
      }
    }

    // check for completion of send
    MPI_Test(&request, &flag, &status);
  }

  //  free up the send buffer
  if (size > PACKSIZE)
    freebuffer(outbuffer);

  // return the success of the operation
  return (errstat == 0);
}


// receive a message from the given node and user tag.  Return a NEW
// Message object if a message arrives, or NULL if no message available.
// node will be set to the node from which the message was sent.
// tag will be set to the 'user tag' for that message.
// etag is the 'encoding' tag, and must be one of the COMM_ tags. 
// Only message sent via the underlying mechanism with the
// given etag are checked.  When one is found, the user tag and sending
// node are extracted from the sent data.
// If node = COMM_ANY_NODE, checks for messages from any node.
// If tag = COMM_ANY_TAG, checks for messages with any user tag.
Message *CommMPI::myreceive(int& node, int& tag, int etag) {
  TAU_PROFILE("CommMPI::myreceive()", "Message *(int, int, int)", TAU_MESSAGE);
  int bufid, size, checknode, checktag, flag = false;
  Message *newmsg = 0;
  MPI_Status stat;

  checknode = (node < 0 || node >= TotalNodes ? MPI_ANY_SOURCE : node);
  checktag = etag;

  // get info about message
  bufid = MPI_Iprobe(checknode, checktag, communicator, &flag, &stat);
  if (bufid < 0) {
    // an error has occurred
    ERRORMSG("CommMPI: cannot receive msg from node " << checknode);
    ERRORMSG(", tag " << checktag << endl);

    if (MAX_MPI_ERRS > 0 && ++numErrors > MAX_MPI_ERRS) {
      ERRORMSG("Maximum number of MPI receive errors (" << numErrors);
      ERRORMSG(") exceeded. MPI is hosed!!" << endl);
      Ippl::abort();
    }
  }

  // if the message is actually available, see if we can get it now
  if (flag == true) {
    MPI_Get_count(&stat,MPI_BYTE,&size);
    if (size < 0) {
      ERRORMSG("CommMPI: received message has size " << size << endl);
    } else if ((stat.MPI_TAG != checktag) || (stat.MPI_TAG < 0)) {
      ERRORMSG("CommMPI: received message with invalid tag ");
      ERRORMSG(stat.MPI_TAG << endl);
    } else if (stat.MPI_SOURCE < 0) {
      ERRORMSG("CommMPI: received message from invalid source ");
      ERRORMSG(stat.MPI_SOURCE << endl);
    } else {
      checknode = stat.MPI_SOURCE;
      checktag = stat.MPI_TAG;

      // malloc the receive buffer
      void *outbuff = makebuffer(size);

      // blocking receive
      //Inform dbgmsg("CommMPI", INFORM_ALL_NODES);
      //dbgmsg << "Receiving MPI message of size " << size << " from node ";
      //dbgmsg << checknode << "." << endl;
      MPI_Recv(outbuff, size, MPI_BYTE, checknode, checktag,
               communicator, &stat);
      TAU_TRACE_RECVMSG(checktag, checknode, size);
      newmsg = unpack_message(node, tag, outbuff);

      // if there was an error unpacking, then the message had a problem
      // and is invalid, so throw this one away
      if (newmsg == 0) {
        // free up the buffer
        cleanupMessage(outbuff);
      } else {
        // tell the message to notify us when its done with the buffer
        newmsg->useCommunicate(this, outbuff);
      }

      // zero out the count of MPI-specific errors
      numErrors = 0;
    }

  } else {
    // no message is available
    DEBUGMSG(level2<<"CommMPI: No Message Received to Match Request"<<endl);
  }

  // return the new Message, or NULL if no message available
  return newmsg;
}


// Synchronize all processors (everybody waits for everybody
// else to get here before returning to calling function).
// Uses MPI barrier for all procs 
void CommMPI::mybarrier(void) {
  TAU_PROFILE("CommMPI::mybarrier()", "void ()", TAU_MESSAGE);

  MPI_Barrier(communicator);
}


// resent a message buffer that has been previously packed and copied
// into the provided buffer.  Return success.
bool CommMPI::resend(void *buf, int buffsize, int node, int etag) {
  TAU_PROFILE("CommMPI::resend()", "void (void *, int, int, int)",
              TAU_MESSAGE);

  //Inform dbgmsg("CommMPI::resend", INFORM_ALL_NODES);
  //dbgmsg << "About to resend buffer of size " << buffsize << " to node ";
  //dbgmsg << node << " with etag = " << etag << endl;

  // this will only work if we're sending to another node
  PInsist(node != myNode(), "Can only retransmit to other nodes");

  // send the buffer out
  MPI_Request request;
  int errstat = MPI_Isend(buf, buffsize, MPI_BYTE, node, etag,
                          communicator, &request);
  TAU_TRACE_SENDMSG(etag, node, size);

  int flag = false;
  MPI_Status status;
  while (!flag) {
    // check for completion of send
    MPI_Test(&request, &flag, &status);
  }

  // return the success of the operation
  return (errstat == 0);
}
 

// clean up after a Message has been used (called by Message).
void CommMPI::cleanupMessage(void *d) {
  TAU_PROFILE("CommMPI::cleanupMessage()", "void (void *)", TAU_MESSAGE);

  // need to free the allocated storage
  freebuffer(d);
}


/***************************************************************************
 * $RCSfile: CommMPI.cpp,v $   $Author: adelmann $
 * $Revision: 1.1.1.1 $   $Date: 2003/01/23 07:40:28 $
 * IPPL_VERSION_ID: $Id: CommMPI.cpp,v 1.1.1.1 2003/01/23 07:40:28 adelmann Exp $ 
 ***************************************************************************/

---