examples/msg/chord/chord.c

   1 /* Copyright (c) 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 <math.h>
   9 #include "msg/msg.h"
  10 #include "xbt/sysdep.h"         /* calloc, printf */
  11
  12 /* Create a log channel to have nice outputs. */
  13 #include "xbt/log.h"
  14 #include "xbt/asserts.h"
  15 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  16                              "Messages specific for this msg example");
  17 #define KEY_BITS  6
  18 #define CHORD_NB_KEYS 64
  19
  20 /*
  21 * Finger Element
  22 */
  23 typedef struct{
  24         int id;
  25         const char *host_name;
  26         const char *mailbox;
  27 } finger_elem;
  28
  29 /*
  30  * Node Data
  31  */
  32 typedef struct{
  33         int id; // my id
  34         const char *host_name; //my host name
  35         const char* mailbox; //my mailbox
  36         int fingers_nb; // size of finger list
  37         finger_elem* fingers; // finger list  [ fingers[0] >> Successor
  38         int next; //next finger to fix
  39         int pred_id; // predecessor id
  40         const char* pred_host_name; // predecessor host name
  41         const char* pred_mailbox; // predecessor mailbox
  42 }data_node;
  43
  44 //global;
  45
  46 static int node(int argc, char *argv[]);
  47 static int sender(int argc,char *argv[]);
  48 static void find_successor_node(data_node *my_data, m_task_t join_task);
  49 static int find_successor(data_node* my_data, int id);
  50 static const char* find_closest_preceding(data_node* n_node, int id); //return a mailbox
  51 static int get_successor_id(m_host_t);
  52
  53 static MSG_error_t test_all(const char *platform_file,
  54                             const char *application_file);
  55 static void init_finger_table(data_node *data, int known_id);
  56
  57 static int is_in_interval(unsigned int id, unsigned int start, unsigned int end) {
  58
  59   id = id % CHORD_NB_KEYS;
  60   start = start % CHORD_NB_KEYS;
  61   end = end % CHORD_NB_KEYS;
  62
  63   /* make sure end >= start and id >= start */
  64   if (end < start) {
  65     end += CHORD_NB_KEYS;
  66   }
  67
  68   if (id < start) {
  69     id += CHORD_NB_KEYS;
  70   }
  71
  72   return id < end;
  73 }
  74
  75 /*
  76  * Node Function
  77  */
  78
  79 /*<process host="host_name" function="node">
  80    <argument value="id"/> <!-- my id -->
  81    <argument value="mailbox"/> <!-- mailbox -->
  82    <!-- optional -->
  83    <argument value="known_host_name" />
  84    <argument value="knwon_host_mailbox" />
  85    >argument value="time_to_sleep"/>
  86   </process>
  87 */
  88 static int cpt = 0;
  89 int node(int argc, char *argv[])
  90 {
  91         m_task_t recv_request = NULL;
  92         int first = 1;
  93         int joined = 0;
  94         int res,create = 0;
  95         if ( argc == 3)   // if no known host are declared >>> first node >>> create chord ring
  96         {
  97                 create = 1;
  98         }
  99
 100         //int id = atoi(argv[1]);
 101         //int mailbox = atoi(argv[2]);
 102         // init data node
 103         data_node *data = xbt_new(data_node,1);
 104         data->host_name = MSG_host_get_name(MSG_host_self());
 105         data->id =  atoi(argv[1]);
 106         data->mailbox = argv[2];
 107         data->fingers_nb = 1;
 108         data->fingers = xbt_new(finger_elem,KEY_BITS);
 109         data->fingers[0].host_name = data->host_name;
 110         data->fingers[0].id = data->id;
 111         data->fingers[0].mailbox = data->mailbox;
 112         data->next = 0;
 113         data->pred_host_name = NULL;
 114         data->pred_id = -1;
 115         data->pred_mailbox = NULL;
 116
 117 /*
 118  * Ring Point Entry Node
 119  */
 120         if (create) // first ring
 121         {
 122                 INFO0("Create new Chord Ring...");
 123                 joined = 1;
 124                 cpt++;
 125                 //sucessor = n
 126                 data->fingers[0].host_name = data->host_name;
 127                 data->fingers[0].id = data->id;
 128                 data->fingers[0].mailbox = data->mailbox;
 129                 while(cpt < MSG_get_host_number()-1)  //make condition!!!
 130                 {
 131                         recv_request = NULL;
 132                         res = MSG_task_receive(&(recv_request),data->mailbox);
 133                         xbt_assert0(res == MSG_OK, "MSG_receiev failed");
 134                         if (!strcmp(MSG_task_get_name(recv_request), "Join Call"))
 135                         {
 136                                 if(MSG_task_get_data(recv_request)==NULL)
 137                                 {
 138                                         WARN0("Receiving an Empty Data");
 139                                 }
 140                                 data_node *recv_data = (data_node*)MSG_task_get_data(recv_request);
 141                                 INFO1("Receiving a Join Call from %s",recv_data->host_name);
 142                                 if (first)
 143                                 {
 144                                         // predecessor(recv_data) >>>> data
 145                                         recv_data->pred_host_name = data->host_name;
 146                                         recv_data->pred_id = data->id;
 147                                         recv_data->pred_mailbox = data->mailbox;
 148                                         data->fingers_nb = 1;
 149                                         // successor(recv_data) >>> data
 150                                         recv_data->fingers[0].id = data->id;
 151                                         recv_data->fingers[0].host_name = data->host_name;
 152                                         recv_data->fingers[0].mailbox = data->mailbox;
 153                                         //successor(data) >>>> recv_data
 154                                         data->fingers[data->fingers_nb - 1].host_name = recv_data->host_name;
 155                                         data->fingers[data->fingers_nb - 1].id = recv_data->id;
 156                                         data->fingers[data->fingers_nb - 1].mailbox = recv_data->mailbox;
 157                                         INFO1("Sending back a Join Request to %s",recv_data->host_name);
 158                                         MSG_task_set_name(recv_request,"Join Response");
 159                                         MSG_task_send(recv_request,recv_data->mailbox);
 160                                         first = 0;
 161                                 }
 162                                 else{
 163                                         find_successor_node(data,recv_request);
 164                                 }
 165
 166                         }
 167                 }
 168         }
 169 /*
 170  * Joining Node
 171  */
 172         else if(!create)
 173         {
 174                 //Sleep Before Starting
 175                 INFO1("Let's Sleep >>%i",atoi(argv[6]));
 176                 MSG_process_sleep(atoi(argv[5]));
 177                 INFO0("Hey! Let's Send a Join Request");
 178                 //send a join task to the known host via its(known host) mailbox
 179                 const char* known_host_name = argv[3];
 180                 const char* known_mailbox = argv[4];
 181                 int known_id = atoi(argv[5]);
 182                 m_task_t join_request = MSG_task_create("Join Call",10000,2000,data);   // define comp size and comm size (#define ...)
 183                 INFO2("Sending a join request to %s via mailbox %s",known_host_name,known_mailbox);
 184                 MSG_task_send(join_request,known_mailbox);
 185                 //wait for answer on my mailbox
 186                 while(cpt < MSG_get_host_number()-1)
 187                 {
 188                         recv_request = NULL;
 189                         int res = MSG_task_receive(&(recv_request),data->mailbox);
 190                         //check if it's the response for my request
 191                         xbt_assert0(res == MSG_OK, "MSG_receiev failed");
 192                         // get data
 193                         data_node *recv_data = (data_node*)MSG_task_get_data(recv_request);
 194                         // Join Call Message
 195                         if(!strcmp(MSG_task_get_name(recv_request), "Join Call"))
 196                         {
 197
 198                                 INFO1("Receiving Join Call From %s",recv_data->host_name);
 199                                 if(!joined)
 200                                 {
 201                                         INFO1("Sorry %s... I'm not yet joined",recv_data->host_name);
 202                                         //No Treatment
 203                                         MSG_task_set_name(recv_request,"Join Failed");
 204                                         MSG_task_send(recv_request,recv_data->mailbox);
 205                                                 }
 206                                         else
 207                                         {
 208                                                 find_successor_node(data,recv_request);
 209                                         }
 210
 211                         }
 212                         // Join Response
 213                         else if(!strcmp(MSG_task_get_name(recv_request), "Join Response"))
 214                         {
 215                                 INFO0("Receiving Join Response!!!");
 216                                 INFO1("My successor is : %s",data->fingers[0].host_name);
 217                                 INFO1("My Predecessor is : %s",data->pred_host_name);
 218                                 cpt++;
 219                                 joined = 1;
 220                                 INFO1("My finger table size : %i",data->fingers_nb);
 221                                 INFO0("***********************************************************************");
 222
 223                                 /*
 224                                 MSG_task_set_name(recv_request,"Fix Fingers");
 225
 226                                 MSG_task_send(recv_request,data->pred_mailbox);
 227                                 MSG_task_send(recv_request,data->fingers[0].mailbox);
 228                                 */
 229                                 init_finger_table(data, known_id);
 230
 231                                 //treatment
 232                         }
 233                         // Join Failure Message
 234                         else if(!strcmp(MSG_task_get_name(recv_request), "Join Failed"))
 235                         {
 236                                 INFO0("My Join call has failed... let's Try Again");
 237                                 // send back
 238                                 //MSG_task_send(join_request,known_mailbox);
 239                                 // !!!!!!!!! YVes Jaques Always...???§§§§**************************
 240
 241                         }
 242                         else if(!strcmp(MSG_task_get_name(recv_request), "Fix Fingers"))
 243                         {
 244                                 int i;
 245                                 for(i = KEY_BITS -1 ; i>= 0;i--)
 246                                 {
 247                                         //data->fingers[i] = find_finger_elem(data,(data->id)+pow(2,i-1));
 248                                 }
 249                         }
 250                 }
 251         }
 252         return 0;
 253 }
 254
 255 /*
 256  * Initializes
 257  */
 258 void init_finger_table(data_node *node, int known_id) {
 259
 260   // ask known_id who is my immediate successor
 261 //  data->fingers[0].id = remote_find_successor(known_id, data->id + 1);
 262 }
 263
 264 /*
 265  *
 266  */
 267 void find_successor_node(data_node* n_data,m_task_t join_task)  //use all data
 268 {
 269         //get recv data
 270         data_node *recv_data = (data_node*)MSG_task_get_data(join_task);
 271         INFO3("recv_data->id : %i , n_data->id :%i , successor->id :%i",recv_data->id,n_data->id,n_data->fingers[0].id);
 272         //if ((recv_data->id >= n_data->id) && (recv_data->id <= n_data->fingers[0].id))
 273         if (is_in_interval(recv_data->id,n_data->id,n_data->fingers[0].id))
 274         {
 275                 INFO1("Successor Is %s",n_data->fingers[0].host_name);
 276                 //predecessor(recv_data) >>>> n_data
 277                 recv_data->pred_host_name = n_data->host_name;
 278                 recv_data->pred_id = n_data->id;
 279                 recv_data->pred_mailbox = n_data->pred_mailbox;
 280                 // successor(recv_data) >>>> n_data.finger[0]
 281                 recv_data->fingers_nb = 1;
 282                 recv_data->fingers[0].host_name = n_data->fingers[0].host_name;
 283                 recv_data->fingers[0].id = n_data->fingers[0].id;
 284                 recv_data->fingers[0].mailbox = n_data->fingers[0].mailbox;
 285                 // successor(n_data) >>>> recv_sucessor
 286                 n_data->fingers[0].id = recv_data->id;
 287                 n_data->fingers[0].host_name = recv_data->host_name;
 288                 n_data->fingers[0].mailbox = recv_data->mailbox;
 289                 // Logs
 290                 INFO1("Sending back a Join Request to %s",recv_data->host_name);
 291                 MSG_task_set_name(join_task,"Join Response");
 292                 MSG_task_send(join_task,recv_data->mailbox);
 293         }
 294
 295         else
 296         {
 297                 const char* closest_preceding_mailbox = find_closest_preceding(n_data,recv_data->id);
 298                 INFO1("Forwarding Join Call to mailbox %s",closest_preceding_mailbox);
 299                 MSG_task_send(join_task,closest_preceding_mailbox);
 300         }
 301 }
 302
 303 const char* find_closest_preceding(data_node* n_node,int id)
 304 {
 305         int i;
 306         for(i = n_node->fingers_nb-1; i >= 0 ; i--)
 307         {
 308                         if (n_node->fingers[i].id <= id)
 309                                 return n_node->fingers[i].mailbox;
 310         }
 311
 312         return n_node->mailbox; // !!!!!!!!!!!!!!
 313 }
 314 /*
 315  * Fin successor id : used to fix finger list
 316  */
 317 static int find_successor(data_node* n_data, int id)
 318 {
 319         if (is_in_interval(id,n_data->id,n_data->fingers[0].id))
 320                 return n_data->fingers[0].id;
 321         else
 322             return 0;
 323
 324 }
 325
 326
 327 /** Test function */
 328 MSG_error_t test_all(const char *platform_file,
 329                      const char *application_file)
 330 {
 331   MSG_error_t res = MSG_OK;
 332
 333   /* MSG_config("workstation/model","KCCFLN05"); */
 334   {                             /*  Simulation setting */
 335     MSG_set_channel_number(0);
 336     MSG_create_environment(platform_file);
 337
 338   }
 339   {                             /*   Application deployment */
 340         MSG_function_register("node",node);
 341     MSG_launch_application(application_file);
 342   }
 343   res = MSG_main();
 344   INFO1("Simulation time %g", MSG_get_clock());
 345
 346   return res;
 347 }                               /* end_of_test_all */
 348
 349 /** Main function */
 350 int main(int argc, char *argv[])
 351 {
 352   MSG_error_t res = MSG_OK;
 353
 354   MSG_global_init(&argc, argv);
 355   if (argc < 3) {
 356     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 357     printf("example: %s msg_platform.xml msg_deployment.xml\n", argv[0]);
 358     exit(1);
 359   }
 360   res = test_all(argv[1], argv[2]);
 361   MSG_clean();
 362
 363   if (res == MSG_OK)
 364     return 0;
 365   else
 366     return 1;
 367 }                               /* end_of_main */