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

// Communicate - common member functions for Communicate object.
// On-node traffic
// is handle here and architecture specific routines are called for off-node
// traffic.  This is the base class for all comm-lib-specific subclasses.

// include files
#include "Message/Communicate.h"
#include "Message/Message.h"
#include "Message/CRC.h"
#include "PETE/IpplExpressions.h"
#include "Profile/Profiler.h"
#include "Utility/IpplInfo.h"
#include "Utility/IpplStats.h"
#include "Utility/RandomNumberGen.h"
#include "Utility/PAssert.h"
#include <stdio.h>


// print summary of this class to the given output stream
ostream& operator<<(ostream& o, const Communicate& c)
{
  TAU_PROFILE("operator<<()", "ostream (ostream, Communicate)", 
              TAU_MESSAGE | TAU_IO);

  o << "Parallel communication method: " << c.name() << "\n";
  o << "  Total nodes: " << c.getNodes() << ", Current node: ";
  o << c.myNode() << "\n";
  o << "  Queued received messages: ";
  o << c.getReceived() <<"\n";

  return o;
}


// 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 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
// Also note: the derived classes should erase Contexts and Processes, and
// put in the proper values.
Communicate::Communicate(int, char **, int)
  : nextMsgNum(1)
{
  TAU_PROFILE("Communicate::Communicate()", "void (int, char **, int)",
              TAU_MESSAGE);

  // initialize data for Communicate
  TotalNodes = 1;
  myHost = 0;
  ErrorStatus = COMM_NOERROR;
  Contexts.push_back(1);
  Processes.push_back(Contexts); // using Contexts is just convenient here
}


// Destructor.  Nothing to do at present.
Communicate::~Communicate(void)
{
  TAU_PROFILE("Communicate::~Communicate()", "void ()", TAU_MESSAGE); 

  // delete the cached messages
  SentCache_t::iterator cachei = sentMsgCache.begin();
  for ( ; cachei != sentMsgCache.end(); ++cachei)
    (*cachei).second.freebuf();
}


// Add a new on-node message to the linked list.  Return success.
bool Communicate::add_msg(Message *msg, int node, int tag)
{
  TAU_PROFILE("Communicate::add_msg()", "bool (Message *, int, int)",
              TAU_MESSAGE);

  recMsgList.push_back(MessageData(node, tag, msg));
  return true;
}


// Looks for a message in the message queue from the specified node
// and tag.  This understands wildcards for node and tag.
// Returns a pointer to the Message object found, and sets node and
// tag equal to the proper values.  Also, this will remove the item from
// the queue.
Message* Communicate::find_msg(int& node, int& tag)
{
  TAU_PROFILE("Communicate::find_msg()", "Message* (int&, int&)",
              TAU_MESSAGE);

  // just find the first message that meets the criteria
  vector<MessageData>::iterator qi   = recMsgList.begin();
  vector<MessageData>::iterator qend = recMsgList.end();
  for ( ; qi != qend ; ++qi) {
    if ((node == COMM_ANY_NODE || (*qi).node == node) &&
        (tag  == COMM_ANY_TAG  || (*qi).tag  == tag)) {
      node = (*qi).node;
      tag = (*qi).tag;
      Message *retval = (*qi).msg;
      recMsgList.erase(qi);
      //INCIPPLSTAT(incMessageReceived);
      //INCIPPLSTAT(incMessageReceivedFromQueue);
      return retval;
    }
  }

  // if we're here, no message was found
  return 0;
}


// Default version of virtual send function ... here, does nothing.
bool Communicate::mysend(Message *, int, int, int)
{
  TAU_PROFILE("Communicate::mysend()", "bool (Message *, int, int, int)",
              TAU_MESSAGE);

  // just return false, since we cannot send a message with this function
  return false;
}


// Default version of virtual receive function ... here, does nothing.
Message* Communicate::myreceive(int&, int&, int)
{
  TAU_PROFILE("Communicate::myreceive()", "Message* (int, int, int)",
              TAU_MESSAGE);

  // just return NULL, since we cannot find a message with this function
  return 0;
}


