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 "simgrid_config.h"     /* getline */
  13
  14 XBT_LOG_NEW_DEFAULT_CATEGORY(actions,
  15                              "Messages specific for this msg example");
  16 int communicator_size = 0;
  17
  18 typedef struct coll_ctr_t {
  19   int bcast_counter;
  20   int reduce_counter;
  21   int allReduce_counter;
  22 } *coll_ctr;
  23
  24 /* Helper function */
  25 static double parse_double(const char *string)
  26 {
  27   double value;
  28   char *endptr;
  29
  30   value = strtod(string, &endptr);
  31   if (*endptr != '\0')
  32     THROW1(unknown_error, 0, "%s is not a double", string);
  33   return value;
  34 }
  35
  36
  37 /* My actions */
  38 static void action_send(xbt_dynar_t action)
  39 {
  40   char *name = NULL;
  41   char to[250];
  42   char *size = xbt_dynar_get_as(action, 3, char *);
  43   double clock = MSG_get_clock();
  44   sprintf(to, "%s_%s", MSG_process_get_name(MSG_process_self()),
  45           xbt_dynar_get_as(action, 2, char *));
  46   //  char *to =  xbt_dynar_get_as(action, 2, char *);
  47
  48   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
  49     name = xbt_str_join(action, " ");
  50
  51   DEBUG2("Entering Send: %s (size: %lg)", name, parse_double(size));
  52   MSG_task_send(MSG_task_create(name, 0, parse_double(size), NULL), to);
  53   VERB2("%s %f", name, MSG_get_clock() - clock);
  54
  55   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
  56     free(name);
  57 }
  58
  59
  60 static int spawned_send(int argc, char *argv[])
  61 {
  62   DEBUG3("%s: Sending %s on %s", MSG_process_get_name(MSG_process_self()),
  63          argv[1], argv[0]);
  64   MSG_task_send(MSG_task_create(argv[0], 0, parse_double(argv[1]), NULL),
  65                 argv[0]);
  66   return 0;
  67 }
  68
  69 static void Isend(xbt_dynar_t action)
  70 {
  71   char spawn_name[80];
  72   char to[250];
  73   //  char *to = xbt_dynar_get_as(action, 2, char *);
  74   char *size = xbt_dynar_get_as(action, 3, char *);
  75   char **myargv;
  76   m_process_t comm_helper;
  77   double clock = MSG_get_clock();
  78   DEBUG1("Isend on %s: spawn process ",
  79          MSG_process_get_name(MSG_process_self()));
  80
  81   sprintf(to, "%s_%s", MSG_process_get_name(MSG_process_self()),
  82           xbt_dynar_get_as(action, 2, char *));
  83   myargv = (char **) calloc(3, sizeof(char *));
  84
  85   myargv[0] = xbt_strdup(to);
  86   myargv[1] = xbt_strdup(size);
  87   myargv[2] = NULL;
  88
  89   //    sprintf(spawn_name,"%s_wait",MSG_process_get_name(MSG_process_self()));
  90   sprintf(spawn_name, "%s_wait", to);
  91   comm_helper =
  92       MSG_process_create_with_arguments(spawn_name, spawned_send,
  93                                         NULL, MSG_host_self(), 2, myargv);
  94   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
  95 }
  96
  97
  98 static void action_recv(xbt_dynar_t action)
  99 {
 100   char *name = NULL;
 101   char mailbox_name[250];
 102   m_task_t task = NULL;
 103   double clock = MSG_get_clock();
 104   //FIXME: argument of action ignored so far; semantic not clear
 105   //char *from=xbt_dynar_get_as(action,2,char*);
 106   sprintf(mailbox_name, "%s_%s", xbt_dynar_get_as(action, 2, char *),
 107           MSG_process_get_name(MSG_process_self()));
 108
 109   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 110     name = xbt_str_join(action, " ");
 111
 112   DEBUG1("Receiving: %s", name);
 113   MSG_task_receive(&task, mailbox_name);
 114   //  MSG_task_receive(&task, MSG_process_get_name(MSG_process_self()));
 115   VERB2("%s %f", name, MSG_get_clock() - clock);
 116   MSG_task_destroy(task);
 117
 118   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 119     free(name);
 120 }
 121
 122 static int spawned_recv(int argc, char *argv[])
 123 {
 124   m_task_t task = NULL;
 125   DEBUG1("Receiving on %s", argv[0]);
 126   MSG_task_receive(&task, argv[0]);
 127   DEBUG1("Received %s", MSG_task_get_name(task));
 128   DEBUG1("waiter on %s", MSG_process_get_name(MSG_process_self()));
 129   MSG_task_send(MSG_task_create("waiter", 0, 0, NULL),
 130                 MSG_process_get_name(MSG_process_self()));
 131
 132   MSG_task_destroy(task);
 133   return 0;
 134 }
 135
 136
 137 static void Irecv(xbt_dynar_t action)
 138 {
 139   char *name;
 140   m_process_t comm_helper;
 141   char mailbox_name[250];
 142   char **myargv;
 143   double clock = MSG_get_clock();
 144   DEBUG1("Irecv on %s: spawn process ",
 145          MSG_process_get_name(MSG_process_self()));
 146
 147   sprintf(mailbox_name, "%s_%s", xbt_dynar_get_as(action, 2, char *),
 148           MSG_process_get_name(MSG_process_self()));
 149   name = bprintf("%s_wait", MSG_process_get_name(MSG_process_self()));
 150   myargv = (char **) calloc(2, sizeof(char *));
 151
 152   myargv[0] = xbt_strdup(mailbox_name);
 153   myargv[1] = NULL;
 154   comm_helper = MSG_process_create_with_arguments(name, spawned_recv,
 155                                                   NULL, MSG_host_self(),
 156                                                   1, myargv);
 157
 158   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 159
 160   free(name);
 161 }
 162
 163
 164 static void action_wait(xbt_dynar_t action)
 165 {
 166   char *name = NULL;
 167   char task_name[80];
 168   m_task_t task = NULL;
 169   double clock = MSG_get_clock();
 170
 171   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 172     name = xbt_str_join(action, " ");
 173
 174   DEBUG1("Entering %s", name);
 175   sprintf(task_name, "%s_wait", MSG_process_get_name(MSG_process_self()));
 176   DEBUG1("wait: %s", task_name);
 177   MSG_task_receive(&task, task_name);
 178   MSG_task_destroy(task);
 179   VERB2("%s %f", name, MSG_get_clock() - clock);
 180   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 181     free(name);
 182 }
 183
 184 /* FIXME: that's a poor man's implementation: we should take the message exchanges into account */
 185 static void barrier(xbt_dynar_t action)
 186 {
 187   char *name = NULL;
 188   static smx_mutex_t mutex = NULL;
 189   static smx_cond_t cond = NULL;
 190   static int processes_arrived_sofar=0;
 191
 192   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 193     name = xbt_str_join(action, " ");
 194
 195   if (mutex == NULL) {       // first arriving on the barrier
 196     mutex = SIMIX_mutex_init();
 197     cond = SIMIX_cond_init();
 198     processes_arrived_sofar=0;
 199   }
 200   DEBUG2("Entering barrier: %s (%d already there)", name,processes_arrived_sofar);
 201
 202   SIMIX_mutex_lock(mutex);
 203   if (++processes_arrived_sofar == communicator_size) {
 204     SIMIX_cond_broadcast(cond);
 205     SIMIX_mutex_unlock(mutex);
 206   } else {
 207     SIMIX_cond_wait(cond,mutex);
 208     SIMIX_mutex_unlock(mutex);
 209   }
 210
 211   DEBUG1("Exiting barrier: %s", name);
 212
 213   processes_arrived_sofar--;
 214   if (!processes_arrived_sofar) {
 215     SIMIX_cond_destroy(cond);
 216     SIMIX_mutex_destroy(mutex);
 217     mutex=NULL;
 218   }
 219
 220   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 221     free(name);
 222
 223 }
 224
 225 static void reduce(xbt_dynar_t action)
 226 {
 227   int i;
 228   char *name;
 229   char task_name[80];
 230   char spawn_name[80];
 231   char **myargv;
 232   char *comm_size = xbt_dynar_get_as(action, 2, char *);
 233   char *comp_size = xbt_dynar_get_as(action, 3, char *);
 234   m_process_t comm_helper = NULL;
 235   m_task_t task = NULL, comp_task = NULL;
 236   const char *process_name;
 237   double clock = MSG_get_clock();
 238
 239   coll_ctr counters = (coll_ctr) MSG_process_get_data(MSG_process_self());
 240
 241   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 242               ", can't use collective operations");
 243
 244   process_name = MSG_process_get_name(MSG_process_self());
 245
 246   if (!counters) {
 247     DEBUG0("Initialize the counters");
 248     counters = (coll_ctr) calloc(1, sizeof(struct coll_ctr_t));
 249   }
 250
 251   name = bprintf("reduce_%d", counters->reduce_counter++);
 252
 253   if (!strcmp(process_name, "p0")) {
 254     DEBUG2("%s: %s is the Root", name, process_name);
 255     for (i = 1; i < communicator_size; i++) {
 256       sprintf(spawn_name, "%s_p%d_%s", name, i,
 257               MSG_process_get_name(MSG_process_self()));
 258       sprintf(task_name, "%s_wait", spawn_name);
 259       myargv = (char **) calloc(2, sizeof(char *));
 260
 261       myargv[0] = xbt_strdup(spawn_name);
 262       myargv[1] = NULL;
 263
 264       comm_helper =
 265           MSG_process_create_with_arguments(task_name, spawned_recv,
 266                                             NULL, MSG_host_self(),
 267                                             1, myargv);
 268     }
 269
 270     for (i = 1; i < communicator_size; i++) {
 271       sprintf(task_name, "%s_p%d_p0_wait", name, i);
 272       MSG_task_receive(&task, task_name);
 273       MSG_task_destroy(task);
 274       task = NULL;
 275     }
 276
 277     comp_task =
 278         MSG_task_create("reduce_comp", parse_double(comp_size), 0, NULL);
 279     DEBUG1("%s: computing 'reduce_comp'", name);
 280     MSG_task_execute(comp_task);
 281     MSG_task_destroy(comp_task);
 282     DEBUG1("%s: computed", name);
 283   } else {
 284     DEBUG2("%s: %s sends", name, process_name);
 285     sprintf(task_name, "%s_%s_p0", name, process_name);
 286     DEBUG1("put on %s", task_name);
 287     MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL),
 288                   task_name);
 289   }
 290
 291   MSG_process_set_data(MSG_process_self(), (void *) counters);
 292   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 293   free(name);
 294 }
 295
 296 static void bcast(xbt_dynar_t action)
 297 {
 298   int i;
 299   char *name;
 300   const char *process_name;
 301   char task_name[80];
 302   char spawn_name[80];
 303   char **myargv;
 304   m_process_t comm_helper = NULL;
 305   m_task_t task = NULL;
 306   char *size = xbt_dynar_get_as(action, 2, char *);
 307   coll_ctr counters = (coll_ctr) MSG_process_get_data(MSG_process_self());
 308   double clock = MSG_get_clock();
 309
 310   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 311               ", can't use collective operations");
 312
 313
 314   process_name = MSG_process_get_name(MSG_process_self());
 315   if (!counters) {
 316     DEBUG0("Initialize the counters");
 317     counters = (coll_ctr) calloc(1, sizeof(struct coll_ctr_t));
 318   }
 319
 320   name = bprintf("bcast_%d", counters->bcast_counter++);
 321   if (!strcmp(process_name, "p0")) {
 322     DEBUG2("%s: %s is the Root", name, process_name);
 323
 324     for (i = 1; i < communicator_size; i++) {
 325       myargv = (char **) calloc(3, sizeof(char *));
 326       myargv[0] = xbt_strdup(name);
 327       myargv[1] = xbt_strdup(size);
 328       myargv[2] = NULL;
 329
 330       sprintf(spawn_name, "%s_%d", myargv[0], i);
 331       comm_helper =
 332           MSG_process_create_with_arguments(spawn_name, spawned_send,
 333                                             NULL, MSG_host_self(), 2,
 334                                             myargv);
 335     }
 336
 337     for (i = 1; i < communicator_size; i++) {
 338       sprintf(task_name, "p%d_wait", i);
 339       DEBUG1("get on %s", task_name);
 340       MSG_task_receive(&task, task_name);
 341       MSG_task_destroy(task);
 342       task = NULL;
 343     }
 344     DEBUG2("%s: all messages sent by %s have been received",
 345            name, process_name);
 346   } else {
 347     DEBUG2("%s: %s receives", name, process_name);
 348     MSG_task_receive(&task, name);
 349     MSG_task_destroy(task);
 350     DEBUG2("%s: %s has received", name, process_name);
 351     sprintf(task_name, "%s_wait", process_name);
 352     DEBUG1("put on %s", task_name);
 353     MSG_task_send(MSG_task_create("waiter", 0, 0, NULL), task_name);
 354   }
 355
 356   MSG_process_set_data(MSG_process_self(), (void *) counters);
 357   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 358   free(name);
 359 }
 360
 361
 362 static void action_sleep(xbt_dynar_t action)
 363 {
 364   char *name = NULL;
 365   char *duration = xbt_dynar_get_as(action, 2, char *);
 366   double clock = MSG_get_clock();
 367
 368   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 369     name = xbt_str_join(action, " ");
 370
 371   DEBUG1("Entering %s", name);
 372   MSG_process_sleep(parse_double(duration));
 373   VERB2("%s %f ", name, MSG_get_clock() - clock);
 374
 375   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 376     free(name);
 377 }
 378
 379 static void allReduce(xbt_dynar_t action)
 380 {
 381   int i;
 382   char *name;
 383   char task_name[80];
 384   char spawn_name[80];
 385   char **myargv;
 386   char *comm_size = xbt_dynar_get_as(action, 2, char *);
 387   char *comp_size = xbt_dynar_get_as(action, 3, char *);
 388   m_process_t comm_helper = NULL;
 389   m_task_t task = NULL, comp_task = NULL;
 390   const char *process_name;
 391   double clock = MSG_get_clock();
 392
 393   coll_ctr counters = (coll_ctr) MSG_process_get_data(MSG_process_self());
 394
 395   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 396               ", can't use collective operations");
 397
 398   process_name = MSG_process_get_name(MSG_process_self());
 399
 400   if (!counters) {
 401     DEBUG0("Initialize the counters");
 402     counters = (coll_ctr) calloc(1, sizeof(struct coll_ctr_t));
 403   }
 404
 405   name = bprintf("allReduce_%d", counters->allReduce_counter++);
 406
 407   if (!strcmp(process_name, "p0")) {
 408     DEBUG2("%s: %s is the Root", name, process_name);
 409     for (i = 1; i < communicator_size; i++) {
 410       sprintf(spawn_name, "%s_p%d_%s", name, i,
 411               MSG_process_get_name(MSG_process_self()));
 412       sprintf(task_name, "%s_wait", spawn_name);
 413       myargv = (char **) calloc(2, sizeof(char *));
 414
 415       myargv[0] = xbt_strdup(spawn_name);
 416       myargv[1] = NULL;
 417
 418       comm_helper =
 419           MSG_process_create_with_arguments(task_name, spawned_recv,
 420                                             NULL, MSG_host_self(),
 421                                             1, myargv);
 422     }
 423
 424     for (i = 1; i < communicator_size; i++) {
 425       sprintf(task_name, "%s_p%d_p0_wait", name, i);
 426       MSG_task_receive(&task, task_name);
 427       MSG_task_destroy(task);
 428       task = NULL;
 429     }
 430
 431     comp_task =
 432         MSG_task_create("allReduce_comp", parse_double(comp_size), 0,
 433                         NULL);
 434     DEBUG1("%s: computing 'reduce_comp'", name);
 435     MSG_task_execute(comp_task);
 436     MSG_task_destroy(comp_task);
 437     DEBUG1("%s: computed", name);
 438
 439     for (i = 1; i < communicator_size; i++) {
 440       myargv = (char **) calloc(3, sizeof(char *));
 441       myargv[0] = xbt_strdup(name);
 442       myargv[1] = xbt_strdup(comm_size);
 443       myargv[2] = NULL;
 444
 445       sprintf(spawn_name, "%s_%d", myargv[0], i);
 446       comm_helper =
 447           MSG_process_create_with_arguments(spawn_name, spawned_send,
 448                                             NULL, MSG_host_self(), 2,
 449                                             myargv);
 450     }
 451
 452     for (i = 1; i < communicator_size; i++) {
 453       sprintf(task_name, "p%d_wait", i);
 454       DEBUG1("get on %s", task_name);
 455       MSG_task_receive(&task, task_name);
 456       MSG_task_destroy(task);
 457       task = NULL;
 458     }
 459     DEBUG2("%s: all messages sent by %s have been received",
 460            name, process_name);
 461
 462   } else {
 463     DEBUG2("%s: %s sends", name, process_name);
 464     sprintf(task_name, "%s_%s_p0", name, process_name);
 465     DEBUG1("put on %s", task_name);
 466     MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL),
 467                   task_name);
 468
 469     MSG_task_receive(&task, name);
 470     MSG_task_destroy(task);
 471     DEBUG2("%s: %s has received", name, process_name);
 472     sprintf(task_name, "%s_wait", process_name);
 473     DEBUG1("put on %s", task_name);
 474     MSG_task_send(MSG_task_create("waiter", 0, 0, NULL), task_name);
 475
 476   }
 477
 478   MSG_process_set_data(MSG_process_self(), (void *) counters);
 479   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 480   free(name);
 481 }
 482
 483 static void comm_size(xbt_dynar_t action)
 484 {
 485   char *name = NULL;
 486   char *size = xbt_dynar_get_as(action, 2, char *);
 487   double clock = MSG_get_clock();
 488
 489   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 490     name = xbt_str_join(action, " ");
 491   communicator_size = parse_double(size);
 492   VERB2("%s %f", name, MSG_get_clock() - clock);
 493   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 494     free(name);
 495 }
 496
 497 static void compute(xbt_dynar_t action)
 498 {
 499   char *name = NULL;
 500   char *amout = xbt_dynar_get_as(action, 2, char *);
 501   m_task_t task = MSG_task_create(name, parse_double(amout), 0, NULL);
 502   double clock = MSG_get_clock();
 503
 504   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 505     name = xbt_str_join(action, " ");
 506   DEBUG1("Entering %s", name);
 507   MSG_task_execute(task);
 508   MSG_task_destroy(task);
 509   VERB2("%s %f", name, MSG_get_clock() - clock);
 510   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 511     free(name);
 512 }
 513
 514 /** Main function */
 515 int main(int argc, char *argv[])
 516 {
 517   MSG_error_t res = MSG_OK;
 518
 519   /* Check the given arguments */
 520   MSG_global_init(&argc, argv);
 521   if (argc < 3) {
 522     printf("Usage: %s platform_file deployment_file [action_files]\n",
 523            argv[0]);
 524     printf
 525         ("example: %s msg_platform.xml msg_deployment.xml actions # if all actions are in the same file\n",
 526          argv[0]);
 527     printf
 528         ("example: %s msg_platform.xml msg_deployment.xml # if actions are in separate files, specified in deployment\n",
 529          argv[0]);
 530     exit(1);
 531   }
 532
 533   /*  Simulation setting */
 534   MSG_create_environment(argv[1]);
 535
 536   /* No need to register functions as in classical MSG programs: the actions get started anyway */
 537   MSG_launch_application(argv[2]);
 538
 539   /*   Action registration */
 540   MSG_action_register("comm_size", comm_size);
 541   MSG_action_register("send", action_send);
 542   MSG_action_register("Isend", Isend);
 543   MSG_action_register("recv", action_recv);
 544   MSG_action_register("Irecv", Irecv);
 545   MSG_action_register("wait", action_wait);
 546   MSG_action_register("barrier", barrier);
 547   MSG_action_register("bcast", bcast);
 548   MSG_action_register("reduce", reduce);
 549   MSG_action_register("allReduce", allReduce);
 550   MSG_action_register("sleep", action_sleep);
 551   MSG_action_register("compute", compute);
 552
 553
 554   /* Actually do the simulation using MSG_action_trace_run */
 555   res = MSG_action_trace_run(argv[3]);  // it's ok to pass a NULL argument here
 556
 557   INFO1("Simulation time %g", MSG_get_clock());
 558   MSG_clean();
 559
 560   if (res == MSG_OK)
 561     return 0;
 562   else
 563     return 1;
 564 }                               /* end_of_main */