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_property_value(host, "stream", (char*)stream, NULL);
  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_property_value(host, "stream");
  56     RngStream_DeleteStream(&stream);
  57   }
  58   xbt_dynar_free(&host_list);
  59
  60   xbt_free(powers2);
  61 }
  62
  63 /* Turns an id into an equivalent id in [0, nb_keys). */
  64 static int normalize(int id)
  65 {
  66   return id % nb_keys;
  67 }
  68
  69 /* Returns whether an id belongs to the interval [start, end].
  70  *
  71  * The parameters are normalized to make sure they are between 0 and nb_keys - 1).
  72  * 1 belongs to [62, 3]
  73  * 1 does not belong to [3, 62]
  74  * 63 belongs to [62, 3]
  75  * 63 does not belong to [3, 62]
  76  * 24 belongs to [21, 29]
  77  * 24 does not belong to [29, 21]
  78  *
  79  * \param id id to check
  80  * \param start lower bound
  81  * \param end upper bound
  82  * \return a non-zero value if id in in [start, end]
  83  */
  84 static int is_in_interval(int id, int start, int end)
  85 {
  86   int i = normalize(id);
  87   int s = normalize(start);
  88   int e = normalize(end);
  89
  90   // make sure end >= start and id >= start
  91   if (e < s) {
  92     e += nb_keys;
  93   }
  94
  95   if (i < s) {
  96     i += nb_keys;
  97   }
  98
  99   return i <= e;
 100 }
 101
 102 /* Gets the mailbox name of a host given its chord id.
 103  * \param node_id id of a node
 104  * \param mailbox pointer to where the mailbox name should be written
 105  * (there must be enough space)
 106  */
 107 static void get_mailbox(int node_id, char* mailbox)
 108 {
 109   snprintf(mailbox, MAILBOX_NAME_SIZE - 1, "%d", node_id);
 110 }
 111
 112 /* Frees the memory used by a task and destroy it */
 113 static void task_free(void* task)
 114 {
 115   // TODO add a parameter data_free_function to MSG_task_create?
 116   if(task != NULL){
 117     xbt_free(MSG_task_get_data(task));
 118     MSG_task_destroy(task);
 119   }
 120 }
 121
 122 /* Displays the finger table of a node. */
 123 static void print_finger_table(node_t node)
 124 {
 125   if (XBT_LOG_ISENABLED(msg_chord, xbt_log_priority_verbose)) {
 126     XBT_VERB("My finger table:");
 127     XBT_VERB("Start | Succ");
 128     for (int i = 0; i < nb_bits; i++) {
 129       XBT_VERB(" %3d  | %3d", (node->id + powers2[i]) % nb_keys, node->fingers[i].id);
 130     }
 131     XBT_VERB("Predecessor: %d", node->pred_id);
 132   }
 133 }
 134
 135 /* Sets a finger of the current node.
 136  *
 137  * \param node the current node
 138  * \param finger_index index of the finger to set (0 to nb_bits - 1)
 139  * \param id the id to set for this finger
 140  */
 141 static void set_finger(node_t node, int finger_index, int id)
 142 {
 143   if (id != node->fingers[finger_index].id) {
 144     node->fingers[finger_index].id = id;
 145     get_mailbox(id, node->fingers[finger_index].mailbox);
 146     node->last_change_date = MSG_get_clock();
 147     XBT_DEBUG("My new finger #%d is %d", finger_index, id);
 148   }
 149 }
 150
 151 /* Sets the predecessor of the current node.
 152  *
 153  * \param node the current node
 154  * \param id the id to predecessor, or -1 to unset the predecessor
 155  */
 156 static void set_predecessor(node_t node, int predecessor_id)
 157 {
 158   if (predecessor_id != node->pred_id) {
 159     node->pred_id = predecessor_id;
 160
 161     if (predecessor_id != -1) {
 162       get_mailbox(predecessor_id, node->pred_mailbox);
 163     }
 164     node->last_change_date = MSG_get_clock();
 165
 166     XBT_DEBUG("My new predecessor is %d", predecessor_id);
 167   }
 168 }
 169
 170 /* Node main Function
 171  *
 172  * Arguments:
 173  * - my id
 174  * - the id of a guy I know in the system (except for the first node)
 175  * - the time to sleep before I join (except for the first node)
 176  */
 177 /* This function is called when the current node receives a task.
 178  *
 179  * \param node the current node
 180  * \param task the task to handle (don't touch it afterward: it will be destroyed, reused or forwarded)
 181  */
 182 static void handle_task(node_t node, msg_task_t task)
 183 {
 184   XBT_DEBUG("Handling task %p", task);
 185   char mailbox[MAILBOX_NAME_SIZE];
 186   task_data_t task_data = (task_data_t) MSG_task_get_data(task);
 187   e_task_type_t type = task_data->type;
 188
 189   switch (type) {
 190   case TASK_FIND_SUCCESSOR:
 191     XBT_DEBUG("Receiving a 'Find Successor' request from %s for id %d",
 192               task_data->issuer_host_name, task_data->request_id);
 193     // is my successor the successor?
 194     if (is_in_interval(task_data->request_id, node->id + 1, node->fingers[0].id)) {
 195       task_data->type = TASK_FIND_SUCCESSOR_ANSWER;
 196       task_data->answer_id = node->fingers[0].id;
 197       XBT_DEBUG("Sending back a 'Find Successor Answer' to %s (mailbox %s): the successor of %d is %d",
 198                 task_data->issuer_host_name, task_data->answer_to, task_data->request_id, task_data->answer_id);
 199       MSG_task_dsend(task, task_data->answer_to, task_free);
 200     } else {
 201       // otherwise, forward the request to the closest preceding finger in my table
 202       int closest = closest_preceding_node(node, task_data->request_id);
 203       XBT_DEBUG("Forwarding the 'Find Successor' request for id %d to my closest preceding finger %d",
 204                 task_data->request_id, closest);
 205       get_mailbox(closest, mailbox);
 206       MSG_task_dsend(task, mailbox, task_free);
 207     }
 208     break;
 209
 210   case TASK_GET_PREDECESSOR:
 211     XBT_DEBUG("Receiving a 'Get Predecessor' request from %s", task_data->issuer_host_name);
 212     task_data->type = TASK_GET_PREDECESSOR_ANSWER;
 213     task_data->answer_id = node->pred_id;
 214     XBT_DEBUG("Sending back a 'Get Predecessor Answer' to %s via mailbox '%s': my predecessor is %d",
 215               task_data->issuer_host_name, task_data->answer_to, task_data->answer_id);
 216     MSG_task_dsend(task, task_data->answer_to, task_free);
 217     break;
 218
 219   case TASK_NOTIFY:
 220     // someone is telling me that he may be my new predecessor
 221     XBT_DEBUG("Receiving a 'Notify' request from %s", task_data->issuer_host_name);
 222     notify(node, task_data->request_id);
 223     task_free(task);
 224     break;
 225
 226   case TASK_PREDECESSOR_LEAVING:
 227     // my predecessor is about to quit
 228     XBT_DEBUG("Receiving a 'Predecessor Leaving' message from %s", task_data->issuer_host_name);
 229     // modify my predecessor
 230     set_predecessor(node, task_data->request_id);
 231     task_free(task);
 232     /*TODO :
 233       >> notify my new predecessor
 234       >> send a notify_predecessors !!
 235     */
 236     break;
 237
 238   case TASK_SUCCESSOR_LEAVING:
 239     // my successor is about to quit
 240     XBT_DEBUG("Receiving a 'Successor Leaving' message from %s", task_data->issuer_host_name);
 241     // modify my successor FIXME : this should be implicit ?
 242     set_finger(node, 0, task_data->request_id);
 243     task_free(task);
 244     /* TODO
 245        >> notify my new successor
 246        >> update my table & predecessors table */
 247     break;
 248
 249   case TASK_FIND_SUCCESSOR_ANSWER:
 250   case TASK_GET_PREDECESSOR_ANSWER:
 251   case TASK_PREDECESSOR_ALIVE_ANSWER:
 252     XBT_DEBUG("Ignoring unexpected task of type %d (%p)", (int)type, task);
 253     task_free(task);
 254     break;
 255
 256   case TASK_PREDECESSOR_ALIVE:
 257     XBT_DEBUG("Receiving a 'Predecessor Alive' request from %s", task_data->issuer_host_name);
 258     task_data->type = TASK_PREDECESSOR_ALIVE_ANSWER;
 259     XBT_DEBUG("Sending back a 'Predecessor Alive Answer' to %s (mailbox %s)",
 260               task_data->issuer_host_name, task_data->answer_to);
 261     MSG_task_dsend(task, task_data->answer_to, task_free);
 262     break;
 263
 264   default:
 265     THROW_IMPOSSIBLE;
 266   }
 267 }
 268
 269 /* Initializes the current node as the first one of the system */
 270 void create(node_t node)
 271 {
 272   XBT_DEBUG("Create a new Chord ring...");
 273   set_predecessor(node, -1); // -1 means that I have no predecessor
 274   print_finger_table(node);
 275 }
 276
 277 /* Makes the current node join the ring, knowing the id of a node already in the ring
 278  *
 279  * \param node the current node
 280  * \param known_id id of a node already in the ring
 281  * \return 1 if the join operation succeeded, 0 otherwise
 282  */
 283 int join(node_t node, int known_id)
 284 {
 285   XBT_INFO("Joining the ring with id %d, knowing node %d", node->id, known_id);
 286   set_predecessor(node, -1); // no predecessor (yet)
 287
 288   int successor_id = remote_find_successor(node, known_id, node->id);
 289   if (successor_id == -1) {
 290     XBT_INFO("Cannot join the ring.");
 291   }
 292   else {
 293     set_finger(node, 0, successor_id);
 294     print_finger_table(node);
 295   }
 296
 297   return successor_id != -1;
 298 }
 299
 300 /* Makes the current node quit the system */
 301 void leave(node_t node)
 302 {
 303   XBT_DEBUG("Well Guys! I Think it's time for me to quit ;)");
 304   quit_notify(node);
 305 }
 306
 307 /* Notifies the successor and the predecessor of the current node before leaving */
 308 void quit_notify(node_t node)
 309 {
 310   char mailbox[MAILBOX_NAME_SIZE];
 311   //send the PREDECESSOR_LEAVING to our successor
 312   task_data_t req_data = xbt_new0(s_task_data_t,1);
 313   req_data->type = TASK_PREDECESSOR_LEAVING;
 314   req_data->request_id = node->pred_id;
 315   get_mailbox(node->id, req_data->answer_to);
 316   req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
 317
 318   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 319   XBT_DEBUG("Sending a 'PREDECESSOR_LEAVING' to my successor %d",node->fingers[0].id);
 320   if (MSG_task_send_with_timeout(task_sent, node->fingers[0].mailbox, timeout)== MSG_TIMEOUT) {
 321     XBT_DEBUG("Timeout expired when sending a 'PREDECESSOR_LEAVING' to my successor %d", node->fingers[0].id);
 322     task_free(task_sent);
 323   }
 324
 325   //send the SUCCESSOR_LEAVING to our predecessor
 326   get_mailbox(node->pred_id, mailbox);
 327   task_data_t req_data_s = xbt_new0(s_task_data_t,1);
 328   req_data_s->type = TASK_SUCCESSOR_LEAVING;
 329   req_data_s->request_id = node->fingers[0].id;
 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 desactivated 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   //   brillant 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(node, 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(node_t node, 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_property_value(MSG_host_self(), "stream");
 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     while (now < init_time + deadline && now < max_simulation_time) {
 742
 743       if (node.comm_receive == NULL) {
 744         task_received = NULL;
 745         node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
 746         // FIXME: do not make MSG_task_irecv() calls from several functions
 747       }
 748
 749       if (!MSG_comm_test(node.comm_receive)) { // no task was received: make some periodic calls
 750         if(MC_is_active() || MC_record_replay_is_active()){
 751           int listen = 0;
 752           int no_op = 0;
 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 }