MPICommunicator.cpp

Go to the documentation of this file.

//
//------------------------------------------------------------------------
//
// File   : MPICommunicator.cpp
// Author : Patrick McCormick (pat@acl.lanl.gov)
// 
// $Date: 2003/08/04 04:57:49 $
// $Revision: 1.16 $
//
// 
// This class implements the details of an MPI based communicator 
// object.  Most of the details in this class and its implementation
// are taken straight from MPI. 
//
// TODO: At the moment we use a map to keep tabs on async message handles.
// This is necessary because we had to keep tabs on more than just the 
// MPI specific handle -- we need to update the KokoBuffer to reflect the
// size of the message recevied...  This leads to a couple of issues in 
// the current implementation:
//
//    * The current implementation allocates an AsyncMesgHandle class 
//      per async call -- if this turns out to be an issue the easiest
//      fix is to build a free list of AsyncMesgHandle objects that can
//      be reused.
//
//    * We need to write some code to test drive the async messages.  The
//      map approach has not be validated with real tests yet. 
//
//    * The current code uses an unsigned integer value for the message
//      handle data.  It attempts to catch a wrap-around condition when
//      a maximum unsigned int value is reached but perhaps it might be 
//      better to consider unsigned long values instead?
//
//------------------------------------------------------------------------

#include <iostream>
#include <mpi.h>

#include "MPICommunicator.h"

const KokoTag KOKO_ANY_TAG             = MPI_ANY_TAG;
const KokoProcID KOKO_ANY_ID           = MPI_ANY_SOURCE;
const KokoMesgHandle KOKO_NULL_HANDLE  = 0;

class AsyncMesgHandle {
  public:
  KokoBuffer    *_kbuf;    // The buffer that was handed to ARecv().
  // Handle some Windows vs. Linux crud...  This may actually be
  // an MPI implementation detail. 
#ifdef WIN32
  unsigned int  _handle;
#else
  void         *_handle;
#endif
};

using namespace std;

// -----------------------
// --- MPICommunicator ---
// -----------------------
//
//
//
MPICommunicator::MPICommunicator(void)
{
  int flag;

  // TODO: Add error checking here! 
  MPI_Init(NULL, NULL);
  MPI_Comm_size(MPI_COMM_WORLD, &_nprocs);
  MPI_Comm_rank(MPI_COMM_WORLD, &_id);

  //cerr << "MPICommunicator: starting up...\n";
  MPI_Attr_get(MPI_COMM_WORLD,MPI_TAG_UB, &_maxtag, &flag);
  if (flag == false)
   cerr << "MPICommunicator::MPICommunicator() : Warning! Unable to get " 
        << "maximum tag value.\n";
   _handlecount = 0;
}


// ------------------------
// --- ~MPICommunicator ---
// ------------------------
//
// 
//
MPICommunicator::~MPICommunicator(void)
{
  MPI_Finalize();

  // Clean out the active handle map.
  AMHIterator first = _handlemap.begin();
  AMHIterator last  = _handlemap.end();
  while(first != last) {
    delete (*first).second;
    first++;
  }
  _handlemap.clear();
}


// ---------------
// --- Barrier ---
// ---------------
//
//
//
bool MPICommunicator::Barrier(void)
{
  int err = MPI_Barrier(MPI_COMM_WORLD);

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::Barrier() : Error on barrier.\n";
    ErrorMesg(err);
    return false;
  } else {
    return true;
  }
}


// ------------
// --- Send ---
// ------------
//
//
//
bool MPICommunicator::Send(KokoProcID id, KokoBuffer &buf, KokoTag tag)
{
  if (tag > _maxtag) {
    cerr << "MPICommunicator::Send() : KokoTag value execeeds upper bound.\n";
    cerr << "\tKokoTags must be <= " << _maxtag << endl;
    return false;
  }

  if (buf.Locked()) {
    cerr << "MPICommunicator::Send() : KokoBuffer is locked.\n";
    return false;
  }

  int err = MPI_Send(buf, buf.Size(), MPI_UNSIGNED_CHAR, id, tag, 
                     MPI_COMM_WORLD);

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::Send() : Error sending message.\n";
    ErrorMesg(err);
    return false;
  } else {
    return true;
  }
}


// ------------
// --- Recv ---
// ------------
//
//
//
int MPICommunicator::Recv(KokoProcID id, KokoBuffer &buf, KokoTag tag)
{
  if (tag > _maxtag) {
    cerr << "MPICommunicator::Recv() : KokoTag value exceeds upper bounds.\n";
    cerr << "\tKokoTags must be <= " << _maxtag << endl;
    return 0;
  }

  if (buf.Locked()) {
    cerr << "MPICommunicator::Recv() : KokoBuffer is locked.\n";
    return false;
  }

  MPI_Status status;

  int err = MPI_Recv(buf, buf.TotalSize(), MPI_UNSIGNED_CHAR, id, tag, 
                     MPI_COMM_WORLD, &status);

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::Recv() : Error receiving message.\n";
    ErrorMesg(err);
    return 0;
  } else {
    MPI_Get_count(&status, MPI_UNSIGNED_CHAR, &_recv_bytes);
    _recv_tag = status.MPI_TAG;
    _recv_src = status.MPI_SOURCE;
    buf.SetCurSize(_recv_bytes); // Update the buffer's size...
    return _recv_bytes;
  }
}


