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 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
  13                              "Messages specific for this msg example");
  14
  15 #define NB_BITS 6
  16 #define NB_KEYS 64
  17 #define COMM_SIZE 10
  18 #define COMP_SIZE 0
  19
  20 /**
  21  * Finger element.
  22  */
  23 typedef struct finger {
  24   int id;
  25   char* mailbox;
  26 } s_finger_t, *finger_t;
  27
  28 /**
  29  * Node data.
  30  */
  31 typedef struct node {
  32   int id;                                 // my id
  33   char* mailbox;                          // my usual mailbox name
  34   s_finger_t fingers[NB_BITS];            // finger table (fingers[0] is my successor)
  35   int pred_id;                            // predecessor id
  36   char* pred_mailbox;                     // predecessor's mailbox name
  37   int next_finger_to_fix;                 // index of the next finger to fix in fix_fingers()
  38   msg_comm_t comm_receive;                // current communication to receive
  39   xbt_dynar_t comms;                      // current communications being sent
  40 } s_node_t, *node_t;
  41
  42 /**
  43  * Task data
  44  */
  45 typedef struct task_data {
  46   int request_id;
  47   int request_finger;
  48   int answer_id;
  49   const char* answer_to;
  50   const char* issuer_host_name; // used for logging
  51   int successor_id;                             // used when quitting
  52   int pred_id;                                  // used when quitting
  53   // FIXME: remove successor_id and pred_id, request_id is enough
  54 } s_task_data_t, *task_data_t;
  55
  56 static int powers2[NB_BITS];
  57
  58 // utility functions
  59 static void chord_initialize(void);
  60 static int normalize(int id);
  61 static int is_in_interval(int id, int start, int end);
  62 static char* get_mailbox(int host_id);
  63 static void print_finger_table(node_t node);
  64 static void set_finger(node_t node, int finger_index, int id);
  65 static void set_predecessor(node_t node, int predecessor_id);
  66
  67 // process functions
  68 static int node(int argc, char *argv[]);
  69 static void handle_task(node_t node, m_task_t task);
  70
  71 // Chord core
  72 static void create(node_t node);
  73 static void join(node_t node, int known_id);
  74 static void leave(node_t node);
  75 static int find_successor(node_t node, int id);
  76 static int remote_find_successor(node_t node, int ask_to_id, int id);
  77 static int remote_get_predecessor(node_t node, int ask_to_id);
  78 static int closest_preceding_node(node_t node, int id);
  79 static void stabilize(node_t node);
  80 static void notify(node_t node, int predecessor_candidate_id);
  81 static void remote_notify(node_t node, int notify_to, int predecessor_candidate_id);
  82 static void fix_fingers(node_t node);
  83 static void check_predecessor(node_t node);
  84 static void quit_notify(node_t node, int to);
  85
  86 static void chord_initialize(void)
  87 {
  88   // compute the powers of 2 once for all
  89   int pow = 1;
  90   int i;
  91   for (i = 0; i < NB_BITS; i++) {
  92     powers2[i] = pow;
  93     pow = pow << 1;
  94   }
  95 }
  96
  97 /**
  98  * \brief Turns an id into an equivalent id in [0, NB_KEYS[
  99  * \param id an id
 100  * \return the corresponding normalized id
 101  */
 102 static int normalize(int id)
 103 {
 104   // make sure id >= 0
 105   while (id < 0) {
 106     id += NB_KEYS;
 107   }
 108   // make sure id < NB_KEYS
 109   id = id % NB_KEYS;
 110
 111   return id;
 112 }
 113
 114 /**
 115  * \brief Returns whether a id belongs to the interval [start, end].
 116  *
 117  * The parameters are noramlized to make sure they are between 0 and CHORD_NB_KEYS - 1).
 118  * 1 belongs to [62, 3]
 119  * 1 does not belong to [3, 62]
 120  * 63 belongs to [62, 3]
 121  * 63 does not belong to [3, 62]
 122  * 24 belongs to [21, 29]
 123  * 24 does not belong to [29, 21]
 124  *
 125  * \param id id to check
 126  * \param start lower bound
 127  * \param end upper bound
 128  * \return a non-zero value if id in in [start, end]
 129  */
 130 static int is_in_interval(int id, int start, int end)
 131 {
 132   id = normalize(id);
 133   start = normalize(start);
 134   end = normalize(end);
 135
 136   // make sure end >= start and id >= start
 137   if (end < start) {
 138     end += NB_KEYS;
 139   }
 140
 141   if (id < start) {
 142     id += NB_KEYS;
 143   }
 144
 145   return id <= end;
 146 }
 147
 148 /**
 149  * \brief Gets the mailbox name of a host given its chord id.
 150  * \param node_id id of a node
 151  * \return the name of its mailbox
 152  * FIXME: free the memory
 153  */
 154 static char* get_mailbox(int node_id)
 155 {
 156   return bprintf("mailbox%d", node_id);
 157 }
 158
 159 /**
 160  * \brief Displays the finger table of a node.
 161  * \param node a node
 162  */
 163 static void print_finger_table(node_t node)
 164 {
 165   int i;
 166   int pow = 1;
 167   INFO0("My finger table:");
 168   INFO0("Start | Succ ");
 169   for (i = 0; i < NB_BITS; i++) {
 170     INFO2(" %3d  | %3d ", (node->id + pow) % NB_KEYS, node->fingers[i].id);
 171     pow = pow << 1;
 172   }
 173   INFO1("Predecessor: %d", node->pred_id);
 174 }
 175
 176 /**
 177  * \brief Sets a finger of the current node.
 178  * \param node the current node
 179  * \param finger_index index of the finger to set (0 to NB_BITS - 1)
 180  * \param id the id to set for this finger
 181  */
 182 static void set_finger(node_t node, int finger_index, int id)
 183 {
 184   node->fingers[finger_index].id = id;
 185   xbt_free(node->fingers[finger_index].mailbox);
 186   node->fingers[finger_index].mailbox = get_mailbox(id);
 187   INFO2("My new finger #%d is %d", finger_index, id);
 188 }
 189
 190 /**
 191  * \brief Sets the predecessor of the current node.
 192  * \param node the current node
 193  * \param id the id to predecessor, or -1 to unset the predecessor
 194  */
 195 static void set_predecessor(node_t node, int predecessor_id)
 196 {
 197   node->pred_id = predecessor_id;
 198   xbt_free(node->pred_mailbox);
 199
 200   if (predecessor_id != -1) {
 201     node->pred_mailbox = get_mailbox(predecessor_id);
 202   }
 203
 204   INFO1("My new predecessor is %d", predecessor_id);
 205 }
 206
 207 /**
 208  * \brief Node Function
 209  * Arguments:
 210  * - my id
 211  * - the id of a guy I know in the system (except for the first node)
 212  * - the time to sleep before I join (except for the first node)
 213  */
 214 int node(int argc, char *argv[])
 215 {
 216   double init_time = MSG_get_clock();
 217   m_task_t task = NULL;
 218   msg_comm_t comm_send = NULL;
 219   int i;
 220   int index;
 221   double deadline;
 222   double next_stabilize_date = init_time + 10;
 223   double next_fix_fingers_date = init_time + 10;
 224   double next_check_predecessor_date = init_time + 10;
 225
 226   xbt_assert0(argc == 3 || argc == 5, "Wrong number of arguments for this node");
 227
 228   // initialize my node
 229   s_node_t node = {0};
 230   node.id = atoi(argv[1]);
 231   node.mailbox = get_mailbox(node.id);
 232   node.next_finger_to_fix = 0;
 233   node.comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
 234
 235   for (i = 0; i < NB_BITS; i++) {
 236     set_finger(&node, i, node.id);
 237   }
 238
 239   if (argc == 3) { // first ring
 240     deadline = atof(argv[2]);
 241     create(&node);
 242   }
 243   else {
 244     int known_id = atoi(argv[2]);
 245     double sleep_time = atof(argv[3]);
 246     deadline = atof(argv[4]);
 247
 248     // sleep before starting
 249     INFO1("Let's sleep during %f", sleep_time);
 250     MSG_process_sleep(sleep_time);
 251     INFO0("Hey! Let's join the system.");
 252
 253     join(&node, known_id);
 254   }
 255
 256   while (MSG_get_clock() < init_time + deadline) {
 257
 258     if (node.comm_receive == NULL) {
 259       task = NULL;
 260       node.comm_receive = MSG_task_irecv(&task, node.mailbox);
 261       // FIXME: do not make MSG_task_irecv() calls from several functions
 262     }
 263
 264     if (!MSG_comm_test(node.comm_receive)) {
 265
 266       // no task was received: make some periodic calls
 267       if (MSG_get_clock() >= next_stabilize_date) {
 268         stabilize(&node);
 269         next_stabilize_date = MSG_get_clock() + 10;
 270       }
 271       else if (MSG_get_clock() >= next_fix_fingers_date) {
 272         fix_fingers(&node);
 273         next_fix_fingers_date = MSG_get_clock() + 10;
 274       }
 275       else if (MSG_get_clock() >= next_check_predecessor_date) {
 276         check_predecessor(&node);
 277         next_check_predecessor_date = MSG_get_clock() + 10;
 278       }
 279       else {
 280         // nothing to do: sleep for a while
 281         MSG_process_sleep(5);
 282       }
 283     }
 284     else {
 285       // a transfer has occured
 286
 287       MSG_error_t status = MSG_comm_get_status(node.comm_receive);
 288       MSG_comm_destroy(node.comm_receive);
 289       node.comm_receive = NULL;
 290
 291       if (status != MSG_OK) {
 292         INFO0("Failed to receive a task. Nevermind.");
 293       }
 294       else {
 295         // the task was successfully received
 296         handle_task(&node, task);
 297       }
 298     }
 299
 300     // see if some communications are finished
 301     while ((index = MSG_comm_testany(node.comms)) != -1) {
 302       comm_send = xbt_dynar_get_as(node.comms, index, msg_comm_t);
 303       xbt_dynar_remove_at(node.comms, index, &comm_send);
 304       DEBUG3("Communication %p is finished with status %d, dynar size is now %lu",
 305             comm_send, MSG_comm_get_status(comm_send), xbt_dynar_length(node.comms));
 306       MSG_comm_destroy(comm_send);
 307     }
 308   }
 309
 310   // clean unfinished comms sent
 311   unsigned int cursor;
 312   xbt_dynar_foreach(node.comms, cursor, comm_send) {
 313     task = MSG_comm_get_task(comm_send);
 314     MSG_task_cancel(task);
 315     xbt_free(MSG_task_get_data(task));
 316     MSG_task_destroy(task);
 317     MSG_comm_destroy(comm_send);
 318   }
 319
 320   // leave the ring and stop the simulation
 321   leave(&node);
 322   xbt_dynar_free(&node.comms);
 323   xbt_free(node.mailbox);
 324   xbt_free(node.pred_mailbox);
 325   for (i = 0; i < NB_BITS - 1; i++) {
 326     xbt_free(node.fingers[i].mailbox);
 327   }
 328   return 0;
 329 }
 330
 331 /**
 332  * \brief This function is called when the current node receives a task.
 333  * \param node the current node
 334  * \param task the task to handle (don't touch it then:
 335  * it will be destroyed, reused or forwarded)
 336  */
 337 static void handle_task(node_t node, m_task_t task) {
 338
 339   msg_comm_t comm = NULL;
 340   char* mailbox = NULL;
 341   const char* task_name = MSG_task_get_name(task);
 342   task_data_t task_data = (task_data_t) MSG_task_get_data(task);
 343
 344   if (!strcmp(task_name, "Find Successor")) {
 345     INFO2("Receiving a 'Find Successor' request from %s for id %d",
 346         task_data->issuer_host_name, task_data->request_id);
 347     // is my successor the successor?
 348     if (is_in_interval(task_data->request_id, node->id + 1, node->fingers[0].id)) {
 349       task_data->answer_id = node->fingers[0].id;
 350       MSG_task_set_name(task, "Find Successor Answer");
 351       comm = MSG_task_isend(task, task_data->answer_to);
 352       xbt_dynar_push(node->comms, &comm);
 353       INFO3("Sending back a 'Find Successor Answer' to %s: the successor of %d is %d",
 354             task_data->issuer_host_name,
 355             task_data->request_id, task_data->answer_id);
 356     }
 357     else {
 358       // otherwise, forward the request to the closest preceding finger in my table
 359       int closest = closest_preceding_node(node, task_data->request_id);
 360       INFO2("Forwarding the 'Find Successor' request for id %d to my closest preceding finger %d",
 361           task_data->request_id, closest);
 362       mailbox = get_mailbox(closest);
 363       comm = MSG_task_isend(task, mailbox);
 364       xbt_dynar_push(node->comms, &comm);
 365       xbt_free(mailbox);
 366     }
 367   }
 368
 369   else if (!strcmp(task_name, "Get Predecessor")) {
 370     INFO1("Receiving a 'Get Predecessor' request from %s", task_data->issuer_host_name);
 371     task_data->answer_id = node->pred_id;
 372     MSG_task_set_name(task, "Get Predecessor Answer");
 373     comm = MSG_task_isend(task, task_data->answer_to);
 374     xbt_dynar_push(node->comms, &comm);
 375     INFO3("Sending back a 'Get Predecessor Answer' to %s via mailbox '%s': my predecessor is %d",
 376           task_data->issuer_host_name,
 377           task_data->answer_to, task_data->answer_id);
 378   }
 379
 380   else if (!strcmp(task_name, "Notify")) {
 381     // someone is telling me that he may be my new predecessor
 382     INFO1("Receiving a 'Notify' request from %s", task_data->issuer_host_name);
 383     notify(node, task_data->request_id);
 384     xbt_free(task_data);
 385     MSG_task_destroy(task);
 386   }
 387
 388   else if (!strcmp(task_name, "Predecessor Leaving")) {
 389     // my predecessor is about to quit
 390     INFO1("Receiving a 'Predecessor Leaving' message from %s", task_data->issuer_host_name);
 391     // modify my predecessor
 392     set_predecessor(node, task_data->pred_id);
 393     xbt_free(task_data);
 394     MSG_task_destroy(task);
 395     /*TODO :
 396     >> notify my new predecessor
 397     >> send a notify_predecessors !!
 398      */
 399   }
 400
 401   else if (!strcmp(task_name, "Successor Leaving")) {
 402     // my successor is about to quit
 403     INFO1("Receiving a 'Successor Leaving' message from %s", task_data->issuer_host_name);
 404     // modify my successor FIXME : this should be implicit ?
 405     set_finger(node, 0, task_data->successor_id);
 406     xbt_free(task_data);
 407     MSG_task_destroy(task);
 408     /* TODO
 409     >> notify my new successor
 410     >> update my table & predecessors table */
 411   }
 412 }
 413
 414 /**
 415  * \brief Initializes the current node as the first one of the system.
 416  * \param node the current node
 417  */
 418 static void create(node_t node)
 419 {
 420   INFO0("Create a new Chord ring...");
 421   set_predecessor(node, -1); // -1 means that I have no predecessor
 422   print_finger_table(node);
 423 }
 424
 425 /**
 426  * \brief Makes the current node join the ring, knowing the id of a node
 427  * already in the ring
 428  * \param node the current node
 429  * \param known_id id of a node already in the ring
 430  */
 431 static void join(node_t node, int known_id)
 432 {
 433   INFO2("Joining the ring with id %d, knowing node %d", node->id, known_id);
 434   set_predecessor(node, -1); // no predecessor (yet)
 435
 436   int successor_id;
 437   do {
 438     successor_id = remote_find_successor(node, known_id, node->id);
 439     if (successor_id == -1) {
 440       INFO0("I really want to join the ring. Let's try again.");
 441     }
 442   } while (successor_id == -1);
 443
 444   set_finger(node, 0, successor_id);
 445   print_finger_table(node);
 446 }
 447
 448 /**
 449  * \brief Makes the current node quit the system
 450  * \param node the current node
 451  */
 452 static void leave(node_t node)
 453 {
 454   INFO0("Well Guys! I Think it's time for me to quit ;)");
 455   quit_notify(node, 1);  // notify to my successor ( >>> 1 );
 456   quit_notify(node, -1); // notify my predecessor  ( >>> -1);
 457   // TODO ...
 458 }
 459
 460 /*
 461  * \brief Notifies the successor or the predecessor of the current node
 462  * of the departure
 463  * \param node the current node
 464  * \param to 1 to notify the successor, -1 to notify the predecessor
 465  * FIXME: notify both nodes with only one call
 466  */
 467 static void quit_notify(node_t node, int to)
 468 {
 469   /* TODO
 470   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 471   req_data->request_id = node->id;
 472   req_data->successor_id = node->fingers[0].id;
 473   req_data->pred_id = node->pred_id;
 474   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 475   const char* task_name = NULL;
 476   const char* to_mailbox = NULL;
 477   if (to == 1) {    // notify my successor
 478     to_mailbox = node->fingers[0].mailbox;
 479     INFO2("Telling my Successor %d about my departure via mailbox %s",
 480           node->fingers[0].id, to_mailbox);
 481     task_name = "Predecessor Leaving";
 482   }
 483   else if (to == -1) {    // notify my predecessor
 484
 485     if (node->pred_id == -1) {
 486       return;
 487     }
 488
 489     to_mailbox = node->pred_mailbox;
 490     INFO2("Telling my Predecessor %d about my departure via mailbox %s",
 491           node->pred_id, to_mailbox);
 492     task_name = "Successor Leaving";
 493   }
 494   m_task_t task = MSG_task_create(task_name, COMP_SIZE, COMM_SIZE, req_data);
 495   //char* mailbox = get_mailbox(to_mailbox);
 496   msg_comm_t comm = MSG_task_isend(task, to_mailbox);
 497   xbt_dynar_push(node->comms, &comm);
 498   */
 499 }
 500
 501 /**
 502  * \brief Makes the current node find the successor node of an id.
 503  * \param node the current node
 504  * \param id the id to find
 505  * \return the id of the successor node, or -1 if the request failed
 506  */
 507 static int find_successor(node_t node, int id)
 508 {
 509   // is my successor the successor?
 510   if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
 511     return node->fingers[0].id;
 512   }
 513
 514   // otherwise, ask the closest preceding finger in my table
 515   int closest = closest_preceding_node(node, id);
 516   return remote_find_successor(node, closest, id);
 517 }
 518
 519 /**
 520  * \brief Asks another node the successor node of an id.
 521  * \param node the current node
 522  * \param ask_to the node to ask to
 523  * \param id the id to find
 524  * \return the id of the successor node, or -1 if the request failed
 525  */
 526 static int remote_find_successor(node_t node, int ask_to, int id)
 527 {
 528   int successor = -1;
 529   int stop = 0;
 530   char* mailbox = get_mailbox(ask_to);
 531   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 532   req_data->request_id = id;
 533   req_data->answer_to =  node->mailbox;
 534   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 535
 536   // send a "Find Successor" request to ask_to_id
 537   INFO2("Sending a 'Find Successor' request to %d for id %d", ask_to, id);
 538   m_task_t task = MSG_task_create("Find Successor", COMP_SIZE, COMM_SIZE, req_data);
 539   MSG_error_t res = MSG_task_send_with_timeout(task, mailbox, 50);
 540
 541   if (res != MSG_OK) {
 542     INFO2("Failed to send the 'Find Successor' request to %d for id %d", ask_to, id);
 543     MSG_task_destroy(task);
 544     xbt_free(req_data);
 545   }
 546   else {
 547
 548     // receive the answer
 549     INFO2("Sent a 'Find Successor' request to %d for key %d, waiting for the answer", ask_to, id);
 550
 551     do {
 552       if (node->comm_receive == NULL) {
 553         task = NULL;
 554         node->comm_receive = MSG_task_irecv(&task, node->mailbox);
 555       }
 556
 557       res = MSG_comm_wait(node->comm_receive, 50);
 558
 559       if (res != MSG_OK) {
 560         INFO1("Failed to receive the answer to my 'Find Successor' request: %d", res);
 561         stop = 1;
 562 //      MSG_comm_destroy(node->comm_receive);
 563       }
 564       else {
 565         task = MSG_comm_get_task(node->comm_receive);
 566         INFO1("Received a task '%s'", MSG_task_get_name(task));
 567
 568         if (strcmp(MSG_task_get_name(task), "Find Successor Answer")) {
 569           handle_task(node, task);
 570         }
 571         else {
 572           INFO2("This is the answer to my 'Find Successor' request: the successor of key %d is %d", id, successor);
 573           task_data_t ans_data = MSG_task_get_data(task);
 574           successor = ans_data->answer_id;
 575           stop = 1;
 576           MSG_task_destroy(task);
 577           xbt_free(req_data);
 578         }
 579       }
 580       node->comm_receive = NULL;
 581     } while (!stop);
 582   }
 583
 584   xbt_free(mailbox);
 585   return successor;
 586 }
 587
 588 /**
 589  * \brief Asks another node its predecessor.
 590  * \param node the current node
 591  * \param ask_to the node to ask to
 592  * \return the id of its predecessor node, or -1 if the request failed
 593  * (or if the node does not know its predecessor)
 594  */
 595 static int remote_get_predecessor(node_t node, int ask_to)
 596 {
 597   int predecessor_id = -1;
 598   int stop = 0;
 599   char* mailbox = get_mailbox(ask_to);
 600   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 601   req_data->answer_to = node->mailbox;
 602   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 603
 604   // send a "Get Predecessor" request to ask_to_id
 605   INFO1("Sending a 'Get Predecessor' request to %d", ask_to);
 606   m_task_t task = MSG_task_create("Get Predecessor", COMP_SIZE, COMM_SIZE, req_data);
 607   MSG_error_t res = MSG_task_send_with_timeout(task, mailbox, 50);
 608
 609   if (res != MSG_OK) {
 610     INFO1("Failed to send the 'Get Predecessor' request to %d", ask_to);
 611     MSG_task_destroy(task);
 612     xbt_free(req_data);
 613   }
 614   else {
 615
 616     // receive the answer
 617     INFO2("Sent 'Get Predecessor' request to %d, waiting for the answer on my mailbox '%s'", ask_to, req_data->answer_to);
 618
 619     do {
 620       if (node->comm_receive == NULL) { // FIXME simplify this
 621         task = NULL;
 622         node->comm_receive = MSG_task_irecv(&task, node->mailbox);
 623       }
 624
 625       res = MSG_comm_wait(node->comm_receive, 50);
 626
 627       if (res != MSG_OK) {
 628         INFO1("Failed to receive the answer to my 'Get Predecessor' request: %d", res);
 629         stop = 1;
 630 //        MSG_comm_destroy(node->comm_receive);
 631       }
 632       else {
 633         task = MSG_comm_get_task(node->comm_receive);
 634         INFO1("Received a task '%s'", MSG_task_get_name(task));
 635
 636         if (strcmp(MSG_task_get_name(task), "Get Predecessor Answer")) {
 637           handle_task(node, task);
 638         }
 639         else {
 640           INFO2("This is the answer to my 'Get Predecessor' request: the predecessor of node %d is %d", ask_to, predecessor_id);
 641           task_data_t ans_data = MSG_task_get_data(task);
 642           predecessor_id = ans_data->answer_id;
 643           stop = 1;
 644           MSG_task_destroy(task);
 645           xbt_free(req_data);
 646         }
 647       }
 648       node->comm_receive = NULL;
 649     } while (!stop);
 650   }
 651
 652   xbt_free(mailbox);
 653   return predecessor_id;
 654 }
 655
 656 /**
 657  * \brief Returns the closest preceding finger of an id
 658  * with respect to the finger table of the current node.
 659  * \param node the current node
 660  * \param id the id to find
 661  * \return the closest preceding finger of that id
 662  */
 663 int closest_preceding_node(node_t node, int id)
 664 {
 665   int i;
 666   for (i = NB_BITS - 1; i >= 0; i--) {
 667     if (is_in_interval(node->fingers[i].id, node->id + 1, id - 1)) {
 668       return node->fingers[i].id;
 669     }
 670   }
 671   return node->id;
 672 }
 673
 674 /**
 675  * \brief This function is called periodically. It checks the immediate
 676  * successor of the current node.
 677  * \param node the current node
 678  */
 679 static void stabilize(node_t node)
 680 {
 681   INFO0("Stabilizing node");
 682
 683   // get the predecessor of my immediate successor
 684   int candidate_id;
 685   int successor_id = node->fingers[0].id;
 686   if (successor_id != node->id) {
 687     candidate_id = remote_get_predecessor(node, successor_id);
 688   }
 689   else {
 690     candidate_id = node->pred_id;
 691   }
 692
 693   // this node is a candidate to become my new successor
 694   if (candidate_id != -1
 695       && is_in_interval(candidate_id, node->id + 1, successor_id - 1)) {
 696     set_finger(node, 0, candidate_id);
 697   }
 698   if (successor_id != node->id) {
 699     remote_notify(node, successor_id, node->id);
 700   }
 701 }
 702
 703 /**
 704  * \brief Notifies the current node that its predecessor may have changed.
 705  * \param node the current node
 706  * \param candidate_id the possible new predecessor
 707  */
 708 static void notify(node_t node, int predecessor_candidate_id) {
 709
 710   if (node->pred_id == -1
 711     || is_in_interval(predecessor_candidate_id, node->pred_id + 1, node->id - 1)) {
 712
 713     set_predecessor(node, predecessor_candidate_id);
 714     print_finger_table(node);
 715   }
 716   else {
 717     INFO1("I don't have to change my predecessor to %d", predecessor_candidate_id);
 718   }
 719 }
 720
 721 /**
 722  * \brief Notifies a remote node that its predecessor may have changed.
 723  * \param node the current node
 724  * \param notify_id id of the node to notify
 725  * \param candidate_id the possible new predecessor
 726  */
 727 static void remote_notify(node_t node, int notify_id, int predecessor_candidate_id) {
 728
 729   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 730   req_data->request_id = predecessor_candidate_id;
 731   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 732
 733   // send a "Notify" request to notify_id
 734   INFO1("Sending a 'Notify' request to %d", notify_id);
 735   m_task_t task = MSG_task_create("Notify", COMP_SIZE, COMM_SIZE, req_data);
 736   char* mailbox = get_mailbox(notify_id);
 737   msg_comm_t comm = MSG_task_isend(task, mailbox);
 738   xbt_dynar_push(node->comms, &comm);
 739   xbt_free(mailbox);
 740 }
 741
 742 /**
 743  * \brief This function is called periodically.
 744  * It refreshes the finger table of the current node.
 745  * \param node the current node
 746  */
 747 static void fix_fingers(node_t node) {
 748
 749   INFO0("Fixing fingers");
 750   int i = node->next_finger_to_fix;
 751   int id = find_successor(node, node->id + powers2[i]);
 752   if (id != -1) {
 753
 754     if (id != node->fingers[i].id) {
 755       set_finger(node, i, id);
 756       print_finger_table(node);
 757     }
 758     node->next_finger_to_fix = (i + 1) % NB_BITS;
 759   }
 760 }
 761
 762 /**
 763  * \brief This function is called periodically.
 764  * It checks whether the predecessor has failed
 765  * \param node the current node
 766  */
 767 static void check_predecessor(node_t node)
 768 {
 769   INFO0("Checking whether my predecessor is alive");
 770   // TODO
 771 }
 772
 773 /**
 774  * \brief Main function.
 775  */
 776 int main(int argc, char *argv[])
 777 {
 778   if (argc < 3) {
 779     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 780     printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
 781     exit(1);
 782   }
 783
 784   const char* platform_file = argv[1];
 785   const char* application_file = argv[2];
 786
 787   chord_initialize();
 788
 789   MSG_global_init(&argc, argv);
 790   MSG_set_channel_number(0);
 791   MSG_create_environment(platform_file);
 792
 793   MSG_function_register("node", node);
 794   MSG_launch_application(application_file);
 795
 796   MSG_error_t res = MSG_main();
 797   INFO1("Simulation time: %g", MSG_get_clock());
 798
 799   MSG_clean();
 800
 801   if (res == MSG_OK)
 802     return 0;
 803   else
 804     return 1;
 805 }