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