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

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