examples/msg/masterslave/masterslave_forwarder.c

   1 /*      $Id$     */
   2
   3 /* Copyright (c) 2002,2003,2004 Arnaud Legrand. 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 "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
  10 #include "xbt/sysdep.h"         /* calloc, printf */
  11
  12 /* Create a log channel to have nice outputs. */
  13 #include "xbt/log.h"
  14 #include "xbt/asserts.h"
  15 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  16                              "Messages specific for this msg example");
  17
  18 int master(int argc, char *argv[]);
  19 int slave(int argc, char *argv[]);
  20 int forwarder(int argc, char *argv[]);
  21 MSG_error_t test_all(const char *platform_file, const char *application_file);
  22
  23 typedef enum {
  24   PORT_22 = 0,
  25   MAX_CHANNEL
  26 } channel_t;
  27
  28 #define FINALIZE ((void*)221297)        /* a magic number to tell people to stop working */
  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, 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       xbt_assert1(slaves[i - 4] != NULL, "Unknown host %s. Stopping Now! ",
  69                   argv[i]);
  70     }
  71   }
  72
  73   INFO2("Got %d slaves and %d tasks to process", slaves_count,
  74         number_of_tasks);
  75   for (i = 0; i < slaves_count; i++)
  76     DEBUG1("%s", slaves[i]->name);
  77
  78   for (i = 0; i < number_of_tasks; i++) {
  79     INFO2("Sending \"%s\" to \"%s\"",
  80           todo[i]->name, slaves[i % slaves_count]->name);
  81     if (MSG_host_self() == slaves[i % slaves_count]) {
  82       INFO0("Hey ! It's me ! :)");
  83     }
  84
  85     MSG_task_put(todo[i], slaves[i % slaves_count], PORT_22);
  86     INFO0("Sent");
  87   }
  88
  89   INFO0
  90     ("All tasks have been dispatched. Let's tell everybody the computation is over.");
  91   for (i = 0; i < slaves_count; i++)
  92     MSG_task_put(MSG_task_create("finalize", 0, 0, FINALIZE),
  93                  slaves[i], PORT_22);
  94
  95   INFO0("Goodbye now!");
  96   free(slaves);
  97   free(todo);
  98   return 0;
  99 }                               /* end_of_master */
 100
 101 /** Receiver function  */
 102 int slave(int argc, char *argv[])
 103 {
 104   m_task_t task = NULL;
 105   int res;
 106   while (1) {
 107     res = MSG_task_get(&(task), PORT_22);
 108     xbt_assert0(res == MSG_OK, "MSG_task_get failed");
 109
 110     INFO1("Received \"%s\"", MSG_task_get_name(task));
 111     if (!strcmp(MSG_task_get_name(task), "finalize")) {
 112       MSG_task_destroy(task);
 113       break;
 114     }
 115
 116     INFO1("Processing \"%s\"", MSG_task_get_name(task));
 117     MSG_task_execute(task);
 118     INFO1("\"%s\" done", MSG_task_get_name(task));
 119     MSG_task_destroy(task);
 120     task = NULL;
 121   }
 122   INFO0("I'm done. See you!");
 123   return 0;
 124 }                               /* end_of_slave */
 125
 126 /** Forwarder function */
 127 int forwarder(int argc, char *argv[])
 128 {
 129   int i;
 130   int slaves_count;
 131   m_host_t *slaves;
 132
 133   {                             /* Process organisation */
 134     slaves_count = argc - 1;
 135     slaves = xbt_new0(m_host_t, slaves_count);
 136
 137     for (i = 1; i < argc; i++) {
 138       slaves[i - 1] = MSG_get_host_by_name(argv[i]);
 139       if (slaves[i - 1] == NULL) {
 140         INFO1("Unknown host %s. Stopping Now! ", argv[i]);
 141         abort();
 142       }
 143     }
 144   }
 145
 146   i = 0;
 147   while (1) {
 148     m_task_t task = NULL;
 149     int a;
 150     a = MSG_task_get(&(task), PORT_22);
 151     if (a == MSG_OK) {
 152       INFO1("Received \"%s\"", MSG_task_get_name(task));
 153       if (MSG_task_get_data(task) == FINALIZE) {
 154         INFO0
 155           ("All tasks have been dispatched. Let's tell everybody the computation is over.");
 156         for (i = 0; i < slaves_count; i++)
 157           MSG_task_put(MSG_task_create("finalize", 0, 0, FINALIZE),
 158                        slaves[i], PORT_22);
 159         MSG_task_destroy(task);
 160         break;
 161       }
 162       INFO2("Sending \"%s\" to \"%s\"",
 163             MSG_task_get_name(task), slaves[i % slaves_count]->name);
 164       MSG_task_put(task, slaves[i % slaves_count], PORT_22);
 165       i++;
 166     } else {
 167       INFO0("Hey ?! What's up ? ");
 168       xbt_assert0(0, "Unexpected behavior");
 169     }
 170   }
 171   xbt_free(slaves);
 172
 173   INFO0("I'm done. See you!");
 174   return 0;
 175 }                               /* end_of_forwarder */
 176
 177 /** Test function */
 178 MSG_error_t test_all(const char *platform_file, const char *application_file)
 179 {
 180   MSG_error_t res = MSG_OK;
 181
 182   /* MSG_config("workstation/model","KCCFLN05"); */
 183   {                             /*  Simulation setting */
 184     MSG_set_channel_number(MAX_CHANNEL);
 185     MSG_create_environment(platform_file);
 186   }
 187   {                             /*   Application deployment */
 188     MSG_function_register("master", master);
 189     MSG_function_register("slave", slave);
 190     MSG_function_register("forwarder", forwarder);
 191     MSG_launch_application(application_file);
 192   }
 193   res = MSG_main();
 194
 195   INFO1("Simulation time %g", MSG_get_clock());
 196   return res;
 197 }                               /* end_of_test_all */
 198
 199
 200 /** Main function */
 201 int main(int argc, char *argv[])
 202 {
 203   MSG_error_t res = MSG_OK;
 204
 205   MSG_global_init(&argc, argv);
 206   if (argc < 3) {
 207     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 208     printf("example: %s msg_platform.xml msg_deployment.xml\n", argv[0]);
 209     exit(1);
 210   }
 211   res = test_all(argv[1], argv[2]);
 212   MSG_clean();
 213
 214   if (res == MSG_OK)
 215     return 0;
 216   else
 217     return 1;
 218 }                               /* end_of_main */