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

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

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

// 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 ACLMPL 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
CommACLMPL::CommACLMPL(int argc, char** argv, int procs)
        : Communicate(argc, argv, procs) {
  TAU_PROFILE("CommACLMPL::CommACLMPL()", "void (int, char **, int)",
              TAU_MESSAGE);
  int i, rep_host, reported, ierror=0;
  // ACLMPL specific stuff...
  if (procs < 0) procs = 0;
  ACLMPL_initialize(procs);       
  myHost = (int) ACLMPL_self();
  TotalNodes = (int) ACLMPL_npes();

  // 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;
  }

  int msize = wordround(sizeof(int));
  void *sendbuf = makebuffer(msize);
  void *recbuff = makebuffer(msize);
  void *pos;

  if (myHost == 0) { // master process
    // send a message to each child node
    pos = sendbuf;
    pack(&myHost, pos, sizeof(int));
    for (i=1; i<TotalNodes; i++) 
      ACLMPL_send(i, sendbuf, msize);

    // wait for child nodes to report ready
    INFOMSG("CommACLMPL: Parent process waiting for children ..." << endl);
    int *child_ready = new int[TotalNodes];
    for (i=0; i<TotalNodes; i++) child_ready[i] = 0;
    reported = 1; // parent is already ready

    while (reported<TotalNodes) {
      ierror = ACLMPL_recv(reported,recbuff,msize);
      if (ierror>0) {
        pos = recbuff;
        unpack(pos,&rep_host,sizeof(int));
        if (rep_host>0 && rep_host<TotalNodes && !(child_ready[rep_host])) {
          child_ready[rep_host] = 1;
          reported++;
          INFOMSG("CommACLMPL: Child " << rep_host << " ready." << endl);
        }
        else {
          ERRORMSG("CommACLMPL: Error with child reporting to parent.  ");
          ERRORMSG("rep_host = " << rep_host);
        }
        ierror = 0;
      }
      else {
        ERRORMSG("CommACLMPL: Message from child has size " << ierror << endl);
      }
    }

    delete [] child_ready;
    INFOMSG("CommACLMPL: Initialization complete." << endl);
  } else {
    // this is a child process; get data from parent
    int checknode;
    ierror = ACLMPL_recv(0, recbuff, msize);
    if (ierror>0) {
      pos = recbuff;
      unpack(pos,&checknode,sizeof(int));
      if (checknode != 0)
        WARNMSG("CommACLMPL: Child received bad message in startup." << endl);
    } else {
      WARNMSG("CommACLMPL: Child received message of size " << ierror << endl);
    }

    // send back an acknowledgement
    pos = sendbuf;
    pack(&myHost,pos,sizeof(int));
    ACLMPL_send(0,sendbuf,msize);
  }

  freebuffer(sendbuf);
  freebuffer(recbuff);

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

}


