src/Message/CommPVM.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
 *
 ***************************************************************************/

/*
 * CommPVM.cpp - member function for PVM-specific Communicate object,
 * which handles node-to-node communication in a parallel environment
 * using the PVM library.
 */

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

// include pvm header file
#include <pvm3.h>

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


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


// typedef used in allocating an array of char pointers
typedef char *charptr;

// static data to keep track of errors
static int numErrors = 0;


// constructor.  Starts up the virtual machine processes.
//      arguments: command-line args, and number of processes
// to start (if < 0, start the 'default' number, i.e. the number of
// hosts in a PVM 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
CommPVM::CommPVM(int argc , char** argv, int procs)
  : Communicate(argc, argv, procs) {
  TAU_PROFILE("CommPVM::CommPVM()", "void (int, char **, int)", TAU_MESSAGE);
  charptr *useargs = argv;

  // PVM specific stuff...
  int i, startNum, mytid, parent_tid, testnumhosts, added;
  int *child_ready, reported, rep_host, rep_tid;
  charptr *spawnargv = 0;

  // initialize useful info
  execName = NULL;
  tids = NULL;

  // Following group enrollment of everybody needed for barrier implement.:
  int ErrRet = pvm_joingroup("all_tasks");
  if (ErrRet==PvmSysErr) {
    ERRORMSG("CommPVM: pvmd not running!!" << endl);
  }
  else if (ErrRet==PvmBadParam) {
    ERRORMSG("CommPVM: bad group name!!" << endl);
  }
  else if (ErrRet==PvmDupGroup) {
    ERRORMSG("CommPVM: already joined this group!!" << endl);
  }

  // determine my tid number, and if this process is the parent process
  mytid = pvm_mytid();
  if(mytid >= 0) {
    // PVM is indeed running, get info about the virtual machine.

    //  Account for old PVM version on Exemplar
    struct pvmhostinfo *hostp;
    struct pvmtaskinfo *taskp;
    parent_tid = pvm_parent();
    pvm_setopt(PvmRoute, PvmRouteDirect);

    // get information about the virtual machine.
    // used to set number of hosts, and to spawn processes if this is the
    // master node.
    int narch;
    if(pvm_config(&TotalNodes, &narch, &hostp) != 0) {
      // error getting config info; go back to being single-node machine
      WARNMSG("CommPVM: cannot get virtual machine configuration.\n");
      WARNMSG("         Going to single-node status." << endl);
      TotalNodes = 1;
      parent_tid = (-1);
    }

    // print out the arguments being used, and make copy of executable name
    char *exnmptr = strrchr(useargs[0],'/');
    if(!exnmptr)
      exnmptr = useargs[0];
    else
      exnmptr++;
    execName = new char[strlen(exnmptr) + 1];
    strcpy(execName,exnmptr);

    // determine the proper number of processes to start.  We usually start
    // one process per node.  If there is just one node, and proc > 1, we
    // start 'proc' proceses on the same node.  Finally, if proc > 1 but
    // less than the total number of nodes, we start processes on just 'proc'
    // nodes (for some arch's, this may mean letting some nodes sit idle)
    if(procs > 0)
      startNum = procs;
    else
      startNum = TotalNodes;

    // make sure, if we are only starting one node
    if(startNum == 1)
      TotalNodes = 1;

    // if this is the first PVM communication object, add hosts to system
    tids = new int[startNum];
    if (parent_tid < 0) {
      // this is the parent process ... spawn others if necessary
      tids[0] = mytid;
      added = 1;
      myHost = 0;
      if(startNum > 1) {
        // For a 'normal' PVM setup, we start a job on each host, unless
        // we are starting up all the processes on the same node

        // first get the argument list
        if (argc > 1) {
          spawnargv = new charptr[argc];
          for(i=0; i < (argc - 1); i++)
            spawnargv[i] = useargs[i+1];
          spawnargv[argc - 1] = NULL;
        }

        // now start a job on each host (may not use ALL the hosts ...)
        // find the parent pvmd tid, so we do not start another job there
        int ntask;
        pvm_tasks(0, &ntask, &taskp);
        int pvmd_tid = (taskp[0]).ti_host;
        int checked_host_once = 0;

        // go to each host and spawn a process; do this round-robin
        int pn;
        for (i=0, pn=0; pn < startNum && added < startNum; pn++) {
          if (checked_host_once == 0 && pvmd_tid == (hostp[i]).hi_tid) {
            checked_host_once = 1;
          }
          else {
            INFOMSG("CommPVM: Starting job " << added + 1 << " on host '");
            INFOMSG((hostp[i]).hi_name << "' ... ");
            pvm_spawn(execName, spawnargv, PvmTaskHost,
                      (hostp[i]).hi_name, 1, tids + added);
            if(tids[added] < 0) {
              ERRORMSG("failed.  PVM Error = " << tids[added] << endl);
            }
            else {
              INFOMSG("new tid = " << tids[added] << endl);
              added++;
            }
          }
          i = ((i + 1) % TotalNodes);
        }

        // free storage used for sending out arguments
        delete [] spawnargv;

        // adjust TotalNodes to the actual number of hosts which responded
        TotalNodes = startNum = added;

        pvm_initsend(PvmDataRaw);
        pvm_pkint(&TotalNodes,1,1);
        pvm_pkint(tids,TotalNodes,1);
        pvm_mcast(tids+1,TotalNodes-1,COMM_HOSTS_TAG);

        pvm_initsend(PvmDataRaw);

        // wait for the spawned processes to report back that they're ready
        INFOMSG("CommPVM: Parent process waiting for children ..." << endl);
        child_ready = new int[TotalNodes];
        for(i=0; i < TotalNodes; child_ready[i++] = 0);
        reported = 1;           // since the parent is already ready
        while(reported < TotalNodes) {
          if(pvm_nrecv((-1),COMM_HOSTS_TAG) > 0) {
            pvm_upkint(&rep_host,1,1);
            pvm_upkint(&rep_tid,1,1);
            if(rep_host>=0 &&rep_host<TotalNodes &&!(child_ready[rep_host])) {
              child_ready[rep_host] = 1;
              reported++;
              INFOMSG("CommPVM: Child " << rep_host << " (tid = " << rep_tid);
              INFOMSG(") ready." << endl);
            }
            else {
              ERRORMSG("CommPVM: Error with child reporting to parent.  ");
              ERRORMSG("rep_host = " << rep_host << ", rep_tid = " << rep_tid);
              ERRORMSG(", child_ready[] = " << child_ready[rep_host] << endl);
            }
          }
        }
        delete [] child_ready;
        INFOMSG("CommPVM: Initialization complete!" << endl);
      }

    }
    else {              // this is a child process; get data from pops
      // get message from parent
      pvm_recv(parent_tid, COMM_HOSTS_TAG);
      pvm_upkint(&testnumhosts,1,1);
      if(testnumhosts != startNum) {
        WARNMSG("CommPVM: For tid " << mytid);
        WARNMSG(": Adjusting # of nodes from " << startNum << " to ");
        WARNMSG(testnumhosts << endl);
        startNum = testnumhosts;
      }
      pvm_upkint(tids, startNum, 1);
      TotalNodes = startNum;

      // find my host index
      for(i=1; i < startNum; i++)
        if(mytid == tids[i]) { myHost = i; break; }

      // check to see if we still need to be executing
      if(i >= startNum) {
        // this is a node which is not needed, so just exit for now
        pvm_exit();
      }

      // reply back to the parent we're ready to go
      pvm_initsend(PvmDataRaw);
      pvm_pkint(&myHost,1,1);
      pvm_pkint(&mytid,1,1);
      pvm_send(parent_tid,COMM_HOSTS_TAG);
    }

  }
  else {                // PVM is not running; make this a single node
    TotalNodes = 1;
    WARNMSG("CommPVM: PVM not running, going to single-node mode." << endl);
  }

  // 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
CommPVM::~CommPVM(void) {
  TAU_PROFILE("CommPVM::~CommPVM()", "void ()", TAU_MESSAGE);

  int i;                        // loop variable

  // 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) {
          WARNMSG("CommPVM: Found extra message from node " << node);
          WARNMSG(", tag " << tag << ": msg = " << *msg << endl);
        }
      } while (msg != 0);
    }
  }

  // if this is the last active CommunicatePVM object, clean up
  if (TotalNodes > 1) {
    // perform barrier to get all nodes ready to die
    mybarrier();

    // if this is the parent process, kill others
    if (myHost == 0) {
      INFOMSG("CommPVM: Parent process killing spawned processes ..." << endl);
      for (i = 1 ; i < TotalNodes; i++)
        pvm_kill(tids[i]);

      pvm_exit();               // and then exit the parent from pvm
      if(tids)
        delete [] tids;
      if(execName)
        delete [] execName;

    } else {
      // on client nodes, just wait to die
      if(tids)
        delete [] tids;
      if(execName)
        delete [] execName;

      while(1)
        sleep(1);               // just sit and wait for the end
    }
  }
}


