= Message Passing =

== First attempt ==

Low-level implementation using malloc, linked lists and so on:

{{{

#define MPI_ANY_SOURCE -1
#define MPI_ANY_TAG -2
#define MPI_INT 1
#define MPI_FLOAT 2
#define MPI_DOUBLE 3
// etc.

$input int NPROCS;
$assume NPROCS >= 1;
$scope top;
$heap mp_heap;
$proc procs[NPROCS];

/* There will be one message queue for each pair of prods (i,j).
 * Each queue is a doubly-linked list of Message objects.
 */
typedef struct MPI_Message {
  struct MPI_Message *<top> next;
  struct MPI_Message *<top> prev;
  int tag;
  int size;
  void *<top> data;
} MPI_Message;

typedef struct MPI_Comm_struct {
  MPI_Message *<top> buf_front[NPROCS][NPROCS];
  MPI_Message *<top> buf_back[NPROCS][NPROCS];
} MPI_Comm_struct;

typedef MPI_Comm_struct *MPI_Comm;

/* As in MPI, when a receive returns, this structure
 * tells you the source and tag of received message,
 * which you need if you used wildcards,  Also size.
 */
typedef struct MPI_Status {
  int source;
  int tag;
  int size;
} MPI_Status;

typedef int MPI_Datatype;

/* This is the actual MPI Comm world structure */
MPI_Comm_struct MPI_Comm_world_struct;

/* The user will use MPI_COMM_WORLD */
MPI_Comm MPI_COMM_WORLD = &MPI_Comm_world_struct;

void init() {
  for (int i=0; i<NPROCS; i++)
    procs[i] = $spawn proc(i);
  for (int i=0; i<NPROCS; i++)
    for (int j=0; j<NPROCS; j++) {
      MPI_COMM_WORLD->buf_front[i][j] = NULL;
      MPI_COMM_WORLD->buf_back[i][j] = NULL;
    }
}

int sizeofDatatype(MPI_Datatype type) {
  switch (type) {
    case MPI_INT: return sizeof(int);
    case MPI_FLOAT: return sizeof(float);
    case MPI_DOUBLE: return sizeof(double);
    default: exit(-1); // not yet implemented
  }
}

void MPI_process(int pid) {

  void MPI_Send(void *buf, int count, MPI_Datatype type, int dest, int tag, MPI_Comm comm) {
    // create a message
    Message message;

    message.tag = tag;
    message.size = count*sizeofDatatype(type);
    message.data = $malloc(&mp_heap, size);
    memcpy(message.data, buf, size);
    // enqueue on comm->buf_front[i][j] …
    // update  message.next, message.prev, buf_back[i][j]
  }

  boolean match(MPI_Message *p, int source, int tag, Comm comm) {
    return (source == MPI_ANY_SOURCE || source == p->source)
         && (tag == MPI_ANY_TAG || tag == p->tag)
         && (comm == p->comm);
  }
  
  boolean probe(int source, int tag, Comm comm) {
    for (MPI_Message *p = comm->buf_front[source][pid]; p != NULL; p=p->next)
      if (match(p, source, dest, tag)) return true;
    return false;
  }

  boolean recv_guard(int source, int tag, Comm comm) {
    if (source == MPI_ANY_SOURCE) {
      for (int i=0; i<NPROCS; i++) {
        if (probe(i, tag, comm)) return true;
      }
      return false;
    } else {
      return probe(source, tag, comm);
    }
  }

  void recv(int dest, void *buf, int size, int source, int tag, Comm comm, Status *status) {
    int message_source;
    int message_dest;
    MPI_Message *message;

    when (recv_guard(source, tag, comm)) ;
    if (source == MPI_ANY_SOURCE) {   
      for (int i=0; i<NPROCS; i++) {
        message = comm->buf_front[source][dest];
        ...
       // search the queue looking for the message
       // also set message_source, message_dest;
       // and remove message from the queue
      }
    } else {
      …
    }
  
    if (message->size > size) error…;
    memcpy(buf, message->data, message.size);
    status->size = message.size;
    status->source = message_source;
    status->dest = message_dest;
    $free(&mp_heap, message->data);
  }

  // main body of process…

  $scope proc_scope;
  $heap proc_heap;
    
  …
}

void main() {
  for (int i=0; i<NPROCS; i++)
    procs[i] = $spawn(i);
  for (int i=0; i<NPROCS; i++)
    $wait procs[i];

}

}}}

== Second attempt ==

