Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Modified platform and deployment files so it's easier to add nodes Fixed chain genera...
[simgrid.git] / examples / msg / kadeploy / kadeploy.c
1 /* Copyright (c) 2007, 2008, 2009, 2010. The SimGrid Team.
2  * Copyright (c) 2012. Maximiliano Geier.
3  * 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 <stdlib.h>
10
11 #include "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
12 #include "xbt/sysdep.h"         /* calloc */
13
14 /* Create a log channel to have nice outputs. */
15 #include "xbt/log.h"
16 #include "xbt/asserts.h"
17
18 /** @addtogroup MSG_examples
19  * 
20  *  - <b>kadeploy/kadeploy.c: Kadeploy implementation</b>.
21  */
22
23
24 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_kadeploy,
25                              "Messages specific for kadeploy");
26
27 #define MESSAGE_SIZE 1
28 #define HOSTNAME_LENGTH 20
29
30 /*
31  Data structures
32  */
33
34 /* Random iterator for xbt_dynar */
35 typedef struct xbt_dynar_iterator_struct {
36   xbt_dynar_t list;
37   xbt_dynar_t indices_list;
38   int current;
39   unsigned long length;
40   int (*criteria_fn)(void* it);
41 } *xbt_dynar_iterator_t;
42 typedef struct xbt_dynar_iterator_struct xbt_dynar_iterator_s;
43
44 /* Messages enum */
45 typedef enum {
46   MESSAGE_BUILD_CHAIN = 0,
47   MESSAGE_SEND_DATA,
48   MESSAGE_END_DATA
49 } e_message_type;
50
51 /* Message struct */
52 typedef struct s_message {
53   e_message_type type;
54   const char *issuer_hostname;
55   const char *mailbox;
56   const char *prev_hostname;
57   const char *next_hostname;
58   const char *data_block;
59   unsigned int data_length;
60 } s_message_t, *message_t;
61
62 /* Iterator methods */
63 xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, int (*criteria_fn)(void*));
64 void *xbt_dynar_iterator_next(xbt_dynar_iterator_t it);
65 void xbt_dynar_iterator_delete(xbt_dynar_iterator_t it);
66 int xbt_dynar_iterator_forward_criteria(void *p);
67
68 /* Message methods */
69 msg_task_t task_message_new(e_message_type type, const char *issuer_hostname, const char *mailbox);
70 msg_task_t task_message_chain_new(const char *issuer_hostname, const char *mailbox, const char* prev, const char *next);
71 msg_task_t task_message_data_new(const char *issuer_hostname, const char *mailbox, const char *block, unsigned int len);
72 msg_task_t task_message_end_data_new(const char *issuer_hostname, const char *mailbox);
73 void task_message_delete(void *);
74
75 /* Tasks */
76 int broadcaster(int argc, char *argv[]);
77 int peer(int argc, char *argv[]);
78
79 xbt_dynar_t build_hostlist_from_hostcount(int hostcount); 
80 /*xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[]);*/
81
82 /* Broadcaster: helper functions */
83 int broadcaster_build_chain(xbt_dynar_t host_list);
84 int broadcaster_send_file(xbt_dynar_t host_list);
85 int broadcaster_finish(xbt_dynar_t host_list);
86
87 /* Peer: helper functions */
88 int peer_wait_for_init();
89
90 /* Initialization stuff */
91 msg_error_t test_all(const char *platform_file,
92                      const char *application_file);
93
94 /* Allocates and initializes a new xbt_dynar iterator for list, using criteria_fn as iteration criteria
95    criteria_fn: given an iterator, it must update the iterator and give the next element's index, 
96    less than 0 otherwise*/
97 xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, int (*criteria_fn)(void*))
98 {
99   xbt_dynar_iterator_t it = xbt_new(xbt_dynar_iterator_s, 1);
100   
101   it->list = list;
102   it->length = xbt_dynar_length(list);
103   it->indices_list = xbt_dynar_new(sizeof(int), NULL);
104   it->criteria_fn = criteria_fn;
105   it->current = -1;
106 }
107
108 /* Returns the next element iterated by iterator it, NULL if there are no more elements */
109 void *xbt_dynar_iterator_next(xbt_dynar_iterator_t it)
110 {
111   int next = it->criteria_fn((xbt_dynar_iterator_t)it);
112   XBT_INFO("%d current\n", next);
113   if (next < 0) {
114     XBT_INFO("Nothing to return!\n");
115     return NULL;
116   } else {
117     xbt_dynar_push(it->indices_list, &next);
118     return xbt_dynar_get_ptr(it->list, next);
119   }
120 }
121
122 void xbt_dynar_iterator_delete(xbt_dynar_iterator_t it)
123 {
124   xbt_dynar_free_container(&(it->indices_list));
125   xbt_free_ref(&it);
126 }
127
128 int xbt_dynar_iterator_forward_criteria(void *p)
129 {
130   xbt_dynar_iterator_t it = (xbt_dynar_iterator_t)p;
131   int r = -1;
132   if (it->current == -1) {
133     /* iterator initialization */
134     it->current = 0;
135   }
136   if (it->current < it->length) {
137     r = it->current;
138     it->current++;
139   }
140
141   return r;
142 }
143
144 msg_task_t task_message_new(e_message_type type, const char *issuer_hostname, const char *mailbox)
145 {
146   message_t msg = xbt_new(s_message_t, 1);
147   msg->type = type;
148   msg->issuer_hostname = issuer_hostname;
149   msg->mailbox = mailbox;
150   msg_task_t task = MSG_task_create(NULL, 0, MESSAGE_SIZE, msg); 
151
152   return task;
153 }
154
155 msg_task_t task_message_chain_new(const char *issuer_hostname, const char *mailbox, const char* prev, const char *next)
156 {
157   msg_task_t task = task_message_new(MESSAGE_BUILD_CHAIN, issuer_hostname, mailbox);
158   message_t msg = MSG_task_get_data(task);
159   msg->prev_hostname = prev;
160   msg->next_hostname = next;
161
162   return task;
163 }
164
165 msg_task_t task_message_data_new(const char *issuer_hostname, const char *mailbox, const char *block, unsigned int len)
166 {
167   msg_task_t task = task_message_new(MESSAGE_SEND_DATA, issuer_hostname, mailbox);
168   message_t msg = MSG_task_get_data(task);
169   msg->data_block = block;
170   msg->data_length = len;
171
172   return task;
173 }
174
175 msg_task_t task_message_end_data_new(const char *issuer_hostname, const char *mailbox)
176 {
177   return task_message_new(MESSAGE_END_DATA, issuer_hostname, mailbox);
178 }
179
180
181 void task_message_delete(void *task)
182 {
183   message_t msg = MSG_task_get_data(task);
184   xbt_free(msg);
185   MSG_task_destroy(task);
186 }
187
188 xbt_dynar_t build_hostlist_from_hostcount(int hostcount)
189 {
190   xbt_dynar_t host_list = xbt_dynar_new(sizeof(char*), NULL);
191   char *hostname = NULL;
192   msg_host_t h = NULL;
193   int i = 1;
194   
195   for (; i < hostcount+1; i++) {
196     hostname = xbt_new(char, HOSTNAME_LENGTH);
197     snprintf(hostname, HOSTNAME_LENGTH, "host%d", i);
198     XBT_INFO("%s", hostname);
199     h = MSG_get_host_by_name(hostname);
200     if (h == NULL) {
201       XBT_INFO("Unknown host %s. Stopping Now! ", hostname);
202       abort();
203     } else {
204       xbt_dynar_push(host_list, &hostname);
205     }
206   }
207   return host_list;
208 }
209
210 /*xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[])
211 {
212   xbt_dynar_t host_list = xbt_dynar_new(sizeof(char*), NULL);
213   msg_host_t h = NULL;
214   int i = 1;
215   
216   for (; i < argc; i++) {
217     XBT_INFO("host%d = %s", i, argv[i]);
218     h = MSG_get_host_by_name(argv[i]);
219     if (h == NULL) {
220       XBT_INFO("Unknown host %s. Stopping Now! ", argv[i]);
221       abort();
222     } else {
223       xbt_dynar_push(host_list, &(argv[i]));
224     }
225   }
226   return host_list;
227 }*/
228
229 void delete_hostlist(xbt_dynar_t h)
230 {
231   xbt_dynar_free(&h);
232 }
233
234 int broadcaster_build_chain(xbt_dynar_t host_list)
235 {
236   xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, xbt_dynar_iterator_forward_criteria);
237   msg_task_t task = NULL;
238   char **cur = (char**)xbt_dynar_iterator_next(it);
239   const char *current_host = NULL;
240   const char *prev = NULL;
241   const char *next = NULL;
242   const char *me = MSG_host_get_name(MSG_host_self());
243   const char *last = NULL;
244
245   /* Build the chain if there's at least one peer */
246   if (cur != NULL) {
247     /* init: prev=NULL, host=current cur, next=next cur */
248     next = *cur;
249
250     /* This iterator iterates one step ahead: cur is current iterated element, 
251        but it's actually the next one in the chain */
252     do {
253       /* following steps: prev=last, host=next, next=cur */
254       cur = (char**)xbt_dynar_iterator_next(it);
255       prev = last;
256       current_host = next;
257       if (cur != NULL)
258         next = *cur;
259       else
260         next = NULL;
261       XBT_INFO("Building chain -- broadcaster:\"%s\" dest:\"%s\" prev:\"%s\" next:\"%s\"", me, current_host, prev, next);
262     
263       /* Send message to current peer */
264       task = task_message_chain_new(me, current_host, prev, next);
265       MSG_task_send(task, current_host);
266
267       last = current_host;
268     } while (cur != NULL);
269   }
270   xbt_dynar_iterator_delete(it);
271
272   return MSG_OK;
273 }
274
275 int broadcaster_send_file(xbt_dynar_t host_list)
276 {
277   /* ... */
278
279   return MSG_OK;
280 }
281
282 int broadcaster_finish(xbt_dynar_t host_list)
283 {
284   xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, xbt_dynar_iterator_forward_criteria);
285   msg_task_t task = NULL;
286   const char *me = MSG_host_get_name(MSG_host_self());
287   const char *current_host = NULL;
288   char **cur = NULL;
289
290   /* Send goodbye message to every peer */
291   for (cur = (char**)xbt_dynar_iterator_next(it); cur != NULL; cur = (char**)xbt_dynar_iterator_next(it)) {
292       /* Send message to current peer */
293       current_host = *cur;
294       task = task_message_end_data_new(me, current_host);
295       MSG_task_send(task, current_host);
296   }
297
298   return MSG_OK;
299 }
300
301
302 /** Emitter function  */
303 int broadcaster(int argc, char *argv[])
304 {
305   xbt_dynar_t host_list = NULL;
306   const char *first = NULL;
307   int status = !MSG_OK;
308
309   XBT_INFO("broadcaster");
310
311   /* Check that every host given by the hostcount in argv[1] exists and add it
312      to a dynamic array */
313   host_list = build_hostlist_from_hostcount(atoi(argv[1]));
314   /*host_list = build_hostlist_from_argv(argc, argv);*/
315   
316   /* TODO: Error checking */
317   status = broadcaster_build_chain(host_list);
318   status = broadcaster_send_file(host_list);
319   status = broadcaster_finish(host_list);
320
321   delete_hostlist(host_list);
322
323   /* Latency */
324   /*time = MSG_get_clock();
325   sprintf(sprintf_buffer_la, "latency task");
326   task_la =
327       MSG_task_create(sprintf_buffer_la, 0.0, task_comm_size_lat, NULL);
328   task_la->data = xbt_new(double, 1);
329   *(double *) task_la->data = time;
330   XBT_INFO("task_la->data = %le", *((double *) task_la->data));
331   MSG_task_send(task_la, argv[1]);*/
332
333   /* Bandwidth */
334   /*time = MSG_get_clock();
335   sprintf(sprintf_buffer_bw, "bandwidth task");
336   task_bw =
337       MSG_task_create(sprintf_buffer_bw, 0.0, task_comm_size_bw, NULL);
338   task_bw->data = xbt_new(double, 1);
339   *(double *) task_bw->data = time;
340   XBT_INFO("task_bw->data = %le", *((double *) task_bw->data));
341   MSG_task_send(task_bw, argv[1]);
342   */
343   return status;
344 }
345
346 int peer_wait_for_init()
347 {
348   msg_task_t task = NULL;
349   const char *me = MSG_host_get_name(MSG_host_self());
350
351   int a = MSG_task_receive(&task, me);
352
353   if (a == MSG_OK) {
354     XBT_INFO("Peer %s got message\n", me);
355   }
356
357   task_message_delete(task);
358
359   return MSG_OK;
360 }
361
362 /** Peer function  */
363 int peer(int argc, char *argv[])
364 {
365   double time, time1, sender_time;
366   msg_task_t task_la = NULL;
367   msg_task_t task_bw = NULL;
368   int a;
369   double communication_time = 0;
370
371   XBT_INFO("peer");
372
373   time = MSG_get_clock();
374
375   a = peer_wait_for_init();
376   /* Get Latency */
377   /*a = MSG_task_receive(&task_la,MSG_host_get_name(MSG_host_self()));
378   if (a == MSG_OK) {
379     time1 = MSG_get_clock();
380     sender_time = *((double *) (task_la->data));
381     time = sender_time;
382     communication_time = time1 - time;
383     XBT_INFO("Task received : %s", task_la->name);
384     xbt_free(task_la->data);
385     MSG_task_destroy(task_la);
386     XBT_INFO("Communic. time %le", communication_time);
387     XBT_INFO("--- la %f ----", communication_time);
388   } else {
389     xbt_die("Unexpected behavior");
390   }*/
391
392
393   /* Get Bandwidth */
394   /*a = MSG_task_receive(&task_bw,MSG_host_get_name(MSG_host_self()));
395   if (a == MSG_OK) {
396     time1 = MSG_get_clock();
397     sender_time = *((double *) (task_bw->data));
398     time = sender_time;
399     communication_time = time1 - time;
400     XBT_INFO("Task received : %s", task_bw->name);
401     xbt_free(task_bw->data);
402     MSG_task_destroy(task_bw);
403     XBT_INFO("Communic. time %le", communication_time);
404     XBT_INFO("--- bw %f ----", task_comm_size_bw / communication_time);
405   } else {
406     xbt_die("Unexpected behavior");
407   }*/
408
409
410   return 0;
411 }                               /* end_of_receiver */
412
413
414 /** Test function */
415 msg_error_t test_all(const char *platform_file,
416                      const char *application_file)
417 {
418
419   msg_error_t res = MSG_OK;
420
421
422
423   XBT_INFO("test_all");
424
425   /*  Simulation setting */
426   MSG_create_environment(platform_file);
427
428   /*   Application deployment */
429   MSG_function_register("broadcaster", broadcaster);
430   MSG_function_register("peer", peer);
431
432   MSG_launch_application(application_file);
433
434   res = MSG_main();
435
436   return res;
437 }                               /* end_of_test_all */
438
439
440 /** Main function */
441 int main(int argc, char *argv[])
442 {
443   msg_error_t res = MSG_OK;
444
445 #ifdef _MSC_VER
446   unsigned int prev_exponent_format =
447       _set_output_format(_TWO_DIGIT_EXPONENT);
448 #endif
449
450   MSG_init(&argc, argv);
451
452
453   /*if (argc <= 3) {
454     XBT_CRITICAL("Usage: %s platform_file deployment_file <model>\n",
455               argv[0]);
456     XBT_CRITICAL
457         ("example: %s msg_platform.xml msg_deployment.xml KCCFLN05_Vegas\n",
458          argv[0]);
459     exit(1);
460   }*/
461
462   /* Options for the workstation/model:
463
464      KCCFLN05              => for maxmin
465      KCCFLN05_proportional => for proportional (Vegas)
466      KCCFLN05_Vegas        => for TCP Vegas
467      KCCFLN05_Reno         => for TCP Reno
468    */
469   //MSG_config("workstation/model", argv[3]);
470
471   res = test_all(argv[1], argv[2]);
472
473   XBT_INFO("Total simulation time: %le", MSG_get_clock());
474
475   MSG_clean();
476
477 #ifdef _MSC_VER
478   _set_output_format(prev_exponent_format);
479 #endif
480
481   if (res == MSG_OK)
482     return 0;
483   else
484     return 1;
485 }                               /* end_of_main */