// 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 CommPVM::pack_message(Message *msg, int tag) {
  TAU_PROFILE("CommPVM::pack_message()", "void (Message *, int)", TAU_MESSAGE);
  int mdata[3];                 //  array to store msg header info
  int nitems = msg->size();     //  Number of items in Message

  // put message header info into the buffer
  mdata[0] = tag;
  mdata[1] = myNode();
  mdata[2] = nitems;
  pvm_pkint(mdata, 3, 1);        // tag, sender, number of items
 
  // scan through the message items, and pack their data
  Message::iterator mbeg = msg->begin();
  Message::iterator mend = msg->end();
  for ( ; mbeg != mend; ++mbeg) {
    Message::MsgItem *msg = *mbeg;
    mdata[0] = msg->elemsize;
    mdata[1] = msg->bytesize;
    pvm_pkint(mdata, 2, 1);
    if (mdata[1] > 0)
      pvm_pkbyte((char *)(msg->item), mdata[1], 1);
  }
}


// unpack the current receive message into a Message,
// and look for a message with the given tag.  If a message is found which
// does not match the search criteria, it is put in the received queue.
// messages are in the format described above for 'pack_message'.
Message* CommPVM::unpack_message(int &node, int &tag) {
  TAU_PROFILE("CommPVM::unpack_message()", "Message * (int, int)", TAU_MESSAGE);
  int mdata[3];                 // array to store msg header info
  int j, nitems;                // loop variable, and # items in msg

  // get the tag, sender, and number of messages
  pvm_upkint(mdata, 3, 1);
  tag = mdata[0];
  node = mdata[1];
  nitems = mdata[2];

  // create data structure for this message
  Message *newmsg = new Message;

  // get all the items and add to the message
  for (j = 0; j < nitems; j++) {
    pvm_upkint(mdata, 2, 1);
    int elemsize = mdata[0];
    int bytesize = mdata[1];

    // for each item, allocate storage, and give to Message object as
    // storage that should NOT be copied, and that SHOULD be freed up
    // when the Message is deleted.
    if ( bytesize > 0  && elemsize > 0 ) {
      //      char *itemdata = new char[bytesize];
      // use malloc instead
      char* itemdata = (char*) malloc(bytesize);
      pvm_upkbyte(itemdata, bytesize, 1);
      newmsg->setCopy(false);
      newmsg->setDelete(true);
      newmsg->putmsg((void *)itemdata, elemsize, bytesize / elemsize );
    }
  }

  // return the new message
  return newmsg;
}


