examples/msg/chord/chord.c

   1 /* Copyright (c) 2010-2014. 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 "msg/msg.h"
   9 #include "xbt/log.h"
  10 #include "xbt/asserts.h"
  11 #include "simgrid/modelchecker.h"
  12 #include <xbt/RngStream.h>
  13
  14 /** @addtogroup MSG_examples
  15  *
  16  *  - <b>chord/chord.c: Classical Chord P2P protocol</b>
  17  *    This example implements the well known Chord P2P protocol. Its
  18  *    main advantage is that it constitute a fully working non-trivial
  19  *    example. In addition, its implementation is rather efficient, as
  20  *    demonstrated in http://hal.inria.fr/inria-00602216/
  21  */
  22
  23
  24 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
  25                              "Messages specific for this msg example");
  26
  27 #define COMM_SIZE 10
  28 #define COMP_SIZE 0
  29 #define MAILBOX_NAME_SIZE 10
  30
  31 static int nb_bits = 24;
  32 static int nb_keys = 0;
  33 static int timeout = 50;
  34 static int max_simulation_time = 1000;
  35 static int periodic_stabilize_delay = 20;
  36 static int periodic_fix_fingers_delay = 120;
  37 static int periodic_check_predecessor_delay = 120;
  38 static int periodic_lookup_delay = 10;
  39
  40 static const double sleep_delay = 4.9999;
  41
  42 extern long int smx_total_comms;
  43
  44 /*
  45  * Finger element.
  46  */
  47 typedef struct s_finger {
  48   int id;
  49   char mailbox[MAILBOX_NAME_SIZE]; // string representation of the id
  50 } s_finger_t, *finger_t;
  51
  52 /*
  53  * Node data.
  54  */
  55 typedef struct s_node {
  56   int id;                                 // my id
  57   char mailbox[MAILBOX_NAME_SIZE];        // my mailbox name (string representation of the id)
  58   s_finger_t *fingers;                    // finger table, of size nb_bits (fingers[0] is my successor)
  59   int pred_id;                            // predecessor id
  60   char pred_mailbox[MAILBOX_NAME_SIZE];   // predecessor's mailbox name
  61   int next_finger_to_fix;                 // index of the next finger to fix in fix_fingers()
  62   msg_comm_t comm_receive;                // current communication to receive
  63   double last_change_date;                // last time I changed a finger or my predecessor
  64   RngStream stream;                       //RngStream for
  65 } s_node_t, *node_t;
  66
  67 /**
  68  * Types of tasks exchanged between nodes.
  69  */
  70 typedef enum {
  71   TASK_FIND_SUCCESSOR,
  72   TASK_FIND_SUCCESSOR_ANSWER,
  73   TASK_GET_PREDECESSOR,
  74   TASK_GET_PREDECESSOR_ANSWER,
  75   TASK_NOTIFY,
  76   TASK_SUCCESSOR_LEAVING,
  77   TASK_PREDECESSOR_LEAVING,
  78   TASK_PREDECESSOR_ALIVE,
  79   TASK_PREDECESSOR_ALIVE_ANSWER
  80 } e_task_type_t;
  81
  82 /*
  83  * Data attached with the tasks sent and received
  84  */
  85 typedef struct s_task_data {
  86   e_task_type_t type;                     // type of task
  87   int request_id;                         // id paramater (used by some types of tasks)
  88   int request_finger;                     // finger parameter (used by some types of tasks)
  89   int answer_id;                          // answer (used by some types of tasks)
  90   char answer_to[MAILBOX_NAME_SIZE];      // mailbox to send an answer to (if any)
  91   const char* issuer_host_name;           // used for logging
  92 } s_task_data_t, *task_data_t;
  93
  94 static int *powers2;
  95
  96 // utility functions
  97 static void chord_initialize(void);
  98 static void chord_exit(void);
  99 static int normalize(int id);
 100 static int is_in_interval(int id, int start, int end);
 101 static void get_mailbox(int host_id, char* mailbox);
 102 static void task_free(void* task);
 103 static void print_finger_table(node_t node);
 104 static void set_finger(node_t node, int finger_index, int id);
 105 static void set_predecessor(node_t node, int predecessor_id);
 106
 107 // process functions
 108 static int node(int argc, char *argv[]);
 109 static void handle_task(node_t node, msg_task_t task);
 110
 111 // Chord core
 112 static void create(node_t node);
 113 static int join(node_t node, int known_id);
 114 static void leave(node_t node);
 115 static int find_successor(node_t node, int id);
 116 static int remote_find_successor(node_t node, int ask_to_id, int id);
 117 static int remote_get_predecessor(node_t node, int ask_to_id);
 118 static int closest_preceding_node(node_t node, int id);
 119 static void stabilize(node_t node);
 120 static void notify(node_t node, int predecessor_candidate_id);
 121 static void remote_notify(node_t node, int notify_to, int predecessor_candidate_id);
 122 static void fix_fingers(node_t node);
 123 static void check_predecessor(node_t node);
 124 static void random_lookup(node_t);
 125 static void quit_notify(node_t node);
 126
 127 /**
 128  * \brief Global initialization of the Chord simulation.
 129  */
 130 static void chord_initialize(void)
 131 {
 132   // compute the powers of 2 once for all
 133   powers2 = xbt_new(int, nb_bits);
 134   int pow = 1;
 135   int i;
 136   for (i = 0; i < nb_bits; i++) {
 137     powers2[i] = pow;
 138     pow = pow << 1;
 139   }
 140   nb_keys = pow;
 141   XBT_DEBUG("Sets nb_keys to %d", nb_keys);
 142 }
 143
 144 static void chord_exit(void)
 145 {
 146   xbt_free(powers2);
 147 }
 148
 149 /**
 150  * \brief Turns an id into an equivalent id in [0, nb_keys).
 151  * \param id an id
 152  * \return the corresponding normalized id
 153  */
 154 static int normalize(int id)
 155 {
 156   // like id % nb_keys, but works with negatives numbers (and faster)
 157   return id & (nb_keys - 1);
 158 }
 159
 160 /**
 161  * \brief Returns whether an id belongs to the interval [start, end].
 162  *
 163  * The parameters are noramlized to make sure they are between 0 and nb_keys - 1).
 164  * 1 belongs to [62, 3]
 165  * 1 does not belong to [3, 62]
 166  * 63 belongs to [62, 3]
 167  * 63 does not belong to [3, 62]
 168  * 24 belongs to [21, 29]
 169  * 24 does not belong to [29, 21]
 170  *
 171  * \param id id to check
 172  * \param start lower bound
 173  * \param end upper bound
 174  * \return a non-zero value if id in in [start, end]
 175  */
 176 static int is_in_interval(int id, int start, int end)
 177 {
 178   id = normalize(id);
 179   start = normalize(start);
 180   end = normalize(end);
 181
 182   // make sure end >= start and id >= start
 183   if (end < start) {
 184     end += nb_keys;
 185   }
 186
 187   if (id < start) {
 188     id += nb_keys;
 189   }
 190
 191   return id <= end;
 192 }
 193
 194 /**
 195  * \brief Gets the mailbox name of a host given its chord id.
 196  * \param node_id id of a node
 197  * \param mailbox pointer to where the mailbox name should be written
 198  * (there must be enough space)
 199  */
 200 static void get_mailbox(int node_id, char* mailbox)
 201 {
 202   snprintf(mailbox, MAILBOX_NAME_SIZE - 1, "%d", node_id);
 203 }
 204
 205 /**
 206  * \brief Frees the memory used by a task.
 207  * \param task the MSG task to destroy
 208  */
 209 static void task_free(void* task)
 210 {
 211   // TODO add a parameter data_free_function to MSG_task_create?
 212   if(task != NULL){
 213     xbt_free(MSG_task_get_data(task));
 214     MSG_task_destroy(task);
 215   }
 216 }
 217
 218 /**
 219  * \brief Displays the finger table of a node.
 220  * \param node a node
 221  */
 222 static void print_finger_table(node_t node)
 223 {
 224   if (XBT_LOG_ISENABLED(msg_chord, xbt_log_priority_verbose)) {
 225     int i;
 226     XBT_VERB("My finger table:");
 227     XBT_VERB("Start | Succ ");
 228     for (i = 0; i < nb_bits; i++) {
 229       XBT_VERB(" %3d  | %3d ", (node->id + powers2[i]) % nb_keys, node->fingers[i].id);
 230     }
 231     XBT_VERB("Predecessor: %d", node->pred_id);
 232   }
 233 }
 234
 235 /**
 236  * \brief Sets a finger of the current node.
 237  * \param node the current node
 238  * \param finger_index index of the finger to set (0 to nb_bits - 1)
 239  * \param id the id to set for this finger
 240  */
 241 static void set_finger(node_t node, int finger_index, int id)
 242 {
 243   if (id != node->fingers[finger_index].id) {
 244     node->fingers[finger_index].id = id;
 245     get_mailbox(id, node->fingers[finger_index].mailbox);
 246     node->last_change_date = MSG_get_clock();
 247     XBT_DEBUG("My new finger #%d is %d", finger_index, id);
 248   }
 249 }
 250
 251 /**
 252  * \brief Sets the predecessor of the current node.
 253  * \param node the current node
 254  * \param id the id to predecessor, or -1 to unset the predecessor
 255  */
 256 static void set_predecessor(node_t node, int predecessor_id)
 257 {
 258   if (predecessor_id != node->pred_id) {
 259     node->pred_id = predecessor_id;
 260
 261     if (predecessor_id != -1) {
 262       get_mailbox(predecessor_id, node->pred_mailbox);
 263     }
 264     node->last_change_date = MSG_get_clock();
 265
 266     XBT_DEBUG("My new predecessor is %d", predecessor_id);
 267   }
 268 }
 269
 270 /**
 271  * \brief Node Function
 272  * Arguments:
 273  * - my id
 274  * - the id of a guy I know in the system (except for the first node)
 275  * - the time to sleep before I join (except for the first node)
 276  */
 277 int node(int argc, char *argv[])
 278 {
 279
 280   /* Reduce the run size for the MC */
 281   if(MC_is_active()){
 282     periodic_stabilize_delay = 8;
 283     periodic_fix_fingers_delay = 8;
 284     periodic_check_predecessor_delay = 8;
 285   }
 286
 287   double init_time = MSG_get_clock();
 288   msg_task_t task_received = NULL;
 289   int i;
 290   int join_success = 0;
 291   double deadline;
 292   double next_stabilize_date = init_time + periodic_stabilize_delay;
 293   double next_fix_fingers_date = init_time + periodic_fix_fingers_delay;
 294   double next_check_predecessor_date = init_time + periodic_check_predecessor_delay;
 295   double next_lookup_date = init_time + periodic_lookup_delay;
 296
 297   int listen = 0;
 298   int no_op = 0;
 299   int sub_protocol = 0;
 300
 301   xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
 302
 303   // initialize my node
 304   s_node_t node = {0};
 305   node.id = atoi(argv[1]);
 306   get_mailbox(node.id, node.mailbox);
 307   node.next_finger_to_fix = 0;
 308   node.fingers = xbt_new0(s_finger_t, nb_bits);
 309   node.last_change_date = init_time;
 310
 311   for (i = 0; i < nb_bits; i++) {
 312     node.fingers[i].id = -1;
 313     set_finger(&node, i, node.id);
 314   }
 315
 316   if (argc == 3) { // first ring
 317     deadline = atof(argv[2]);
 318     create(&node);
 319     join_success = 1;
 320   }
 321   else {
 322     int known_id = atoi(argv[2]);
 323     //double sleep_time = atof(argv[3]);
 324     deadline = atof(argv[4]);
 325
 326     /*
 327     // sleep before starting
 328     XBT_DEBUG("Let's sleep during %f", sleep_time);
 329     MSG_process_sleep(sleep_time);
 330     */
 331     XBT_DEBUG("Hey! Let's join the system.");
 332
 333     join_success = join(&node, known_id);
 334   }
 335
 336   if (join_success) {
 337     while (MSG_get_clock() < init_time + deadline
 338 //      && MSG_get_clock() < node.last_change_date + 1000
 339         && MSG_get_clock() < max_simulation_time) {
 340
 341       if (node.comm_receive == NULL) {
 342         task_received = NULL;
 343         node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
 344         // FIXME: do not make MSG_task_irecv() calls from several functions
 345       }
 346
 347       //XBT_INFO("Node %d is ring member : %d", node.id, is_ring_member(known_id, node.id) != -1);
 348
 349       if (!MSG_comm_test(node.comm_receive)) {
 350
 351         // no task was received: make some periodic calls
 352
 353         if(MC_is_active()){
 354           if(!MC_visited_reduction() && no_op){
 355               MC_cut();
 356           }
 357           if(listen == 0 && (sub_protocol = MC_random(0, 4)) > 0){
 358             if(sub_protocol == 1)
 359               stabilize(&node);
 360             else if(sub_protocol == 2)
 361               fix_fingers(&node);
 362             else if(sub_protocol == 3)
 363               check_predecessor(&node);
 364             else
 365               random_lookup(&node);
 366             listen = 1;
 367           }else{
 368             MSG_process_sleep(sleep_delay);
 369             if(!MC_visited_reduction())
 370               no_op = 1;
 371           }
 372         }else{
 373           if (MSG_get_clock() >= next_stabilize_date) {
 374             stabilize(&node);
 375             next_stabilize_date = MSG_get_clock() + periodic_stabilize_delay;
 376           }else if (MSG_get_clock() >= next_fix_fingers_date) {
 377             fix_fingers(&node);
 378             next_fix_fingers_date = MSG_get_clock() + periodic_fix_fingers_delay;
 379           }else if (MSG_get_clock() >= next_check_predecessor_date) {
 380             check_predecessor(&node);
 381             next_check_predecessor_date = MSG_get_clock() + periodic_check_predecessor_delay;
 382           }else if (MSG_get_clock() >= next_lookup_date) {
 383             random_lookup(&node);
 384             next_lookup_date = MSG_get_clock() + periodic_lookup_delay;
 385           }else {
 386             // nothing to do: sleep for a while
 387             MSG_process_sleep(sleep_delay);
 388           }
 389         }
 390
 391       } else {
 392         // a transfer has occurred
 393
 394         msg_error_t status = MSG_comm_get_status(node.comm_receive);
 395
 396         if (status != MSG_OK) {
 397           XBT_DEBUG("Failed to receive a task. Nevermind.");
 398           MSG_comm_destroy(node.comm_receive);
 399           node.comm_receive = NULL;
 400         }
 401         else {
 402           // the task was successfully received
 403           MSG_comm_destroy(node.comm_receive);
 404           node.comm_receive = NULL;
 405           handle_task(&node, task_received);
 406         }
 407       }
 408     }
 409
 410     if (node.comm_receive) {
 411       /* handle last task if any */
 412       if (MSG_comm_wait(node.comm_receive, 0) == MSG_OK)
 413         task_free(task_received);
 414       MSG_comm_destroy(node.comm_receive);
 415       node.comm_receive = NULL;
 416     }
 417
 418     // leave the ring
 419     leave(&node);
 420   }
 421
 422   // stop the simulation
 423   xbt_free(node.fingers);
 424   return 0;
 425 }
 426
 427 /**
 428  * \brief This function is called when the current node receives a task.
 429  * \param node the current node
 430  * \param task the task to handle (don't touch it then:
 431  * it will be destroyed, reused or forwarded)
 432  */
 433 static void handle_task(node_t node, msg_task_t task) {
 434
 435   XBT_DEBUG("Handling task %p", task);
 436   char mailbox[MAILBOX_NAME_SIZE];
 437   task_data_t task_data = (task_data_t) MSG_task_get_data(task);
 438   e_task_type_t type = task_data->type;
 439
 440   switch (type) {
 441
 442   case TASK_FIND_SUCCESSOR:
 443     XBT_DEBUG("Receiving a 'Find Successor' request from %s for id %d",
 444               task_data->issuer_host_name, task_data->request_id);
 445     // is my successor the successor?
 446     if (is_in_interval(task_data->request_id, node->id + 1, node->fingers[0].id)) {
 447       task_data->type = TASK_FIND_SUCCESSOR_ANSWER;
 448       task_data->answer_id = node->fingers[0].id;
 449       XBT_DEBUG("Sending back a 'Find Successor Answer' to %s (mailbox %s): the successor of %d is %d",
 450                 task_data->issuer_host_name,
 451                 task_data->answer_to,
 452                 task_data->request_id, task_data->answer_id);
 453       MSG_task_dsend(task, task_data->answer_to, task_free);
 454     }
 455     else {
 456       // otherwise, forward the request to the closest preceding finger in my table
 457       int closest = closest_preceding_node(node, task_data->request_id);
 458       XBT_DEBUG("Forwarding the 'Find Successor' request for id %d to my closest preceding finger %d",
 459                 task_data->request_id, closest);
 460       get_mailbox(closest, mailbox);
 461       MSG_task_dsend(task, mailbox, task_free);
 462     }
 463     break;
 464
 465   case TASK_GET_PREDECESSOR:
 466     XBT_DEBUG("Receiving a 'Get Predecessor' request from %s", task_data->issuer_host_name);
 467     task_data->type = TASK_GET_PREDECESSOR_ANSWER;
 468     task_data->answer_id = node->pred_id;
 469     XBT_DEBUG("Sending back a 'Get Predecessor Answer' to %s via mailbox '%s': my predecessor is %d",
 470               task_data->issuer_host_name,
 471               task_data->answer_to, task_data->answer_id);
 472     MSG_task_dsend(task, task_data->answer_to, task_free);
 473     break;
 474
 475   case TASK_NOTIFY:
 476     // someone is telling me that he may be my new predecessor
 477     XBT_DEBUG("Receiving a 'Notify' request from %s", task_data->issuer_host_name);
 478     notify(node, task_data->request_id);
 479     task_free(task);
 480     break;
 481
 482   case TASK_PREDECESSOR_LEAVING:
 483     // my predecessor is about to quit
 484     XBT_DEBUG("Receiving a 'Predecessor Leaving' message from %s", task_data->issuer_host_name);
 485     // modify my predecessor
 486     set_predecessor(node, task_data->request_id);
 487     task_free(task);
 488     /*TODO :
 489       >> notify my new predecessor
 490       >> send a notify_predecessors !!
 491     */
 492     break;
 493
 494   case TASK_SUCCESSOR_LEAVING:
 495     // my successor is about to quit
 496     XBT_DEBUG("Receiving a 'Successor Leaving' message from %s", task_data->issuer_host_name);
 497     // modify my successor FIXME : this should be implicit ?
 498     set_finger(node, 0, task_data->request_id);
 499     task_free(task);
 500     /* TODO
 501        >> notify my new successor
 502        >> update my table & predecessors table */
 503     break;
 504
 505   case TASK_FIND_SUCCESSOR_ANSWER:
 506   case TASK_GET_PREDECESSOR_ANSWER:
 507   case TASK_PREDECESSOR_ALIVE_ANSWER:
 508     XBT_DEBUG("Ignoring unexpected task of type %d (%p)", (int)type, task);
 509     task_free(task);
 510     break;
 511
 512   case TASK_PREDECESSOR_ALIVE:
 513     XBT_DEBUG("Receiving a 'Predecessor Alive' request from %s", task_data->issuer_host_name);
 514     task_data->type = TASK_PREDECESSOR_ALIVE_ANSWER;
 515     XBT_DEBUG("Sending back a 'Predecessor Alive Answer' to %s (mailbox %s)",
 516               task_data->issuer_host_name,
 517               task_data->answer_to);
 518     MSG_task_dsend(task, task_data->answer_to, task_free);
 519     break;
 520
 521   default:
 522     THROW_IMPOSSIBLE;
 523   }
 524 }
 525
 526 /**
 527  * \brief Initializes the current node as the first one of the system.
 528  * \param node the current node
 529  */
 530 static void create(node_t node)
 531 {
 532   XBT_DEBUG("Create a new Chord ring...");
 533   set_predecessor(node, -1); // -1 means that I have no predecessor
 534   print_finger_table(node);
 535 }
 536
 537 /**
 538  * \brief Makes the current node join the ring, knowing the id of a node
 539  * already in the ring
 540  * \param node the current node
 541  * \param known_id id of a node already in the ring
 542  * \return 1 if the join operation succeeded, 0 otherwise
 543  */
 544 static int join(node_t node, int known_id)
 545 {
 546   XBT_INFO("Joining the ring with id %d, knowing node %d", node->id, known_id);
 547   set_predecessor(node, -1); // no predecessor (yet)
 548
 549   /*
 550   int i;
 551   for (i = 0; i < nb_bits; i++) {
 552     set_finger(node, i, known_id);
 553   }
 554   */
 555
 556   int successor_id = remote_find_successor(node, known_id, node->id);
 557   if (successor_id == -1) {
 558     XBT_INFO("Cannot join the ring.");
 559   }
 560   else {
 561     set_finger(node, 0, successor_id);
 562     print_finger_table(node);
 563   }
 564
 565   return successor_id != -1;
 566 }
 567
 568 /**
 569  * \brief Makes the current node quit the system
 570  * \param node the current node
 571  */
 572 static void leave(node_t node)
 573 {
 574   XBT_DEBUG("Well Guys! I Think it's time for me to quit ;)");
 575   quit_notify(node);
 576   RngStream_DeleteStream(&node->stream);
 577 }
 578
 579 /**
 580  * \brief Notifies the successor and the predecessor of the current node
 581  * of the departure
 582  * \param node the current node
 583  */
 584 static void quit_notify(node_t node)
 585 {
 586   char mailbox[MAILBOX_NAME_SIZE];
 587   //send the PREDECESSOR_LEAVING to our successor
 588   task_data_t req_data = xbt_new0(s_task_data_t,1);
 589   req_data->type = TASK_PREDECESSOR_LEAVING;
 590   req_data->request_id = node->pred_id;
 591   get_mailbox(node->id, req_data->answer_to);
 592   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 593
 594   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 595   XBT_DEBUG("Sending a 'PREDECESSOR_LEAVING' to my successor %d",node->fingers[0].id);
 596   if (MSG_task_send_with_timeout(task_sent, node->fingers[0].mailbox, timeout)==
 597       MSG_TIMEOUT) {
 598     XBT_DEBUG("Timeout expired when sending a 'PREDECESSOR_LEAVING' to my successor %d",
 599         node->fingers[0].id);
 600     task_free(task_sent);
 601   }
 602
 603   //send the SUCCESSOR_LEAVING to our predecessor
 604   get_mailbox(node->pred_id, mailbox);
 605   task_data_t req_data_s = xbt_new0(s_task_data_t,1);
 606   req_data_s->type = TASK_SUCCESSOR_LEAVING;
 607   req_data_s->request_id = node->fingers[0].id;
 608   req_data_s->request_id = node->pred_id;
 609   get_mailbox(node->id, req_data_s->answer_to);
 610   req_data_s->issuer_host_name = MSG_host_get_name(MSG_host_self());
 611
 612   msg_task_t task_sent_s = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data_s);
 613   XBT_DEBUG("Sending a 'SUCCESSOR_LEAVING' to my predecessor %d",node->pred_id);
 614   if (MSG_task_send_with_timeout(task_sent_s, mailbox, timeout)==
 615       MSG_TIMEOUT) {
 616     XBT_DEBUG("Timeout expired when sending a 'SUCCESSOR_LEAVING' to my predecessor %d",
 617         node->pred_id);
 618     task_free(task_sent_s);
 619   }
 620
 621 }
 622
 623 /**
 624  * \brief Makes the current node find the successor node of an id.
 625  * \param node the current node
 626  * \param id the id to find
 627  * \return the id of the successor node, or -1 if the request failed
 628  */
 629 static int find_successor(node_t node, int id)
 630 {
 631   // is my successor the successor?
 632   if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
 633     return node->fingers[0].id;
 634   }
 635
 636   // otherwise, ask the closest preceding finger in my table
 637   int closest = closest_preceding_node(node, id);
 638   return remote_find_successor(node, closest, id);
 639 }
 640
 641 /**
 642  * \brief Asks another node the successor node of an id.
 643  * \param node the current node
 644  * \param ask_to the node to ask to
 645  * \param id the id to find
 646  * \return the id of the successor node, or -1 if the request failed
 647  */
 648 static int remote_find_successor(node_t node, int ask_to, int id)
 649 {
 650   int successor = -1;
 651   int stop = 0;
 652   char mailbox[MAILBOX_NAME_SIZE];
 653   get_mailbox(ask_to, mailbox);
 654   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 655   req_data->type = TASK_FIND_SUCCESSOR;
 656   req_data->request_id = id;
 657   get_mailbox(node->id, req_data->answer_to);
 658   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 659
 660   // send a "Find Successor" request to ask_to_id
 661   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 662   XBT_DEBUG("Sending a 'Find Successor' request (task %p) to %d for id %d", task_sent, ask_to, id);
 663   msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 664
 665   if (res != MSG_OK) {
 666     XBT_DEBUG("Failed to send the 'Find Successor' request (task %p) to %d for id %d",
 667         task_sent, ask_to, id);
 668     task_free(task_sent);
 669   }
 670   else {
 671
 672     // receive the answer
 673     XBT_DEBUG("Sent a 'Find Successor' request (task %p) to %d for key %d, waiting for the answer",
 674         task_sent, ask_to, id);
 675
 676     do {
 677       if (node->comm_receive == NULL) {
 678         msg_task_t task_received = NULL;
 679         node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
 680       }
 681
 682       res = MSG_comm_wait(node->comm_receive, timeout);
 683
 684       if (res != MSG_OK) {
 685         XBT_DEBUG("Failed to receive the answer to my 'Find Successor' request (task %p): %d",
 686                   task_sent, (int)res);
 687         stop = 1;
 688         MSG_comm_destroy(node->comm_receive);
 689         node->comm_receive = NULL;
 690       }
 691       else {
 692         msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
 693         XBT_DEBUG("Received a task (%p)", task_received);
 694         task_data_t ans_data = MSG_task_get_data(task_received);
 695
 696         // Once upon a time, our code assumed that here, task_received != task_sent all the time
 697         //
 698         // This assumption is wrong (as messages from differing round can interleave), leading to a bug in our code.
 699         // We failed to find this bug directly, as it only occured on large platforms, leading to hardly usable traces.
 700         // Instead, we used the model-checker to track down the issue by adding the following test here in the code:
 701         //   if (MC_is_active()) {
 702         //      MC_assert(task_received == task_sent);
 703         //   }
 704         // That explained the bug in a snap, with a very cool example and everything.
 705         //
 706         // This MC_assert is now desactivated as the case is now properly handled in our code and we don't want the
 707         //   MC to fail any further under that condition, but this comment is here to as a memorial for this first
 708         //   brillant victory of the model-checking in the SimGrid community :)
 709
 710         if (task_received != task_sent ||
 711             ans_data->type != TASK_FIND_SUCCESSOR_ANSWER) {
 712           // this is not the expected answer
 713           MSG_comm_destroy(node->comm_receive);
 714           node->comm_receive = NULL;
 715           handle_task(node, task_received);
 716         }
 717         else {
 718           // this is our answer
 719           XBT_DEBUG("Received the answer to my 'Find Successor' request for id %d (task %p): the successor of key %d is %d",
 720               ans_data->request_id, task_received, id, ans_data->answer_id);
 721           successor = ans_data->answer_id;
 722           stop = 1;
 723           MSG_comm_destroy(node->comm_receive);
 724           node->comm_receive = NULL;
 725           task_free(task_received);
 726         }
 727       }
 728     } while (!stop);
 729   }
 730
 731   return successor;
 732 }
 733
 734 /**
 735  * \brief Asks another node its predecessor.
 736  * \param node the current node
 737  * \param ask_to the node to ask to
 738  * \return the id of its predecessor node, or -1 if the request failed
 739  * (or if the node does not know its predecessor)
 740  */
 741 static int remote_get_predecessor(node_t node, int ask_to)
 742 {
 743   int predecessor_id = -1;
 744   int stop = 0;
 745   char mailbox[MAILBOX_NAME_SIZE];
 746   get_mailbox(ask_to, mailbox);
 747   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 748   req_data->type = TASK_GET_PREDECESSOR;
 749   get_mailbox(node->id, req_data->answer_to);
 750   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 751
 752   // send a "Get Predecessor" request to ask_to_id
 753   XBT_DEBUG("Sending a 'Get Predecessor' request to %d", ask_to);
 754   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 755   msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 756
 757   if (res != MSG_OK) {
 758     XBT_DEBUG("Failed to send the 'Get Predecessor' request (task %p) to %d",
 759         task_sent, ask_to);
 760     task_free(task_sent);
 761   }
 762   else {
 763
 764     // receive the answer
 765     XBT_DEBUG("Sent 'Get Predecessor' request (task %p) to %d, waiting for the answer on my mailbox '%s'",
 766         task_sent, ask_to, req_data->answer_to);
 767
 768     do {
 769       if (node->comm_receive == NULL) { // FIXME simplify this
 770         msg_task_t task_received = NULL;
 771         node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
 772       }
 773
 774       res = MSG_comm_wait(node->comm_receive, timeout);
 775
 776       if (res != MSG_OK) {
 777         XBT_DEBUG("Failed to receive the answer to my 'Get Predecessor' request (task %p): %d",
 778             task_sent, (int)res);
 779         stop = 1;
 780         MSG_comm_destroy(node->comm_receive);
 781         node->comm_receive = NULL;
 782       }
 783       else {
 784         msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
 785         task_data_t ans_data = MSG_task_get_data(task_received);
 786
 787         /*if (MC_is_active()) {
 788           MC_assert(task_received == task_sent);
 789           }*/
 790
 791         if (task_received != task_sent ||
 792             ans_data->type != TASK_GET_PREDECESSOR_ANSWER) {
 793           MSG_comm_destroy(node->comm_receive);
 794           node->comm_receive = NULL;
 795           handle_task(node, task_received);
 796         }
 797         else {
 798           XBT_DEBUG("Received the answer to my 'Get Predecessor' request (task %p): the predecessor of node %d is %d",
 799               task_received, ask_to, ans_data->answer_id);
 800           predecessor_id = ans_data->answer_id;
 801           stop = 1;
 802           MSG_comm_destroy(node->comm_receive);
 803           node->comm_receive = NULL;
 804           task_free(task_received);
 805         }
 806       }
 807     } while (!stop);
 808   }
 809
 810   return predecessor_id;
 811 }
 812
 813 /**
 814  * \brief Returns the closest preceding finger of an id
 815  * with respect to the finger table of the current node.
 816  * \param node the current node
 817  * \param id the id to find
 818  * \return the closest preceding finger of that id
 819  */
 820 int closest_preceding_node(node_t node, int id)
 821 {
 822   int i;
 823   for (i = nb_bits - 1; i >= 0; i--) {
 824     if (is_in_interval(node->fingers[i].id, node->id + 1, id - 1)) {
 825       return node->fingers[i].id;
 826     }
 827   }
 828   return node->id;
 829 }
 830
 831 /**
 832  * \brief This function is called periodically. It checks the immediate
 833  * successor of the current node.
 834  * \param node the current node
 835  */
 836 static void stabilize(node_t node)
 837 {
 838   XBT_DEBUG("Stabilizing node");
 839
 840   // get the predecessor of my immediate successor
 841   int candidate_id;
 842   int successor_id = node->fingers[0].id;
 843   if (successor_id != node->id) {
 844     candidate_id = remote_get_predecessor(node, successor_id);
 845   }
 846   else {
 847     candidate_id = node->pred_id;
 848   }
 849
 850   // this node is a candidate to become my new successor
 851   if (candidate_id != -1
 852       && is_in_interval(candidate_id, node->id + 1, successor_id - 1)) {
 853     set_finger(node, 0, candidate_id);
 854   }
 855   if (successor_id != node->id) {
 856     remote_notify(node, successor_id, node->id);
 857   }
 858 }
 859
 860 /**
 861  * \brief Notifies the current node that its predecessor may have changed.
 862  * \param node the current node
 863  * \param candidate_id the possible new predecessor
 864  */
 865 static void notify(node_t node, int predecessor_candidate_id) {
 866
 867   if (node->pred_id == -1
 868     || is_in_interval(predecessor_candidate_id, node->pred_id + 1, node->id - 1)) {
 869
 870     set_predecessor(node, predecessor_candidate_id);
 871     print_finger_table(node);
 872   }
 873   else {
 874     XBT_DEBUG("I don't have to change my predecessor to %d", predecessor_candidate_id);
 875   }
 876 }
 877
 878 /**
 879  * \brief Notifies a remote node that its predecessor may have changed.
 880  * \param node the current node
 881  * \param notify_id id of the node to notify
 882  * \param candidate_id the possible new predecessor
 883  */
 884 static void remote_notify(node_t node, int notify_id, int predecessor_candidate_id) {
 885
 886       task_data_t req_data = xbt_new0(s_task_data_t, 1);
 887       req_data->type = TASK_NOTIFY;
 888       req_data->request_id = predecessor_candidate_id;
 889       req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 890
 891       // send a "Notify" request to notify_id
 892       msg_task_t task = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 893       XBT_DEBUG("Sending a 'Notify' request (task %p) to %d", task, notify_id);
 894       char mailbox[MAILBOX_NAME_SIZE];
 895       get_mailbox(notify_id, mailbox);
 896       MSG_task_dsend(task, mailbox, task_free);
 897     }
 898
 899 /**
 900  * \brief This function is called periodically.
 901  * It refreshes the finger table of the current node.
 902  * \param node the current node
 903  */
 904 static void fix_fingers(node_t node) {
 905
 906   XBT_DEBUG("Fixing fingers");
 907   int i = node->next_finger_to_fix;
 908   int id = find_successor(node, node->id + powers2[i]);
 909   if (id != -1) {
 910
 911     if (id != node->fingers[i].id) {
 912       set_finger(node, i, id);
 913       print_finger_table(node);
 914     }
 915     node->next_finger_to_fix = (i + 1) % nb_bits;
 916   }
 917 }
 918
 919 /**
 920  * \brief This function is called periodically.
 921  * It checks whether the predecessor has failed
 922  * \param node the current node
 923  */
 924 static void check_predecessor(node_t node)
 925 {
 926   XBT_DEBUG("Checking whether my predecessor is alive");
 927
 928   if(node->pred_id == -1)
 929     return;
 930
 931   int stop = 0;
 932
 933   char mailbox[MAILBOX_NAME_SIZE];
 934   get_mailbox(node->pred_id, mailbox);
 935   task_data_t req_data = xbt_new0(s_task_data_t,1);
 936   req_data->type = TASK_PREDECESSOR_ALIVE;
 937   req_data->request_id = node->pred_id;
 938   get_mailbox(node->id, req_data->answer_to);
 939   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 940
 941   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 942   XBT_DEBUG("Sending a 'Predecessor Alive' request to my predecessor %d", node->pred_id);
 943
 944   msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 945
 946   if (res != MSG_OK) {
 947     XBT_DEBUG("Failed to send the 'Predecessor Alive' request (task %p) to %d", task_sent, node->pred_id);
 948     task_free(task_sent);
 949   }else{
 950
 951     // receive the answer
 952     XBT_DEBUG("Sent 'Predecessor Alive' request (task %p) to %d, waiting for the answer on my mailbox '%s'",
 953               task_sent, node->pred_id, req_data->answer_to);
 954
 955     do {
 956       if (node->comm_receive == NULL) { // FIXME simplify this
 957         msg_task_t task_received = NULL;
 958         node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
 959       }
 960
 961       res = MSG_comm_wait(node->comm_receive, timeout);
 962
 963       if (res != MSG_OK) {
 964         XBT_DEBUG("Failed to receive the answer to my 'Predecessor Alive' request (task %p): %d",
 965                   task_sent, (int)res);
 966         stop = 1;
 967         MSG_comm_destroy(node->comm_receive);
 968         node->comm_receive = NULL;
 969         node->pred_id = -1;
 970       }else {
 971         msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
 972         if (task_received != task_sent) {
 973           MSG_comm_destroy(node->comm_receive);
 974           node->comm_receive = NULL;
 975           handle_task(node, task_received);
 976         }else{
 977           XBT_DEBUG("Received the answer to my 'Predecessor Alive' request (task %p) : my predecessor %d is alive", task_received, node->pred_id);
 978           stop = 1;
 979           MSG_comm_destroy(node->comm_receive);
 980           node->comm_receive = NULL;
 981           task_free(task_received);
 982         }
 983       }
 984     } while (!stop);
 985   }
 986 }
 987
 988 /**
 989  * \brief Performs a find successor request to a random id.
 990  * \param node the current node
 991  */
 992 static void random_lookup(node_t node)
 993 {
 994
 995   int id = 1337;
 996   find_successor(node, id);
 997
 998   /*** Random lookup disabled for tesh examples ***/
 999   /*if(node->stream == NULL)
1000     node->stream = RngStream_CreateStream("");
1001   int random_index = RngStream_RandInt (node->stream, 0, nb_bits - 1);
1002   int random_id = node->fingers[random_index].id;
1003   XBT_DEBUG("Making a lookup request for id %d", random_id);
1004   int res = find_successor(node, random_id);
1005   XBT_DEBUG("The successor of node %d is %d", random_id, res);*/
1006
1007 }
1008
1009 /**
1010  * \brief Main function.
1011  */
1012 int main(int argc, char *argv[])
1013 {
1014   MSG_init(&argc, argv);
1015   if (argc < 3) {
1016     printf("Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n", argv[0]);
1017     printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
1018     exit(1);
1019   }
1020
1021   char **options = &argv[1];
1022   while (!strncmp(options[0], "-", 1)) {
1023
1024     int length = strlen("-nb_bits=");
1025     if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
1026       nb_bits = atoi(options[0] + length);
1027       XBT_DEBUG("Set nb_bits to %d", nb_bits);
1028     }
1029     else {
1030
1031       length = strlen("-timeout=");
1032       if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
1033         timeout = atoi(options[0] + length);
1034         XBT_DEBUG("Set timeout to %d", timeout);
1035       }
1036       else {
1037         xbt_die("Invalid chord option '%s'", options[0]);
1038       }
1039     }
1040     options++;
1041   }
1042
1043   const char* platform_file = options[0];
1044   const char* application_file = options[1];
1045
1046   chord_initialize();
1047
1048   MSG_create_environment(platform_file);
1049
1050   MSG_function_register("node", node);
1051   MSG_launch_application(application_file);
1052
1053   msg_error_t res = MSG_main();
1054   XBT_CRITICAL("Messages created: %ld", smx_total_comms);
1055   XBT_INFO("Simulated time: %g", MSG_get_clock());
1056
1057   chord_exit();
1058
1059   if (res == MSG_OK)
1060     return 0;
1061   else
1062     return 1;
1063 }