// class destructor
CommACLMPL::~CommACLMPL(void) {
  TAU_PROFILE("CommACLMPL::~CommACLMPL()", "void ()", TAU_MESSAGE);

  // 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("CommACLMPL: Found extra message from node " << node);
          WARNMSG(", tag " << tag << ": msg = " << *msg << endl);
        }
      } while (msg != 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  (int)
//              size of item 1, in number of elements   (int)
//              item 1 data     (various)
//              ...
//              type of item N  (int)
//              size of item N, in number of elements   (int)
//              item N data     (various)
AsyncMsgInfo *
CommACLMPL::pack_message(Message *msg, int tag, int &buffsize, int node) {
  TAU_PROFILE("CommACLMPL::pack_message()", "(Message *, int, int, int)",
              TAU_MESSAGE);

  // calculate size of buffer
  buffsize = find_msg_length(*msg);

  // allocate storage for buffer
  AsyncMsgInfo *outbuffer = ACLMPL_asend_alloc(buffsize);
  void *pos = ACLMPL_agetbuffer(outbuffer);

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


// 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 CommACLMPL::mysend(Message* msg, int node, int tag, int etag) {
  TAU_TYPE_STRING(taustr, "bool (Message *, int, int, int)");
  TAU_PROFILE("CommACLMPL::mysend()", taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(packtimer, "  ACLMPL::mysend::pack", taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(sendtimer, "  ACLMPL::mysend::send", taustr, TAU_MESSAGE);

  // the size of the buffer we are creating
  int size = 0;

  // pack and send the message
  TAU_PROFILE_START(packtimer);
  AsyncMsgInfo *outbuffer = pack_message(msg, tag, size, node);
  TAU_PROFILE_STOP(packtimer);

  TAU_PROFILE_START(sendtimer);
  int sentsize = ACLMPL_alink(node, etag, outbuffer);
  TAU_PROFILE_STOP(sendtimer);

  // check for an error, and return
  if (sentsize == 0)
    ERRORMSG("CommACLMPL: Failure during asynchronous send!" << endl);
  return (sentsize != 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* CommACLMPL::myreceive(int& node, int& tag, int etag) {
  TAU_TYPE_STRING(taustr, "Message *(int, int, int)" );
  TAU_PROFILE("CommACLMPL::myreceive()", taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(findtimer, "  ACLMPL::myrec::find", taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(infotimer, "  ACLMPL::myrec::getinfo",taustr, TAU_MESSAGE);
  TAU_PROFILE_TIMER(unpktimer, "  ACLMPL::myrec::unpack", taustr, TAU_MESSAGE);

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

  // get info about message
  Message* newmsg = NULL;
  TAU_PROFILE_START(findtimer);
  AsyncMsgInfo *msg;
  msg = ACLMPL_afind(checknode, checktag);
  TAU_PROFILE_STOP(findtimer);

  if (msg != 0) {
    checknode = ACLMPL_agetsource(msg);
    checktag = ACLMPL_agettag(msg);
    //    int size = ACLMPL_agetsize(msg);

    TAU_PROFILE_START(unpktimer);
    // make a copy of the buffer
    //void *outbuff = makebuffer(size);
    //memcpy(outbuff, ACLMPL_agetbuffer(msg), size);
    void *outbuff = ACLMPL_agetbuffer(msg);

    // unpack the data from the buffer into a Message
    newmsg = unpack_message(node, tag, outbuff);

    if (newmsg == 0) {
      // there was a problem with the message body, so we just throw this
      // one away
      cleanupMessage(static_cast<void *>(msg));
    } else {
      // tell the Message object that we should be notified when the buffer
      // holding this data should be deallocated
      //newmsg->useCommunicate(this, outbuff);
      //ACLMPL_afree(msg);
      newmsg->useCommunicate(this, static_cast<void *>(msg));
    }

    TAU_PROFILE_STOP(unpktimer);
  }
  else {
    // no message available
    DEBUGMSG(level2 << "CommACLMPL: No Message Found 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 ACLMPL barrier for all procs 
void CommACLMPL::mybarrier(void) {
  TAU_PROFILE("CommACLMPL::mybarrier()", "void ()", TAU_MESSAGE);

  ACLMPL_barrier();
}


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

  //Inform dbgmsg("CommACLMPL::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");

  // allocate storage for buffer
  AsyncMsgInfo *outbuffer = ACLMPL_asend_alloc(buffsize);
  void *pos = ACLMPL_agetbuffer(outbuffer);

  // copy the buffer into storage that can be used to send
  memcpy(pos, buf, buffsize);

  // now send the buffer
  int sentsize = ACLMPL_alink(node, etag, outbuffer);

  //dbgmsg << "Message resent; sentsize = " << sentsize << endl;

  // check for an error, and return
  if (sentsize == 0)
    ERRORMSG("CommACLMPL: Failure during asynchronous resend!" << endl);
  return (sentsize != 0);
}


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

  // must free up the buffers from ACLMPL
  ACLMPL_afree(static_cast<AsyncMsgInfo *>(d));
}


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

---