examples/msg/actions/actions.c

   1 /* Copyright (c) 2009, 2010. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include <stdio.h>
   8 #include <stdlib.h>
   9 #include "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
  10 #include "simix/simix.h"        /* semaphores for the barrier */
  11 #include "xbt.h"                /* calloc, printf */
  12 #include "instr/instr_private.h"
  13
  14 void SIMIX_ctx_raw_factory_init(smx_context_factory_t *factory);
  15
  16 XBT_LOG_NEW_DEFAULT_CATEGORY(actions,
  17                              "Messages specific for this msg example");
  18 int communicator_size = 0;
  19
  20 static void action_Isend(const char *const *action);
  21
  22 typedef struct  {
  23   int last_Irecv_sender_id;
  24   int bcast_counter;
  25   int reduce_counter;
  26   int allReduce_counter;
  27   xbt_dynar_t isends; /* of msg_comm_t */
  28   /* Used to implement irecv+wait */
  29   xbt_dynar_t irecvs; /* of msg_comm_t */
  30   xbt_dynar_t tasks; /* of m_task_t */
  31 } s_process_globals_t, *process_globals_t;
  32
  33 /* Helper function */
  34 static double parse_double(const char *string)
  35 {
  36   double value;
  37   char *endptr;
  38
  39   value = strtod(string, &endptr);
  40   if (*endptr != '\0')
  41     THROW1(unknown_error, 0, "%s is not a double", string);
  42   return value;
  43 }
  44
  45 static int get_rank (const char *process_name)
  46 {
  47   return atoi(&(process_name[1]));
  48 }
  49
  50 static void asynchronous_cleanup(void) {
  51   process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
  52
  53   /* Destroy any isend which correspond to completed communications */
  54   int found;
  55   msg_comm_t comm;
  56   while ((found = MSG_comm_testany(globals->isends)) != -1) {
  57     xbt_dynar_remove_at(globals->isends,found,&comm);
  58     MSG_comm_destroy(comm);
  59   }
  60 }
  61
  62 /* My actions */
  63 static void action_send(const char *const *action)
  64 {
  65   char *name = NULL;
  66   char to[250];
  67   const char *size_str = action[3];
  68   double size=parse_double(size_str);
  69   double clock = MSG_get_clock(); /* this "call" is free thanks to inlining */
  70
  71   sprintf(to, "%s_%s", MSG_process_get_name(MSG_process_self()),action[2]);
  72
  73   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
  74     name = xbt_str_join_array(action, " ");
  75
  76 #ifdef HAVE_TRACING
  77   int rank = get_rank(MSG_process_get_name(MSG_process_self()));
  78   int dst_traced = get_rank(action[2]);
  79   TRACE_smpi_ptp_in(rank, rank, dst_traced, "send");
  80   TRACE_smpi_send(rank, rank, dst_traced);
  81 #endif
  82
  83   XBT_DEBUG("Entering Send: %s (size: %lg)", name, size);
  84    if (size<65536) {
  85      action_Isend(action);
  86    } else {
  87      MSG_task_send(MSG_task_create(name, 0, size, NULL), to);
  88    }
  89
  90    XBT_VERB("%s %f", name, MSG_get_clock() - clock);
  91
  92   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
  93     free(name);
  94
  95 #ifdef HAVE_TRACING
  96   TRACE_smpi_ptp_out(rank, rank, dst_traced, "send");
  97 #endif
  98
  99   asynchronous_cleanup();
 100 }
 101
 102 static void action_Isend(const char *const *action)
 103 {
 104   char to[250];
 105   const char *size = action[3];
 106   double clock = MSG_get_clock();
 107   process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
 108
 109
 110   sprintf(to, "%s_%s", MSG_process_get_name(MSG_process_self()),action[2]);
 111   msg_comm_t comm =
 112       MSG_task_isend( MSG_task_create(to,0,parse_double(size),NULL), to);
 113   xbt_dynar_push(globals->isends,&comm);
 114
 115   XBT_DEBUG("Isend on %s", MSG_process_get_name(MSG_process_self()));
 116   XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
 117
 118   asynchronous_cleanup();
 119 }
 120
 121
 122 static void action_recv(const char *const *action)
 123 {
 124   char *name = NULL;
 125   char mailbox_name[250];
 126   m_task_t task = NULL;
 127   double clock = MSG_get_clock();
 128
 129   sprintf(mailbox_name, "%s_%s", action[2],
 130           MSG_process_get_name(MSG_process_self()));
 131
 132   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 133     name = xbt_str_join_array(action, " ");
 134
 135 #ifdef HAVE_TRACING
 136   int rank = get_rank(MSG_process_get_name(MSG_process_self()));
 137   int src_traced = get_rank(action[2]);
 138   TRACE_smpi_ptp_in(rank, src_traced, rank, "recv");
 139 #endif
 140
 141   XBT_DEBUG("Receiving: %s", name);
 142   MSG_task_receive(&task, mailbox_name);
 143   //  MSG_task_receive(&task, MSG_process_get_name(MSG_process_self()));
 144   XBT_VERB("%s %f", name, MSG_get_clock() - clock);
 145   MSG_task_destroy(task);
 146
 147   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 148     free(name);
 149 #ifdef HAVE_TRACING
 150   TRACE_smpi_ptp_out(rank, src_traced, rank, "recv");
 151   TRACE_smpi_recv(rank, src_traced, rank);
 152 #endif
 153
 154   asynchronous_cleanup();
 155 }
 156
 157 static void action_Irecv(const char *const *action)
 158 {
 159   char mailbox[250];
 160   double clock = MSG_get_clock();
 161   process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
 162
 163   XBT_DEBUG("Irecv on %s", MSG_process_get_name(MSG_process_self()));
 164 #ifdef HAVE_TRACING
 165   int rank = get_rank(MSG_process_get_name(MSG_process_self()));
 166   int src_traced = get_rank(action[2]);
 167   globals->last_Irecv_sender_id = src_traced;
 168   MSG_process_set_data(MSG_process_self(), (void *) globals);
 169
 170   TRACE_smpi_ptp_in(rank, src_traced, rank, "Irecv");
 171 #endif
 172
 173   sprintf(mailbox, "%s_%s", action[2],
 174           MSG_process_get_name(MSG_process_self()));
 175   m_task_t t=NULL;
 176   xbt_dynar_push(globals->tasks,&t);
 177   msg_comm_t c =
 178       MSG_task_irecv(
 179           xbt_dynar_get_ptr(globals->tasks, xbt_dynar_length(globals->tasks)-1),
 180           mailbox);
 181   xbt_dynar_push(globals->irecvs,&c);
 182
 183   XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
 184
 185 #ifdef HAVE_TRACING
 186   TRACE_smpi_ptp_out(rank, src_traced, rank, "Irecv");
 187 #endif
 188
 189   asynchronous_cleanup();
 190 }
 191
 192
 193 static void action_wait(const char *const *action)
 194 {
 195   char *name = NULL;
 196   m_task_t task = NULL;
 197   msg_comm_t comm;
 198   double clock = MSG_get_clock();
 199   process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
 200
 201   xbt_assert1(xbt_dynar_length(globals->irecvs),
 202       "action wait not preceded by any irecv: %s", xbt_str_join_array(action," "));
 203
 204   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 205     name = xbt_str_join_array(action, " ");
 206 #ifdef HAVE_TRACING
 207   process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self());
 208   int src_traced = counters->last_Irecv_sender_id;
 209   int rank = get_rank(MSG_process_get_name(MSG_process_self()));
 210   TRACE_smpi_ptp_in(rank, src_traced, rank, "wait");
 211 #endif
 212
 213   XBT_DEBUG("Entering %s", name);
 214   comm = xbt_dynar_pop_as(globals->irecvs,msg_comm_t);
 215   MSG_comm_wait(comm,-1);
 216   task = xbt_dynar_pop_as(globals->tasks,m_task_t);
 217   MSG_comm_destroy(comm);
 218   MSG_task_destroy(task);
 219
 220   XBT_VERB("%s %f", name, MSG_get_clock() - clock);
 221   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 222     free(name);
 223 #ifdef HAVE_TRACING
 224   TRACE_smpi_ptp_out(rank, src_traced, rank, "wait");
 225   TRACE_smpi_recv(rank, src_traced, rank);
 226 #endif
 227
 228 }
 229
 230 /* FIXME: that's a poor man's implementation: we should take the message exchanges into account */
 231 static void action_barrier(const char *const *action)
 232 {
 233   char *name = NULL;
 234   static smx_mutex_t mutex = NULL;
 235   static smx_cond_t cond = NULL;
 236   static int processes_arrived_sofar=0;
 237
 238   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 239     name = xbt_str_join_array(action, " ");
 240
 241   if (mutex == NULL) {       // first arriving on the barrier
 242     mutex = SIMIX_req_mutex_init();
 243     cond = SIMIX_req_cond_init();
 244     processes_arrived_sofar=0;
 245   }
 246   XBT_DEBUG("Entering barrier: %s (%d already there)", name,processes_arrived_sofar);
 247
 248   SIMIX_req_mutex_lock(mutex);
 249   if (++processes_arrived_sofar == communicator_size) {
 250     SIMIX_req_cond_broadcast(cond);
 251     SIMIX_req_mutex_unlock(mutex);
 252   } else {
 253     SIMIX_req_cond_wait(cond,mutex);
 254     SIMIX_req_mutex_unlock(mutex);
 255   }
 256
 257   XBT_DEBUG("Exiting barrier: %s", name);
 258
 259   processes_arrived_sofar--;
 260   if (!processes_arrived_sofar) {
 261     SIMIX_req_cond_destroy(cond);
 262     SIMIX_req_mutex_destroy(mutex);
 263     mutex=NULL;
 264   }
 265
 266   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 267     free(name);
 268
 269 }
 270
 271 static void action_reduce(const char *const *action)
 272 {
 273         int i;
 274         char *reduce_identifier;
 275         char mailbox[80];
 276         double comm_size = parse_double(action[2]);
 277         double comp_size = parse_double(action[3]);
 278         m_task_t comp_task = NULL;
 279         const char *process_name;
 280         double clock = MSG_get_clock();
 281
 282         process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self());
 283
 284         xbt_assert0(communicator_size, "Size of Communicator is not defined, "
 285                         "can't use collective operations");
 286
 287         process_name = MSG_process_get_name(MSG_process_self());
 288
 289         reduce_identifier = bprintf("reduce_%d", counters->reduce_counter++);
 290
 291         if (!strcmp(process_name, "p0")) {
 292                 XBT_DEBUG("%s: %s is the Root", reduce_identifier, process_name);
 293
 294                 msg_comm_t *comms = xbt_new0(msg_comm_t,communicator_size-1);
 295             m_task_t *tasks = xbt_new0(m_task_t,communicator_size-1);
 296             for (i = 1; i < communicator_size; i++) {
 297               sprintf(mailbox, "%s_p%d_p0", reduce_identifier, i);
 298               comms[i-1] = MSG_task_irecv(&(tasks[i-1]),mailbox);
 299             }
 300             MSG_comm_waitall(comms,communicator_size-1,-1);
 301             for (i = 1; i < communicator_size; i++) {
 302                 MSG_comm_destroy(comms[i-1]);
 303                 MSG_task_destroy(tasks[i-1]);
 304             }
 305             free(tasks);
 306
 307             comp_task = MSG_task_create("reduce_comp", comp_size, 0, NULL);
 308             XBT_DEBUG("%s: computing 'reduce_comp'", reduce_identifier);
 309             MSG_task_execute(comp_task);
 310             MSG_task_destroy(comp_task);
 311             XBT_DEBUG("%s: computed", reduce_identifier);
 312
 313         } else {
 314                 XBT_DEBUG("%s: %s sends", reduce_identifier, process_name);
 315                 sprintf(mailbox, "%s_%s_p0", reduce_identifier, process_name);
 316             XBT_DEBUG("put on %s", mailbox);
 317             MSG_task_send(MSG_task_create(reduce_identifier, 0, comm_size, NULL),
 318                           mailbox);
 319         }
 320
 321         XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
 322         free(reduce_identifier);
 323 }
 324
 325 static void action_bcast(const char *const *action)
 326 {
 327         int i;
 328         char *bcast_identifier;
 329         char mailbox[80];
 330         double comm_size = parse_double(action[2]);
 331         m_task_t task = NULL;
 332         const char *process_name;
 333         double clock = MSG_get_clock();
 334
 335         process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self());
 336
 337         xbt_assert0(communicator_size, "Size of Communicator is not defined, "
 338                         "can't use collective operations");
 339
 340         process_name = MSG_process_get_name(MSG_process_self());
 341
 342         bcast_identifier = bprintf("bcast_%d", counters->bcast_counter++);
 343
 344         if (!strcmp(process_name, "p0")) {
 345                 XBT_DEBUG("%s: %s is the Root", bcast_identifier, process_name);
 346
 347             msg_comm_t *comms = xbt_new0(msg_comm_t,communicator_size-1);
 348
 349             for (i = 1; i < communicator_size; i++) {
 350               sprintf(mailbox, "%s_p0_p%d", bcast_identifier, i);
 351               comms[i-1] =
 352                   MSG_task_isend(MSG_task_create(mailbox,0,comm_size,NULL),
 353                       mailbox);
 354             }
 355             MSG_comm_waitall(comms,communicator_size-1,-1);
 356                 for (i = 1; i < communicator_size; i++)
 357                MSG_comm_destroy(comms[i-1]);
 358             free(comms);
 359
 360             XBT_DEBUG("%s: all messages sent by %s have been received",
 361                    bcast_identifier, process_name);
 362
 363         } else {
 364             sprintf(mailbox, "%s_p0_%s", bcast_identifier, process_name);
 365             MSG_task_receive(&task, mailbox);
 366             MSG_task_destroy(task);
 367             XBT_DEBUG("%s: %s has received", bcast_identifier, process_name);
 368         }
 369
 370         XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
 371         free(bcast_identifier);
 372 }
 373
 374
 375 static void action_sleep(const char *const *action)
 376 {
 377   char *name = NULL;
 378   const char *duration = action[2];
 379   double clock = MSG_get_clock();
 380
 381   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 382     name = xbt_str_join_array(action, " ");
 383
 384   XBT_DEBUG("Entering %s", name);
 385   MSG_process_sleep(parse_double(duration));
 386   XBT_VERB("%s %f ", name, MSG_get_clock() - clock);
 387
 388   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 389     free(name);
 390 }
 391
 392 static void action_allReduce(const char *const *action) {
 393   int i;
 394   char *allreduce_identifier;
 395   char mailbox[80];
 396   double comm_size = parse_double(action[2]);
 397   double comp_size = parse_double(action[3]);
 398   m_task_t task = NULL, comp_task = NULL;
 399   const char *process_name;
 400   double clock = MSG_get_clock();
 401
 402   process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self());
 403
 404   xbt_assert0(communicator_size, "Size of Communicator is not defined, "
 405               "can't use collective operations");
 406
 407   process_name = MSG_process_get_name(MSG_process_self());
 408
 409   allreduce_identifier = bprintf("allReduce_%d", counters->allReduce_counter++);
 410
 411   if (!strcmp(process_name, "p0")) {
 412     XBT_DEBUG("%s: %s is the Root", allreduce_identifier, process_name);
 413
 414     msg_comm_t *comms = xbt_new0(msg_comm_t,communicator_size-1);
 415     m_task_t *tasks = xbt_new0(m_task_t,communicator_size-1);
 416     for (i = 1; i < communicator_size; i++) {
 417       sprintf(mailbox, "%s_p%d_p0", allreduce_identifier, i);
 418       comms[i-1] = MSG_task_irecv(&(tasks[i-1]),mailbox);
 419     }
 420     MSG_comm_waitall(comms,communicator_size-1,-1);
 421     for (i = 1; i < communicator_size; i++) {
 422       MSG_comm_destroy(comms[i-1]);
 423       MSG_task_destroy(tasks[i-1]);
 424     }
 425     free(tasks);
 426
 427     comp_task = MSG_task_create("allReduce_comp", comp_size, 0, NULL);
 428     XBT_DEBUG("%s: computing 'reduce_comp'", allreduce_identifier);
 429     MSG_task_execute(comp_task);
 430     MSG_task_destroy(comp_task);
 431     XBT_DEBUG("%s: computed", allreduce_identifier);
 432
 433     for (i = 1; i < communicator_size; i++) {
 434       sprintf(mailbox, "%s_p0_p%d", allreduce_identifier, i);
 435       comms[i-1] =
 436           MSG_task_isend(MSG_task_create(mailbox,0,comm_size,NULL),
 437               mailbox);
 438     }
 439     MSG_comm_waitall(comms,communicator_size-1,-1);
 440     for (i = 1; i < communicator_size; i++)
 441        MSG_comm_destroy(comms[i-1]);
 442     free(comms);
 443
 444     XBT_DEBUG("%s: all messages sent by %s have been received",
 445            allreduce_identifier, process_name);
 446
 447   } else {
 448     XBT_DEBUG("%s: %s sends", allreduce_identifier, process_name);
 449     sprintf(mailbox, "%s_%s_p0", allreduce_identifier, process_name);
 450     XBT_DEBUG("put on %s", mailbox);
 451     MSG_task_send(MSG_task_create(allreduce_identifier, 0, comm_size, NULL),
 452                   mailbox);
 453
 454     sprintf(mailbox, "%s_p0_%s", allreduce_identifier, process_name);
 455     MSG_task_receive(&task, mailbox);
 456     MSG_task_destroy(task);
 457     XBT_DEBUG("%s: %s has received", allreduce_identifier, process_name);
 458   }
 459
 460   XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
 461   free(allreduce_identifier);
 462 }
 463
 464 static void action_comm_size(const char *const *action)
 465 {
 466   char *name = NULL;
 467   const char *size = action[2];
 468   double clock = MSG_get_clock();
 469
 470   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 471     name = xbt_str_join_array(action, " ");
 472   communicator_size = parse_double(size);
 473   XBT_VERB("%s %f", name, MSG_get_clock() - clock);
 474   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 475     free(name);
 476 }
 477
 478 static void action_compute(const char *const *action)
 479 {
 480   char *name = NULL;
 481   const char *amout = action[2];
 482   m_task_t task = MSG_task_create(name, parse_double(amout), 0, NULL);
 483   double clock = MSG_get_clock();
 484
 485   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 486     name = xbt_str_join_array(action, " ");
 487   XBT_DEBUG("Entering %s", name);
 488   MSG_task_execute(task);
 489   MSG_task_destroy(task);
 490   XBT_VERB("%s %f", name, MSG_get_clock() - clock);
 491   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 492     free(name);
 493 }
 494
 495 static void action_init(const char *const *action)
 496 {
 497 #ifdef HAVE_TRACING
 498   TRACE_smpi_init(get_rank(MSG_process_get_name(MSG_process_self())));
 499 #endif
 500   XBT_DEBUG("Initialize the counters");
 501   process_globals_t globals = (process_globals_t) calloc(1, sizeof(s_process_globals_t));
 502   globals->isends = xbt_dynar_new(sizeof(msg_comm_t),NULL);
 503   globals->irecvs = xbt_dynar_new(sizeof(msg_comm_t),NULL);
 504   globals->tasks  = xbt_dynar_new(sizeof(m_task_t),NULL);
 505   MSG_process_set_data(MSG_process_self(),globals);
 506
 507 }
 508
 509 static void action_finalize(const char *const *action)
 510 {
 511 #ifdef HAVE_TRACING
 512   TRACE_smpi_finalize(get_rank(MSG_process_get_name(MSG_process_self())));
 513 #endif
 514   process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
 515   if (globals){
 516     xbt_dynar_free_container(&(globals->isends));
 517     xbt_dynar_free_container(&(globals->irecvs));
 518     xbt_dynar_free_container(&(globals->tasks));
 519     free(globals);
 520   }
 521 }
 522
 523 /** Main function */
 524 int main(int argc, char *argv[])
 525 {
 526   MSG_error_t res = MSG_OK;
 527
 528   smx_factory_initializer_to_use = SIMIX_ctx_raw_factory_init;
 529
 530   /* Check the given arguments */
 531   MSG_global_init(&argc, argv);
 532   if (argc < 3) {
 533     printf("Usage: %s platform_file deployment_file [action_files]\n",
 534            argv[0]);
 535     printf
 536         ("example: %s msg_platform.xml msg_deployment.xml actions # if all actions are in the same file\n",
 537          argv[0]);
 538     printf
 539         ("example: %s msg_platform.xml msg_deployment.xml # if actions are in separate files, specified in deployment\n",
 540          argv[0]);
 541     exit(1);
 542   }
 543
 544   /*  Simulation setting */
 545   MSG_create_environment(argv[1]);
 546
 547   /* No need to register functions as in classical MSG programs: the actions get started anyway */
 548   MSG_launch_application(argv[2]);
 549
 550   /*   Action registration */
 551   MSG_action_register("init",     action_init);
 552   MSG_action_register("finalize", action_finalize);
 553   MSG_action_register("comm_size",action_comm_size);
 554   MSG_action_register("send",     action_send);
 555   MSG_action_register("Isend",    action_Isend);
 556   MSG_action_register("recv",     action_recv);
 557   MSG_action_register("Irecv",    action_Irecv);
 558   MSG_action_register("wait",     action_wait);
 559   MSG_action_register("barrier",  action_barrier);
 560   MSG_action_register("bcast",    action_bcast);
 561   MSG_action_register("reduce",   action_reduce);
 562   MSG_action_register("allReduce",action_allReduce);
 563   MSG_action_register("sleep",    action_sleep);
 564   MSG_action_register("compute",  action_compute);
 565
 566
 567   /* Actually do the simulation using MSG_action_trace_run */
 568   res = MSG_action_trace_run(argv[3]);  // it's ok to pass a NULL argument here
 569
 570   XBT_INFO("Simulation time %g", MSG_get_clock());
 571   MSG_clean();
 572
 573   if (res == MSG_OK)
 574     return 0;
 575   else
 576     return 1;
 577 }                               /* end_of_main */