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->pred_id;
 331   get_mailbox(node->id, req_data_s->answer_to);
 332   req_data_s->issuer_host_name = MSG_host_get_name(MSG_host_self());
 333
 334   msg_task_t task_sent_s = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data_s);
 335   XBT_DEBUG("Sending a 'SUCCESSOR_LEAVING' to my predecessor %d",node->pred_id);
 336   if (MSG_task_send_with_timeout(task_sent_s, mailbox, timeout)== MSG_TIMEOUT) {
 337     XBT_DEBUG("Timeout expired when sending a 'SUCCESSOR_LEAVING' to my predecessor %d", node->pred_id);
 338     task_free(task_sent_s);
 339   }
 340 }
 341
 342 /* Makes the current node find the successor node of an id.
 343  *
 344  * \param node the current node
 345  * \param id the id to find
 346  * \return the id of the successor node, or -1 if the request failed
 347  */
 348 int find_successor(node_t node, int id)
 349 {
 350   // is my successor the successor?
 351   if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
 352     return node->fingers[0].id;
 353   }
 354
 355   // otherwise, ask the closest preceding finger in my table
 356   int closest = closest_preceding_node(node, id);
 357   return remote_find_successor(node, closest, id);
 358 }
 359
 360 /* \brief Asks another node the successor node of an id.
 361  *
 362  * \param node the current node
 363  * \param ask_to the node to ask to
 364  * \param id the id to find
 365  * \return the id of the successor node, or -1 if the request failed
 366  */
 367 int remote_find_successor(node_t node, int ask_to, int id)
 368 {
 369   int successor = -1;
 370   int stop = 0;
 371   char mailbox[MAILBOX_NAME_SIZE];
 372   get_mailbox(ask_to, mailbox);
 373   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 374   req_data->type = TASK_FIND_SUCCESSOR;
 375   req_data->request_id = id;
 376   get_mailbox(node->id, req_data->answer_to);
 377   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 378
 379   // send a "Find Successor" request to ask_to_id
 380   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 381   XBT_DEBUG("Sending a 'Find Successor' request (task %p) to %d for id %d", task_sent, ask_to, id);
 382   msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 383
 384   if (res != MSG_OK) {
 385     XBT_DEBUG("Failed to send the 'Find Successor' request (task %p) to %d for id %d", task_sent, ask_to, id);
 386     task_free(task_sent);
 387   } else {
 388     // receive the answer
 389     XBT_DEBUG("Sent a 'Find Successor' request (task %p) to %d for key %d, waiting for the answer",
 390         task_sent, ask_to, id);
 391     do {
 392       if (node->comm_receive == NULL) {
 393         msg_task_t task_received = NULL;
 394         node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
 395       }
 396
 397       res = MSG_comm_wait(node->comm_receive, timeout);
 398
 399       if (res != MSG_OK) {
 400         XBT_DEBUG("Failed to receive the answer to my 'Find Successor' request (task %p): %d",
 401                   task_sent, (int)res);
 402         stop = 1;
 403         MSG_comm_destroy(node->comm_receive);
 404         node->comm_receive = NULL;
 405       }
 406       else {
 407         msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
 408         XBT_DEBUG("Received a task (%p)", task_received);
 409         task_data_t ans_data = MSG_task_get_data(task_received);
 410
 411   // Once upon a time, our code assumed that here, task_received != task_sent all the time
 412   //
 413   // This assumption is wrong (as messages from differing round can interleave), leading to a bug in our code.
 414   // We failed to find this bug directly, as it only occurred on large platforms, leading to hardly usable traces.
 415   // Instead, we used the model-checker to track down the issue by adding the following test here in the code:
 416   //   if (MC_is_active()) {
 417   //      MC_assert(task_received == task_sent);
 418         //   }
 419   // That explained the bug in a snap, with a very cool example and everything.
 420   //
 421   // This MC_assert is now deactivated as the case is now properly handled in our code and we don't want the
 422   //   MC to fail any further under that condition, but this comment is here to as a memorial for this first
 423   //   brilliant victory of the model-checking in the SimGrid community :)
 424
 425         if (task_received != task_sent ||
 426             ans_data->type != TASK_FIND_SUCCESSOR_ANSWER) {
 427           // this is not the expected answer
 428           MSG_comm_destroy(node->comm_receive);
 429           node->comm_receive = NULL;
 430           handle_task(node, task_received);
 431         }
 432         else {
 433           // this is our answer
 434           XBT_DEBUG("Received the answer to my 'Find Successor' request for id %d (task %p): the successor of key %d is %d",
 435               ans_data->request_id, task_received, id, ans_data->answer_id);
 436           successor = ans_data->answer_id;
 437           stop = 1;
 438           MSG_comm_destroy(node->comm_receive);
 439           node->comm_receive = NULL;
 440           task_free(task_received);
 441         }
 442       }
 443     } while (!stop);
 444   }
 445
 446   return successor;
 447 }
 448
 449 /* Asks its predecessor to a remote node
 450  *
 451  * \param node the current node
 452  * \param ask_to the node to ask to
 453  * \return the id of its predecessor node, or -1 if the request failed
 454  * (or if the node does not know its predecessor)
 455  */
 456 int remote_get_predecessor(node_t node, int ask_to)
 457 {
 458   int predecessor_id = -1;
 459   int stop = 0;
 460   char mailbox[MAILBOX_NAME_SIZE];
 461   get_mailbox(ask_to, mailbox);
 462   task_data_t req_data = xbt_new0(s_task_data_t, 1);
 463   req_data->type = TASK_GET_PREDECESSOR;
 464   get_mailbox(node->id, req_data->answer_to);
 465   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 466
 467   // send a "Get Predecessor" request to ask_to_id
 468   XBT_DEBUG("Sending a 'Get Predecessor' request to %d", ask_to);
 469   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 470   msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 471
 472   if (res != MSG_OK) {
 473     XBT_DEBUG("Failed to send the 'Get Predecessor' request (task %p) to %d",
 474         task_sent, ask_to);
 475     task_free(task_sent);
 476   }
 477   else {
 478
 479     // receive the answer
 480     XBT_DEBUG("Sent 'Get Predecessor' request (task %p) to %d, waiting for the answer on my mailbox '%s'",
 481         task_sent, ask_to, req_data->answer_to);
 482
 483     do {
 484       if (node->comm_receive == NULL) { // FIXME simplify this
 485         msg_task_t task_received = NULL;
 486         node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
 487       }
 488
 489       res = MSG_comm_wait(node->comm_receive, timeout);
 490
 491       if (res != MSG_OK) {
 492         XBT_DEBUG("Failed to receive the answer to my 'Get Predecessor' request (task %p): %d",
 493             task_sent, (int)res);
 494         stop = 1;
 495         MSG_comm_destroy(node->comm_receive);
 496         node->comm_receive = NULL;
 497       }
 498       else {
 499         msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
 500         task_data_t ans_data = MSG_task_get_data(task_received);
 501
 502         /*if (MC_is_active()) {
 503           MC_assert(task_received == task_sent);
 504           }*/
 505
 506         if (task_received != task_sent ||
 507             ans_data->type != TASK_GET_PREDECESSOR_ANSWER) {
 508           MSG_comm_destroy(node->comm_receive);
 509           node->comm_receive = NULL;
 510           handle_task(node, task_received);
 511         }
 512         else {
 513           XBT_DEBUG("Received the answer to my 'Get Predecessor' request (task %p): the predecessor of node %d is %d",
 514               task_received, ask_to, ans_data->answer_id);
 515           predecessor_id = ans_data->answer_id;
 516           stop = 1;
 517           MSG_comm_destroy(node->comm_receive);
 518           node->comm_receive = NULL;
 519           task_free(task_received);
 520         }
 521       }
 522     } while (!stop);
 523   }
 524
 525   return predecessor_id;
 526 }
 527
 528 /* Returns the closest preceding finger of an id with respect to the finger table of the current node.
 529  *
 530  * \param node the current node
 531  * \param id the id to find
 532  * \return the closest preceding finger of that id
 533  */
 534 int closest_preceding_node(node_t node, int id)
 535 {
 536   int i;
 537   for (i = nb_bits - 1; i >= 0; i--) {
 538     if (is_in_interval(node->fingers[i].id, node->id + 1, id - 1)) {
 539       return node->fingers[i].id;
 540     }
 541   }
 542   return node->id;
 543 }
 544
 545 /* This function is called periodically. It checks the immediate successor of the current node. */
 546 void stabilize(node_t node)
 547 {
 548   XBT_DEBUG("Stabilizing node");
 549
 550   // get the predecessor of my immediate successor
 551   int candidate_id;
 552   int successor_id = node->fingers[0].id;
 553   if (successor_id != node->id) {
 554     candidate_id = remote_get_predecessor(node, successor_id);
 555   }
 556   else {
 557     candidate_id = node->pred_id;
 558   }
 559
 560   // this node is a candidate to become my new successor
 561   if (candidate_id != -1
 562       && is_in_interval(candidate_id, node->id + 1, successor_id - 1)) {
 563     set_finger(node, 0, candidate_id);
 564   }
 565   if (successor_id != node->id) {
 566     remote_notify(successor_id, node->id);
 567   }
 568 }
 569
 570 /* Notifies the current node that its predecessor may have changed. */
 571 void notify(node_t node, int predecessor_candidate_id) {
 572
 573   if (node->pred_id == -1
 574     || is_in_interval(predecessor_candidate_id, node->pred_id + 1, node->id - 1)) {
 575
 576     set_predecessor(node, predecessor_candidate_id);
 577     print_finger_table(node);
 578   }
 579   else {
 580     XBT_DEBUG("I don't have to change my predecessor to %d", predecessor_candidate_id);
 581   }
 582 }
 583
 584 /* Notifies a remote node that its predecessor may have changed. */
 585 void remote_notify(int notify_id, int predecessor_candidate_id) {
 586
 587       task_data_t req_data = xbt_new0(s_task_data_t, 1);
 588       req_data->type = TASK_NOTIFY;
 589       req_data->request_id = predecessor_candidate_id;
 590       req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 591
 592       // send a "Notify" request to notify_id
 593       msg_task_t task = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 594       XBT_DEBUG("Sending a 'Notify' request (task %p) to %d", task, notify_id);
 595       char mailbox[MAILBOX_NAME_SIZE];
 596       get_mailbox(notify_id, mailbox);
 597       MSG_task_dsend(task, mailbox, task_free);
 598     }
 599
 600 /* refreshes the finger table of the current node (called periodically) */
 601   void fix_fingers(node_t node) {
 602
 603   XBT_DEBUG("Fixing fingers");
 604   int i = node->next_finger_to_fix;
 605   int id = find_successor(node, node->id + powers2[i]);
 606   if (id != -1) {
 607
 608     if (id != node->fingers[i].id) {
 609       set_finger(node, i, id);
 610       print_finger_table(node);
 611     }
 612     node->next_finger_to_fix = (i + 1) % nb_bits;
 613   }
 614 }
 615
 616 /* checks whether the predecessor has failed (called periodically) */
 617 void check_predecessor(node_t node)
 618 {
 619   XBT_DEBUG("Checking whether my predecessor is alive");
 620
 621   if(node->pred_id == -1)
 622     return;
 623
 624   int stop = 0;
 625
 626   char mailbox[MAILBOX_NAME_SIZE];
 627   get_mailbox(node->pred_id, mailbox);
 628   task_data_t req_data = xbt_new0(s_task_data_t,1);
 629   req_data->type = TASK_PREDECESSOR_ALIVE;
 630   req_data->request_id = node->pred_id;
 631   get_mailbox(node->id, req_data->answer_to);
 632   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 633
 634   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 635   XBT_DEBUG("Sending a 'Predecessor Alive' request to my predecessor %d", node->pred_id);
 636
 637   msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 638
 639   if (res != MSG_OK) {
 640     XBT_DEBUG("Failed to send the 'Predecessor Alive' request (task %p) to %d", task_sent, node->pred_id);
 641     task_free(task_sent);
 642   } else {
 643     // receive the answer
 644     XBT_DEBUG("Sent 'Predecessor Alive' request (task %p) to %d, waiting for the answer on my mailbox '%s'",
 645               task_sent, node->pred_id, req_data->answer_to);
 646
 647     do {
 648       if (node->comm_receive == NULL) { // FIXME simplify this
 649         msg_task_t task_received = NULL;
 650         node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
 651       }
 652
 653       res = MSG_comm_wait(node->comm_receive, timeout);
 654
 655       if (res != MSG_OK) {
 656         XBT_DEBUG("Failed to receive the answer to my 'Predecessor Alive' request (task %p): %d",
 657                   task_sent, (int)res);
 658         stop = 1;
 659         MSG_comm_destroy(node->comm_receive);
 660         node->comm_receive = NULL;
 661         node->pred_id = -1;
 662       } else {
 663         msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
 664         if (task_received != task_sent) {
 665           MSG_comm_destroy(node->comm_receive);
 666           node->comm_receive = NULL;
 667           handle_task(node, task_received);
 668         }else{
 669           XBT_DEBUG("Received the answer to my 'Predecessor Alive' request (task %p) : my predecessor %d is alive",
 670                     task_received, node->pred_id);
 671           stop = 1;
 672           MSG_comm_destroy(node->comm_receive);
 673           node->comm_receive = NULL;
 674           task_free(task_received);
 675         }
 676       }
 677     } while (!stop);
 678   }
 679 }
 680
 681 /* Performs a find successor request to a random id */
 682 void random_lookup(node_t node)
 683 {
 684   int random_index = RngStream_RandInt (node->stream, 0, nb_bits - 1);
 685   int random_id = node->fingers[random_index].id;
 686   XBT_DEBUG("Making a lookup request for id %d", random_id);
 687   int res = find_successor(node, random_id);
 688   XBT_DEBUG("The successor of node %d is %d", random_id, res);
 689 }
 690
 691 static int node(int argc, char *argv[])
 692 {
 693   /* Reduce the run size for the MC */
 694   if(MC_is_active() || MC_record_replay_is_active()){
 695     periodic_stabilize_delay = 8;
 696     periodic_fix_fingers_delay = 8;
 697     periodic_check_predecessor_delay = 8;
 698   }
 699
 700   double init_time = MSG_get_clock();
 701   msg_task_t task_received = NULL;
 702   int i;
 703   int join_success = 0;
 704   double deadline;
 705   double next_stabilize_date = init_time + periodic_stabilize_delay;
 706   double next_fix_fingers_date = init_time + periodic_fix_fingers_delay;
 707   double next_check_predecessor_date = init_time + periodic_check_predecessor_delay;
 708   double next_lookup_date = init_time + periodic_lookup_delay;
 709
 710   xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
 711
 712   // initialize my node
 713   s_node_t node = {0};
 714   node.id = xbt_str_parse_int(argv[1],"Invalid ID: %s");
 715   node.stream   = (RngStream)MSG_host_get_data(MSG_host_self());
 716   get_mailbox(node.id, node.mailbox);
 717   node.next_finger_to_fix = 0;
 718   node.fingers = xbt_new0(s_finger_t, nb_bits);
 719   node.last_change_date = init_time;
 720
 721   for (i = 0; i < nb_bits; i++) {
 722     node.fingers[i].id = -1;
 723     set_finger(&node, i, node.id);
 724   }
 725
 726   if (argc == 3) { // first ring
 727     deadline = xbt_str_parse_double(argv[2],"Invalid deadline: %s");
 728     create(&node);
 729     join_success = 1;
 730   } else {
 731     int known_id = xbt_str_parse_int(argv[2],"Invalid root ID: %s");
 732     deadline = xbt_str_parse_double(argv[4],"Invalid deadline: %s");
 733
 734     XBT_DEBUG("Hey! Let's join the system.");
 735
 736     join_success = join(&node, known_id);
 737   }
 738
 739   if (join_success) {
 740     double now = MSG_get_clock();
 741     int listen = 0;
 742     int no_op = 0;
 743     while (now < init_time + deadline && now < max_simulation_time) {
 744       if (node.comm_receive == NULL) {
 745         task_received = NULL;
 746         node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
 747         // FIXME: do not make MSG_task_irecv() calls from several functions
 748       }
 749
 750       if (!MSG_comm_test(node.comm_receive)) { // no task was received: make some periodic calls
 751         if(MC_is_active() || MC_record_replay_is_active()){
 752           int sub_protocol = MC_random(0, 4);
 753           if(MC_is_active() && !MC_visited_reduction() && no_op)
 754             MC_cut();
 755           if(listen == 0 && (sub_protocol > 0)){
 756             if(sub_protocol == 1)
 757               stabilize(&node);
 758             else if(sub_protocol == 2)
 759               fix_fingers(&node);
 760             else if(sub_protocol == 3)
 761               check_predecessor(&node);
 762             else
 763               random_lookup(&node);
 764             listen = 1;
 765           } else {
 766             MSG_process_sleep(sleep_delay);
 767             if(!MC_visited_reduction())
 768               no_op = 1;
 769           }
 770         }else{
 771           if (now >= next_stabilize_date) {
 772             stabilize(&node);
 773             next_stabilize_date = MSG_get_clock() + periodic_stabilize_delay;
 774           }else if (now >= next_fix_fingers_date) {
 775             fix_fingers(&node);
 776             next_fix_fingers_date = MSG_get_clock() + periodic_fix_fingers_delay;
 777           }else if (now >= next_check_predecessor_date) {
 778             check_predecessor(&node);
 779             next_check_predecessor_date = MSG_get_clock() + periodic_check_predecessor_delay;
 780           }else if (now >= next_lookup_date) {
 781             random_lookup(&node);
 782             next_lookup_date = MSG_get_clock() + periodic_lookup_delay;
 783           }else {
 784             // nothing to do: sleep for a while
 785             MSG_process_sleep(sleep_delay);
 786           }
 787         }
 788       } else { // a transfer has occurred
 789         msg_error_t status = MSG_comm_get_status(node.comm_receive);
 790         MSG_comm_destroy(node.comm_receive);
 791         node.comm_receive = NULL;
 792
 793         if (status == MSG_OK)
 794           handle_task(&node, task_received);
 795         else
 796           XBT_DEBUG("Failed to receive a task. Nevermind.");
 797       }
 798       now = MSG_get_clock();
 799     }
 800
 801     if (node.comm_receive) {
 802       /* handle last task if any */
 803       if (MSG_comm_wait(node.comm_receive, 0) == MSG_OK)
 804         task_free(task_received);
 805       MSG_comm_destroy(node.comm_receive);
 806       node.comm_receive = NULL;
 807     }
 808
 809     // leave the ring
 810     leave(&node);
 811   }
 812
 813   // stop the simulation
 814   xbt_free(node.fingers);
 815   return 0;
 816 }
 817
 818 int main(int argc, char *argv[])
 819 {
 820   MSG_init(&argc, argv);
 821   xbt_assert(argc > 2, "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
 822                        "\tExample: %s ../msg_platform.xml chord.xml\n", argv[0], argv[0]);
 823
 824   char **options = &argv[1];
 825   while (!strncmp(options[0], "-", 1)) {
 826     int length = strlen("-nb_bits=");
 827     if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
 828       nb_bits = xbt_str_parse_int(options[0] + length, "Invalid nb_bits parameter: %s");
 829       XBT_DEBUG("Set nb_bits to %d", nb_bits);
 830     } else {
 831       length = strlen("-timeout=");
 832       if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
 833         timeout = xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
 834         XBT_DEBUG("Set timeout to %d", timeout);
 835       } else {
 836         xbt_die("Invalid chord option '%s'", options[0]);
 837       }
 838     }
 839     options++;
 840   }
 841
 842   MSG_create_environment(options[0]);
 843
 844   chord_initialize();
 845
 846   MSG_function_register("node", node);
 847   MSG_launch_application(options[1]);
 848
 849   msg_error_t res = MSG_main();
 850   XBT_INFO("Simulated time: %g", MSG_get_clock());
 851
 852   chord_exit();
 853
 854   return res != MSG_OK;
 855 }