Logo AND Algorithmique Numérique Distribuée

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