Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Added files to the MASTER/SLAVE parser bypassing example. It is not
[simgrid.git] / 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