3.7   Message passing

• Sending a message is composed of three steps
  1. A send buffer must be initialized
  2. The message must be ``packed'' in this buffer
  3. The entire buffer is
     • Send to another task, or
     • Multicast to a list of tasks, or
     • Broadcast to a group (if using the group library)
     Any number of times without repacking
• Send operation is non-blocking:
  • Returns ASAP
  • Does not implies receipt
• Messages sent between any two tasks A and B will arrive in the order they are sent

• A message is received by
  1. Calling either
     • a blocking, or
     • a non-blocking
     receive routine
  2. ``Unpacking'' each of the packed items from the buffer
• The received routines can be set to accept
  • Any message, or
  • Any message from a specified source, or
  • Any message with a specified message tag, or
  • Only messages with a given message tag from a given source
• PVM also allows users to define their own receive contexts

Message buffers

• Message buffer routines are required only if the user wishes to manage multiple message buffers At any given moment, there is only one active send buffer and one active received buffer The packing, sending, receiving, and unpacking routines only affect the active buffer
• int bufid = pvm_initsend (int encoding)
call pvmfinitsend (encoding, bufid)
  
  
  • Basic buffer manipulation
  • Clears the send buffer and prepares it for packing a new message
  • Specifies the encoding method used for packing messages:

    PvmDataDefault

    XDR encoding is used if the virtual machine configuration is heterogeneous, otherwise no encoding is done

    PvmDataRaw

    No encoding is done

    PvmDataInPlace

    Data left in place. Buffer only contains size and pointers to the items to be send

• int bufid = pvm_mkbuf (int encoding)
  • Creates a new empty send buffer
  • encoding is one of PvmDataDefault, PvmDataRaw, PvmDataInPlace
  • Returns the buffer ID
• int info = pvm_freebuf (int bufid)
  • Disposes the buffer bufid
  • Should be done after a message has been sent and is no longer needed
  • Then call pvm_mkbuf() to create a buffer for a new message if required
  • Neither of these is required when using pvm_initsend()
• int bufid = pvm_getsbuf (void)
  • Returns the active send buffer ID
• int bufid = pvm_getrbuf (void)
  • Returns the active receive buffer ID

• int oldbuf = pvm_setsbuf (int bufid)
  • Sets the active send buffer to bufid
  • Saves the state of the previous buffer
  • Returns the previous active buffer ID
• int oldbuf = pvm_setrbuf (int bufid)
  • Sets the active receive buffer to bufid
  • Saves the state of the previous buffer
  • Returns the previous active buffer ID
• If bufid is set to 0 in pvm_set?buf(), then the present buffer is save and there is no active buffer This can be used to save the present state of an application's messages So that a math library or graphical interface which also used PVM messages will not interfere with the state of the application's buffers
  
  
• Example, forward a message without repacking it

  oldid = pvm_setsbuf (0) ; 
  bufid = pvm_recv (src, tag) ; 
  info  = pvm_send (dst, tag) ; 
  info  = pvm_setrbuf (oldid) ;
          

Packing data

• Each of the following routines packs an array of the given data type into the active send buffer
• They can be called multiple times to pack a single message
• Thus, a message can contain several arrays each of different data type
• An application must unpack the messages exactly like they were packed

• C structures must be passed by packing their individual elements
• The arguments for each of the routines of the C interface are
  • a pointer to the first item to be packed
  • nitem which is the total number of items to pack from this array
  • stride which is the stride to use when packing

• pvm_pkstr() is an exception; its packs a NULL terminated character string and so does not need nitem nor stride

  int info = pvm_pkbyte   (char   *cp, int nitem, int stride)            
  int info = pvm_pkcplx   (float  *fp, int nitem, int stride)            
  int info = pvm_pkdcplx  (double *dp, int nitem, int stride)            
  int info = pvm_pkdouble (double *dp, int nitem, int stride)            
  int info = pvm_pkfloat  (float  *fp, int nitem, int stride)            
  int info = pvm_pkint    (int    *np, int nitem, int stride)            
  int info = pvm_pklong   (long   *np, int nitem, int stride)            
  int info = pvm_pkshort  (short  *np, int nitem, int stride)            
  int info = pvm_pkstr    (char   *cp)            
          

• call pvmfpack (datatype, data, item,
   stride, info)
  
  
  • datatype is one of

    STRING	0	REAL4	4
    BYTE1	1	REAL8	6
    INTEGER2	2	COMPLEX8	5
    INTEGER4	3	COMPLEX16	7

  • data is the first item to be packed
  • nitem is the total number of item to be packed
  • stride is the stride to use when packing

• Example/exercise: consider a N × N matrix M
  • Two PVM tasks want to exchange
    • The matrix
    • The rth row of the matrix
    • The cth column of the matrix
    • The diagonal of the matrix
    • The r_l to r_u rows of the matrix
    • The c_l to c_u columns of the matrix

Unpacking data

• The following routines unpack (multiple) data types from the active receive buffer
• In an application, they should match their corresponding pack routines in
  • Types
  • Number of items
  
  