It would be much better if CIVL-C provided some abstract datatypes for message queues and so on.  What should these datatypes be?  Proposal:

* `$message`
 * an abstract datatype
* `$message $message_pack(int source, int dest, int tag, void *data, int size)`
 * creates a new message, copying data from the specified buffer
* `int $message_source($message * message)`
 * returns the message source
* `int $message_tag($message * message)`
 * returns the message tag
* `int $message_dest($message * message)`
 * returns the message destination
* `int $message_size($message * message)`
 * returns the message size
* `void $message_unpack($message * message, void *buf, int size)`
 * transfers message data to but, throwing exception if message size exceeds specified size
* `$comm`
 * an abstract datatype representing a communicator, or set of message channels between every pair of processes in a set of processes
* `$comm $comm_create(int nprocs, $proc * procs)`
 * creates a new comm from the given sequence of processes, by allocating memory and copying the process sequence; the new comm has no messages
* `void $comm_destroy($comm * comm)`
 * destroys the comm, deallocating whatever was allocated in its creation
* `int $comm_nprocs($comm * comm)`
 * returns the number of processes associated to the comm
* `$proc * $comm_procs($comm * comm)`
 * returns a pointer to the procs array in comm
* `void $comm_enqueue($comm * comm, $message * message)`
 * adds the message to the comm
* `boolean $comm_probe($comm * comm, int source, int dest, int tag)`
 * returns true iff a matching message exists in comm
* `$message * $comm_seek($comm * comm, int source, int dest, int tag)`
 * finds the first matching message and returns pointer to it without modifying comm
* `$message * $comm_dequeue($comm * comm, int source, int dest, int tag)`
 * finds the first matching message, removes it from comm, and returns pointer to message
* `int $comm_chan_size($comm * comm, int source, int dest)`
 * returns the number of messages from source to dest stored in comm
* `int $comm_total_size($comm * comm)`
 * returns the total number of messages in the comm
* constants defined:
 * `int $COMM_ANY_SOURCE`
 * `int $COMM_ANY_TAG`

Here is how these primitives could be used to implement an MPI program:

{{{
#define MPI_ANY_SOURCE $COMM_ANY_SOURCE
#define MPI_ANY_TAG $COMM_ANY_TAG
#define MPI_INT 1
#define MPI_FLOAT 2
#define MPI_DOUBLE 3
// etc.

$input int NPROCS;
$assume NPROCS >= 1;
$scope top;
$proc procs[NPROCS];

typedef $comm *MPI_Comm;
$comm MPI_Comm_world_comm;
MPI_Comm *MPI_COMM_WORLD = &_MPI_COMM_WORLD;
boolean initialized = $false;

/* This structure tells you the source and tag of a received message,
 * which you need if you used wildcards,  Also size.
 */
typedef struct MPI_Status {
  int source;
  int tag;
  int size;
} MPI_Status;

typedef int MPI_Datatype;

int sizeofDatatype(MPI_Datatype type) {
  switch (type) {
    case MPI_INT: return sizeof(int);
    case MPI_FLOAT: return sizeof(float);
    case MPI_DOUBLE: return sizeof(double);
    default: exit(-1); // not yet implemented
  }
}

void MPI_Comm_init(MPI_Comm comm) {
  $when (initialized) ;
}

void MPI_Send(int pid, void *buf, int count, MPI_Datatype type, int dest, int tag, MPI_Comm comm) {
  $message m = $message_pack(pid, dest, tag, buf, sizeofDatatype(type)*count);

  $comm_enqueue(comm, &m);
}

void MPI_Recv(int pid, void *buf, int count, MPI_Datataype type, int source, int tag, MPI_Comm comm, MPI_Status *status) {
  int size = sizeofDatatype(type)*count;
  // the following is a system function with built-in guard:
  $message m = $comm_dequeue(comm, source, pid, tag);

  status->size = $message_size(&m);
  status->source = $message_source(&m);
  status->dest = $message_dest(&m);
  $message_unpack(&m, buf, size); // will throw exception if message too big
}

void MPI_process(int pid) {
  ...
  MPI_Comm_init(MPI_COMM_WORLD);
  ...
}

void main() {
  for (int i=0; i<NPROCS; i++) procs[i] = $spawn MPI_process(i);
  MPI_Comm_world_comm = $comm_create(NPROCS, &procs[0]);
  initialized = $true;
  for (int i=0; i<NPROCS; i++) $wait procs[i];
}
}}}