examples/msg/actions/actions.c

   1 /*      $Id$     */
   2
   3 /* Copyright (c) 2009. The SimGrid team. All rights reserved.               */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #include <stdio.h>
   9 #include <stdlib.h>
  10 #include "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
  11 #include "xbt.h"                /* calloc, printf */
  12
  13
  14 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  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   double value;
  27   char *endptr;
  28
  29   value=strtod(string, &endptr);
  30   if (*endptr != '\0')
  31           THROW1(unknown_error, 0, "%s is not a double", string);
  32   return value;
  33 }
  34
  35
  36 /* My actions */
  37 static void send(xbt_dynar_t action)
  38 {
  39   char *name = xbt_str_join(action, " ");
  40   char *to = xbt_dynar_get_as(action, 2, char *);
  41   char *size = xbt_dynar_get_as(action, 3, char *);
  42   INFO2("Send: %s (size: %lg)", name, parse_double(size));
  43   MSG_task_send(MSG_task_create(name, 0, parse_double(size), NULL), to);
  44   INFO1("Sent %s", name);
  45   free(name);
  46 }
  47
  48
  49 static int spawned_send(int argc, char *argv[])
  50 {
  51   INFO3("%s: Sending %s on %s", MSG_process_self()->name,
  52         argv[1],argv[0]);
  53   MSG_task_send(MSG_task_create(argv[0], 0, parse_double(argv[1]), NULL),
  54                 argv[0]);
  55   return 0;
  56 }
  57
  58 static void Isend(xbt_dynar_t action)
  59 {
  60   char spawn_name[80];
  61   char *to = xbt_dynar_get_as(action, 2, char *);
  62   char *size = xbt_dynar_get_as(action, 3, char *);
  63   char **myargv;
  64   m_process_t comm_helper;
  65
  66   INFO1("Isend on %s: spawn process ",
  67          MSG_process_get_name(MSG_process_self()));
  68
  69   myargv = (char**) calloc (3, sizeof (char*));
  70
  71   myargv[0] = xbt_strdup(to);
  72   myargv[1] = xbt_strdup(size);
  73   myargv[2] = NULL;
  74
  75   sprintf(spawn_name,"%s_wait",MSG_process_get_name(MSG_process_self()));
  76   comm_helper =
  77     MSG_process_create_with_arguments(spawn_name, spawned_send,
  78                                       NULL, MSG_host_self(), 2, myargv);
  79 }
  80
  81
  82 static void recv(xbt_dynar_t action)
  83 {
  84   char *name = xbt_str_join(action, " ");
  85   m_task_t task = NULL;
  86   INFO1("Receiving: %s", name);
  87   //FIXME: argument of action ignored so far; semantic not clear
  88   //char *from=xbt_dynar_get_as(action,2,char*);
  89   MSG_task_receive(&task, MSG_process_get_name(MSG_process_self()));
  90   INFO1("Received %s", MSG_task_get_name(task));
  91   MSG_task_destroy(task);
  92   free(name);
  93 }
  94
  95 static int spawned_recv(int argc, char *argv[])
  96 {
  97   m_task_t task = NULL;
  98   char* name = (char *) MSG_process_get_data(MSG_process_self());
  99   INFO1("Receiving on %s", name);
 100   MSG_task_receive(&task, name);
 101   INFO1("Received %s", MSG_task_get_name(task));
 102   INFO1("waiter on %s", MSG_process_get_name(MSG_process_self()));
 103   MSG_task_send(MSG_task_create("waiter",0,0,NULL),MSG_process_get_name(MSG_process_self()));
 104
 105   MSG_task_destroy(task);
 106   return 0;
 107 }
 108
 109
 110 static void Irecv(xbt_dynar_t action)
 111 {
 112   char *name = xbt_str_join(action, " ");
 113   m_process_t comm_helper;
 114
 115   INFO1("Irecv on %s: spawn process ",
 116          MSG_process_get_name(MSG_process_self()));
 117
 118   sprintf(name,"%s_wait",MSG_process_get_name(MSG_process_self()));
 119   comm_helper = MSG_process_create(name,spawned_recv,
 120                  (void *) MSG_process_get_name(MSG_process_self()),
 121                  MSG_host_self());
 122
 123
 124   free(name);
 125 }
 126
 127
 128 static void wait(xbt_dynar_t action)
 129 {
 130   char *name = xbt_str_join(action, " ");
 131   char task_name[80];
 132   m_task_t task = NULL;
 133
 134   INFO1("wait: %s", name);
 135   sprintf(task_name,"%s_wait",MSG_process_get_name(MSG_process_self()));
 136   MSG_task_receive(&task,task_name);
 137   MSG_task_destroy(task);
 138   INFO1("waited: %s", name);
 139   free(name);
 140 }
 141
 142 static void barrier (xbt_dynar_t action)
 143 {
 144   char *name = xbt_str_join(action, " ");
 145   INFO1("barrier: %s", name);
 146
 147
 148   free(name);
 149
 150 }
 151
 152 static void reduce(xbt_dynar_t action)
 153 {
 154   int i;
 155   char *name;
 156   char task_name[80];
 157   char spawn_name[80];
 158   char *comm_size = xbt_dynar_get_as(action, 2, char *);
 159   char *comp_size = xbt_dynar_get_as(action, 3, char *);
 160   m_process_t comm_helper=NULL;
 161   m_task_t task=NULL, comp_task=NULL;
 162   const char* process_name;
 163
 164   coll_ctr counters =  (coll_ctr) MSG_process_get_data(MSG_process_self());
 165
 166   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 167               ", can't use collective operations");
 168
 169   process_name = MSG_process_get_name(MSG_process_self());
 170
 171   if (!counters){
 172     DEBUG0("Initialize the counters");
 173     counters = (coll_ctr) calloc (1, sizeof(struct coll_ctr_t));
 174   }
 175
 176   name = bprintf("reduce_%d", counters->reduce_counter++);
 177
 178   if (!strcmp(process_name, "process0")){
 179     INFO2("%s: %s is the Root",name, process_name);
 180     for(i=1;i<communicator_size;i++){
 181       sprintf(spawn_name,"%s_process%d", name, i);
 182       sprintf(task_name,"%s_wait", spawn_name);
 183       comm_helper =
 184         MSG_process_create(task_name, spawned_recv,
 185                            (void *) xbt_strdup(spawn_name),
 186                            MSG_host_self());
 187     }
 188
 189     for(i=1;i<communicator_size;i++){
 190       sprintf(task_name,"%s_process%d_wait", name, i);
 191       MSG_task_receive(&task,task_name);
 192       MSG_task_destroy(task);
 193       task=NULL;
 194     }
 195
 196     comp_task =
 197       MSG_task_create("reduce_comp", parse_double(comp_size), 0, NULL);
 198     INFO1("%s: computing 'reduce_comp'", name);
 199     MSG_task_execute(comp_task);
 200     MSG_task_destroy(comp_task);
 201     INFO1("%s: computed", name);
 202   } else {
 203     INFO2("%s: %s sends", name, process_name);
 204     sprintf(task_name,"%s_%s", name, process_name);
 205     INFO1("put on %s", task_name);
 206     MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL),
 207                   task_name);
 208   }
 209
 210   MSG_process_set_data(MSG_process_self(), (void*)counters);
 211   free(name);
 212 }
 213
 214 static void bcast (xbt_dynar_t action)
 215 {
 216   int i;
 217   char *name;
 218   const char* process_name;
 219   char task_name[80];
 220   char spawn_name[80];
 221   char **myargv;
 222   m_process_t comm_helper=NULL;
 223   m_task_t task=NULL;
 224   char *size = xbt_dynar_get_as(action, 2, char *);
 225   coll_ctr counters =  (coll_ctr) MSG_process_get_data(MSG_process_self());
 226
 227   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 228               ", can't use collective operations");
 229
 230
 231   process_name = MSG_process_get_name(MSG_process_self());
 232   if (!counters){
 233     DEBUG0("Initialize the counters");
 234     counters = (coll_ctr) calloc (1, sizeof(struct coll_ctr_t));
 235   }
 236
 237   name = bprintf("bcast_%d", counters->bcast_counter++);
 238   if (!strcmp(process_name, "process0")){
 239     INFO2("%s: %s is the Root",name, process_name);
 240
 241     for(i=1;i<communicator_size;i++){
 242       myargv = (char**) calloc (3, sizeof (char*));
 243       myargv[0] = xbt_strdup(name);
 244       myargv[1] = xbt_strdup(size);
 245       myargv[2] = NULL;
 246
 247       sprintf(spawn_name,"%s_%d", myargv[0], i);
 248       comm_helper =
 249         MSG_process_create_with_arguments(spawn_name, spawned_send,
 250                                           NULL, MSG_host_self(), 2, myargv);
 251     }
 252
 253     for(i=1;i<communicator_size;i++){
 254       sprintf(task_name,"process%d_wait", i);
 255       DEBUG1("get on %s", task_name);
 256       MSG_task_receive(&task,task_name);
 257       MSG_task_destroy(task);
 258       task=NULL;
 259     }
 260     INFO2("%s: all messages sent by %s have been received",
 261           name, process_name);
 262   } else {
 263     INFO2("%s: %s receives", name, process_name);
 264     MSG_task_receive(&task, name);
 265     MSG_task_destroy(task);
 266     INFO2("%s: %s has received", name,process_name);
 267     sprintf(task_name,"%s_wait", process_name);
 268     DEBUG1("put on %s", task_name);
 269     MSG_task_send(MSG_task_create("waiter",0,0,NULL),task_name);
 270   }
 271
 272   MSG_process_set_data(MSG_process_self(), (void*)counters);
 273   free(name);
 274 }
 275
 276
 277 static void sleep(xbt_dynar_t action)
 278 {
 279   char *name = xbt_str_join(action, " ");
 280   char *duration = xbt_dynar_get_as(action, 2, char *);
 281   INFO1("sleep: %s", name);
 282   MSG_process_sleep(parse_double(duration));
 283   INFO1("sleept: %s", name);
 284   free(name);
 285 }
 286
 287 static void allReduce(xbt_dynar_t action)
 288 {
 289   int i;
 290   char *name;
 291   char task_name[80];
 292   char spawn_name[80];
 293   char **myargv;
 294   char *comm_size = xbt_dynar_get_as(action, 2, char *);
 295   char *comp_size = xbt_dynar_get_as(action, 3, char *);
 296   m_process_t comm_helper=NULL;
 297   m_task_t task=NULL, comp_task=NULL;
 298   const char* process_name;
 299
 300   coll_ctr counters =  (coll_ctr) MSG_process_get_data(MSG_process_self());
 301
 302   xbt_assert0(communicator_size, "Size of Communicator is not defined"
 303               ", can't use collective operations");
 304
 305   process_name = MSG_process_get_name(MSG_process_self());
 306
 307   if (!counters){
 308     DEBUG0("Initialize the counters");
 309     counters = (coll_ctr) calloc (1, sizeof(struct coll_ctr_t));
 310   }
 311
 312   name = bprintf("allReduce_%d", counters->allReduce_counter++);
 313
 314   if (!strcmp(process_name, "process0")){
 315     INFO2("%s: %s is the Root",name, process_name);
 316     for(i=1;i<communicator_size;i++){
 317       sprintf(spawn_name,"%s_process%d", name, i);
 318       sprintf(task_name,"%s_wait", spawn_name);
 319       comm_helper =
 320         MSG_process_create(task_name, spawned_recv,
 321                            (void *) xbt_strdup(spawn_name),
 322                            MSG_host_self());
 323     }
 324
 325     for(i=1;i<communicator_size;i++){
 326       sprintf(task_name,"%s_process%d_wait", name, i);
 327       MSG_task_receive(&task,task_name);
 328       MSG_task_destroy(task);
 329       task=NULL;
 330     }
 331
 332     comp_task =
 333       MSG_task_create("allReduce_comp", parse_double(comp_size), 0, NULL);
 334     INFO1("%s: computing 'reduce_comp'", name);
 335     MSG_task_execute(comp_task);
 336     MSG_task_destroy(comp_task);
 337     INFO1("%s: computed", name);
 338
 339     for(i=1;i<communicator_size;i++){
 340       myargv = (char**) calloc (3, sizeof (char*));
 341       myargv[0] = xbt_strdup(name);
 342       myargv[1] = xbt_strdup(comm_size);
 343       myargv[2] = NULL;
 344
 345       sprintf(spawn_name,"%s_%d", myargv[0], i);
 346       comm_helper =
 347         MSG_process_create_with_arguments(spawn_name, spawned_send,
 348                                           NULL, MSG_host_self(), 2, myargv);
 349     }
 350
 351     for(i=1;i<communicator_size;i++){
 352       sprintf(task_name,"process%d_wait", i);
 353       DEBUG1("get on %s", task_name);
 354       MSG_task_receive(&task,task_name);
 355       MSG_task_destroy(task);
 356       task=NULL;
 357     }
 358     INFO2("%s: all messages sent by %s have been received",
 359           name, process_name);
 360
 361   } else {
 362     INFO2("%s: %s sends", name, process_name);
 363     sprintf(task_name,"%s_%s", name, process_name);
 364     INFO1("put on %s", task_name);
 365     MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL),
 366                   task_name);
 367
 368     MSG_task_receive(&task, name);
 369     MSG_task_destroy(task);
 370     INFO2("%s: %s has received", name,process_name);
 371     sprintf(task_name,"%s_wait", process_name);
 372     DEBUG1("put on %s", task_name);
 373     MSG_task_send(MSG_task_create("waiter",0,0,NULL),task_name);
 374
 375   }
 376
 377   MSG_process_set_data(MSG_process_self(), (void*)counters);
 378   free(name);
 379
 380 }
 381
 382 static void comm_size(xbt_dynar_t action)
 383 {
 384   char *size = xbt_dynar_get_as(action, 2, char *);
 385   communicator_size = parse_double(size);
 386 }
 387
 388 static void compute(xbt_dynar_t action)
 389 {
 390   char *name = xbt_str_join(action, " ");
 391   char *amout = xbt_dynar_get_as(action, 2, char *);
 392   m_task_t task = MSG_task_create(name, parse_double(amout), 0, NULL);
 393   INFO1("computing: %s", name);
 394   MSG_task_execute(task);
 395   MSG_task_destroy(task);
 396   INFO1("computed: %s", name);
 397   free(name);
 398 }
 399
 400 /** Main function */
 401 int main(int argc, char *argv[])
 402 {
 403   MSG_error_t res = MSG_OK;
 404
 405   /* Check the given arguments */
 406   MSG_global_init(&argc, argv);
 407   if (argc < 4) {
 408     printf("Usage: %s platform_file deployment_file action_files\n", argv[0]);
 409     printf("example: %s msg_platform.xml msg_deployment.xml actions\n",
 410            argv[0]);
 411     exit(1);
 412   }
 413
 414   /*  Simulation setting */
 415   MSG_create_environment(argv[1]);
 416
 417   /* No need to register functions as in classical MSG programs: the actions get started anyway */
 418   MSG_launch_application(argv[2]);
 419
 420   /*   Action registration */
 421   MSG_action_register("comm_size", comm_size);
 422   MSG_action_register("send", send);
 423   MSG_action_register("Isend", Isend);
 424   MSG_action_register("recv", recv);
 425   MSG_action_register("Irecv", Irecv);
 426   MSG_action_register("wait", wait);
 427   MSG_action_register("barrier", barrier);
 428   MSG_action_register("bcast", bcast);
 429   MSG_action_register("reduce", reduce);
 430   MSG_action_register("allReduce", allReduce);
 431   MSG_action_register("sleep", sleep);
 432   MSG_action_register("compute", compute);
 433
 434
 435   /* Actually do the simulation using MSG_action_trace_run */
 436   res = MSG_action_trace_run(argv[3]);
 437
 438   INFO1("Simulation time %g", MSG_get_clock());
 439   MSG_clean();
 440
 441   if (res == MSG_OK)
 442     return 0;
 443   else
 444     return 1;
 445 }                               /* end_of_main */