examples/msg/masterslave/masterslave_bypass.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 "surf/surfxml_parse.h" /* to override surf_parse */
  11 #include "surf/surfxml.h"       /* to hijack surf_parse_lex */
  12
  13 /* Create a log channel to have nice outputs. */
  14 #include "xbt/log.h"
  15 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,"Messages specific for this msg example");
  16 #define FINALIZE ((void*)221297) /* a magic number to tell people to stop working */
  17
  18 static int surf_parse_bypass(void)
  19 {
  20
  21   /* allocating memory to the buffer, I think 2MB should be enough */
  22   surfxml_buffer = xbt_new0(char, 2048);
  23
  24   /* <platform_description> */
  25   strcpy(A_surfxml_platform_description_version, "1");
  26   STag_surfxml_platform_description();
  27
  28
  29 /*   <cpu name="Cpu A" power="100000000.00" availability_file="trace_A.txt"/> */
  30   strcpy(A_surfxml_cpu_name,"Cpu A");
  31   strcpy(A_surfxml_cpu_power, "100000000.00");
  32   strcpy(A_surfxml_cpu_availability, "1.0");
  33   strcpy(A_surfxml_cpu_availability_file, "");
  34   A_surfxml_cpu_state = A_surfxml_cpu_state_ON;
  35   //bypassing the parser file
  36   AX_surfxml_cpu_state_file = 0;
  37   strcpy(A_surfxml_cpu_interference_send, "1.0");
  38   strcpy(A_surfxml_cpu_interference_recv, "1.0");
  39   strcpy(A_surfxml_cpu_interference_send_recv, "1.0");
  40   strcpy(A_surfxml_cpu_max_outgoing_rate, "-1.0");
  41
  42   STag_surfxml_cpu();
  43   ETag_surfxml_cpu();
  44
  45 /*   <cpu name="Cpu B" power="100000000.00" availability_file="trace_B.txt"/> */
  46   strcpy(A_surfxml_cpu_name, "Cpu B");
  47   strcpy(A_surfxml_cpu_power, "100000000.00");
  48   strcpy(A_surfxml_cpu_availability, "1.0");
  49   strcpy(A_surfxml_cpu_availability_file, "");
  50   A_surfxml_cpu_state = A_surfxml_cpu_state_ON;
  51   //bypassing the parser file
  52   AX_surfxml_cpu_state_file = 0;
  53   strcpy(A_surfxml_cpu_interference_send, "1.0");
  54   strcpy(A_surfxml_cpu_interference_recv, "1.0");
  55   strcpy(A_surfxml_cpu_interference_send_recv, "1.0");
  56   strcpy(A_surfxml_cpu_max_outgoing_rate, "-1.0");
  57
  58   STag_surfxml_cpu();
  59   ETag_surfxml_cpu();
  60
  61 /*   <network_link name="LinkA" bandwidth="10000000.0" latency="0.2"/> */
  62   strcpy(A_surfxml_network_link_name, "LinkA");
  63   strcpy(A_surfxml_network_link_bandwidth, "10000000.0");
  64   AX_surfxml_network_link_bandwidth_file = 0;
  65   strcpy(A_surfxml_network_link_latency, "0.2");
  66   AX_surfxml_network_link_latency_file = 0;
  67   A_surfxml_network_link_state = A_surfxml_network_link_state_ON;
  68   AX_surfxml_network_link_state_file = 0;
  69   A_surfxml_network_link_sharing_policy = A_surfxml_network_link_sharing_policy_SHARED;
  70   STag_surfxml_network_link();
  71   ETag_surfxml_network_link();
  72
  73 /*   <route src="Cpu A" dst="Cpu B"><route_element name="LinkA"/></route> */
  74   strcpy(A_surfxml_route_src, "Cpu A");
  75   strcpy(A_surfxml_route_dst, "Cpu B");
  76   strcpy(A_surfxml_route_impact_on_src, "0.0");
  77   strcpy(A_surfxml_route_impact_on_dst, "0.0");
  78   strcpy(A_surfxml_route_impact_on_src_with_other_recv, "0.0");
  79   strcpy(A_surfxml_route_impact_on_dst_with_other_send, "0.0");
  80
  81   STag_surfxml_route();
  82
  83   strcpy(A_surfxml_route_element_name, "LinkA");
  84   STag_surfxml_route_element();
  85   ETag_surfxml_route_element();
  86
  87   ETag_surfxml_route();
  88
  89 /*   <route src="Cpu B" dst="Cpu A"><route_element name="LinkA"/></route> */
  90   strcpy(A_surfxml_route_src, "Cpu B");
  91   strcpy(A_surfxml_route_dst, "Cpu A");
  92   strcpy(A_surfxml_route_impact_on_src, "0.0");
  93   strcpy(A_surfxml_route_impact_on_dst, "0.0");
  94   strcpy(A_surfxml_route_impact_on_src_with_other_recv, "0.0");
  95   strcpy(A_surfxml_route_impact_on_dst_with_other_send, "0.0");
  96
  97   STag_surfxml_route();
  98
  99   strcpy(A_surfxml_route_element_name, "LinkA");
 100   STag_surfxml_route_element();
 101   ETag_surfxml_route_element();
 102
 103   ETag_surfxml_route();
 104
 105 /*   <process host="Cpu A" function="master"> */
 106   strcpy(A_surfxml_process_host, "Cpu A");
 107   strcpy(A_surfxml_process_function, "master");
 108   strcpy(A_surfxml_process_start_time, "-1.0");
 109   strcpy(A_surfxml_process_kill_time, "-1.0");
 110   STag_surfxml_process();
 111
 112 /*      <argument value="20"/> */
 113   strcpy(A_surfxml_argument_value, "20");
 114   STag_surfxml_argument();
 115   ETag_surfxml_argument();
 116
 117 /*      <argument value="5000000"/> */
 118   strcpy(A_surfxml_argument_value, "5000000");
 119   STag_surfxml_argument();
 120   ETag_surfxml_argument();
 121
 122 /*      <argument value="100000"/> */
 123   strcpy(A_surfxml_argument_value, "100000");
 124   STag_surfxml_argument();
 125   ETag_surfxml_argument();
 126
 127 /*      <argument value="Cpu B"/> */
 128   strcpy(A_surfxml_argument_value, "Cpu B");
 129   STag_surfxml_argument();
 130   ETag_surfxml_argument();
 131
 132 /*   </process> */
 133   ETag_surfxml_process();
 134
 135 /*   <process host="Cpu B" function="slave"/> */
 136   strcpy(A_surfxml_process_host, "Cpu B");
 137   strcpy(A_surfxml_process_function, "slave");
 138   strcpy(A_surfxml_process_start_time, "-1.0");
 139   strcpy(A_surfxml_process_kill_time, "-1.0");
 140   STag_surfxml_process();
 141   ETag_surfxml_process();
 142
 143 /* </platform_description> */
 144   ETag_surfxml_platform_description();
 145
 146   return 0;
 147 }
 148
 149 int master(int argc, char *argv[]);
 150 int slave(int argc, char *argv[]);
 151 MSG_error_t test_all(void);
 152
 153 typedef enum {
 154   PORT_22 = 0,
 155   MAX_CHANNEL
 156 } channel_t;
 157
 158 /** Emitter function  */
 159 int master(int argc, char *argv[])
 160 {
 161   int slaves_count = 0;
 162   m_host_t *slaves = NULL;
 163   m_task_t *todo = NULL;
 164   int number_of_tasks = 0;
 165   double task_comp_size = 0;
 166   double task_comm_size = 0;
 167
 168
 169   int i;
 170
 171   xbt_assert1(sscanf(argv[1],"%d", &number_of_tasks),
 172          "Invalid argument %s\n",argv[1]);
 173   xbt_assert1(sscanf(argv[2],"%lg", &task_comp_size),
 174          "Invalid argument %s\n",argv[2]);
 175   xbt_assert1(sscanf(argv[3],"%lg", &task_comm_size),
 176          "Invalid argument %s\n",argv[3]);
 177
 178   {                  /*  Task creation */
 179     char sprintf_buffer[64];
 180
 181     todo = calloc(number_of_tasks, sizeof(m_task_t));
 182
 183     for (i = 0; i < number_of_tasks; i++) {
 184       sprintf(sprintf_buffer, "Task_%d", i);
 185       todo[i] = MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size, NULL);
 186     }
 187   }
 188
 189   {                  /* Process organisation */
 190     slaves_count = argc - 4;
 191     slaves = calloc(slaves_count, sizeof(m_host_t));
 192
 193     for (i = 4; i < argc; i++) {
 194       slaves[i-4] = MSG_get_host_by_name(argv[i]);
 195       if(slaves[i-4]==NULL) {
 196         INFO1("Unknown host %s. Stopping Now! ", argv[i]);
 197         abort();
 198       }
 199     }
 200   }
 201
 202   INFO1("Got %d slave(s) :", slaves_count);
 203   for (i = 0; i < slaves_count; i++)
 204     INFO1("\t %s", slaves[i]->name);
 205
 206   INFO1("Got %d task to process :", number_of_tasks);
 207
 208   for (i = 0; i < number_of_tasks; i++)
 209     INFO1("\t\"%s\"", todo[i]->name);
 210
 211   for (i = 0; i < number_of_tasks; i++) {
 212     INFO2("Sending \"%s\" to \"%s\"",
 213                   todo[i]->name,
 214                   slaves[i % slaves_count]->name);
 215     if(MSG_host_self()==slaves[i % slaves_count]) {
 216       INFO0("Hey ! It's me ! :)");
 217     }
 218     MSG_task_put(todo[i], slaves[i % slaves_count],
 219                  PORT_22);
 220     INFO0("Send completed");
 221   }
 222
 223   INFO0("All tasks have been dispatched. Let's tell everybody the computation is over.");
 224   for (i = 0; i < slaves_count; i++)
 225     MSG_task_put(MSG_task_create("finalize", 0, 0, FINALIZE),
 226                  slaves[i], PORT_22);
 227
 228   INFO0("Goodbye now!");
 229   free(slaves);
 230   free(todo);
 231   return 0;
 232 } /* end_of_master */
 233
 234 /** Receiver function  */
 235 int slave(int argc, char *argv[])
 236 {
 237   while(1) {
 238     m_task_t task = NULL;
 239     int a;
 240     a = MSG_task_get(&(task), PORT_22);
 241     if (a == MSG_OK) {
 242       INFO1("Received \"%s\" ", MSG_task_get_name(task));
 243       if(MSG_task_get_data(task)==FINALIZE) {
 244         MSG_task_destroy(task);
 245         break;
 246       }
 247       INFO1("Processing \"%s\" ", MSG_task_get_name(task));
 248       MSG_task_execute(task);
 249       INFO1("\"%s\" done ", MSG_task_get_name(task));
 250       MSG_task_destroy(task);
 251     } else {
 252       INFO0("Hey ?! What's up ? ");
 253       xbt_assert0(0,"Unexpected behavior");
 254     }
 255   }
 256   INFO0("I'm done. See you!");
 257   return 0;
 258 } /* end_of_slave */
 259
 260 /** Test function */
 261 MSG_error_t test_all(void)
 262 {
 263   MSG_error_t res = MSG_OK;
 264
 265   {                             /*  Simulation setting */
 266     MSG_set_channel_number(MAX_CHANNEL);
 267     MSG_paje_output("msg_test.trace");
 268     surf_parse = surf_parse_bypass;
 269     MSG_create_environment(NULL);
 270   }
 271   {                            /*   Application deployment */
 272     MSG_function_register("master", master);
 273     MSG_function_register("slave", slave);
 274     MSG_launch_application(NULL);
 275   }
 276   res = MSG_main();
 277
 278   INFO1("Simulation time %g",MSG_get_clock());
 279   return res;
 280 } /* end_of_test_all */
 281
 282 /** Main function */
 283 int main(int argc, char *argv[])
 284 {
 285   MSG_error_t res = MSG_OK;
 286
 287   MSG_global_init(&argc,argv);
 288   res = test_all();
 289   MSG_clean();
 290
 291   if(res==MSG_OK) return 0;
 292   else return 1;
 293 } /* end_of_main */
 294