examples/msg/dht-chord/dht-chord.c

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