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