// Default version of virtual barrier function ... here, does nothing.
void Communicate::mybarrier(void)
{
  TAU_PROFILE("Communicate::mybarrier()", "void ()", TAU_MESSAGE);

  // just return NULL, since we cannot find a message with this function
  return;
}


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

  // just return false, since we cannot resend a message with this function
  return false;
}


// Send data to the given node, with given tag.  If delmsg==true, the
// message will be deleted after it is sent, otherwise it will be left alone.
bool Communicate::send(Message *msg, int node, int tag, bool delmsg)
{
  TAU_PROFILE("Communicate::send()", "bool (Message *, int, int, bool)",
              TAU_MESSAGE);

  bool retval;

  // process list of resend requests
  //process_resend_requests();

  // check for problems ...
  if ( node < 0 || node >= getNodes() || tag < 0 ) {
    ERRORMSG("Communicate: illegal send node " << node << endl);
    ErrorStatus = COMM_ERROR;
    return false;
  }

  // if the message is addressed to this node, put it in the local receive
  // queue immediately
  if ( node == myNode() ) {
    retval = add_msg(msg, node, tag);
    //INCIPPLSTAT(incMessageSent);
    //INCIPPLSTAT(incMessageSentToSelf);
  } else {
    // the message must be sent elsewhere ... call the proper function
    retval = mysend(msg, node, tag, COMM_SEND_TAG);

    // if the send was successful, delete the message if requested
    if (retval) {
      //INCIPPLSTAT(incMessageSent);
      //INCIPPLSTAT(incMessageSentToOthers);
      if (delmsg)
        delete msg;
    }
  }

  // set error code
  ErrorStatus = (retval != 0 ? COMM_NOERROR : COMM_NOSEND);

  // return the success of the operation
  return retval;
}


// Receive data from another node.  Returns newly created Message object
// with received message, or NULL if no message is available.
// If node == COMM_ANY_NODE, this will receive the next message with the given
// tag from any node.
// If tag == COMM_ANY_TAG, this will receive the next message with
// any tag from the given node.  If both are wildcards, this will receive the
// next message, period.  node and tag are passed by reference; if either
// is a wildcard, and a message is received, they are changed to their actual
// values.
// Messages are searched for in this order (if node == COMM_ANY_NODE) :
//      1. Pending in network
//      2. In receive queue
Message* Communicate::receive(int& node, int& tag)
{
  TAU_PROFILE("Communicate::receive()", "Message* (int, int)", TAU_MESSAGE);

  //Inform dbgmsg("Comm::receive", INFORM_ALL_NODES);
  //dbgmsg << "Doing receive from node " << node << ", tag " << tag << endl;

  // process list of resend requests
  process_resend_requests();

  // do a check for a message from another node
  //dbgmsg << "Checking for queued message ..." << endl;
  Message *msg = find_msg(node, tag);
  //dbgmsg << "Found one? " << (msg != 0 ? "yes" : "no") << endl;

  if (msg == 0 && myNode() != node) {
    int checknode = node;
    int checktag = tag;
    //INCIPPLSTAT(incMessageReceiveChecks);
    //dbgmsg << "Checking for remote message ..." << endl;
    if ((msg = myreceive(checknode, checktag, COMM_SEND_TAG)) != 0) {
      // see if the message matches our criteria for searching
      //dbgmsg << "Message found from node " << checknode << " with tag ";
      //dbgmsg << checktag << endl;
      if ((node != COMM_ANY_NODE && node != checknode) ||
          (tag  != COMM_ANY_TAG  && tag  != checktag )) {
        // the message does not match; queue it and report no msg found
        //dbgmsg << "But it's not what we want." << endl;
        add_msg(msg, checknode, checktag);
        msg = 0;
      } else {
        // the message matches; save the node and tag and return the msg
        //dbgmsg << "And it is what we want!" << endl;
        node = checknode;
        tag = checktag;
        //INCIPPLSTAT(incMessageReceived);
        //INCIPPLSTAT(incMessageReceivedFromNetwork);
      }
    } else {
      //INCIPPLSTAT(incMessageReceiveChecksFailed);
    }
  }

  // set error code
  ErrorStatus = (msg != 0 ? COMM_NOERROR : COMM_NORECEIVE);

  // return the message, or NULL if none was found
  return msg;
}