// 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 CommPVM::mysend(Message *msg, int node, int tag, int etag) {
  TAU_PROFILE("CommPVM::mysend()", "bool (Message *, int, int, int)", TAU_MESSAGE);
  int newbufid, errstat = (-1);

  // create a new buffer for the message
  newbufid = pvm_initsend(PvmDataRaw);
  if ( newbufid >= 0 )    {
    // pack the message data into the buffer
    pack_message(msg, tag);

    // send the message
    errstat = pvm_send(tids[node], etag);
  }
  else {
    ERRORMSG("CommPVM: Error in pvm_initsend = " << newbufid << endl);
    return false;
  }

  // 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 *CommPVM::myreceive(int& node, int& tag, int etag) {
  TAU_PROFILE("CommPVM::myreceive()", "Message * (int, int, int)", TAU_MESSAGE);
  int bufid, size, checknode, checktag;
  Message *newmsg = NULL;

  checknode = (node < 0 || node >= TotalNodes ? (-1) : tids[node]);
  checktag = etag;

  // clear out any current message
  if ((bufid = pvm_getrbuf()) > 0)
    pvm_freebuf(bufid);

  // get the message, if available
  bufid = pvm_probe(checknode, checktag);

  // check if there really was a message
  if (bufid > 0) {

    // yes there was .. get info about message
    if (pvm_bufinfo(bufid, &size, &checktag, &checknode) < 0) {
      ERRORMSG("CommPVM: myreceive shows msg, but ");
      ERRORMSG("cannot get info about this buffer." << endl);
      ERRORMSG("   id = " << bufid << endl);
    }
    else if (size <= 0) {
      ERRORMSG("CommPVM: received message has size " << size << endl);
    }
    else {
      // we have a valid message.  break it up into Messages,
      // and look for a message with a matching tag.
      bufid = pvm_recv(checknode, checktag);
      newmsg = unpack_message(node, tag);
      pvm_freebuf(bufid);
      numErrors = 0;
    }

  }
  else if (bufid < 0) {
    // an error has occurred
    ERRORMSG("CommPVM: myreceive cannot receive msg from node " << node);
    ERRORMSG(", tag " << tag << endl);

    if (MAX_PVM_ERRS > 0 && ++numErrors > MAX_PVM_ERRS) {
      ERRORMSG("Maximum number of PVM receive errors (" << numErrors);
      ERRORMSG(") exceeded.  PVM is hosed!!" << endl);
      exit(1);
    }
  }

  // return the message if it was found, or NULL if no message
  return newmsg;
}


// Synchronize all processors (everybody waits for everybody
// else to get here before returning to calling function).
// Uses PVM barrier for all procs.
void CommPVM::mybarrier(void) {
  TAU_PROFILE("CommPVM::mybarrier()", "void ()", TAU_MESSAGE);
  pvm_barrier("all_tasks", TotalNodes);
  return;
}
/***************************************************************************
 * $RCSfile: CommPVM.cpp,v $   $Author: adelmann $
 * $Revision: 1.1.1.1 $   $Date: 2003/01/23 07:40:28 $
 * IPPL_VERSION_ID: $Id: CommPVM.cpp,v 1.1.1.1 2003/01/23 07:40:28 adelmann Exp $ 
 ***************************************************************************/

---