examples/msg/msg_test.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 /** \file msg_test.c
   9  *  \ingroup MSG_examples
  10  *  \brief Simulation of a master-slave application using a realistic platform
  11  *  and an external description of the deployment.
  12 */
  13
  14 /** Yeah! If you want to use msg, you need to include msg/msg.h */
  15 #include "msg/msg.h"
  16
  17 /** This flag enable the debugging messages from #PRINT_DEBUG_MESSAGE() */
  18 #define VERBOSE
  19 #include "messages.h"
  20
  21 int master(int argc, char *argv[]);
  22 int slave(int argc, char *argv[]);
  23 int forwarder(int argc, char *argv[]);
  24 void test_all(const char *platform_file, const char *application_file);
  25
  26
  27 typedef enum {
  28   PORT_22 = 0,
  29   MAX_CHANNEL
  30 } channel_t;
  31
  32 /** Print arguments
  33  * This function is just used so that users can check that each process
  34  *  has received the arguments it was supposed to receive.
  35  */
  36 static void print_args(int argc, char** argv)
  37 {
  38   int i ;
  39
  40   fprintf(stderr,"<");
  41   for(i=0; i<argc; i++)
  42     fprintf(stderr,"%s ",argv[i]);
  43   fprintf(stderr,">\n");
  44 }
  45
  46 /** Emitter function
  47  * This function has to be assigned to a m_process_t that will behave as the master.
  48    It should not be called directly but either given as a parameter to
  49    #MSG_process_create() or registered as a public function through
  50    #MSG_function_register() and then automatically assigned to a process through
  51    #MSG_launch_application().
  52
  53    C style arguments (argc/argv) are interpreted as
  54    \li the number of tasks to distribute
  55    \li the computation size of each task
  56    \li the size of the files associated to each task
  57    \li a list of host that will accept those tasks.
  58
  59    Tasks are dumbly sent in a round-robin style.
  60   */
  61 int master(int argc, char *argv[])
  62 {
  63   int slaves_count = 0;
  64   m_host_t *slaves = NULL;
  65   m_task_t *todo = NULL;
  66   int number_of_tasks = 0;
  67   double task_comp_size = 0;
  68   double task_comm_size = 0;
  69
  70
  71   int i;
  72
  73   print_args(argc,argv);
  74
  75   ASSERT(sscanf(argv[1],"%d", &number_of_tasks),
  76          "Invalid argument %s\n",argv[1]);
  77   ASSERT(sscanf(argv[2],"%lg", &task_comp_size),
  78          "Invalid argument %s\n",argv[2]);
  79   ASSERT(sscanf(argv[3],"%lg", &task_comm_size),
  80          "Invalid argument %s\n",argv[3]);
  81
  82   {                  /*  Task creation */
  83     char sprintf_buffer[64];
  84
  85     todo = calloc(number_of_tasks, sizeof(m_task_t));
  86
  87     for (i = 0; i < number_of_tasks; i++) {
  88       sprintf(sprintf_buffer, "Task_%d", i);
  89       todo[i] = MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size, NULL);
  90     }
  91   }
  92
  93   {                  /* Process organisation */
  94     slaves_count = argc - 4;
  95     slaves = calloc(slaves_count, sizeof(m_host_t));
  96
  97     for (i = 4; i < argc; i++) {
  98       slaves[i-4] = MSG_get_host_by_name(argv[i]);
  99       if(slaves[i-4]==NULL) {
 100         PRINT_MESSAGE("Unknown host %s. Stopping Now! \n", argv[i]);
 101         abort();
 102       }
 103     }
 104   }
 105
 106   PRINT_MESSAGE("Got %d slave(s) :\n", slaves_count);
 107   for (i = 0; i < slaves_count; i++)
 108     PRINT_MESSAGE("\t %s\n", slaves[i]->name);
 109
 110   PRINT_MESSAGE("Got %d task to process :\n", number_of_tasks);
 111
 112   for (i = 0; i < number_of_tasks; i++)
 113     PRINT_MESSAGE("\t\"%s\"\n", todo[i]->name);
 114
 115   for (i = 0; i < number_of_tasks; i++) {
 116     PRINT_MESSAGE("Sending \"%s\" to \"%s\"\n",
 117                   todo[i]->name,
 118                   slaves[i % slaves_count]->name);
 119     MSG_task_put(todo[i], slaves[i % slaves_count],
 120                  PORT_22);
 121     PRINT_MESSAGE("Send completed\n");
 122   }
 123
 124   PRINT_MESSAGE("All tasks have been dispatched. Bye!\n");
 125   free(slaves);
 126   free(todo);
 127   return 0;
 128 }
 129
 130 /** Receiver function
 131  * This function has to be assigned to a #m_process_t that has to behave as a slave.
 132    Just like #master(), it should not be called directly.
 133
 134    This function keeps waiting for tasks and executes them as it receives them.
 135   */
 136 int slave(int argc, char *argv[])
 137 {
 138   print_args(argc,argv);
 139
 140   while(1) {
 141     m_task_t task = NULL;
 142     int a;
 143     a = MSG_task_get(&(task), PORT_22);
 144     if (a == MSG_OK) {
 145       PRINT_MESSAGE("Received \"%s\" \n", task->name);
 146       PRINT_MESSAGE("Processing \"%s\" \n", task->name);
 147       MSG_task_execute(task);
 148       PRINT_MESSAGE("\"%s\" done \n", task->name);
 149       MSG_task_destroy(task);
 150     } else {
 151       PRINT_MESSAGE("Hey ?! What's up ? \n");
 152       DIE("Unexpected behaviour");
 153     }
 154   }
 155   PRINT_MESSAGE("I'm done. See you!\n");
 156   return 0;
 157 }
 158
 159 /** Receiver function
 160  * This function has to be assigned to a #m_process_t that has to behave as a forwarder.
 161    Just like #master(), it should not be called directly.
 162
 163    C style arguments (argc/argv) are interpreted as a list of host
 164    that will accept those tasks.
 165
 166    This function keeps waiting for tasks and dispathes them to its slaves.
 167   */
 168 int forwarder(int argc, char *argv[])
 169 {
 170   int i;
 171   int slaves_count = argc - 1;
 172   m_host_t *slaves = calloc(slaves_count, sizeof(m_host_t));
 173
 174   print_args(argc,argv);
 175
 176   {                  /* Process organisation */
 177     slaves_count = argc - 1;
 178     slaves = calloc(slaves_count, sizeof(m_host_t));
 179
 180     for (i = 1; i < argc; i++) {
 181       slaves[i-1] = MSG_get_host_by_name(argv[i]);
 182       if(slaves[i-1]==NULL) {
 183         PRINT_MESSAGE("Unknown host %s. Stopping Now! \n", argv[i]);
 184         abort();
 185       }
 186     }
 187   }
 188
 189   i=0;
 190   while(1) {
 191     m_task_t task = NULL;
 192     int a;
 193     a = MSG_task_get(&(task), PORT_22);
 194     if (a == MSG_OK) {
 195       PRINT_MESSAGE("Received \"%s\" \n", task->name);
 196       PRINT_MESSAGE("Sending \"%s\" to \"%s\"\n",
 197                     task->name,
 198                     slaves[i % slaves_count]->name);
 199       MSG_task_put(task, slaves[i % slaves_count],
 200                    PORT_22);
 201     } else {
 202       PRINT_MESSAGE("Hey ?! What's up ? \n");
 203       DIE("Unexpected behaviour");
 204     }
 205   }
 206
 207   PRINT_MESSAGE("I'm done. See you!\n");
 208   return 0;
 209 }
 210
 211
 212 /** Test function
 213  * This function is the core of the simulation and is divided only into 3 parts
 214  * thanks to MSG_create_environment() and MSG_launch_application().
 215  *      -# Simulation settings : MSG_create_environment() creates a realistic
 216  *         environment
 217  *      -# Application deployment : create the agents on the right locations with
 218  *         MSG_launch_application()
 219  *      -# The simulation is run with #MSG_main()
 220  * @param platform_file the name of a file containing an valid surfxml platform
 221  *        description.
 222  * @param application_file the name of a file containing a valid surfxml application
 223  *        description
 224  */
 225 void test_all(const char *platform_file,const char *application_file)
 226 {
 227   {                             /*  Simulation setting */
 228     MSG_set_channel_number(MAX_CHANNEL);
 229     MSG_create_environment(platform_file);
 230   }
 231   {                            /*   Application deployment */
 232     MSG_function_register("master", master);
 233     MSG_function_register("slave", slave);
 234     MSG_function_register("forwarder", forwarder);
 235     MSG_launch_application(application_file);
 236   }
 237   MSG_main();
 238   printf("Simulation time %g\n",MSG_getClock());
 239 }
 240
 241
 242 /** Main function
 243  * This initializes MSG, runs a simulation, and free all data-structures created
 244  * by MSG.
 245  */
 246 int main(int argc, char *argv[])
 247 {
 248   MSG_global_init_args(&argc,argv);
 249   if (argc < 3) {
 250      printf ("Usage: %s platform_file deployment_file\n",argv[0]);
 251      printf ("example: %s msg_platform.xml msg_deployment.xml\n",argv[0]);
 252      exit(1);
 253   }
 254   test_all(argv[1],argv[2]);
 255   MSG_clean();
 256   return (0);
 257 }