Changes between Version 4 and Version 5 of MessagePassing

Timestamp:

07/11/13 21:14:57 (13 years ago)

Author:

siegel

Comment:

--

MessagePassing

@@ -75 +75 @@
 void MPI_process(int pid) {
 
-  void MPI_Send(void *buf, int size, int tag, int dest, MPI_Comm comm) {
-  // create a message
-  Message message;
+  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 = size;
-  message.data = $malloc<top>(&mp_heap, size);
-  memcpy(message.data, buf, size);
-  // enqueue on comm->buf_front[i][j] …
-  // update  message.next, message.prev, buf_back[i][j]
-}
+    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]
+  }
 
-// can I use a function as a guard?  not now
-// can always use busy-wait loop?
-boolean recv_guard(int source, int dest, int tag, Comm comm) {
+  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...
 
 }
 
-void recv(int dest, void *buf, int size, int tag, int source, Comm comm, Status *status) {
-  Message *message;  
-  int message_source;
-  int message_dest;
-  Message *queue = comm->buf_front[source][dest];
-
-  // first,  the guard:
-  when ($exists {int i | 0<= i && 
-  // search the queue looking for the message
-  // also set message_source, message_dest;
-  // and remove message from the queue
-  
-  if (message.size > size) error…;
-  memcpy(buf, message.data, message.size);
-  status->size = message.size;
-  status->source = message_source;
-  status->dest = message_dest;
-  $free(message->data);
-}
-
 }}}