Logo AND Algorithmique Numérique Distribuée

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