src/Message/Communicate.h

Go to the documentation of this file.

// -*- C++ -*-
/***************************************************************************
 *
 * The IPPL Framework
 * 
 *
 * Visit http://people.web.psi.ch/adelmann/ for more details
 *
 ***************************************************************************/

#ifndef COMMUNICATE_H
#define COMMUNICATE_H

/***************************************************************************
 * Communicate.h - communications object for use with Ippl framework.  Allows
 * user to establish id's for available nodes, establish connections, and
 * send/receive data.
 ***************************************************************************/

// include files
#include "Message/TagMaker.h"
#include "Message/Tags.h"

#ifdef IPPL_STDSTL
#include <vector>
#include <utility>
using std::vector;
using std::pair;
#else
#include <vector.h>
#include <pair.h>
#endif // IPPL_STDSTL

#ifdef IPPL_USE_STANDARD_HEADERS
#include <iostream>
using namespace std;
#else
#include <iostream.h>
#endif

// forward declarations
class Message;
class Communicate;
ostream& operator<<(ostream&, const Communicate&);

// special codes used as 'wildcards' to match any node or tag
const int COMM_ANY_NODE = (-1);
const int COMM_ANY_TAG  = (-1);


// A simple class used to store information for caching sent messages.  This
// is only used if the 'retransmit' option is active.
class CommSendInfo
{
public:
  CommSendInfo()
    : size_m(0), buf_m(0)
    {
    }

  CommSendInfo(int size, char *buf, int node)
    : size_m(size), buf_m(buf), node_m(node)
    {
    }

  CommSendInfo(const CommSendInfo &c)
    : size_m(c.size_m), buf_m(c.buf_m), node_m(c.node_m)
    {
    }

  ~CommSendInfo()
    {
      // the user is actually responsible for freeing the buffer.  We
      // do not do this automatically here
    }

  CommSendInfo &operator=(const CommSendInfo &c)
    {
      size_m = c.size_m;
      buf_m = c.buf_m;
      node_m = c.node_m;
      return *this;
    }

  int size() const { return size_m; }

  int node() const { return node_m; }

  char *buf() { return buf_m; }
  const char *buf() const { return buf_m; }

  void freebuf()
    {
      if (buf_m != 0)
        delete [] buf_m;
      buf_m = 0;
    }

private:
  int size_m;
  int node_m;
  char *buf_m;
};


// The base class for all specific Communicate objects
class Communicate : public TagMaker
{

public:
  // default error codes, may be overridden by inherited classes.
  enum CommErrors { COMM_NOERROR, COMM_ERROR, COMM_NOSEND, COMM_NORECEIVE };

  // special tags used by this class ... 32000 is arbitrary
  enum CommTags { COMM_HOSTS_TAG = 32000, COMM_DIE_TAG, COMM_SEND_TAG };

  // special codes used as 'wildcards' to match any node or tag
  // These are listed again because they should be here, but the global
  // values are kept for compatibility.
  enum CommCodes { COMM_ANY_NODE = (-1), COMM_ANY_TAG = (-1) };

public:
  // constructor and destructor
  // 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)
  Communicate(int argc = 0, char** argv = NULL, int procs = (-1));
  virtual ~Communicate(void);

  // return the name of this item
  virtual const char *name() const { return "Serial"; }

  // return info about connections in general
  int getNodes() const { return TotalNodes; }
  int getContexts(const int n) const { return Contexts[n]; }
  int getProcesses(const int n, const int c) const { return Processes[n][c]; }
  int myNode() const { return myHost; }
  int getError() const { return ErrorStatus; }
  int getReceived() const { return recMsgList.size(); }

  //
  //    nonvirtual routines to send/receive data
  //
 
  // send data to another node.  Returns success (T or F).
  // last argument specifies whether to delete the Message after sending
  // (if message is for another node).  Note that if the send is not
  // successful, the message will NOT be deleted, regardless of delmsg.
  bool send(Message *, int node, int tag, bool delmsg = true);

  // receive data from another node.  Returns newly created Message object
  // with received message, or NULL if no message is available.
  // If node is < 0, this will receive the next message with the given tag
  // from any node.  If tag < 0, this will receive the next message with
  // any tag from the given node.  If both are < 0, this will receive the
  // next message, period.  node and tag are passed by reference; if either
  // is < 0, and a message is received, they are changed to their actual
  // values.
  Message *receive(int& node, int& tag);
 
  // a blocking version of receive;
  Message *receive_block(int& node, int& tag);

  //
  //    virtual routines to broadcast data
  //
 
  // broadcast the current message to other nodes.
  // Return number of nodes actually sent to.
  // The first version sends to all nodes including this node.
  // The second version sends to all nodes except this node.
  // The first argument is the Message; the last argument is the tag.
  virtual int broadcast_all(Message *, int);
  virtual int broadcast_others(Message *, int, bool delmsg=true);


  //
  //    routines to synchronize processors at a barrier
  //

  // Synchronize all processors (everybody waits for everybody
  // else to get here before returning to calling function).
  void barrier(void);

  //
  //    virtual routines to deal with memory management
  //

