* change from any-source style messaging to source-specified receives.

* Update cofs module to match

This commit was SVN r288.

src/mca/lam/oob/cofs/src/oob_cofs.c

@@ -18,7 +18,7 @@
 #include <unistd.h>
 
 static int blocking_recv_posted = 0;
-static int do_recv(lam_job_handle_t job_handle, int* tag, int* vpid, 
+static int do_recv(lam_job_handle_t job_handle, int vpid, int* tag,
                    void** data, size_t* data_len);
 
 int
@@ -69,13 +69,13 @@ mca_oob_cofs_send(lam_job_handle_t job_handle, int vpid, int tag,
 
 
 int
-mca_oob_cofs_recv(lam_job_handle_t job_handle, int* tag, int* vpid, 
+mca_oob_cofs_recv(lam_job_handle_t job_handle, int vpid, int* tag,
                   void** data, size_t* data_len)
 {
   int ret = LAM_ERR_WOULD_BLOCK;
   blocking_recv_posted = 1;
   while (ret == LAM_ERR_WOULD_BLOCK) {
-    ret = do_recv(job_handle, tag, vpid, data, data_len);
+    ret = do_recv(job_handle, vpid, tag, data, data_len);
   }
   blocking_recv_posted = 0;
   return ret;
@@ -83,19 +83,19 @@ mca_oob_cofs_recv(lam_job_handle_t job_handle, int* tag, int* vpid,
 
 
 int
-mca_oob_cofs_recv_nb(lam_job_handle_t job_handle, int* tag, int* vpid, 
+mca_oob_cofs_recv_nb(lam_job_handle_t job_handle, int vpid, int* tag,
                      void** data, size_t* data_len)
 {
   if (blocking_recv_posted != 0) {
     return LAM_ERR_WOULD_BLOCK;
   }
 
-  return do_recv(job_handle, tag, vpid, data, data_len);
+  return do_recv(job_handle, vpid, tag, data, data_len);
 }
 
 
 int
-mca_oob_cofs_recv_cb(lam_job_handle_t job_handle, int tag, 
+mca_oob_cofs_recv_cb(lam_job_handle_t job_handle, int vpid, int tag, 
                      mca_oob_base_recv_cb_t callback)
 {
   return LAM_ERR_NOT_SUPPORTED;
@@ -103,7 +103,7 @@ mca_oob_cofs_recv_cb(lam_job_handle_t job_handle, int tag,
 
 
 static char*
-find_match(lam_job_handle_t job_handle, int* tag, int* vpid)
+find_match(lam_job_handle_t job_handle, int vpid, int* tag)
 {
   DIR* dir;
   struct dirent *ent;
@@ -113,7 +113,7 @@ find_match(lam_job_handle_t job_handle, int* tag, int* vpid)
   int ret;
   bool found = false;
   char best_name[LAM_PATH_MAX];
-  int best_tag, best_vpid;
+  int best_tag;
   uint64_t best_serial = ((1ULL << 63) - 1);
 
   dir = opendir(mca_oob_cofs_comm_loc);
@@ -139,13 +139,15 @@ find_match(lam_job_handle_t job_handle, int* tag, int* vpid)
     if (*tag != MCA_OOB_ANY_TAG && tmp_tag != *tag) {
       continue;
     }
+    if (tmp_vpid != vpid) {
+      continue;
+    }
 
     /* do best one here... */
     found = true;
     if (tmp_serial < best_serial) {
       strcpy(best_name, ent->d_name);
       best_tag = tmp_tag;
-      best_vpid = tmp_vpid;
       best_serial = tmp_serial;
     }
   }
@@ -153,7 +155,6 @@ find_match(lam_job_handle_t job_handle, int* tag, int* vpid)
   closedir(dir);
   if (found) {
     *tag = best_tag;
-    *vpid = best_vpid;
     return strdup(best_name);
   } else {
     return NULL;
@@ -162,7 +163,7 @@ find_match(lam_job_handle_t job_handle, int* tag, int* vpid)
 
 
 static int
-do_recv(lam_job_handle_t job_handle, int* tag, int* vpid, 
+do_recv(lam_job_handle_t job_handle, int vpid, int* tag,
         void** data, size_t* data_len)
 {
   char *fname;
@@ -170,7 +171,7 @@ do_recv(lam_job_handle_t job_handle, int* tag, int* vpid,
   FILE *fp;
   size_t rlen;
 
-  fname = find_match(job_handle, tag, vpid);
+  fname = find_match(job_handle, vpid, tag);
   if (fname == NULL) {
     return LAM_ERR_WOULD_BLOCK;
   }

src/mca/lam/oob/cofs/src/oob_cofs.h

@@ -28,11 +28,11 @@ int mca_oob_cofs_finalize(void);
  */
 int mca_oob_cofs_send(lam_job_handle_t job_handle, int vpid, int tag, 
                       void* data, size_t data_len);
-int mca_oob_cofs_recv(lam_job_handle_t job_handle, int* tag, int* vpid, 
+int mca_oob_cofs_recv(lam_job_handle_t job_handle, int vpid, int* tag,
                       void** data, size_t* data_len);
-int mca_oob_cofs_recv_nb(lam_job_handle_t job_handle, int* tag, int* vpid, 
+int mca_oob_cofs_recv_nb(lam_job_handle_t job_handle,  int vpid, int* tag,
                          void** data, size_t* data_len);
-int mca_oob_cofs_recv_cb(lam_job_handle_t job_handle, int tag, 
+int mca_oob_cofs_recv_cb(lam_job_handle_t job_handle, int vpid, int tag, 
                          mca_oob_base_recv_cb_t callback);
 
 

src/mca/lam/oob/oob.h

@@ -59,11 +59,11 @@ typedef int (*mca_oob_base_query_fn_t)(int *priority);
 typedef struct mca_oob_1_0_0_t* (*mca_oob_base_init_fn_t)(void);
 typedef int (*mca_oob_base_send_fn_t)(lam_job_handle_t job_handle, int vpid, int tag, 
                                       void* data, size_t data_len);
-typedef int (*mca_oob_base_recv_fn_t)(lam_job_handle_t job_handle, int* tag, int* vpid, 
+typedef int (*mca_oob_base_recv_fn_t)(lam_job_handle_t job_handle,  int vpid, int* tag, 
                                       void** data, size_t* data_len);
-typedef int (*mca_oob_base_recv_nb_fn_t)(lam_job_handle_t job_handle, int* tag, int* vpid, 
+typedef int (*mca_oob_base_recv_nb_fn_t)(lam_job_handle_t job_handle, int vpid, int* tag,  
                                          void** data, size_t* data_len);
-typedef int (*mca_oob_base_recv_cb_fn_t)(lam_job_handle_t job_handle, int tag, 
+typedef int (*mca_oob_base_recv_cb_fn_t)(lam_job_handle_t job_handle, int vpid, int tag, 
                                          mca_oob_base_recv_cb_t callback);
 typedef int (*mca_oob_base_finalize_fn_t)(void);
 

test/unit/lam/oob_cofs/oob_cofs_test.c

@@ -63,7 +63,7 @@ main(int argc, char* argv[])
   printf("#\n# Receiving Messages\n#\n\n");
   for (msg_count = 0 ; msg_count < 20 ; ++msg_count) {
     source_tag = 1;
-    ret = mca_oob_cofs_recv("foobar", &source_tag, &source_vpid, (void**) &source_data, &source_len);
+    ret = mca_oob_cofs_recv("foobar", target_vpid, &source_tag, (void**) &source_data, &source_len);
     if (ret != LAM_SUCCESS) {
       printf("mca_oob_cofs_recv failed on msg_count %d, %d\n", msg_count, ret);
       exit(1);