// ----------------
// --- SendRecv ---
// ----------------
//
//
//
int MPICommunicator::SendRecv(int dest, KokoBuffer &sendbuf, KokoTag stag,
                              int  src, KokoBuffer &recvbuf, KokoTag rtag)
{
  if (stag > _maxtag || rtag > _maxtag) {
    cerr << "MPICommunicator::SendRecv() : Warning, send or recveive tag "
         << "exceeds upper bound.\n";
    cerr << "\tKokoTags must be <= " << _maxtag << endl;
    return 0;
  }

  if (sendbuf.Locked()) {
    cerr << "MPICommunicator::SendRecv() : send KokoBuffer is locked.\n";
    return false;
  }

  if (recvbuf.Locked()) {
    cerr << "MPICommunicator::SendRecv() : send KokoBuffer is locked.\n";
    return false;
  }

  MPI_Status status;
  int err = MPI_Sendrecv(sendbuf, sendbuf.Size(), MPI_UNSIGNED_CHAR, dest, 
                         stag, recvbuf, recvbuf.Size(), MPI_UNSIGNED_CHAR, 
                         src, rtag, MPI_COMM_WORLD, &status);

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::SendRecv() : Error sending/receiving message.\n";
    ErrorMesg(err);
    return 0;
  } else {
    MPI_Get_count(&status, MPI_UNSIGNED_CHAR, &_recv_bytes);
    _recv_tag = status.MPI_TAG;
    _recv_src = status.MPI_SOURCE;
    recvbuf.SetCurSize(_recv_bytes); // Update the buffer's size... 
    return _recv_bytes;
  }
}

// -------------
// --- ASend ---
// -------------
//
//
//
KokoMesgHandle MPICommunicator::ASend(KokoProcID id, KokoBuffer &buf, 
                                      KokoTag tag)
{
  if (tag > _maxtag) {
    cerr << "MPICommunicator::ASend() : Warning, tag exceeds upper bound.\n";
    cerr << "\tKokoTags must be <= " << _maxtag << endl;
    return KOKO_NULL_HANDLE;
  }

  if (buf.Locked()) {
    cerr << "MPICommunicator::ASend() : send KokoBuffer is locked.\n";
    return false;
  }

  MPI_Request req;
  int err = MPI_Isend(buf, buf.Size(), MPI_UNSIGNED_CHAR, id, tag, 
                      MPI_COMM_WORLD, &req);

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::ASend() : Error sending message.\n";
    ErrorMesg(err);
    return KOKO_NULL_HANDLE;
  } else {
    AsyncMesgHandle *amhand = new AsyncMesgHandle;
    amhand->_handle = req;
    amhand->_kbuf   = &buf;
    _handlecount++;
    if (_handlecount == 0) // we've wrapped around a full uint's worth
      _handlecount = 1;    //   of messages... (whew!)
    AMHPair amhpair(_handlecount, amhand);
    AMHBool flag = _handlemap.insert(amhpair);

    buf.Lock(); // Lock the buffer down until the async operation successfully 
                //  completes.

    if (flag.second == true) 
      return amhpair.first;
    else 
      return KOKO_NULL_HANDLE;
  }
}


// -------------
// --- ARecv ---
// -------------
//
//
//
KokoMesgHandle MPICommunicator::ARecv(KokoProcID id, KokoBuffer &buf, 
                                      KokoTag tag)
{
  if (tag > _maxtag) {
    cerr << "MPICommunicator::ARecv() : Warning, tag exceeds upper bound.\n";
    cerr << "\tKokoTags must be <= " << _maxtag << endl;
    return KOKO_NULL_HANDLE;
  }

  if (buf.Locked()) {
    cerr << "MPICommunicator::ARecv() : send KokoBuffer is locked.\n";
    return false;
  }

  MPI_Request req;
  int err = MPI_Irecv(buf, buf.Size(), MPI_UNSIGNED_CHAR, id, tag, 
                      MPI_COMM_WORLD, &req);
  
  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::ARecv() : Error receiving message.\n";
    ErrorMesg(err);
    return KOKO_NULL_HANDLE;
  } else {
    AsyncMesgHandle *amhand = new AsyncMesgHandle;
    amhand->_handle = req;
    amhand->_kbuf   = &buf;
    _handlecount++;
    if (_handlecount == 0) // we've wrapped around a full uint's worth
      _handlecount = 1;    //   of messages... (whew!)
    AMHPair amhpair(_handlecount, amhand);
    AMHBool flag = _handlemap.insert(amhpair);

    buf.Lock();  // Lock the buffer down until the async operation successfully
                 //   completes.

    if (flag.second == true)
      return amhpair.first;
    else
      return KOKO_NULL_HANDLE;
  }
}


