examples/msg/chord/chord.c

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