Make the pmm robust to the change of amount of slaves

[simgrid.git] / examples / gras / pmm / pmm.c

diff --git a/examples/gras/pmm/pmm.c b/examples/gras/pmm/pmm.c
old mode 100755 (executable)
new mode 100644 (file)
index b29af45..7e7d35d
--- a/examples/gras/pmm/pmm.c
+++ b/examples/gras/pmm/pmm.c
@@ -10,7 +10,7 @@
 #include "xbt/matrix.h"
 #include "amok/peermanagement.h"
 
-#define PROC_MATRIX_SIZE 3
+#define PROC_MATRIX_SIZE 2
 #define NEIGHBOR_COUNT PROC_MATRIX_SIZE - 1
 #define SLAVE_COUNT (PROC_MATRIX_SIZE*PROC_MATRIX_SIZE)
 
@@ -21,8 +21,9 @@ XBT_LOG_NEW_DEFAULT_CATEGORY(pmm, "Parallel Matrix Multiplication");
 
 /* struct for recovering results */
 GRAS_DEFINE_TYPE(s_result, struct s_result {
-                 int linepos;
-                 int rowpos; xbt_matrix_t C GRAS_ANNOTE(subtype, double);});
+                 int linepos; int rowpos;
+                 xbt_matrix_t C GRAS_ANNOTE(subtype, double);
+                 });
 
 typedef struct s_result result_t;
 
@@ -33,7 +34,8 @@ GRAS_DEFINE_TYPE(s_pmm_assignment, struct s_pmm_assignment {
                  xbt_peer_t line[NEIGHBOR_COUNT];
                  xbt_peer_t row[NEIGHBOR_COUNT];
                  xbt_matrix_t A GRAS_ANNOTE(subtype, double);
-                 xbt_matrix_t B GRAS_ANNOTE(subtype, double);});
+                 xbt_matrix_t B GRAS_ANNOTE(subtype, double);
+                 });
 
 typedef struct s_pmm_assignment s_pmm_assignment_t;
 
@@ -55,11 +57,14 @@ static void register_messages(void)
 
   /* send data between slaves */
   gras_msgtype_declare("dataA",
-                       gras_datadesc_matrix(gras_datadesc_by_name("double"),
-                                            NULL));
+                       gras_datadesc_matrix(gras_datadesc_by_name
+                                            ("double"), NULL));
   gras_msgtype_declare("dataB",
-                       gras_datadesc_matrix(gras_datadesc_by_name("double"),
-                                            NULL));
+                       gras_datadesc_matrix(gras_datadesc_by_name
+                                            ("double"), NULL));
+
+  /* synchronization message */
+  gras_msgtype_declare("pmm_sync", 0);
 }
 
 /* Function prototypes */
@@ -110,10 +115,11 @@ int master(int argc, char *argv[])
   /* friends, we're ready. Come and play */
   INFO0("Wait for peers for 2 sec");
   gras_msg_handleall(2);
-  while (xbt_dynar_length(peers)<9) {
-    INFO1("Got only %ld pals. Wait 2 more seconds", xbt_dynar_length(peers));
+  do {
+    INFO2("Got only %ld pals (of %d). Wait 2 more seconds",
+        xbt_dynar_length(peers),SLAVE_COUNT);
     gras_msg_handleall(2);
-  }
+  } while (xbt_dynar_length(peers) < SLAVE_COUNT);
   INFO1("Good. Got %ld pals", xbt_dynar_length(peers));
 
   for (i = 0; i < xbt_dynar_length(peers) && i < SLAVE_COUNT; i++) {
@@ -164,8 +170,9 @@ int master(int argc, char *argv[])
                                       submatrix_size * line,
                                       submatrix_size * row, NULL);
     assignment.B =
-      xbt_matrix_new_sub(B, submatrix_size, submatrix_size,
-                         submatrix_size * line, submatrix_size * row, NULL);
+        xbt_matrix_new_sub(B, submatrix_size, submatrix_size,
+                           submatrix_size * line, submatrix_size * row,
+                           NULL);
     row++;
     if (row >= PROC_MATRIX_SIZE) {
       row = 0;
@@ -177,12 +184,20 @@ int master(int argc, char *argv[])
     xbt_matrix_free(assignment.B);
   }
 
