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)
  53 {
  54   xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, forward_indices_list);
  55   msg_task_t task = NULL;
  56   char **cur = (char**)xbt_dynar_iterator_next(it);
  57   const char *me = MSG_host_get_name(MSG_host_self());
  58   const char *current_host = NULL;
  59   const char *prev = NULL;
  60   const char *next = NULL;
  61   const char *last = NULL;
  62
  63   /* Build the chain if there's at least one peer */
  64   if (cur != NULL) {
  65     /* init: prev=NULL, host=current cur, next=next cur */
  66     next = *cur;
  67     *first = next;
  68
  69     /* This iterator iterates one step ahead: cur is current iterated element,
  70        but it's actually the next one in the chain */
  71     do {
  72       /* following steps: prev=last, host=next, next=cur */
  73       cur = (char**)xbt_dynar_iterator_next(it);
  74       prev = last;
  75       current_host = next;
  76       if (cur != NULL)
  77         next = *cur;
  78       else
  79         next = NULL;
  80       //XBT_INFO("Building chain -- broadcaster:\"%s\" dest:\"%s\" prev:\"%s\" next:\"%s\"", me, current_host, prev, next);
  81
  82       /* Send message to current peer */
  83       task = task_message_chain_new(me, current_host, prev, next);
  84       //MSG_task_set_category(task, current_host);
  85       MSG_task_send(task, current_host);
  86
  87       last = current_host;
  88     } while (cur != NULL);
  89   }
  90   xbt_dynar_iterator_delete(it);
  91
  92   return MSG_OK;
  93 }
  94
  95 int broadcaster_send_file(const char *first)
  96 {
  97   const char *me = MSG_host_get_name(MSG_host_self());
  98   msg_task_t task = NULL;
  99   msg_comm_t comm = NULL;
 100   int status;
 101
 102   int piece_count = PIECE_COUNT;
 103   int cur = 0;
 104
 105   for (; cur < piece_count; cur++) {
 106     task = task_message_data_new(me, first, NULL, 0);
 107     XBT_INFO("Sending (send) from %s into mailbox %s", me, first);
 108     status = MSG_task_send(task, first);
 109
 110     xbt_assert(status == MSG_OK, __FILE__ ": broadcaster_send_file() failed");
 111   }
 112
 113   return MSG_OK;
 114 }
 115
 116 /* FIXME: I should iterate nodes in the same order as the one used to build the chain */
 117 int broadcaster_finish(xbt_dynar_t host_list)
 118 {
 119   xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, forward_indices_list);
 120   msg_task_t task = NULL;
 121   const char *me = MSG_host_get_name(MSG_host_self());
 122   const char *current_host = NULL;
 123   char **cur = NULL;
 124
 125   /* Send goodbye message to every peer */
 126   for (cur = (char**)xbt_dynar_iterator_next(it); cur != NULL; cur = (char**)xbt_dynar_iterator_next(it)) {
 127     /* Send message to current peer */
 128     current_host = *cur;
 129     task = task_message_end_data_new(me, current_host);
 130     //MSG_task_set_category(task, current_host);
 131     MSG_task_send(task, current_host);
 132   }
 133
 134   return MSG_OK;
 135 }
 136
 137
 138 /** Emitter function  */
 139 int broadcaster(int argc, char *argv[])
 140 {
 141   xbt_dynar_t host_list = NULL;
 142   const char *first = NULL;
 143   int status = !MSG_OK;
 144
 145   XBT_INFO("broadcaster");
 146
 147   /* Check that every host given by the hostcount in argv[1] exists and add it
 148      to a dynamic array */
 149   host_list = build_hostlist_from_hostcount(atoi(argv[1]));
 150   /*host_list = build_hostlist_from_argv(argc, argv);*/
 151
 152   /* TODO: Error checking */
 153   status = broadcaster_build_chain(&first, host_list);
 154   status = broadcaster_send_file(first);
 155   status = broadcaster_finish(host_list);
 156
 157   delete_hostlist(host_list);
 158
 159   return status;
 160 }