src/Message/CommPM.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/CommPM.h"
#include "Message/Message.h"
#include "Utility/IpplInfo.h"
#include "Profile/Profiler.h"

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

// include PM header file. need to invoke PM library.
// #include <Pm.h>
// define interface directly, instead of using Pm.h.
// (because KCC couldn't parse Pm.h)
// The following definitions are written in Pm.h
typedef void* pmCtx;
extern "C" {
  int _pmGetSendBuf(pmCtx pmc, caddr_t *bufp, size_t length);
  int _pmSend(pmCtx pmc, int dst_node);
  int _pmSendDone(pmCtx pmc);
  int _pmReceive(pmCtx pmc, caddr_t *bufp, size_t *length);
  int _pmPutReceiveBuf(pmCtx pmc);
  extern pmCtx _pm_subnet[];
  extern int _pm_subnet_count;
};
#define PM_MTU                  (8192 + 32)

// include score header file. need to invoke sub-network functions.
#include <score.h>

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

// if an error occurs during myreceive more times than this, CommPM
// 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 size_of_MPI_INT; /* needed for tracing */

// sub-network for message passing using PM directly.
static pmCtx pm_network;

// 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
CommPM::CommPM(int& argc , char**& argv, int procs)
        : Communicate(argc, argv, procs) {
  TAU_PROFILE("CommPM::CommPM()", "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);
  }

  // allocate a sub-network for message passing using PM directly.
  if (_score_alloc_subnet()) {
    ERRORMSG("CommPM: Error with allocating sub-network.");
  }
  pm_network = _pm_subnet[_pm_subnet_count - 1];

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

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

  // determine the number of nodes running and my node number
  MPI_Comm_size(MPI_COMM_WORLD,&TotalNodes);
  MPI_Comm_rank(MPI_COMM_WORLD,&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, (MPI_Aint*)&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, MPI_COMM_WORLD);
      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("CommPM: 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, MPI_COMM_WORLD, &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("CommPM: Child " << rep_host << " ready." << endl);
       }
       else {
         ERRORMSG("CommPM: Error with child reporting to parent.  ");
         ERRORMSG("rep_host = " << rep_host);
         ERRORMSG(", child_ready[] = " << child_ready[rep_host] << endl);
       }
    }

    delete [] child_ready;
    INFOMSG("CommPM: 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("CommPM: Started job " << myHost << " on host `");
      INFOMSG(host_name <<  "'." << endl);
    }
    else {
      ERRORMSG("CommPM: failed" << endl);
    }

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

    // send back an acknowledgement
    MPI_Send(&myHost, 1, MPI_INT, 0, COMM_HOSTS_TAG, MPI_COMM_WORLD);
    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
CommPM::~CommPM(void) {
  TAU_PROFILE("CommPM::~CommPM()", "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) {
          WARNMSG("CommPM: 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, MPI_COMM_WORLD);
      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, MPI_COMM_WORLD, &stat);
    TAU_TRACE_RECVMSG(COMM_DIE_TAG, 0, 1*size_of_MPI_INT);
  }

  // release sub-network
  _score_free_subnet();

  MPI_Finalize();
}


struct PM_Message {
  int node;
  int tag;
};

#if 0

// 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 *CommPM::pack_message(Message *msg, int tag, int &buffsize) {
  TAU_PROFILE("CommPM::pack_message()", "(Message *, 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);
  return pos;
}
#endif

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

  int nerr = 0;

  //  printf("mysend\n");

  // calculate size of buffer
  int length;
  length = find_msg_length(*msg) + sizeof(PM_Message);

  //  printf("length = %d\n", length);

  // allocate storage for buffer
  PM_Message* msgbuf;
  int timeout_counter = 0;
  while (_pmGetSendBuf(pm_network, (caddr_t*)&msgbuf, length) == ENOBUFS) {
    timeout_counter++;
    if (timeout_counter > 10000000) {
      ERRORMSG("CommPM: _pmGetSendBuf TIMEOUT");
      timeout_counter = 0;
      nerr++;
    }
    // receive a message when the buffer couldn't be allocated.
    nerr += pickup_message();
  }

  //  printf("buff %x is allocated\n", msgbuf);
  // pack message
  msgbuf->node = myHost;
  msgbuf->tag = tag;
  //  printf("invoke fill_msg_buffer\n");
  fill_msg_buffer((void*) (&(msgbuf->tag) + 1), *msg, tag);
  //  printf("fill_msg_buffer done\n");

  // send message
  if (_pmSend(pm_network, node)) {
    ERRORMSG("CommPM: _pmSend Error");
    nerr++;
  }
  // receive message waiting the sending done.
  while (_pmSendDone(pm_network) == EBUSY) {
    nerr += pickup_message();
  }

  //  printf("mysend done\n");
  // return the succsess of the operation
  return (nerr == 0);
}

int CommPM::pickup_message(void) {
  int nerr = 0;
  int length;
  PM_Message* msgbuf;
  int error;
  Message* newmsg = 0;
  // int rec_tag; 
  int src_node, rec_size, rec_utag;
  void* rec_buff;

  // pickup message
  if (error = _pmReceive(pm_network, (caddr_t*)&msgbuf, (size_t*)&length)) {
    // no message is received
    if (error != ENOBUFS) {
      ERRORMSG("CommPM: _pmReceive Error (in pickup_message)");
      nerr++;
    }
  } else {
    // a message is received, unpack it
    TAU_TRACE_RECVMSG(rec_tag, src_node, rec_size);
    src_node = msgbuf->node;
    // rec_tag = msgbuf->tag;
    rec_size = length - sizeof(PM_Message);
    rec_buff = makebuffer(rec_size);
    //    bcopy((void*) (&(msgbuf->tag) + 1), rec_buff, rec_size);
    memcpy(rec_buff, (void*) (&(msgbuf->tag) + 1), rec_size);
    _pmPutReceiveBuf(pm_network);
    newmsg = unpack_message(src_node, rec_utag, rec_buff);
    newmsg->useCommunicate(this, rec_buff);

    // put message in my message queue
    if (add_msg(newmsg, src_node, rec_utag)) {
      newmsg = NULL; // reset message pointer
    }
  }

  // return the number of errors;
  return nerr;
}


// 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 *CommPM::myreceive(int& node, int& tag, int etag) {
  TAU_PROFILE("CommPM::myreceive()", "Message *(int, int, int)", TAU_MESSAGE);
  int error;
  PM_Message* msgbuf;
  int length;
  int rec_tag, src_node, rec_size, rec_utag;
  void* rec_buff;
  Message* newmsg = 0;

  // pickup message
  if (error = _pmReceive(pm_network, (caddr_t*)&msgbuf, (size_t*)&length)) {
    // no message is received
    if (error != ENOBUFS) {
      ERRORMSG("CommPM: _pmReceive Error (in myreceive)");
    } else {
      // no message is available
      DEBUGMSG(level2<<"CommPM: No Message Received to Match Request"<<endl);
    }
  } else {
    // a message is received, unpack it
    TAU_TRACE_RECVMSG(rec_tag, src_node, rec_size);
    src_node = msgbuf->node;
    rec_tag = msgbuf->tag;
    rec_size = length - sizeof(PM_Message);
    rec_buff = makebuffer(rec_size);
    //    bcopy((void*) (&(msgbuf->tag) + 1), rec_buff, rec_size);
    memcpy(rec_buff, (void*) (&(msgbuf->tag) + 1), rec_size);
    _pmPutReceiveBuf(pm_network);
    newmsg = unpack_message(src_node, rec_utag, rec_buff);
    newmsg->useCommunicate(this, rec_buff);
  }

  // Retrun the new Message, or NULL if no message available
  node = src_node;
  tag = rec_tag;
  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 CommPM::mybarrier(void) {
  TAU_PROFILE("CommPM::mybarrier()", "void ()", TAU_MESSAGE);

  MPI_Barrier(MPI_COMM_WORLD);
}


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

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


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

---