doc/msg-tuto-src/masterworker0.c

   1 /* Copyright (c) 2007-2010, 2013-2014. 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 "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
   9 #include "xbt/sysdep.h"         /* calloc, printf */
  10
  11 /* Create a log channel to have nice outputs. */
  12 #include "xbt/log.h"
  13 #include "xbt/asserts.h"
  14 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  15                              "Messages specific for this msg example");
  16
  17 int master(int argc, char *argv[]);
  18 int worker(int argc, char *argv[]);
  19 msg_error_t test_all(const char *platform_file,
  20                      const char *application_file);
  21
  22 #define FINALIZE ((void*)221297)        /* a magic number to tell people to stop working */
  23
  24 /** Emitter function  */
  25 int master(int argc, char *argv[])
  26 {
  27   int workers_count = 0;
  28   msg_host_t *workers = NULL;
  29   msg_task_t *todo = NULL;
  30   int number_of_tasks = 0;
  31   double task_comp_size = 0;
  32   double task_comm_size = 0;
  33
  34   int i;
  35
  36   _XBT_GNUC_UNUSED int res = sscanf(argv[1], "%d", &number_of_tasks);
  37   xbt_assert(res,"Invalid argument %s\n", argv[1]);
  38   res = sscanf(argv[2], "%lg", &task_comp_size);
  39   xbt_assert(res, "Invalid argument %s\n", argv[2]);
  40   res = sscanf(argv[3], "%lg", &task_comm_size);
  41   xbt_assert(res, "Invalid argument %s\n", argv[3]);
  42
  43   {                             /*  Task creation */
  44     char sprintf_buffer[64];
  45
  46     todo = xbt_new0(msg_task_t, number_of_tasks);
  47
  48     for (i = 0; i < number_of_tasks; i++) {
  49       sprintf(sprintf_buffer, "Task_%d", i);
  50       todo[i] =
  51           MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size,
  52                           NULL);
  53     }
  54   }
  55
  56   {                             /* Process organisation */
  57     workers_count = argc - 4;
  58     workers = xbt_new0(msg_host_t, workers_count);
  59
  60     for (i = 4; i < argc; i++) {
  61       workers[i - 4] = MSG_get_host_by_name(argv[i]);
  62       xbt_assert(workers[i - 4] != NULL, "Unknown host %s. Stopping Now! ",
  63                   argv[i]);
  64     }
  65   }
  66
  67   XBT_INFO("Got %d workers and %d tasks to process", workers_count,
  68         number_of_tasks);
  69
  70   for (i = 0; i < number_of_tasks; i++) {
  71     XBT_INFO("Sending \"%s\" to \"%s\"",
  72           todo[i]->name, MSG_host_get_name(workers[i % workers_count]));
  73     if (MSG_host_self() == workers[i % workers_count]) {
  74       XBT_INFO("Hey ! It's me ! :)");
  75     }
  76
  77     MSG_task_send(todo[i], MSG_host_get_name(workers[i % workers_count]));
  78     XBT_INFO("Sent");
  79   }
  80
  81   XBT_INFO
  82       ("All tasks have been dispatched. Let's tell everybody the computation is over.");
  83   for (i = 0; i < workers_count; i++) {
  84     msg_task_t finalize = MSG_task_create("finalize", 0, 0, FINALIZE);
  85     MSG_task_send(finalize, MSG_host_get_name(workers[i]));
  86   }
  87
  88   XBT_INFO("Goodbye now!");
  89   free(workers);
  90   free(todo);
  91   return 0;
  92 }                               /* end_of_master */
  93
  94 /** Receiver function  */
  95 int worker(int argc, char *argv[])
  96 {
  97   msg_task_t task = NULL;
  98   _XBT_GNUC_UNUSED int res;
  99   while (1) {
 100     res = MSG_task_receive(&(task),MSG_host_get_name(MSG_host_self()));
 101     xbt_assert(res == MSG_OK, "MSG_task_receive failed");
 102
 103     XBT_INFO("Received \"%s\"", MSG_task_get_name(task));
 104     if (!strcmp(MSG_task_get_name(task), "finalize")) {
 105       MSG_task_destroy(task);
 106       break;
 107     }
 108
 109     XBT_INFO("Processing \"%s\"", MSG_task_get_name(task));
 110     MSG_task_execute(task);
 111     XBT_INFO("\"%s\" done", MSG_task_get_name(task));
 112     MSG_task_destroy(task);
 113     task = NULL;
 114   }
 115   XBT_INFO("I'm done. See you!");
 116   return 0;
 117 }                               /* end_of_worker */
 118
 119 /** Test function */
 120 msg_error_t test_all(const char *platform_file,
 121                      const char *application_file)
 122 {
 123   msg_error_t res = MSG_OK;
 124
 125   {                             /*  Simulation setting */
 126     MSG_create_environment(platform_file);
 127   }
 128   {                             /*   Application deployment */
 129     MSG_function_register("master", master);
 130     MSG_function_register("worker", worker);
 131     MSG_launch_application(application_file);
 132   }
 133   res = MSG_main();
 134
 135   XBT_INFO("Simulation time %g", MSG_get_clock());
 136   return res;
 137 }                               /* end_of_test_all */
 138
 139
 140 /** Main function */
 141 int main(int argc, char *argv[])
 142 {
 143   msg_error_t res = MSG_OK;
 144
 145   MSG_init(&argc, argv);
 146   if (argc < 3) {
 147     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 148     printf("example: %s msg_platform.xml msg_deployment.xml\n", argv[0]);
 149     exit(1);
 150   }
 151   res = test_all(argv[1], argv[2]);
 152
 153   if (res == MSG_OK)
 154     return 0;
 155   else
 156     return 1;
 157 }                               /* end_of_main */