-  /* (have a rest while the slave perform the multiplication) */
+  /* synchronize slaves */
+  for (i = 0; i < PROC_MATRIX_SIZE; i++) {
+    int j;
+    for (j = 0; j < SLAVE_COUNT; j++)
+      gras_msg_wait(600, "pmm_sync", NULL, NULL);
+    for (j = 0; j < SLAVE_COUNT; j++)
+      gras_msg_send(socket[j], "pmm_sync", NULL);
+  }
 
   /* Retrieve the results */
   for (i = 0; i < SLAVE_COUNT; i++) {
     gras_msg_wait(6000, "result", &from, &result);
-    VERB2("%d slaves are done already. Waiting for %d", i + 1, SLAVE_COUNT);
+    VERB2("%d slaves are done already. Waiting for %d", i + 1,
+          SLAVE_COUNT);
     xbt_matrix_copy_values(C, result.C, submatrix_size, submatrix_size,
                            submatrix_size * result.linepos,
                            submatrix_size * result.rowpos, 0, 0, NULL);
@@ -194,13 +209,13 @@ int master(int argc, char *argv[])
     INFO0("XXXXXXXXXXXXXXXXXXXXXX Ok, the result matches expectations");
   else {
     WARN0("the result seems wrong");
-  if (DATA_MATRIX_SIZE < 30) {
-    INFO0("The Result of Multiplication is :");
-    xbt_matrix_dump(C, "C:res", 0, xbt_matrix_dump_display_double);
-  } else {
-    INFO1("Matrix size too big (%d>30) to be displayed here",
-          DATA_MATRIX_SIZE);
-  }
+    if (DATA_MATRIX_SIZE < 30) {
+      INFO0("The Result of Multiplication is :");
+      xbt_matrix_dump(C, "C:res", 0, xbt_matrix_dump_display_double);
+    } else {
+      INFO1("Matrix size too big (%d>30) to be displayed here",
+            DATA_MATRIX_SIZE);
+    }
   }
 
   amok_pm_group_shutdown("pmm");        /* Ok, we're out of here */
@@ -236,7 +251,7 @@ static int pmm_worker_cb(gras_msg_cb_ctx_t ctx, void *payload)
   int myline, myrow;
   xbt_matrix_t mydataA, mydataB;
   xbt_matrix_t bC =
-    xbt_matrix_double_new_zeros(submatrix_size, submatrix_size);
+      xbt_matrix_double_new_zeros(submatrix_size, submatrix_size);
 
   result_t result;
 
@@ -275,10 +290,13 @@ static int pmm_worker_cb(gras_msg_cb_ctx_t ctx, void *payload)
   }
 
   for (step = 0; step < PROC_MATRIX_SIZE; step++) {
+    gras_msg_send(master, "pmm_sync", NULL);
+    gras_msg_wait(600, "pmm_sync", NULL, NULL);
 
     /* a line brodcast */
     if (myline == step) {
-      VERB2("LINE: step(%d) = Myline(%d). Broadcast my data.", step, myline);
+      VERB2("LINE: step(%d) = Myline(%d). Broadcast my data.", step,
+            myline);
       for (l = 0; l < PROC_MATRIX_SIZE - 1; l++) {
         VERB1("LINE:   Send to %s", gras_socket_peer_name(socket_row[l]));
         gras_msg_send(socket_row[l], "dataB", &mydataB);
@@ -304,7 +322,8 @@ static int pmm_worker_cb(gras_msg_cb_ctx_t ctx, void *payload)
     if (myrow == step) {
       VERB2("ROW: step(%d)=myrow(%d). Broadcast my data.", step, myrow);
       for (l = 1; l < PROC_MATRIX_SIZE; l++) {
-        VERB1("ROW:   Send to %s", gras_socket_peer_name(socket_line[l - 1]));
+        VERB1("ROW:   Send to %s",
+              gras_socket_peer_name(socket_line[l - 1]));
         gras_msg_send(socket_line[l - 1], "dataA", &mydataA);
       }
       xbt_matrix_free(bA);
@@ -396,6 +415,7 @@ int slave(int argc, char *argv[])
       gras_os_sleep(0.5);
     }
   }
+  INFO2("Connected to master: %s:%d",gras_socket_peer_name(master),gras_socket_peer_port(master));
 
   /* Join and run the group */
   rank = amok_pm_group_join(master, "pmm");