  // clean up after a Message has been used (called by Message).  By
  // default, does nothing.
  virtual void cleanupMessage(void *);

protected:
  // struct used to store messages, tags, and nodes
  struct MessageData {
    int node;                   // sending/receiving node
    int tag;                    // tag of the message
    Message *msg;               // pointer to the message itself
    MessageData(int n, int t, Message *m) : node(n),tag(t),msg(m) {}
    MessageData() : node(-1), tag(-1), msg(0) { }
    MessageData(const MessageData& m) : node(m.node),tag(m.tag),msg(m.msg) {}
    ~MessageData() {}
  };

  // a list of messages which have already been received, but not yet
  // delivered
  vector<MessageData> recMsgList;

  // the following items should be filled in by the derived classes
  int TotalNodes;               // number of nodes available (0 ... # nodes-1)
  int myHost;                   // which node am I?
  int ErrorStatus;              // error code, from above enumeration
  vector<int> Contexts;         // the number of contexts per node
  vector< vector<int> > Processes;   // number of running processes per context

  // An integer message number identifier; this is included in each
  // message, and continually increases as more messages are sent.
  typedef long MsgNum_t;
  MsgNum_t nextMsgNum;

  // An optional sent-message cache, used to attempt to retransmit
  // messages if they are corrupted in-transit.  Messages are keyed on
  // a message number, which is is unique for each message.
  typedef map<MsgNum_t, CommSendInfo> SentCache_t;
  SentCache_t sentMsgCache;

  // a list of things to resend at the next opportunity
  vector<MsgNum_t> resendList;

  // a list of messages which have been received OK
  vector<MsgNum_t> sentOKList;

  // a list of messages which should be cleared out on other nodes
  vector<pair<int,MsgNum_t> > informOKList;

  // a list of requests we must make to other nodes to resend messages
  vector<pair<int,MsgNum_t> > requestList;
  
  // add a new message to the received message queues.  Return success.
  // arguments: message, sending node, tag
  bool add_msg(Message *, int, int);

  // 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* find_msg(int&, int&);

  //
  // implementation-specific routines (which begin with 'my')
  //    these should be provided in a derived class, and contain the
  //    comm-library-specific code
  //

  // 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.
  virtual bool mysend(Message *, int node, int utag, int etag);

  // 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.
  virtual Message *myreceive(int& node, int& tag, int etag);

  // Synchronize all processors (everybody waits for everybody
  // else to get here before returning to calling function).
  virtual void mybarrier(void);

  // resent a message buffer that has been previously packed and copied
  // into the provided buffer.  Return success.
  virtual bool resend(void *buf, int size, int node, int etag);

  //
  // utility functions used to serialize data into and out of byte buffers
  //

  // standard way to create and free buffer storage
  static inline void *makebuffer(int size) { return malloc(size); }
  static inline void freebuffer(void *buf) { free(buf); }

  // compute the size of storage needed to add 'size' bytes to a buffer,
  // in order to keep everything word-aligned
#if defined(IPPL_LONGLONG)
  static inline unsigned int wordround(int size) {
    return sizeof(long long) *
      ((size + sizeof(long long) - 1)/sizeof(long long));
  }
#else
  static inline unsigned int wordround(int size) {
    return sizeof(long) *
      ((size + sizeof(long) - 1)/sizeof(long));
  }
#endif

  // compute a wordround value for 'size' bytes, then add that to the
  // given 'pos' pointer
  static inline void addwordround(void * &pos, int size) {
    pos = static_cast<void *>(wordround(size) + static_cast<char *>(pos));
  }

  // memcpy data into the given location, and then increment the pointer
  static inline void pack(void *packdata, void * &pos, int size) {
    memcpy(pos, packdata, size);
    addwordround(pos, size);
  }

  // memcpy data out of a given location to another, updating 'pos'
  static inline void unpack(void * &pos, void *packdata, int size) {
    memcpy(packdata, pos, size);
    addwordround(pos, size);
  }

  //
  // utility functions used in packing and unpacking Message data
  //

  // calculate how big the buffer must be to send the given message
  int find_msg_length(Message &);

  // put data from the given Message into the given buffer, with tag value.
  // the final arguments are the buffer size, in bytes, and the dest node.
  void fill_msg_buffer(void *, Message &, int, int, int);

  // take data out of the current receive buf and create a new Message
  Message *unpack_message(int &node, int &tag, void *pos);

  //
  // utility functions used for message caching/retransmit
  //

  // put the given message buffer in the sent-message cache, as a new
  // CommSendInfo object storing the buffer and other information.
  void add_to_send_cache(void *pos, MsgNum_t mnum, int size, int node);

  // send off a request to have this message retransmitted to us
  void request_retransmission(int node, MsgNum_t mnum);

  // resend the data for message mnum ... calls the virtual 'resend'
  void perform_resend(MsgNum_t mnum);

  // get the resend information from a buffer sent in a message requesting
  // retransmission
  void unpack_retransmission_request(int nitems, void *pos);

  // tell the sender that we received this message OK
  void send_ok_message(int node, MsgNum_t mnum);

  // unpack message with a list of OK message numbers, and delete them
  // from our cache
  void clear_ok_messages(int nitems, void *pos);

  // remove a single OK message
  void remove_single_ok_message(MsgNum_t mnum);

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

#endif // COMMUNICATE_H


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

---