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