examples/msg/masterslave/masterslave_cluster.c

   1 /* Copyright (c) 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 "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
   9 #include "surf/surfxml_parse.h" /* to override surf_parse and bypass the parser */
  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 #define FINALIZE ((void*)221297)        /* a magic number to tell people to stop working */
  18
  19 MSG_error_t test_all(const char *platform_file);
  20
  21
  22 int master(int argc, char *argv[]);
  23 int slave(int argc, char *argv[]);
  24
  25 typedef enum {
  26   PORT_22 = 0,
  27   MAX_CHANNEL
  28 } channel_t;
  29
  30 /** Emitter function  */
  31 int master(int argc, char *argv[])
  32 {
  33   int slaves_count = 0;
  34   m_host_t *slaves = NULL;
  35   m_task_t *todo = NULL;
  36   int number_of_tasks = 0;
  37   double task_comp_size = 0;
  38   double task_comm_size = 0;
  39
  40
  41   int i;
  42
  43   xbt_assert1(sscanf(argv[1], "%d", &number_of_tasks),
  44               "Invalid argument %s\n", argv[1]);
  45   xbt_assert1(sscanf(argv[2], "%lg", &task_comp_size),
  46               "Invalid argument %s\n", argv[2]);
  47   xbt_assert1(sscanf(argv[3], "%lg", &task_comm_size),
  48               "Invalid argument %s\n", argv[3]);
  49
  50   {                             /*  Task creation */
  51     char sprintf_buffer[64];
  52
  53     todo = xbt_new0(m_task_t, number_of_tasks);
  54
  55     for (i = 0; i < number_of_tasks; i++) {
  56       sprintf(sprintf_buffer, "Task_%d", i);
  57       todo[i] =
  58           MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size,
  59                           NULL);
  60     }
  61   }
  62
  63   {                             /* Process organisation */
  64     slaves_count = argc - 4;
  65     slaves = xbt_new0(m_host_t, slaves_count);
  66
  67     for (i = 4; i < argc; i++) {
  68       slaves[i - 4] = MSG_get_host_by_name(argv[i]);
  69       if (slaves[i - 4] == NULL) {
  70         INFO1("Unknown host %s. Stopping Now! ", argv[i]);
  71         abort();
  72       }
  73     }
  74   }
  75
  76   INFO1("Got %d slave(s) :", slaves_count);
  77   for (i = 0; i < slaves_count; i++)
  78     INFO1("\t %s", slaves[i]->name);
  79
  80   INFO1("Got %d task to process :", number_of_tasks);
  81
  82   for (i = 0; i < number_of_tasks; i++)
  83     INFO1("\t\"%s\"", todo[i]->name);
  84
  85   for (i = 0; i < number_of_tasks; i++) {
  86     INFO2("Sending \"%s\" to \"%s\"",
  87           todo[i]->name, slaves[i % slaves_count]->name);
  88     if (MSG_host_self() == slaves[i % slaves_count]) {
  89       INFO0("Hey ! It's me ! :)");
  90     }
  91     MSG_task_put(todo[i], slaves[i % slaves_count], PORT_22);
  92     INFO0("Send completed");
  93   }
  94
  95   INFO0
  96       ("All tasks have been dispatched. Let's tell everybody the computation is over.");
  97   for (i = 0; i < slaves_count; i++)
  98     MSG_task_put(MSG_task_create("finalize", 0, 0, FINALIZE),
  99                  slaves[i], PORT_22);
 100
 101   INFO0("Goodbye now!");
 102   free(slaves);
 103   free(todo);
 104   return 0;
 105 }                               /* end_of_master */
 106
 107 /** Receiver function  */
 108 int slave(int argc, char *argv[])
 109 {
 110   INFO0("I'm a slave");
 111   while (1) {
 112     m_task_t task = NULL;
 113     int a;
 114     a = MSG_task_get(&(task), PORT_22);
 115     if (a == MSG_OK) {
 116       INFO1("Received \"%s\" ", MSG_task_get_name(task));
 117       if (MSG_task_get_data(task) == FINALIZE) {
 118         MSG_task_destroy(task);
 119         break;
 120       }
 121       INFO1("Processing \"%s\" ", MSG_task_get_name(task));
 122       MSG_task_execute(task);
 123       INFO1("\"%s\" done ", MSG_task_get_name(task));
 124       MSG_task_destroy(task);
 125     } else {
 126       INFO0("Hey ?! What's up ? ");
 127       xbt_assert0(0, "Unexpected behavior");
 128     }
 129   }
 130   INFO0("I'm done. See you!");
 131   return 0;
 132 }                               /* end_of_slave */
 133
 134
 135 /** Bypass deployment **/
 136 static int bypass_deployment(void)
 137 {
 138         int nb_host,i;
 139         static int AX_ptr;
 140         static int surfxml_bufferstack_size = 2048;
 141         static int surfxml_buffer_stack_stack_ptr = 0;
 142         static int surfxml_buffer_stack_stack[1024];
 143         /* allocating memory to the buffer, I think 2MB should be enough */
 144         surfxml_bufferstack = xbt_new0(char, surfxml_bufferstack_size);
 145         nb_host = MSG_get_host_number();
 146
 147         /* <platform> */
 148         SURFXML_BUFFER_SET(platform_version, "3");
 149         SURFXML_START_TAG(platform);
 150         DEBUG0("<platform version=\"3\">");
 151
 152           DEBUG0("      <process host=\"c-0.me\" function=\"master\">");
 153           SURFXML_BUFFER_SET(process_host, "c-0.me");
 154           SURFXML_BUFFER_SET(process_function, "master");
 155           SURFXML_BUFFER_SET(process_start_time, "-1.0");
 156           SURFXML_BUFFER_SET(process_kill_time, "-1.0");
 157           SURFXML_START_TAG(process);
 158
 159           DEBUG1("              <argument value=\"%s\"/>",bprintf("%d",nb_host-1));
 160           SURFXML_BUFFER_SET(argument_value, bprintf("%d",nb_host-1));
 161           SURFXML_START_TAG(argument);
 162           SURFXML_END_TAG(argument);
 163
 164           DEBUG0("              <argument value=\"5000000\"/>");
 165           SURFXML_BUFFER_SET(argument_value, "5000000");
 166           SURFXML_START_TAG(argument);
 167           SURFXML_END_TAG(argument);
 168
 169           DEBUG0("              <argument value=\"100000\"/>");
 170           SURFXML_BUFFER_SET(argument_value, "100000");
 171           SURFXML_START_TAG(argument);
 172           SURFXML_END_TAG(argument);
 173
 174         for(i=1 ; i<nb_host ; i++)
 175         {
 176           DEBUG1("              <argument value=\"%s.me\"/>",bprintf("c-%d",i));
 177           SURFXML_BUFFER_SET(argument_value, bprintf("c-%d.me",i));
 178           SURFXML_START_TAG(argument);
 179           SURFXML_END_TAG(argument);
 180         }
 181         DEBUG0("        </process>");
 182         SURFXML_END_TAG(process);
 183
 184         for(i=1 ; i<nb_host ; i++)
 185         {
 186           DEBUG1("      <process host=\"%s.me\" function=\"slave\"/>",bprintf("c-%d",i));
 187           SURFXML_BUFFER_SET(process_host, bprintf("c-%d.me",i));
 188           SURFXML_BUFFER_SET(process_function, "slave");
 189           SURFXML_BUFFER_SET(process_start_time, "-1.0");
 190           SURFXML_BUFFER_SET(process_kill_time, "-1.0");
 191           SURFXML_START_TAG(process);
 192           SURFXML_END_TAG(process);
 193         }
 194
 195         DEBUG0("</platform>");
 196         SURFXML_END_TAG(platform);
 197
 198         free(surfxml_bufferstack);
 199         return 0;
 200 }
 201
 202 /** Test function */
 203 MSG_error_t test_all(const char *platform_file)
 204 {
 205         MSG_error_t res = MSG_OK;
 206         MSG_set_channel_number(MAX_CHANNEL);
 207         MSG_create_environment(platform_file);
 208         MSG_function_register("master", master);
 209         MSG_function_register("slave", slave);
 210         surf_parse = bypass_deployment;
 211         MSG_launch_application(NULL);
 212
 213         res = MSG_main();
 214
 215         INFO1("Simulation time %g", MSG_get_clock());
 216         return res;
 217 }                               /* end_of_test_all */
 218
 219
 220 /** Main function */
 221 int main(int argc, char *argv[])
 222 {
 223         MSG_error_t res = MSG_OK;
 224
 225         MSG_global_init(&argc, argv);
 226         res = test_all(argv[1]);
 227         MSG_clean();
 228
 229         if (res == MSG_OK)
 230                 return 0;
 231         else
 232                 return 1;
 233 }                               /* end_of_main */