// A blocking version of receive.
Message *Communicate::receive_block(int& node, int &tag)
{
  TAU_TYPE_STRING(taustr, "Message *(int, int)" );
  TAU_PROFILE("Communicate::receive_block()", taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(findtimer, "  Comm::recblock::find", taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(looptimer, "  Comm::recblock::loop", taustr, TAU_MESSAGE);

  // process list of resend requests
  process_resend_requests();

  //Inform dbgmsg("Comm::receive_block", INFORM_ALL_NODES);
  //dbgmsg << "Doing blocking receive from node " << node;
  //dbgmsg << ", tag " << tag << endl;

  // If we haven't already found a message, check the local messages
  //dbgmsg << "Checking for queued message ..." << endl;
  TAU_PROFILE_START(findtimer);
  Message *msg = find_msg(node, tag);
  TAU_PROFILE_STOP(findtimer);
  //dbgmsg << "Found one? " << (msg != 0 ? "yes" : "no") << endl;

  // keep checking for remote msgs until we get one
  TAU_PROFILE_START(looptimer);
  if (myNode() != node) {
    while (msg == 0) {
      // process list of resend requests
      process_resend_requests();

      int checknode = node;
      int checktag = tag;
      //INCIPPLSTAT(incMessageReceiveChecks);
      //dbgmsg << "Checking for remote message ..." << endl;
      if ((msg = myreceive(checknode, checktag, COMM_SEND_TAG)) != 0) {
        // see if the message matches our criteria for searching
        //dbgmsg << "Message found from node " << checknode << " with tag ";
        //dbgmsg << checktag << endl;
        if ((node != COMM_ANY_NODE && node != checknode) ||
            (tag  != COMM_ANY_TAG  && tag  != checktag )) {
          // the message does not match; queue it and report no msg found
          //dbgmsg << "But it's not what we want." << endl;
          add_msg(msg, checknode, checktag);
          msg = 0;
        } else {
          // the message matches; save the node and tag and return the msg
          //dbgmsg << "And it is what we want!" << endl;
          node = checknode;
          tag = checktag;
          //INCIPPLSTAT(incMessageReceived);
          //INCIPPLSTAT(incMessageReceivedFromNetwork);
        }
      } else {
        //INCIPPLSTAT(incMessageReceiveChecksFailed);
        if (Ippl::retransmit())
          msg = find_msg(node, tag);
      }
    }
  }
  TAU_PROFILE_STOP(looptimer);

  // If we're on just one node, and we did not find a message, this is
  // a big problem.
  PInsist(!(myNode() == node && msg == 0),
          "Local message not found in Communicate::receive_block!!");

  // set error code
  ErrorStatus = (msg != 0 ? COMM_NOERROR : COMM_NORECEIVE);

  // return the message, or NULL if none was found
  return msg;
}


// Broadcast the given message to ALL nodes, including this node.
// Return number of nodes sent to.
// Arguments are the Message, and the tag for the message.
int Communicate::broadcast_all(Message *msg, int tag)
{
  TAU_PROFILE("Communicate::broadcast_all()", "int (Message *, int)",
              TAU_MESSAGE);
  int i;                        // loop variable

  // send message to all other nodes
  for (i=(getNodes() - 1); i >= 0; i--) {
    if (i != myNode()) {
      mysend(msg, i, tag, COMM_SEND_TAG);
      //INCIPPLSTAT(incMessageSent);
      //INCIPPLSTAT(incMessageSentToOthers);
    }
  }

  // send message to this node; since we do this, don't need to delete msg
  add_msg(msg, myNode(), tag);
  //INCIPPLSTAT(incMessageSent);
  //INCIPPLSTAT(incMessageSentToSelf);

  return getNodes();
}


// Broadcast the given message to all OTHER nodes, but not this node.
// Return number of nodes sent to.
// Arguments are the Message, and the tag for the message, and whether
// we should delete the given message object.
int Communicate::broadcast_others(Message *msg, int tag, bool delmsg)
{
  TAU_PROFILE("Communicate::broadcast_others()", "int (Message *, int, bool)",
              TAU_MESSAGE);
  int i;                        // loop variable

  // send message to all other nodes
  for (i=(getNodes() - 1); i >= 0; i--) {
    if (i != myNode()) {
      mysend(msg, i, tag, COMM_SEND_TAG);
      //INCIPPLSTAT(incMessageSent);
      //INCIPPLSTAT(incMessageSentToOthers);
    }
  }

  // delete message
  if (delmsg)
    delete msg;

  return getNodes() - 1;
}


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

  mybarrier();
  //INCIPPLSTAT(incBarriers);
}


// clean up after a Message has been used (called by Message).  By
// default, does nothing.
void Communicate::cleanupMessage(void *) { }


// calculate how big the buffer must be to send the given message
int Communicate::find_msg_length(Message &msg)
{
  TAU_PROFILE("Communicate::find_msg_length()", "int (Message)",
              TAU_MESSAGE);

  static const int longsize = wordround(sizeof(MsgNum_t));
  static const int intsize4 = wordround(4 * sizeof(int));
  static const int intsize2 = wordround(2 * sizeof(int));

  // the message contains a long and three integers at the start with the
  // msg num, node, tag, and number of items
  int buffsize = longsize + intsize4;

  // now include the sizes of the elements themselves.  For each item,
  // we also include two integers with size information.
  int nitems = msg.size();
  for (int i=0; i < nitems; ++i)
    buffsize += (intsize2 + wordround(msg.item(i).numBytes()));

  // if checksums are to be performed, add in space for the 32-bit checksum
  if (Ippl::useChecksums())
    buffsize += sizeof(CRCTYPE);

  return buffsize;
}


// put data from the given Message into the given buffer
void Communicate::fill_msg_buffer(void *buffer, Message &msg, int tag,
                                  int bufsize, int node)
{
  TAU_PROFILE("Communicate::fill_msg_buffer()",
              "void (void *, Message, int, int, int)", TAU_MESSAGE);

  void *pos = buffer;             // location in buffer to pack data
  int nitems = msg.size();        // Number of items in Message
  int mdata[4];                   // Array to store msg header info
  MsgNum_t mnum = (nextMsgNum++); // Message ID

  //Inform dbgmsg("***Communicate::fill_msg_buffer", INFORM_ALL_NODES);
  //dbgmsg << "Preparing to send out message " << mnum;
  //dbgmsg << " with tag " << tag << " of size " << bufsize << endl;

  // we must make sure to zero out the buffer if we're using checksums,
  // so that random data values do not occur in the spaces where word
  // alignment padding is used
  if (Ippl::useChecksums())
    memset(pos, 0, bufsize);

  // put message ID info into the buffer
  pack(&mnum, pos, sizeof(MsgNum_t));

  // put message header info into the buffer
  mdata[0] = tag;
  mdata[1] = myNode();
  mdata[2] = nitems;
  mdata[3] = bufsize;
  pack(mdata, pos, 4*sizeof(int));

  // finally pack in the data
  for (int i=0; i < nitems; ++i) {
    Message::MsgItem &msgitem = msg.item(i);
    mdata[0] = msgitem.numElems();
    mdata[1] = msgitem.numBytes();
    pack(mdata, pos, 2*sizeof(int));
    if (mdata[1] > 0)
      pack(msgitem.data(), pos, mdata[1]);
  }

  // if checksums are on, find the checksum and append it to the buffer
  if (Ippl::useChecksums()) {
    // calculate the crc
    int crcsize = bufsize - sizeof(CRCTYPE);
    CRCTYPE crcval = crc(buffer, crcsize);

    // append it to the end of the buffer
    *(static_cast<CRCTYPE *>(pos)) = crcval;

    // if we're trying to retransmit, cache the message
    if (Ippl::retransmit()) {
      if (tag != IPPL_RETRANSMIT_TAG && tag != IPPL_MSG_OK_TAG) {
        //dbgmsg << "Adding message " << mnum << " of size " << bufsize;
        //dbgmsg << " with tag " << tag << " to sent cache." << endl;
        add_to_send_cache(buffer, mnum, bufsize, node);
      } else {
        //dbgmsg << "NOT adding msg with tag " << tag << " to cache" << endl;
      }
    }
  }

  ADDIPPLSTAT(incMessageBytesSent,bufsize);
}


// get data out of a buffer and create a Message
Message* Communicate::unpack_message(int &node, int &tag, void *buffer)
{
  TAU_PROFILE("Communicate::unpack_message()",
              "Message* (int, int, long, void*)", TAU_MESSAGE);

  Message *newmsg = 0;

  // pos will always point to the next location in the buffer to get data
  void *pos = buffer;

  // get the message ID number
  MsgNum_t mnum = *(static_cast<MsgNum_t *>(pos));
  addwordround(pos, sizeof(MsgNum_t));

  // get the tag, sender, and number of messages
  int *mdata = static_cast<int *>(pos);
  tag = mdata[0];
  node = mdata[1];
  int nitems = mdata[2];
  int bufsize = mdata[3];
  addwordround(pos, 4*sizeof(int));

  //WARNMSG("Received message " << mnum << " from node " << node);
  //WARNMSG(" with tag " << tag << " of size " << bufsize << endl);

  // check for special tags, to abort, retransmit, or just receive
  if (tag == IPPL_ABORT_TAG) {
    ERRORMSG("Stopping due to abort request sent from node " << node << endl);
    ::abort();

  } else if (tag == IPPL_EXIT_TAG) {
    ERRORMSG("Exiting due to exit request sent from node " << node << endl);
    ::exit(1);

  } else if (tag == IPPL_RETRANSMIT_TAG) {
    // get the retransmit message number and tag out of the current buffer
    unpack_retransmission_request(nitems, pos);

  } else if (tag == IPPL_MSG_OK_TAG) {
    // clear out the messages that this message lists are OK to be deleted
    clear_ok_messages(nitems, pos);

  } else {
    // this is just a regular message

    // do checksum comparison, if checksums are on
    if (Ippl::useChecksums()) {
      // calculate the crc
      int crcsize = bufsize - sizeof(CRCTYPE);
      CRCTYPE crcval = crc(buffer, crcsize);

      // as a test, randomly change crcval
      //if (IpplRandom() < 0.1)
      //  crcval += 1;

      // compare this crc to the original one
      void *origloc = static_cast<void *>(static_cast<char *>(buffer)+crcsize);
      CRCTYPE origcrc = *(static_cast<CRCTYPE *>(origloc));
      if (crcval != origcrc) {
        ERRORMSG("Failed CRC check (" << crcval << " != " << origcrc);
        ERRORMSG(") on node " << Ippl::myNode());
        ERRORMSG(" for message " << mnum << " of size " << bufsize);
        ERRORMSG(" bytes sent from node ");
        ERRORMSG(node << " with tag " << tag << endl);
        if (Ippl::retransmit()) {
          // send off a request to have message 'mnum' resent to us by 'node'
          requestList.push_back(pair<int,MsgNum_t>(node, mnum));
        } else {
          // since we're not trying to retransmit, we just quit.
          PInsist(crcval == origcrc, "Exiting due to CRC check failure.");
        }

        // and then return 0 so that the caller knows there was a problem
        return 0;
      }
    }

    // if we're here, the checksums (if enabled) were OK, so receive the
    // message

    // create data structure for this message
    newmsg = new Message(nitems);

    // get all the items and add to the message
    for (int j = 0; j < nitems; j++) {
      int *hdr = static_cast<int *>(pos);
      int elements = hdr[0];
      int bytesize = hdr[1];
      addwordround(pos, 2*sizeof(int));

      // for each item, find the pointer to the actual data and give
      // that pointer to the Message object.  The Message object then
      // does not delete the data until the very end, when the Message
      // is deleted.
      if (bytesize > 0 && elements > 0) {
        newmsg->setCopy(false);
        newmsg->setDelete(false);
        newmsg->putmsg(pos, bytesize/elements, elements);
        addwordround(pos, bytesize);
      }
    }

    // indicate we've received a normal message
    ADDIPPLSTAT(incMessageBytesReceived,bufsize);

    // tell the sender that we received this message OK
    if (Ippl::retransmit())
      informOKList.push_back(pair<int,MsgNum_t>(node, mnum));
  }

  // return the new message, or zero to indicate the buffer contained
  // something else than an actual message
  return newmsg;
}


// put the given message buffer in the sent-message cache, as a new
// CommSendInfo object storing the buffer and other information.
void Communicate::add_to_send_cache(void *msgbuf, MsgNum_t mnum, int msgsize,
                                    int node)
{
  TAU_PROFILE("Communicate::add_to_send_cache",
              "void (void*, MsgNum_t, int)", TAU_MESSAGE);

  // make sure we do not already have this message
  SentCache_t::iterator senti = sentMsgCache.find(mnum);
  if (senti != sentMsgCache.end()) {
    ERRORMSG("ERROR: Trying to cache an already-cached message with num = ");
    ERRORMSG(mnum << endl);
    return;
  }

  //Inform dbgmsg("***Communicate::add_to_send_cache", INFORM_ALL_NODES);
  //dbgmsg << "Adding message " << mnum << " to cache with size = " << msgsize;
  //dbgmsg << endl;

  // make a copy of the message
  char *copybuf = new char[msgsize];
  memcpy(copybuf, msgbuf, msgsize);

  // add the message to the cache list
  CommSendInfo csi(msgsize, copybuf, node);
  sentMsgCache.insert(SentCache_t::value_type(mnum, csi));

  //dbgmsg<<"Sent message cached; now " << sentMsgCache.size() << " buffers ";
  //dbgmsg << "in the cache." << endl;
}


// send off a request to have this message retransmitted to us
void Communicate::request_retransmission(int node, MsgNum_t mnum)
{
  Inform dbgmsg("***Communicate::request_retransmission", INFORM_ALL_NODES);
  dbgmsg << "Requesting retransmission of message " << mnum << " from node ";
  dbgmsg << node << endl;

  // create a regular message, but with the
  // special retransmit tag and the message number as the one item
  // in the Message
  Message msg(1);
  msg.put(mnum);
  send(&msg, node, IPPL_RETRANSMIT_TAG, false); // does not delete message
}


// get the resend information from a buffer sent in a message requesting
// retransmission
void Communicate::unpack_retransmission_request(int nitems, void *pos)
{
  Inform dbgmsg("***Communicate::unpack_retrans_req", INFORM_ALL_NODES);
  //dbgmsg << "Unpacking retransmission request ..." << endl;

  // retransmission messages have the following information as separate
  // items:
  //   message number to retransmit (type == MsgNum_t)
  // so, nitems should be one, and the bytesize should match
  PInsist(nitems == 1, "Wrong number of items in retransmit request.");

  // get the retransmit message number item header info
  int *hdr = static_cast<int *>(pos);
  PInsist(hdr[0] == 1 && hdr[1] == sizeof(MsgNum_t),
          "Wrong message info for retransmit message number.");
  addwordround(pos, 2*sizeof(int));

  // get the actual retransmit message number
  MsgNum_t mnum = *(static_cast<MsgNum_t *>(pos));
  dbgmsg << "Received request to resend message " << mnum << endl; 
  resendList.push_back(mnum);
}


// for message mnum, resend the data
void Communicate::perform_resend(MsgNum_t mnum)
{
  // get the message info from our cache.
  SentCache_t::iterator senti = sentMsgCache.find(mnum);
  PInsist(senti != sentMsgCache.end(),
          "Could not find message in local sent cache to retransmit.");

  // get the node and size
  int size = (*senti).second.size();
  int node = (*senti).second.node();

  // resend the data
  ERRORMSG("WARNING: Resending message number " << mnum);
  ERRORMSG(" of size " << size << " from node ");
  ERRORMSG(myNode() << " to node " << node << " ..." << endl);
  resend((*senti).second.buf(), size, node, COMM_SEND_TAG);
}

 
// tell the sender that we received this message OK
void Communicate::send_ok_message(int node, MsgNum_t mnum)
{
  Inform dbgmsg("***Communicate::send_ok_message", INFORM_ALL_NODES);
  dbgmsg << "Informing node " << node << " that message " << mnum;
  dbgmsg << " was received ok." << endl;

  Message msg(1);
  msg.put(mnum);                // the list of message numbers, one at a time

  send(&msg, node, IPPL_MSG_OK_TAG, false); // does not delete message
}


// unpack message with a list of OK message numbers, and delete them
// from our cache
void Communicate::clear_ok_messages(int nitems, void *pos)
{
  Inform dbgmsg("***Communicate::clear_ok_messages", INFORM_ALL_NODES);
  //dbgmsg << "Unpacking messages-ok information for " << nitems;
  //dbgmsg << " messages ..." << endl;

  // message-ok messages have the following information as separate
  // items:
  //   the number of OK messages (type == int)
  //   the first OK message number (type == MsgNum_t)
  //   the second OK message number (type == MsgNum_t)
  //   etc
  PInsist(nitems >= 1, "Wrong number of items in retransmit request.");

  // loop through the list of items, get the message number from each,
  // and remove that message from our queue
  for (int i=0; i < nitems; ++i) {
    // get the message-ok header
    int *hdr = static_cast<int *>(pos);
    PInsist(hdr[0] == 1 && hdr[1] == sizeof(MsgNum_t),
            "Wrong message info for message-ok number.");
    addwordround(pos, 2*sizeof(int));

    // get the message-ok number
    MsgNum_t mnum = *(static_cast<MsgNum_t *>(pos));
    addwordround(pos, sizeof(MsgNum_t));

    // add this number to our list of messages to say are OK
    dbgmsg << "Will clear message " << mnum << " as OK." << endl;
    sentOKList.push_back(mnum);
  }
}


// unpack message with a list of OK message numbers, and delete them
// from our cache
void Communicate::remove_single_ok_message(MsgNum_t mnum)
{
  Inform dbgmsg("***Communicate::remove_single_ok_message", INFORM_ALL_NODES);

  // check if we have that message
  SentCache_t::iterator senti = sentMsgCache.find(mnum);
  if (senti == sentMsgCache.end()) {
    // we do not have it; print an error message
    ERRORMSG("ERROR: Received 'message ok' for message " << mnum);
    ERRORMSG(", but this node does not have that message in cache." << endl);

  } else {
    // we have it, so remove it after freeing the buffer
    (*senti).second.freebuf();
    sentMsgCache.erase(senti);
    dbgmsg << "Removed message " << mnum << " from send cache; now ";
    dbgmsg << sentMsgCache.size() << " messages in list." << endl;
  }
}


// process list of resend requests
void Communicate::process_resend_requests()
{
  if (resendList.size() > 0) {
    Inform dbgmsg("***Communicate::process_resend_reqs", INFORM_ALL_NODES);
    dbgmsg << "Clearing " << sentOKList.size() << " and resending ";
    dbgmsg << resendList.size() << " messages ..." << endl;
  }

  // clear out OK messages
  while (sentOKList.size() > 0) {
    MsgNum_t mnum = *(sentOKList.begin());
    sentOKList.erase(sentOKList.begin());
    remove_single_ok_message(mnum);
  }

  // resend a message, if necessary
  while (resendList.size() > 0) {
    MsgNum_t mnum = *(resendList.begin());
    resendList.erase(resendList.begin());
    perform_resend(mnum);
  }

  // inform other nodes that we've received their messages ok
  while (informOKList.size() > 0) {
    int node = (*(informOKList.begin())).first;
    MsgNum_t mnum = (*(informOKList.begin())).second;
    informOKList.erase(informOKList.begin());
    send_ok_message(node, mnum);
  }
    
  // request resends from other nodes
  while (requestList.size() > 0) {
    int node = (*(requestList.begin())).first;
    MsgNum_t mnum = (*(requestList.begin())).second;
    requestList.erase(requestList.begin());
    request_retransmission(node, mnum);
  }
}


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

---