Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
3e3560e5aa163da94341d76cf8941f3cad08eabd
[simgrid.git] / examples / msg / masterslave / masterslave_forwarder.c
1 /* Copyright (c) 2007, 2008, 2009, 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 "xbt/sysdep.h"         /* calloc, printf */
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 /** @addtogroup MSG_examples
18  * 
19  *  - <b>masterslave/masterslave_forwarder.c: Master/slaves
20  *    example</b>. This good old example is also very simple. Its basic
21  *    version is fully commented on this page: \ref MSG_ex_master_slave,
22  *    but several variants can be found in the same directory. 
23  */
24
25
26 int master(int argc, char *argv[]);
27 int slave(int argc, char *argv[]);
28 int forwarder(int argc, char *argv[]);
29 MSG_error_t test_all(const char *platform_file,
30                      const char *application_file);
31
32 #define FINALIZE ((void*)221297)        /* a magic number to tell people to stop working */
33
34 /** Emitter function  */
35 int master(int argc, char *argv[])
36 {
37   int slaves_count = 0;
38   m_host_t *slaves = NULL;
39   m_task_t *todo = NULL;
40   int number_of_tasks = 0;
41   double task_comp_size = 0;
42   double task_comm_size = 0;
43
44   int i;
45
46   _XBT_GNUC_UNUSED int res = sscanf(argv[1], "%d", &number_of_tasks);
47   xbt_assert(res,"Invalid argument %s\n", argv[1]);
48   res = sscanf(argv[2], "%lg", &task_comp_size);
49   xbt_assert(res, "Invalid argument %s\n", argv[2]);
50   res = sscanf(argv[3], "%lg", &task_comm_size);
51   xbt_assert(res, "Invalid argument %s\n", argv[3]);
52
53   {                             /*  Task creation */
54     char sprintf_buffer[64];
55
56     todo = xbt_new0(m_task_t, number_of_tasks);
57
58     for (i = 0; i < number_of_tasks; i++) {
59       sprintf(sprintf_buffer, "Task_%d", i);
60       todo[i] =
61           MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size,
62                           NULL);
63     }
64   }
65
66   {                             /* Process organisation */
67     slaves_count = argc - 4;
68     slaves = xbt_new0(m_host_t, slaves_count);
69
70     for (i = 4; i < argc; i++) {
71       slaves[i - 4] = MSG_get_host_by_name(argv[i]);
72       xbt_assert(slaves[i - 4] != NULL, "Unknown host %s. Stopping Now! ",
73                   argv[i]);
74     }
75   }
76
77   XBT_INFO("Got %d slaves and %d tasks to process", slaves_count,
78         number_of_tasks);
79   for (i = 0; i < slaves_count; i++)
80     XBT_DEBUG("%s", MSG_host_get_name(slaves[i]));
81
82   for (i = 0; i < number_of_tasks; i++) {
83     XBT_INFO("Sending \"%s\" to \"%s\"",
84           todo[i]->name, MSG_host_get_name(slaves[i % slaves_count]));
85     if (MSG_host_self() == slaves[i % slaves_count]) {
86       XBT_INFO("Hey ! It's me ! :)");
87     }
88
89     MSG_task_send(todo[i], MSG_host_get_name(slaves[i % slaves_count]));
90     XBT_INFO("Sent");
91   }
92
93   XBT_INFO
94       ("All tasks have been dispatched. Let's tell everybody the computation is over.");
95   for (i = 0; i < slaves_count; i++) {
96     m_task_t finalize = MSG_task_create("finalize", 0, 0, FINALIZE);
97     MSG_task_send(finalize, MSG_host_get_name(slaves[i]));
98   }
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   m_task_t task = NULL;
110   _XBT_GNUC_UNUSED int res;
111   while (1) {
112     res = MSG_task_receive(&(task),MSG_host_get_name(MSG_host_self()));
113     xbt_assert(res == MSG_OK, "MSG_task_get failed");
114
115     XBT_INFO("Received \"%s\"", MSG_task_get_name(task));
116     if (!strcmp(MSG_task_get_name(task), "finalize")) {
117       MSG_task_destroy(task);
118       break;
119     }
120
121     XBT_INFO("Processing \"%s\"", MSG_task_get_name(task));
122     MSG_task_execute(task);
123     XBT_INFO("\"%s\" done", MSG_task_get_name(task));
124     MSG_task_destroy(task);
125     task = NULL;
126   }
127   XBT_INFO("I'm done. See you!");
128   return 0;
129 }                               /* end_of_slave */
130
131 /** Forwarder function */
132 int forwarder(int argc, char *argv[])
133 {
134   int i;
135   int slaves_count;
136   m_host_t *slaves;
137
138   {                             /* Process organisation */
139     slaves_count = argc - 1;
140     slaves = xbt_new0(m_host_t, slaves_count);
141
142     for (i = 1; i < argc; i++) {
143       slaves[i - 1] = MSG_get_host_by_name(argv[i]);
144       if (slaves[i - 1] == NULL) {
145         XBT_INFO("Unknown host %s. Stopping Now! ", argv[i]);
146         abort();
147       }
148     }
149   }
150
151   i = 0;
152   while (1) {
153     m_task_t task = NULL;
154     int a;
155     a = MSG_task_receive(&(task),MSG_host_get_name(MSG_host_self()));
156     if (a == MSG_OK) {
157       XBT_INFO("Received \"%s\"", MSG_task_get_name(task));
158       if (MSG_task_get_data(task) == FINALIZE) {
159         XBT_INFO
160             ("All tasks have been dispatched. Let's tell everybody the computation is over.");
161         for (i = 0; i < slaves_count; i++)
162           MSG_task_send(MSG_task_create("finalize", 0, 0, FINALIZE),
163               MSG_host_get_name(slaves[i]));
164         MSG_task_destroy(task);
165         break;
166       }
167       XBT_INFO("Sending \"%s\" to \"%s\"",
168             MSG_task_get_name(task), MSG_host_get_name(slaves[i % slaves_count]));
169       MSG_task_send(task, MSG_host_get_name(slaves[i % slaves_count]));
170       i++;
171     } else {
172       XBT_INFO("Hey ?! What's up ? ");
173       xbt_die("Unexpected behavior");
174     }
175   }
176   xbt_free(slaves);
177
178   XBT_INFO("I'm done. See you!");
179   return 0;
180 }                               /* end_of_forwarder */
181
182 /** Test function */
183 MSG_error_t test_all(const char *platform_file,
184                      const char *application_file)
185 {
186   MSG_error_t res = MSG_OK;
187
188   /* MSG_config("workstation/model","KCCFLN05"); */
189   {                             /*  Simulation setting */
190     MSG_create_environment(platform_file);
191   }
192   {                             /*   Application deployment */
193     MSG_function_register("master", master);
194     MSG_function_register("slave", slave);
195     MSG_function_register("forwarder", forwarder);
196     MSG_launch_application(application_file);
197   }
198   res = MSG_main();
199
200   XBT_INFO("Simulation time %g", MSG_get_clock());
201   return res;
202 }                               /* end_of_test_all */
203
204
205 /** Main function */
206 int main(int argc, char *argv[])
207 {
208   MSG_error_t res = MSG_OK;
209
210   MSG_global_init(&argc, argv);
211   if (argc < 3) {
212     printf("Usage: %s platform_file deployment_file\n", argv[0]);
213     printf("example: %s msg_platform.xml msg_deployment.xml\n", argv[0]);
214     exit(1);
215   }
216   res = test_all(argv[1], argv[2]);
217   MSG_clean();
218
219   if (res == MSG_OK)
220     return 0;
221   else
222     return 1;
223 }                               /* end_of_main */