• C interface

  int info = pvm_upkbyte   (char   *cp, int nitem, int stride)            
  int info = pvm_upkcplx   (float  *fp, int nitem, int stride)            
  int info = pvm_upkdcplx  (double *dp, int nitem, int stride)            
  int info = pvm_upkdouble (double *dp, int nitem, int stride)            
  int info = pvm_upkfloat  (float  *fp, int nitem, int stride)            
  int info = pvm_upkint    (int    *np, int nitem, int stride)            
  int info = pvm_upklong   (long   *np, int nitem, int stride)            
  int info = pvm_upkshort  (short  *np, int nitem, int stride)            
  int info = pvm_upkstr    (char   *cp)                        
        

• Fortran interface
  
  call pvmfunpack (datatype, data, nitems, stride, info)

Sending and receiving data

• int info = pvm_send (int tid, int msgtag)
call pvmfsend(tid, mstag, info)
  • Labels the message with an integer identifier msgtag
  • Sends it immediately to the process tid
• int info = pvm_mcast (int *tids, int ntask,
   int msgtag)
call pvmfmcast (ntask, tids, mstag, info)
  • Labels the message with an integer identifier msgtag
  • Broadcasts the message to the ntask tasks specified in the array tids

• int bufid = pvm_nrecv (int tid, int msgtag)
call pvmfnrecv (tid, mstag, info)
  
  
  • Non-bloking receive routine
  • If the requested message has not arrived, returns 0
  • Can be called multiple times for the same message to check
  • If a message with label msgtag has arrived from tid, places this message in a new active receive buffer which is created
  • The previous active receive buffer is cleared, unless it has been saved
  • A value of -1 in msgtag or tid matches anything (wildcard)
• int bufid = pvm_recv (int tid, int msgtag)
call pvmfrecv (tid, mstag, info)
  
  
  • Blocking receive routine
  • It will wait until a message with label msgtag has arrived from tid
  • Otherwise acts like pvm_nrecv()

• Example:

  C     ** sender ** 
  C        send a int N and an array of N reals
        CALL PVMFINITSEND (PVMDEFAULT, IBUF)
        CALL PVMFPACK (INTEGER4, N, 1, 1, IERR)
        CALL PVMFPACK (REAL8, Y(1), N, 1, IERR)
        CALL PVMFSEND (TIDS(I), METAG, IERR)
  
  C     ** receiver **
        CALL PVMFRECV (-1, METAG, IBUF)
        CALL PVMFUNPACK (INTEGER4, N, 1, 1, IERR)
        CALL PVMFUNPACK (REAL8, Z(1), N, 1, IERR)
  
  

• int info = pvm_bufinfo (int bufid, int *bytes,
   int *msgtag, int *tid)
call pvmfbufinfo (bufid, bytes, mstag, tid, info)
  
  
  • Returns informations about the message in the buffer bufid:
    • Actual message tag
    • Actual source tid
    • Length in bytes
  • It can be used to determine the label and source of a message that was received with wildcards
  • Example

  bufid = pvm_recv (-1, -1) ; 
  info = pvm_bufinfo (bufid, &bytes, &type, &source) ; 
  printf ("message from %d, tag %d\n", source, type) ; 
        

• int bufid = pvm_probe (int tid, int msgtag)
call pvmfprobe (tid, mstag, bufid)
  • Checks if a message with label msgtag has arrived from tid
  • tid and msgtag may be wildcarded by the value -1
  • If the requested message has not arrived, returns 0
  • Otherwise returns the buffer ID for the message, but does not ``received'' it
  • After a message has arrived, pvm_recv() must be call before the message can be unpacked
  • In addition, pvm_bufinfo() can be called with the returned buffer ID to determine information about the message before receiving it
  • Example

    tid = pvm_parent () ;
    msgtag = 4 ; 
    arrived = pvm_probe (tid, msgtag) ;
    if (arrived)
      info = pvm_bufinfo (arrived, &len, &tag, &tid) ;
    else 
      /* go to other computing */
                

• int (*old)() = pvm_recvf (
   int (*new)(int buf, int tid, int tag))
  • No Fortran interface
  • Modifies the receive context used by the receive functions
  • Can be used to extend PVM
  • The default receive context is to match on source and message tag
  • This can be modified by any user defined comparison function
  • The matching function returns preference to pvm_recv()

  < 0	return immediately with this error code
  0	do not pick this message
  1	pick this message and stop now
  > 1	pick this highest preference after scanning them all

• 
  
  Example, implementing probe with pvm_recvf()

  #include "pvm3.h"
  
  static foundit = 0 ; 
  
  static int foo_match (int mid, int tid, int code)
  {
    int t, c, cc ; 
    
    if ((cc = pvm_bufinfo (mid, (int *) 0, &c, &t)) < 0)
      return cc; 
    if ((tid == -1 || tid == t)
    && (code == -1 || code == c))
      foundit = 1 ; 
    return 0 ; 
  }
  
  int probe (int src, int code) 
  {
    int (*omatch)(), cc ; 
  
    omatch = pvm_recvf (foo_match) ; 
    foundit = 0 ; 
    if ((cc = pvm_nrecv (src, code) < 0) 
      return cc ; 
    pvm_recvf (omatch) ; 
    return foundit ; 
  }