examples/msg/kadeploy/broadcaster.c

   1 #include "broadcaster.h"
   2
   3 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_broadcaster,
   4                              "Messages specific for kadeploy");
   5
   6 xbt_dynar_t build_hostlist_from_hostcount(int hostcount)
   7 {
   8   xbt_dynar_t host_list = xbt_dynar_new(sizeof(char*), NULL);
   9   char *hostname = NULL;
  10   msg_host_t h = NULL;
  11   int i = 1;
  12
  13   for (; i < hostcount+1; i++) {
  14     hostname = xbt_new(char, HOSTNAME_LENGTH);
  15     snprintf(hostname, HOSTNAME_LENGTH, "host%d", i);
  16     //XBT_INFO("%s", hostname);
  17     /*h = MSG_get_host_by_name(hostname);
  18     if (h == NULL) {
  19       XBT_INFO("Unknown host %s. Stopping Now! ", hostname);
  20       abort();
  21     } else {*/
  22       xbt_dynar_push(host_list, &hostname);
  23     /*}*/
  24   }
  25   return host_list;
  26 }
  27
  28 /*xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[])
  29 {
  30   xbt_dynar_t host_list = xbt_dynar_new(sizeof(char*), NULL);
  31   msg_host_t h = NULL;
  32   int i = 1;
  33
  34   for (; i < argc; i++) {
  35     XBT_INFO("host%d = %s", i, argv[i]);
  36     h = MSG_get_host_by_name(argv[i]);
  37     if (h == NULL) {
  38       XBT_INFO("Unknown host %s. Stopping Now! ", argv[i]);
  39       abort();
  40     } else {
  41       xbt_dynar_push(host_list, &(argv[i]));
  42     }
  43   }
  44   return host_list;
  45 }*/
  46
  47 void delete_hostlist(xbt_dynar_t h)
  48 {
  49   xbt_dynar_free(&h);
  50 }
  51
  52 int broadcaster_build_chain(const char **first, xbt_dynar_t host_list, xbt_dynar_iterator_t it)
  53 {
  54   msg_task_t task = NULL;
  55   char **cur = (char**)xbt_dynar_iterator_next(it);
  56   const char *me = "host0"; /* FIXME: hardcoded*/ /*MSG_host_get_name(MSG_host_self());*/
  57   const char *current_host = NULL;
  58   const char *prev = NULL;
  59   const char *next = NULL;
  60   const char *last = NULL;
  61
  62   /* Build the chain if there's at least one peer */
  63   if (cur != NULL) {
  64     /* init: prev=NULL, host=current cur, next=next cur */
  65     next = *cur;
  66     *first = next;
  67
  68     /* This iterator iterates one step ahead: cur is current iterated element,
  69        but it's actually the next one in the chain */
  70     do {
  71       /* following steps: prev=last, host=next, next=cur */
  72       cur = (char**)xbt_dynar_iterator_next(it);
  73       prev = last;
  74       current_host = next;
  75       if (cur != NULL)
  76         next = *cur;
  77       else
  78         next = NULL;
  79       //XBT_INFO("Building chain -- broadcaster:\"%s\" dest:\"%s\" prev:\"%s\" next:\"%s\"", me, current_host, prev, next);
  80
  81       /* Send message to current peer */
  82       task = task_message_chain_new(me, current_host, prev, next);
  83       //MSG_task_set_category(task, current_host);
  84       MSG_task_send(task, current_host);
  85
  86       last = current_host;
  87     } while (cur != NULL);
  88   }
  89
  90   return MSG_OK;
  91 }
  92
  93 int broadcaster_send_file(const char *first)
  94 {
  95   const char *me = MSG_host_get_name(MSG_host_self());
  96   msg_task_t task = NULL;
  97   msg_comm_t comm = NULL;
  98   int status;
  99
 100   int piece_count = PIECE_COUNT;
 101   int cur = 0;
 102
 103   for (; cur < piece_count; cur++) {
 104     task = task_message_data_new(me, first, NULL, 0);
 105     XBT_INFO("Sending (send) from %s into mailbox %s", me, first);
 106     status = MSG_task_send(task, first);
 107
 108     xbt_assert(status == MSG_OK, "broadcaster_send_file() failed");
 109   }
 110
 111   return MSG_OK;
 112 }
 113
 114 int broadcaster_finish(xbt_dynar_iterator_t it)
 115 {
 116   msg_task_t task = NULL;
 117   const char *me = "host0"; /* FIXME: hardcoded*/ /*MSG_host_get_name(MSG_host_self());*/
 118   const char *current_host = NULL;
 119   char **cur = NULL;
 120
 121   xbt_dynar_iterator_seek(it, 0);
 122
 123   /* Send goodbye message to every peer in the order generated by iterator it */
 124   for (cur = (char**)xbt_dynar_iterator_next(it); cur != NULL; cur = (char**)xbt_dynar_iterator_next(it)) {
 125     /* Send message to current peer */
 126     current_host = *cur;
 127     task = task_message_end_data_new(me, current_host);
 128     //MSG_task_set_category(task, current_host);
 129     MSG_task_send(task, current_host);
 130   }
 131
 132   return MSG_OK;
 133 }
 134
 135
 136 /** Emitter function  */
 137 int broadcaster(int argc, char *argv[])
 138 {
 139   xbt_dynar_t host_list = NULL;
 140   const char *first = NULL;
 141   int status = !MSG_OK;
 142
 143   XBT_INFO("broadcaster");
 144
 145   /* Add every mailbox given by the hostcount in argv[1] to a dynamic array */
 146   host_list = build_hostlist_from_hostcount(atoi(argv[1]));
 147   /*host_list = build_hostlist_from_argv(argc, argv);*/
 148
 149   /* Initialize iterator */
 150   xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, forward_indices_list);
 151
 152   /* TODO: Error checking */
 153   status = broadcaster_build_chain(&first, host_list, it);
 154   status = broadcaster_send_file(first);
 155   status = broadcaster_finish(it);
 156
 157   /* Destroy iterator and hostlist */
 158   xbt_dynar_iterator_delete(it);
 159   delete_hostlist(host_list);
 160
 161   return status;
 162 }