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 smx_sem_t barrier_semaphore = NULL;
 186 static void barrier(xbt_dynar_t action)
 187 {
 188   char *name = NULL;
 189
 190   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 191     name = xbt_str_join(action, " ");
 192
 193   DEBUG1("Entering barrier: %s", name);
 194   if (barrier_semaphore == NULL)        // first arriving on the barrier
 195     barrier_semaphore = SIMIX_sem_init(0);
 196
 197   if (SIMIX_sem_get_capacity(barrier_semaphore) == -communicator_size + 1) {    // last arriving
 198     SIMIX_sem_release_forever(barrier_semaphore);
 199     SIMIX_sem_destroy(barrier_semaphore);
 200     barrier_semaphore = NULL;
 201   } else {                      // not last
 202     SIMIX_sem_acquire(barrier_semaphore);
 203   }
 204
 205   DEBUG1("Exiting barrier: %s", name);
 206
 207   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 208     free(name);
 209
 210 }
 211
 212 static void reduce(xbt_dynar_t action)
 213 {
 214   int i;
 215   char *name;
 216   char task_name[80];
 217   char spawn_name[80];
 218   char **myargv;
 219   char *comm_size = xbt_dynar_get_as(action, 2, char *);
 220   char *comp_size = xbt_dynar_get_as(action, 3, char *);
 221   m_process_t comm_helper = NULL;
 222   m_task_t task = NULL, comp_task = NULL;
 223   const char *process_name;
 224   double clock = MSG_get_clock();
 225
 226   coll_ctr counters = (coll_ctr) MSG_process_get_data(MSG_process_self());
 227
 228   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 229               ", can't use collective operations");
 230
 231   process_name = MSG_process_get_name(MSG_process_self());
 232
 233   if (!counters) {
 234     DEBUG0("Initialize the counters");
 235     counters = (coll_ctr) calloc(1, sizeof(struct coll_ctr_t));
 236   }
 237
 238   name = bprintf("reduce_%d", counters->reduce_counter++);
 239
 240   if (!strcmp(process_name, "p0")) {
 241     DEBUG2("%s: %s is the Root", name, process_name);
 242     for (i = 1; i < communicator_size; i++) {
 243       sprintf(spawn_name, "%s_p%d_%s", name, i,
 244               MSG_process_get_name(MSG_process_self()));
 245       sprintf(task_name, "%s_wait", spawn_name);
 246       myargv = (char **) calloc(2, sizeof(char *));
 247
 248       myargv[0] = xbt_strdup(spawn_name);
 249       myargv[1] = NULL;
 250
 251       comm_helper =
 252           MSG_process_create_with_arguments(task_name, spawned_recv,
 253                                             NULL, MSG_host_self(),
 254                                             1, myargv);
 255     }
 256
 257     for (i = 1; i < communicator_size; i++) {
 258       sprintf(task_name, "%s_p%d_p0_wait", name, i);
 259       MSG_task_receive(&task, task_name);
 260       MSG_task_destroy(task);
 261       task = NULL;
 262     }
 263
 264     comp_task =
 265         MSG_task_create("reduce_comp", parse_double(comp_size), 0, NULL);
 266     DEBUG1("%s: computing 'reduce_comp'", name);
 267     MSG_task_execute(comp_task);
 268     MSG_task_destroy(comp_task);
 269     DEBUG1("%s: computed", name);
 270   } else {
 271     DEBUG2("%s: %s sends", name, process_name);
 272     sprintf(task_name, "%s_%s_p0", name, process_name);
 273     DEBUG1("put on %s", task_name);
 274     MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL),
 275                   task_name);
 276   }
 277
 278   MSG_process_set_data(MSG_process_self(), (void *) counters);
 279   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 280   free(name);
 281 }
 282
 283 static void bcast(xbt_dynar_t action)
 284 {
 285   int i;
 286   char *name;
 287   const char *process_name;
 288   char task_name[80];
 289   char spawn_name[80];
 290   char **myargv;
 291   m_process_t comm_helper = NULL;
 292   m_task_t task = NULL;
 293   char *size = xbt_dynar_get_as(action, 2, char *);
 294   coll_ctr counters = (coll_ctr) MSG_process_get_data(MSG_process_self());
 295   double clock = MSG_get_clock();
 296
 297   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 298               ", can't use collective operations");
 299
 300
 301   process_name = MSG_process_get_name(MSG_process_self());
 302   if (!counters) {
 303     DEBUG0("Initialize the counters");
 304     counters = (coll_ctr) calloc(1, sizeof(struct coll_ctr_t));
 305   }
 306
 307   name = bprintf("bcast_%d", counters->bcast_counter++);
 308   if (!strcmp(process_name, "p0")) {
 309     DEBUG2("%s: %s is the Root", name, process_name);
 310
 311     for (i = 1; i < communicator_size; i++) {
 312       myargv = (char **) calloc(3, sizeof(char *));
 313       myargv[0] = xbt_strdup(name);
 314       myargv[1] = xbt_strdup(size);
 315       myargv[2] = NULL;
 316
 317       sprintf(spawn_name, "%s_%d", myargv[0], i);
 318       comm_helper =
 319           MSG_process_create_with_arguments(spawn_name, spawned_send,
 320                                             NULL, MSG_host_self(), 2,
 321                                             myargv);
 322     }
 323
 324     for (i = 1; i < communicator_size; i++) {
 325       sprintf(task_name, "p%d_wait", i);
 326       DEBUG1("get on %s", task_name);
 327       MSG_task_receive(&task, task_name);
 328       MSG_task_destroy(task);
 329       task = NULL;
 330     }
 331     DEBUG2("%s: all messages sent by %s have been received",
 332            name, process_name);
 333   } else {
 334     DEBUG2("%s: %s receives", name, process_name);
 335     MSG_task_receive(&task, name);
 336     MSG_task_destroy(task);
 337     DEBUG2("%s: %s has received", name, process_name);
 338     sprintf(task_name, "%s_wait", process_name);
 339     DEBUG1("put on %s", task_name);
 340     MSG_task_send(MSG_task_create("waiter", 0, 0, NULL), task_name);
 341   }
 342
 343   MSG_process_set_data(MSG_process_self(), (void *) counters);
 344   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 345   free(name);
 346 }
 347
 348
 349 static void action_sleep(xbt_dynar_t action)
 350 {
 351   char *name = NULL;
 352   char *duration = xbt_dynar_get_as(action, 2, char *);
 353   double clock = MSG_get_clock();
 354
 355   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 356     name = xbt_str_join(action, " ");
 357
 358   DEBUG1("Entering %s", name);
 359   MSG_process_sleep(parse_double(duration));
 360   VERB2("%s %f ", name, MSG_get_clock() - clock);
 361
 362   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 363     free(name);
 364 }
 365
 366 static void allReduce(xbt_dynar_t action)
 367 {
 368   int i;
 369   char *name;
 370   char task_name[80];
 371   char spawn_name[80];
 372   char **myargv;
 373   char *comm_size = xbt_dynar_get_as(action, 2, char *);
 374   char *comp_size = xbt_dynar_get_as(action, 3, char *);
 375   m_process_t comm_helper = NULL;
 376   m_task_t task = NULL, comp_task = NULL;
 377   const char *process_name;
 378   double clock = MSG_get_clock();
 379
 380   coll_ctr counters = (coll_ctr) MSG_process_get_data(MSG_process_self());
 381
 382   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 383               ", can't use collective operations");
 384
 385   process_name = MSG_process_get_name(MSG_process_self());
 386
 387   if (!counters) {
 388     DEBUG0("Initialize the counters");
 389     counters = (coll_ctr) calloc(1, sizeof(struct coll_ctr_t));
 390   }
 391
 392   name = bprintf("allReduce_%d", counters->allReduce_counter++);
 393
 394   if (!strcmp(process_name, "p0")) {
 395     DEBUG2("%s: %s is the Root", name, process_name);
 396     for (i = 1; i < communicator_size; i++) {
 397       sprintf(spawn_name, "%s_p%d_%s", name, i,
 398               MSG_process_get_name(MSG_process_self()));
 399       sprintf(task_name, "%s_wait", spawn_name);
 400       myargv = (char **) calloc(2, sizeof(char *));
 401
 402       myargv[0] = xbt_strdup(spawn_name);
 403       myargv[1] = NULL;
 404
 405       comm_helper =
 406           MSG_process_create_with_arguments(task_name, spawned_recv,
 407                                             NULL, MSG_host_self(),
 408                                             1, myargv);
 409     }
 410
 411     for (i = 1; i < communicator_size; i++) {
 412       sprintf(task_name, "%s_p%d_p0_wait", name, i);
 413       MSG_task_receive(&task, task_name);
 414       MSG_task_destroy(task);
 415       task = NULL;
 416     }
 417
 418     comp_task =
 419         MSG_task_create("allReduce_comp", parse_double(comp_size), 0,
 420                         NULL);
 421     DEBUG1("%s: computing 'reduce_comp'", name);
 422     MSG_task_execute(comp_task);
 423     MSG_task_destroy(comp_task);
 424     DEBUG1("%s: computed", name);
 425
 426     for (i = 1; i < communicator_size; i++) {
 427       myargv = (char **) calloc(3, sizeof(char *));
 428       myargv[0] = xbt_strdup(name);
 429       myargv[1] = xbt_strdup(comm_size);
 430       myargv[2] = NULL;
 431
 432       sprintf(spawn_name, "%s_%d", myargv[0], i);
 433       comm_helper =
 434           MSG_process_create_with_arguments(spawn_name, spawned_send,
 435                                             NULL, MSG_host_self(), 2,
 436                                             myargv);
 437     }
 438
 439     for (i = 1; i < communicator_size; i++) {
 440       sprintf(task_name, "p%d_wait", i);
 441       DEBUG1("get on %s", task_name);
 442       MSG_task_receive(&task, task_name);
 443       MSG_task_destroy(task);
 444       task = NULL;
 445     }
 446     DEBUG2("%s: all messages sent by %s have been received",
 447            name, process_name);
 448
 449   } else {
 450     DEBUG2("%s: %s sends", name, process_name);
 451     sprintf(task_name, "%s_%s_p0", name, process_name);
 452     DEBUG1("put on %s", task_name);
 453     MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL),
 454                   task_name);
 455
 456     MSG_task_receive(&task, name);
 457     MSG_task_destroy(task);
 458     DEBUG2("%s: %s has received", name, process_name);
 459     sprintf(task_name, "%s_wait", process_name);
 460     DEBUG1("put on %s", task_name);
 461     MSG_task_send(MSG_task_create("waiter", 0, 0, NULL), task_name);
 462
 463   }
 464
 465   MSG_process_set_data(MSG_process_self(), (void *) counters);
 466   VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock);
 467   free(name);
 468 }
 469
 470 static void comm_size(xbt_dynar_t action)
 471 {
 472   char *name = NULL;
 473   char *size = xbt_dynar_get_as(action, 2, char *);
 474   double clock = MSG_get_clock();
 475
 476   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 477     name = xbt_str_join(action, " ");
 478   communicator_size = parse_double(size);
 479   VERB2("%s %f", name, MSG_get_clock() - clock);
 480   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 481     free(name);
 482 }
 483
 484 static void compute(xbt_dynar_t action)
 485 {
 486   char *name = NULL;
 487   char *amout = xbt_dynar_get_as(action, 2, char *);
 488   m_task_t task = MSG_task_create(name, parse_double(amout), 0, NULL);
 489   double clock = MSG_get_clock();
 490
 491   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 492     name = xbt_str_join(action, " ");
 493   DEBUG1("Entering %s", name);
 494   MSG_task_execute(task);
 495   MSG_task_destroy(task);
 496   VERB2("%s %f", name, MSG_get_clock() - clock);
 497   if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
 498     free(name);
 499 }
 500
 501 /** Main function */
 502 int main(int argc, char *argv[])
 503 {
 504   MSG_error_t res = MSG_OK;
 505
 506   /* Check the given arguments */
 507   MSG_global_init(&argc, argv);
 508   if (argc < 3) {
 509     printf("Usage: %s platform_file deployment_file [action_files]\n",
 510            argv[0]);
 511     printf
 512         ("example: %s msg_platform.xml msg_deployment.xml actions # if all actions are in the same file\n",
 513          argv[0]);
 514     printf
 515         ("example: %s msg_platform.xml msg_deployment.xml # if actions are in separate files, specified in deployment\n",
 516          argv[0]);
 517     exit(1);
 518   }
 519
 520   /*  Simulation setting */
 521   MSG_create_environment(argv[1]);
 522
 523   /* No need to register functions as in classical MSG programs: the actions get started anyway */
 524   MSG_launch_application(argv[2]);
 525
 526   /*   Action registration */
 527   MSG_action_register("comm_size", comm_size);
 528   MSG_action_register("send", action_send);
 529   MSG_action_register("Isend", Isend);
 530   MSG_action_register("recv", action_recv);
 531   MSG_action_register("Irecv", Irecv);
 532   MSG_action_register("wait", action_wait);
 533   MSG_action_register("barrier", barrier);
 534   MSG_action_register("bcast", bcast);
 535   MSG_action_register("reduce", reduce);
 536   MSG_action_register("allReduce", allReduce);
 537   MSG_action_register("sleep", action_sleep);
 538   MSG_action_register("compute", compute);
 539
 540
 541   /* Actually do the simulation using MSG_action_trace_run */
 542   res = MSG_action_trace_run(argv[3]);  // it's ok to pass a NULL argument here
 543
 544   INFO1("Simulation time %g", MSG_get_clock());
 545   MSG_clean();
 546
 547   if (res == MSG_OK)
 548     return 0;
 549   else
 550     return 1;
 551 }                               /* end_of_main */