// ------------
// --- Wait ---
// ------------
//
//
//
bool MPICommunicator::Wait(KokoMesgHandle &handle)
{
  if (handle == KOKO_NULL_HANDLE) {
    cerr << "MPICommunicator::Wait() : Warning!  Invalid message handle.\n";
    return true;
  }

  MPI_Status status;
  // Danger Will Robinson!   This cast isn't great but it works... 
  int err = MPI_Wait((MPI_Request *)&handle, &status); 

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::Wait() : Error waiting...\n";
    ErrorMesg(err);
    return true;
  }

  // Set recv_bytes to the number of bytes of the completed
  // operation. 
  MPI_Get_count(&status, MPI_UNSIGNED_CHAR, &_recv_bytes);

  // We're done waiting for the message -- clean the handle 
  // out of the map. 

  // TODO: We might want to consider making a
  // free list of AsyncMesgHandle objects to reuse vs. always
  // using dynamic allocation. 
  AMHIterator it = _handlemap.find(handle);
  if (it != _handlemap.end()) {
    (*it).second->_kbuf->SetCurSize(_recv_bytes);
    (*it).second->_kbuf->Unlock();  // Async op. done, unlock buffer. 
    delete (*it).second;
    _handlemap.erase(it);
  } else {
    cerr << "MPICommunicator::Wait() : Warning! Message handle not found!\n";
    cerr << "\tMessage recevied successfully but we may have a corrupt map.\n";
  }

  handle = KOKO_NULL_HANDLE;
  
  return false;
}


// ------------
// --- Test ---
// ------------
//
//
//
bool MPICommunicator::Test(KokoMesgHandle &handle)
{
  if (handle == KOKO_NULL_HANDLE) {
    cerr << "MPICommunicator::Test() : Warning, invalid message handle.\n";
    return false;
  }

  int        flag;
  MPI_Status status;
  // Danger Will Robinson!  This cast isn't great but it works... 
  int err = MPI_Test((MPI_Request *)&handle, &flag, &status); 
  
  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::Test() : Error testing completion status.\n";
    ErrorMesg(err);
    return false;
  }

  if (flag) {
    // operation is finished, store the size of the transfer
    // in recv_bytes. 

    // TODO: Need to add a handle to the koko buffer to the 
    // message handle class/struct so we can update the 
    // buffer's size...
    MPI_Get_count(&status, MPI_UNSIGNED_CHAR, &_recv_bytes);

    // We've recevied the message -- clean the handle 
    // out of the map. 

    // TODO: We might want to consider making a
    // free list of AsyncMesgHandle objects to reuse vs. always
    // using dynamic allocation. 
    AMHIterator it = _handlemap.find(handle);
    if (it != _handlemap.end()) {
      (*it).second->_kbuf->SetCurSize(_recv_bytes);
      (*it).second->_kbuf->Unlock(); // Async op. done, unlock buffer. 
      delete (*it).second;
      _handlemap.erase(it);
    } else {
      cerr << "MPICommunicator::Test() : Warning! Message handle not found!\n";
      cerr << "\tMessage received but we may have a corrupt map.\n";
    }

    return true;
  } else {
    return false;
  }
}


// -----------------
// --- Broadcast ---
// -----------------
//
//
//
bool MPICommunicator::Broadcast(KokoBuffer &buf, KokoProcID root, KokoTag tag)
{

  if (buf.Locked()) {
    cerr << "MPICommunicator::Broadcast() : KokoBuffer is locked.\n";
    return false;
  }

  // 
  // MPI doesn't support tags in broadcast.  In order to support them
  // we use the Tag region in the KokoBuffer object.   If we are the
  // root (sending) node in the broadcast we stuff the tag value into
  // the buffer. 
  //
  if (root == this->ID())
    buf.SetTag(tag);

  // Call MPI's broadcast function -- note that all processors must use the 
  // same 'root' value.   Also note that under MPI it must be the case that
  // all buffers have the same size on all processes (buf.size() should be
  // identical).
  int err = MPI_Bcast(buf.GetBuffer(), buf.TotalSize(), MPI_UNSIGNED_CHAR, 
                      root, MPI_COMM_WORLD);

  if (err != MPI_SUCCESS) {
    cerr << "MPICommunicator::Broadcast() : Error sending broadcast.\n";
    ErrorMesg(err);
    return false;
  } else {
    // If we were not the originator of the broadcast we snag the 
    // tag value out of the buffer. 
    if (this->ID() != root) {
      _recv_tag = buf.GetTag();
      if (!buf.SetCurSize(buf.TotalSize())) {
        cerr << "MPICommunicator::Broadcast() : Warning! "
             << "Can't set buffer size.\n";
      }
    }
    return true;
  }
}


// -----------------
// --- ErrorMesg ---
// -----------------
//
//
//
void MPICommunicator::ErrorMesg(int code)
{
  char mesg[MPI_MAX_ERROR_STRING];
  int  len;
  MPI_Error_string(code, mesg, &len);
  cerr << "MPI Error: " << mesg << endl;
}

---

Send questions, comments, and bug reports to: