examples/msg/pmm/msg_pmm.c

   1 #include "msg/msg.h"
   2 #include "xbt/matrix.h"
   3 #include "xbt/log.h"
   4 #include "xbt/xbt_os_time.h"
   5
   6 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pmm,
   7                              "Messages specific for this msg example");
   8
   9 #define MAILBOX_NAME_SIZE 10
  10 #define MATRIX_SIZE 18
  11 #define GRID_SIZE 3
  12 #define GRID_NUM_NODES GRID_SIZE * GRID_SIZE
  13 #define NODE_MATRIX_SIZE MATRIX_SIZE / GRID_SIZE
  14 #define NEIGHBOURS_COUNT GRID_SIZE - 1
  15
  16 /*
  17  * Task data
  18  */
  19 typedef struct s_task_data{
  20   int row;
  21   int col;
  22   int nodes_in_row[NEIGHBOURS_COUNT];
  23   int nodes_in_col[NEIGHBOURS_COUNT];
  24   xbt_matrix_t A;
  25   xbt_matrix_t B;
  26 } s_task_data_t, *task_data_t;
  27
  28 /*
  29  * Node data
  30  */
  31 typedef struct s_node{
  32   int id;
  33   char mailbox[MAILBOX_NAME_SIZE];
  34   task_data_t job;
  35   xbt_matrix_t C;
  36 } s_node_t, *node_t;
  37
  38 /**
  39  * Structure for recovering results
  40  */
  41 typedef struct s_result {
  42   int row;
  43   int col;
  44   xbt_matrix_t C;
  45 } s_result_t, *result_t;
  46
  47 int node(int argc, char **argv);
  48 static void assign_tasks(xbt_matrix_t A, xbt_matrix_t B);
  49 static task_data_t wait_task(int selfid);
  50 static void broadcast_matrix(xbt_matrix_t M, int num_nodes, int *nodes);
  51 static void get_sub_matrix(xbt_matrix_t *sM, int selfid);
  52 static void task_cleanup(void *arg);
  53
  54 int node(int argc, char **argv)
  55 {
  56   int j,k;
  57   xbt_matrix_t A, B, C, sA, sB;
  58   result_t result;
  59
  60   xbt_assert0(argc != 1, "Wrong number of arguments for this node");
  61
  62   /* Initialize node information (id and mailbox) */
  63   s_node_t mydata = {0};
  64   mydata.id = atoi(argv[1]);
  65   snprintf(mydata.mailbox, MAILBOX_NAME_SIZE - 1, "%d", mydata.id);
  66   mydata.C = xbt_matrix_double_new_zeros(NODE_MATRIX_SIZE, NODE_MATRIX_SIZE);
  67
  68   if(mydata.id == 0){
  69     /* Initialize data matrices */
  70     A = xbt_matrix_double_new_id(MATRIX_SIZE, MATRIX_SIZE);
  71     B = xbt_matrix_double_new_seq(MATRIX_SIZE, MATRIX_SIZE);
  72     C = xbt_matrix_double_new_zeros(MATRIX_SIZE, MATRIX_SIZE);
  73
  74     /* Get own job first */
  75     mydata.job = xbt_new0(s_task_data_t, 1);
  76     mydata.job->row = 0;
  77     mydata.job->col = 0;
  78     mydata.job->A =
  79       xbt_matrix_new_sub(A, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE, 0, 0, NULL);
  80     mydata.job->B =
  81       xbt_matrix_new_sub(B, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE, 0, 0, NULL);
  82
  83     for (j = 0, k = 0; j < GRID_SIZE; j++) {
  84       if (j != 0) {
  85         mydata.job->nodes_in_row[k] = j;
  86         k++;
  87       }
  88     }
  89
  90     for (j = 0, k = 0; j < GRID_SIZE; j++) {
  91       if (GRID_SIZE * j != 0) {
  92         mydata.job->nodes_in_col[k] = GRID_SIZE * j;
  93         k++;
  94       }
  95     }
  96
  97     /* Broadcast the rest of the jobs to the other nodes */
  98     assign_tasks(A,B);
  99
 100   }else{
 101     mydata.job = wait_task(mydata.id);
 102   }
 103
 104   /* Multiplication main-loop */
 105   XBT_CRITICAL("Start Multiplication's Main-loop");
 106   for(k=0; k < GRID_SIZE; k++){
 107     if(k == mydata.job->col){
 108       XBT_VERB("Broadcast sA(%d,%d) to row %d", mydata.job->row, k, mydata.job->row);
 109       broadcast_matrix(mydata.job->A, NEIGHBOURS_COUNT, mydata.job->nodes_in_row);
 110     }
 111
 112     if(k == mydata.job->row){
 113       XBT_VERB("Broadcast sB(%d,%d) to col %d", k, mydata.job->col, mydata.job->col);
 114       broadcast_matrix(mydata.job->B, NEIGHBOURS_COUNT, mydata.job->nodes_in_col);
 115     }
 116
 117     if(mydata.job->row == k && mydata.job->col == k){
 118       xbt_matrix_double_addmult(mydata.job->A, mydata.job->B, mydata.C);
 119     }else if(mydata.job->row == k){
 120       get_sub_matrix(&sA, mydata.id);
 121       xbt_matrix_double_addmult(sA, mydata.job->B, mydata.C);
 122       xbt_matrix_free(sA);
 123     }else if(mydata.job->col == k){
 124       get_sub_matrix(&sB, mydata.id);
 125       xbt_matrix_double_addmult(mydata.job->A, sB, mydata.C);
 126       xbt_matrix_free(sB);
 127     }else{
 128       get_sub_matrix(&sA, mydata.id);
 129       get_sub_matrix(&sB, mydata.id);
 130       xbt_matrix_double_addmult(sA, sB, mydata.C);
 131       xbt_matrix_free(sA);
 132       xbt_matrix_free(sB);
 133     }
 134   }
 135
 136   /* Node 0: gather the results and reconstruct the final matrix */
 137   if(mydata.id == 0){
 138     int node;
 139     msg_comm_t comms[GRID_NUM_NODES-1] = {0};
 140     m_task_t tasks[GRID_NUM_NODES-1] = {0};
 141
 142     XBT_CRITICAL("Multiplication done. Reconstruct the result.");
 143
 144     /* First add our results */
 145     xbt_matrix_copy_values(C, mydata.C, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE,
 146                            0, 0, 0, 0, NULL);
 147
 148     /* Get the result from the nodes in the GRID */
 149     for (node = 1; node < GRID_NUM_NODES; node++){
 150       comms[node-1] = MSG_task_irecv(&tasks[node-1], mydata.mailbox);
 151     }
 152     MSG_comm_waitall(comms, GRID_NUM_NODES - 1, -1);
 153
 154     /* Reconstruct the result matrix */
 155     for (node = 1; node < GRID_NUM_NODES; node++){
 156       result = (result_t)MSG_task_get_data(tasks[node-1]);
 157       xbt_matrix_copy_values(C, result->C, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE,
 158           NODE_MATRIX_SIZE * result->row, NODE_MATRIX_SIZE * result->col, 0, 0, NULL);
 159       xbt_matrix_free(result->C);
 160       xbt_free(result);
 161       MSG_task_destroy(tasks[node-1]);
 162     }
 163
 164     xbt_matrix_dump(C, "C:res", 0, xbt_matrix_dump_display_double);
 165
 166   /* The rest: return the result to node 0 */
 167   }else{
 168     m_task_t task;
 169
 170     XBT_CRITICAL("Multiplication done. Send the sub-result.");
 171
 172     result = xbt_new0(s_result_t, 1);
 173     result->row = mydata.job->row;
 174     result->col = mydata.job->col;
 175     result->C =
 176       xbt_matrix_new_sub(mydata.C, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE, 0, 0, NULL);
 177     task = MSG_task_create("result",100,100,result);
 178     MSG_task_dsend(task, "0", NULL);
 179   }
 180
 181   /* Clean up and finish*/
 182   xbt_matrix_free(mydata.job->A);
 183   xbt_matrix_free(mydata.job->B);
 184   xbt_free(mydata.job);
 185   return 0;
 186 }
 187
 188 /*
 189  * Assign the tasks to the GRID
 190  */
 191 static void assign_tasks(xbt_matrix_t A, xbt_matrix_t B)
 192 {
 193   int node, j, k, row = 0, col = 1;
 194   char node_mbox[MAILBOX_NAME_SIZE];
 195   m_task_t task;
 196   task_data_t assignment;
 197   msg_comm_t comms[GRID_NUM_NODES - 1] = {0};
 198
 199   XBT_CRITICAL("Assign tasks");
 200   for (node = 1; node < GRID_NUM_NODES; node++){
 201     assignment = xbt_new0(s_task_data_t, 1);
 202     assignment->row = row;
 203     assignment->col = col;
 204
 205     /* Compute who are the peers in the same row and column */
 206     /* than the node receiving this task and include this
 207      * information in the assignment */
 208     for (j = 0, k = 0; j < GRID_SIZE; j++) {
 209       if (node != (GRID_SIZE * row) + j) {
 210         assignment->nodes_in_row[k] = (GRID_SIZE * row) + j;
 211         k++;
 212       }
 213     }
 214
 215     for (j = 0, k = 0; j < GRID_SIZE; j++) {
 216       if (node != (GRID_SIZE * j) + col) {
 217         assignment->nodes_in_col[k] = (GRID_SIZE * j) + col;
 218         k++;
 219       }
 220     }
 221
 222     assignment->A =
 223       xbt_matrix_new_sub(A, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE,
 224                          NODE_MATRIX_SIZE * row, NODE_MATRIX_SIZE * col,
 225                          NULL);
 226     assignment->B =
 227       xbt_matrix_new_sub(B, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE,
 228                          NODE_MATRIX_SIZE * row, NODE_MATRIX_SIZE * col,
 229                          NULL);
 230
 231     col++;
 232     if (col >= GRID_SIZE){
 233       col = 0;
 234       row++;
 235     }
 236
 237     task = MSG_task_create("Job", 100, 100, assignment);
 238     snprintf(node_mbox, MAILBOX_NAME_SIZE - 1, "%d", node);
 239     comms[node-1] = MSG_task_isend(task, node_mbox);
 240   }
 241
 242   MSG_comm_waitall(comms, GRID_NUM_NODES-1, -1);
 243 }
 244
 245 static task_data_t wait_task(int selfid)
 246 {
 247   m_task_t task = NULL;
 248   char self_mbox[MAILBOX_NAME_SIZE];
 249   task_data_t assignment;
 250   snprintf(self_mbox, MAILBOX_NAME_SIZE - 1, "%d", selfid);
 251   MSG_task_receive(&task, self_mbox);
 252   assignment = (task_data_t)MSG_task_get_data(task);
 253   MSG_task_destroy(task);
 254   XBT_CRITICAL("Got Job (%d,%d)", assignment->row, assignment->col);
 255
 256   return assignment;
 257 }
 258
 259 static void broadcast_matrix(xbt_matrix_t M, int num_nodes, int *nodes)
 260 {
 261   int node;
 262   char node_mbox[MAILBOX_NAME_SIZE];
 263   m_task_t task;
 264   xbt_matrix_t sM;
 265
 266   for(node=0; node < num_nodes; node++){
 267     snprintf(node_mbox, MAILBOX_NAME_SIZE - 1, "%d", nodes[node]);
 268     sM = xbt_matrix_new_sub(M, NODE_MATRIX_SIZE, NODE_MATRIX_SIZE, 0, 0, NULL);
 269     task = MSG_task_create("sub-matrix", 100, 100, sM);
 270     MSG_task_dsend(task, node_mbox, task_cleanup);
 271     XBT_DEBUG("sub-matrix sent to %s", node_mbox);
 272   }
 273
 274 }
 275
 276 static void get_sub_matrix(xbt_matrix_t *sM, int selfid)
 277 {
 278   m_task_t task = NULL;
 279   char node_mbox[MAILBOX_NAME_SIZE];
 280
 281   XBT_VERB("Get sub-matrix");
 282
 283   snprintf(node_mbox, MAILBOX_NAME_SIZE - 1, "%d", selfid);
 284   MSG_task_receive(&task, node_mbox);
 285   *sM = (xbt_matrix_t)MSG_task_get_data(task);
 286   MSG_task_destroy(task);
 287 }
 288
 289 static void task_cleanup(void *arg){
 290   m_task_t task = (m_task_t)arg;
 291   xbt_matrix_t m = (xbt_matrix_t)MSG_task_get_data(task);
 292   xbt_matrix_free(m);
 293   MSG_task_destroy(task);
 294 }
 295
 296 /**
 297  * \brief Main function.
 298  */
 299 int main(int argc, char *argv[])
 300 {
 301   xbt_os_timer_t timer = xbt_os_timer_new();
 302
 303   MSG_global_init(&argc, argv);
 304
 305   char **options = &argv[1];
 306   const char* platform_file = options[0];
 307   const char* application_file = options[1];
 308
 309   MSG_set_channel_number(0);
 310   MSG_create_environment(platform_file);
 311
 312   MSG_function_register("node", node);
 313   MSG_launch_application(application_file);
 314
 315   xbt_os_timer_start(timer);
 316   MSG_error_t res = MSG_main();
 317   xbt_os_timer_stop(timer);
 318   XBT_CRITICAL("Simulated time: %g", MSG_get_clock());
 319
 320   MSG_clean();
 321
 322   if (res == MSG_OK)
 323     return 0;
 324   else
 325     return 1;
 326 }
 327
 328
 329