examples/gras/pmm/pmm.c

   1 /* $Id$ */
   2 /* pmm - parallel matrix multiplication "double diffusion"                  */
   3
   4 /* Copyright (c) 2006-2008 The SimGrid team. All rights reserved.           */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 #include "gras.h"
  10 #include "xbt/matrix.h"
  11 #include "amok/peermanagement.h"
  12
  13 #define PROC_MATRIX_SIZE 3
  14 #define NEIGHBOR_COUNT PROC_MATRIX_SIZE - 1
  15 #define SLAVE_COUNT (PROC_MATRIX_SIZE*PROC_MATRIX_SIZE)
  16
  17 #define DATA_MATRIX_SIZE 18
  18 const int submatrix_size = DATA_MATRIX_SIZE / PROC_MATRIX_SIZE;
  19
  20 XBT_LOG_NEW_DEFAULT_CATEGORY(pmm, "Parallel Matrix Multiplication");
  21
  22 /* struct for recovering results */
  23 GRAS_DEFINE_TYPE(s_result, struct s_result {
  24                  int linepos;
  25                  int rowpos; xbt_matrix_t C GRAS_ANNOTE(subtype, double);});
  26
  27 typedef struct s_result result_t;
  28
  29 /* struct to send initial data to slave */
  30 GRAS_DEFINE_TYPE(s_pmm_assignment, struct s_pmm_assignment {
  31                  int linepos;
  32                  int rowpos;
  33                  xbt_peer_t line[NEIGHBOR_COUNT];
  34                  xbt_peer_t row[NEIGHBOR_COUNT];
  35                  xbt_matrix_t A GRAS_ANNOTE(subtype, double);
  36                  xbt_matrix_t B GRAS_ANNOTE(subtype, double);});
  37
  38 typedef struct s_pmm_assignment s_pmm_assignment_t;
  39
  40 /* register messages which may be sent (common to client and server) */
  41 static void register_messages(void)
  42 {
  43   gras_datadesc_type_t result_type;
  44   gras_datadesc_type_t pmm_assignment_type;
  45
  46   gras_datadesc_set_const("NEIGHBOR_COUNT", NEIGHBOR_COUNT);
  47   result_type = gras_datadesc_by_symbol(s_result);
  48   pmm_assignment_type = gras_datadesc_by_symbol(s_pmm_assignment);
  49
  50   /* receive a final result from slave */
  51   gras_msgtype_declare("result", result_type);
  52
  53   /* send from master to slave to assign a position and some data */
  54   gras_msgtype_declare("pmm_slave", pmm_assignment_type);
  55
  56   /* send data between slaves */
  57   gras_msgtype_declare("dataA",
  58                        gras_datadesc_matrix(gras_datadesc_by_name("double"),
  59                                             NULL));
  60   gras_msgtype_declare("dataB",
  61                        gras_datadesc_matrix(gras_datadesc_by_name("double"),
  62                                             NULL));
  63 }
  64
  65 /* Function prototypes */
  66 int slave(int argc, char *argv[]);
  67 int master(int argc, char *argv[]);
  68
  69
  70 /* **********************************************************************
  71  * master code
  72  * **********************************************************************/
  73
  74 /* Global private data */
  75 typedef struct {
  76   int nbr_row, nbr_line;
  77   int remaining_step;
  78   int remaining_ack;
  79 } master_data_t;
  80
  81
  82 int master(int argc, char *argv[])
  83 {
  84
  85   int i;
  86
  87   xbt_matrix_t A, B, C;
  88   result_t result;
  89
  90   gras_socket_t from;
  91
  92   xbt_dynar_t peers;            /* group of slaves */
  93   xbt_peer_t grid[SLAVE_COUNT]; /* The slaves as an array */
  94   gras_socket_t socket[SLAVE_COUNT];    /* sockets for brodcast to slaves */
  95
  96   /* Init the GRAS's infrastructure */
  97   gras_init(&argc, argv);
  98   amok_pm_init();
  99   register_messages();
 100
 101   /* Initialize data matrices */
 102   A = xbt_matrix_double_new_id(DATA_MATRIX_SIZE, DATA_MATRIX_SIZE);
 103   B = xbt_matrix_double_new_seq(DATA_MATRIX_SIZE, DATA_MATRIX_SIZE);
 104   C = xbt_matrix_double_new_zeros(DATA_MATRIX_SIZE, DATA_MATRIX_SIZE);
 105
 106   /* Create the connexions */
 107   xbt_assert0(argc > 1, "Usage: master <port>");
 108   gras_socket_server(atoi(argv[1]));
 109   peers = amok_pm_group_new("pmm");
 110
 111   /* friends, we're ready. Come and play */
 112   INFO0("Wait for peers for 5 sec");
 113   gras_msg_handleall(5);
 114   INFO1("Got %ld pals", xbt_dynar_length(peers));
 115
 116   for (i = 0; i < xbt_dynar_length(peers) && i < SLAVE_COUNT; i++) {
 117
 118     xbt_dynar_get_cpy(peers, i, &grid[i]);
 119     socket[i] = gras_socket_client(grid[i]->name, grid[i]->port);
 120   }
 121   xbt_assert2(i == SLAVE_COUNT,
 122               "Not enough slaves for this setting (got %d of %d). Change the deployment file",
 123               i, SLAVE_COUNT);
 124
 125   /* Kill surnumerous slaves */
 126   for (i = SLAVE_COUNT; i < xbt_dynar_length(peers);) {
 127     xbt_peer_t h;
 128
 129     xbt_dynar_remove_at(peers, i, &h);
 130     INFO2("Too much slaves. Killing %s:%d", h->name, h->port);
 131     amok_pm_kill_hp(h->name, h->port);
 132     free(h);
 133   }
 134
 135
 136   /* Assign job to slaves */
 137   int row = 0, line = 0;
 138   INFO0("XXXXXXXXXXXXXXXXXXXXXX begin Multiplication");
 139   for (i = 0; i < SLAVE_COUNT; i++) {
 140     s_pmm_assignment_t assignment;
 141     int j, k;
 142
 143     assignment.linepos = line;  // assigned line
 144     assignment.rowpos = row;    // assigned row
 145
 146     /* Neiborhood */
 147     for (j = 0, k = 0; j < PROC_MATRIX_SIZE; j++) {
 148       if (i != j * PROC_MATRIX_SIZE + (row)) {
 149         assignment.row[k] = grid[j * PROC_MATRIX_SIZE + (row)];
 150         k++;
 151       }
 152     }
 153     for (j = 0, k = 0; j < PROC_MATRIX_SIZE; j++) {
 154       if (i != (line) * PROC_MATRIX_SIZE + j) {
 155         assignment.line[k] = grid[(line) * PROC_MATRIX_SIZE + j];
 156         k++;
 157       }
 158     }
 159
 160     assignment.A = xbt_matrix_new_sub(A,
 161                                       submatrix_size, submatrix_size,
 162                                       submatrix_size * line,
 163                                       submatrix_size * row, NULL);
 164     assignment.B =
 165       xbt_matrix_new_sub(B, submatrix_size, submatrix_size,
 166                          submatrix_size * line, submatrix_size * row, NULL);
 167     row++;
 168     if (row >= PROC_MATRIX_SIZE) {
 169       row = 0;
 170       line++;
 171     }
 172
 173     gras_msg_send(socket[i], "pmm_slave", &assignment);
 174     xbt_matrix_free(assignment.A);
 175     xbt_matrix_free(assignment.B);
 176   }
 177
 178   /* (have a rest while the slave perform the multiplication) */
 179
 180   /* Retrieve the results */
 181   for (i = 0; i < SLAVE_COUNT; i++) {
 182     gras_msg_wait(6000, "result", &from, &result);
 183     VERB2("%d slaves are done already. Waiting for %d", i + 1, SLAVE_COUNT);
 184     xbt_matrix_copy_values(C, result.C, submatrix_size, submatrix_size,
 185                            submatrix_size * result.linepos,
 186                            submatrix_size * result.rowpos, 0, 0, NULL);
 187     xbt_matrix_free(result.C);
 188   }
 189   /*    end of gather   */
 190
 191   if (DATA_MATRIX_SIZE < 30) {
 192     INFO0("The Result of Multiplication is :");
 193     xbt_matrix_dump(C, "C:res", 0, xbt_matrix_dump_display_double);
 194   } else {
 195     INFO1("Matrix size too big (%d>30) to be displayed here",
 196           DATA_MATRIX_SIZE);
 197   }
 198
 199   amok_pm_group_shutdown("pmm");        /* Ok, we're out of here */
 200
 201   for (i = 0; i < SLAVE_COUNT; i++) {
 202     gras_socket_close(socket[i]);
 203   }
 204
 205   xbt_matrix_free(A);
 206   xbt_matrix_free(B);
 207   xbt_matrix_free(C);
 208   gras_exit();
 209   return 0;
 210 }                               /* end_of_master */
 211
 212 /* **********************************************************************
 213  * slave code
 214  * **********************************************************************/
 215
 216 static int pmm_worker_cb(gras_msg_cb_ctx_t ctx, void *payload)
 217 {
 218   /* Recover my initialized Data and My Position */
 219   s_pmm_assignment_t assignment = *(s_pmm_assignment_t *) payload;
 220   gras_socket_t master = gras_msg_cb_ctx_from(ctx);
 221
 222   xbt_ex_t e;
 223
 224   int step, l;
 225   xbt_matrix_t bA = xbt_matrix_new(submatrix_size, submatrix_size,
 226                                    sizeof(double), NULL);
 227   xbt_matrix_t bB = xbt_matrix_new(submatrix_size, submatrix_size,
 228                                    sizeof(double), NULL);
 229
 230   int myline, myrow;
 231   xbt_matrix_t mydataA, mydataB;
 232   xbt_matrix_t bC =
 233     xbt_matrix_double_new_zeros(submatrix_size, submatrix_size);
 234
 235   result_t result;
 236
 237   gras_socket_t from;           /* to exchange data with my neighbor */
 238
 239   /* sockets for brodcast to other slave */
 240   gras_socket_t socket_line[PROC_MATRIX_SIZE - 1];
 241   gras_socket_t socket_row[PROC_MATRIX_SIZE - 1];
 242   memset(socket_line, 0, sizeof(socket_line));
 243   memset(socket_row, 0, sizeof(socket_row));
 244
 245   int i;
 246
 247   gras_os_sleep(1);             /* wait for my pals */
 248
 249   myline = assignment.linepos;
 250   myrow = assignment.rowpos;
 251   mydataA = assignment.A;
 252   mydataB = assignment.B;
 253
 254   INFO2("Receive my pos (%d,%d) and assignment", myline, myrow);
 255
 256   /* Get my neighborhood from the assignment message (skipping myself) */
 257   for (i = 0; i < PROC_MATRIX_SIZE - 1; i++) {
 258     socket_line[i] = gras_socket_client(assignment.line[i]->name,
 259                                         assignment.line[i]->port);
 260     xbt_peer_free(assignment.line[i]);
 261   }
 262   for (i = 0; i < PROC_MATRIX_SIZE - 1; i++) {
 263     socket_row[i] = gras_socket_client(assignment.row[i]->name,
 264                                        assignment.row[i]->port);
 265     xbt_peer_free(assignment.row[i]);
 266   }
 267
 268   for (step = 0; step < PROC_MATRIX_SIZE; step++) {
 269
 270     /* a line brodcast */
 271     if (myline == step) {
 272       INFO2("LINE: step(%d) = Myline(%d). Broadcast my data.", step, myline);
 273       for (l = 0; l < PROC_MATRIX_SIZE - 1; l++) {
 274         INFO1("LINE:   Send to %s", gras_socket_peer_name(socket_row[l]));
 275         gras_msg_send(socket_row[l], "dataB", &mydataB);
 276       }
 277
 278
 279       xbt_matrix_free(bB);
 280       bB = xbt_matrix_new_sub(mydataB,
 281                               submatrix_size, submatrix_size, 0, 0, NULL);
 282     } else {
 283       TRY {
 284         xbt_matrix_free(bB);
 285         gras_msg_wait(600, "dataB", &from, &bB);
 286       }
 287       CATCH(e) {
 288         RETHROW0("Can't get a data message from line : %s");
 289       }
 290       INFO3("LINE: step(%d) <> Myline(%d). Receive data from %s", step,
 291             myline, gras_socket_peer_name(from));
 292     }
 293
 294     /* a row brodcast */
 295     if (myrow == step) {
 296       INFO2("ROW: step(%d)=myrow(%d). Broadcast my data.", step, myrow);
 297       for (l = 1; l < PROC_MATRIX_SIZE; l++) {
 298         INFO1("ROW:   Send to %s", gras_socket_peer_name(socket_line[l - 1]));
 299         gras_msg_send(socket_line[l - 1], "dataA", &mydataA);
 300       }
 301       xbt_matrix_free(bA);
 302       bA = xbt_matrix_new_sub(mydataA,
 303                               submatrix_size, submatrix_size, 0, 0, NULL);
 304     } else {
 305       TRY {
 306         xbt_matrix_free(bA);
 307         gras_msg_wait(1200, "dataA", &from, &bA);
 308       }
 309       CATCH(e) {
 310         RETHROW0("Can't get a data message from row : %s");
 311       }
 312       INFO3("ROW: step(%d)<>myrow(%d). Receive data from %s", step, myrow,
 313             gras_socket_peer_name(from));
 314     }
 315     xbt_matrix_double_addmult(bA, bB, bC);
 316
 317   };
 318
 319   /* send Result to master */
 320   result.C = bC;
 321   result.linepos = myline;
 322   result.rowpos = myrow;
 323
 324   TRY {
 325     gras_msg_send(master, "result", &result);
 326   }
 327   CATCH(e) {
 328     RETHROW0("Failed to send answer to server: %s");
 329   }
 330   INFO2(">>>>>>>> Result sent to %s:%d <<<<<<<<",
 331         gras_socket_peer_name(master), gras_socket_peer_port(master));
 332   /*  Free the allocated resources, and shut GRAS down */
 333
 334   xbt_matrix_free(bA);
 335   xbt_matrix_free(bB);
 336   xbt_matrix_free(bC);
 337
 338   xbt_matrix_free(mydataA);
 339   xbt_matrix_free(mydataB);
 340   /* FIXME: some are said to be unknown
 341      gras_socket_close(master);
 342      gras_socket_close(from);
 343      for (l=0; l < PROC_MATRIX_SIZE-1; l++) {
 344      if (socket_line[l])
 345      gras_socket_close(socket_line[l]);
 346      if (socket_row[l])
 347      gras_socket_close(socket_row[l]);
 348      } */
 349
 350   return 0;
 351 }
 352
 353 int slave(int argc, char *argv[])
 354 {
 355   gras_socket_t mysock;
 356   gras_socket_t master = NULL;
 357   int connected = 0;
 358   int rank;
 359
 360   /* Init the GRAS's infrastructure */
 361   gras_init(&argc, argv);
 362   amok_pm_init();
 363   if (argc != 3 && argc != 2)
 364     xbt_die("Usage: slave masterhost:masterport [rank]");
 365   if (argc == 2)
 366     rank = -1;
 367   else
 368     rank = atoi(argv[2]);
 369
 370   /*  Register the known messages and my callback */
 371   register_messages();
 372   gras_cb_register("pmm_slave", pmm_worker_cb);
 373
 374   /* Create the connexions */
 375   mysock = gras_socket_server_range(3000, 9999, 0, 0);
 376   INFO1("Sensor %d starting", rank);
 377   while (!connected) {
 378     xbt_ex_t e;
 379     TRY {
 380       master = gras_socket_client_from_string(argv[1]);
 381       connected = 1;
 382     }
 383     CATCH(e) {
 384       if (e.category != system_error)
 385         RETHROW;
 386       xbt_ex_free(e);
 387       gras_os_sleep(0.5);
 388     }
 389   }
 390
 391   /* Join and run the group */
 392   amok_pm_group_join(master, "pmm", rank);
 393   amok_pm_mainloop(600);
 394
 395   /* housekeeping */
 396   gras_socket_close(mysock);
 397   //  gras_socket_close(master); Unknown
 398   gras_exit();
 399   return 0;
 400 }                               /* end_of_slave */