Logo AND Algorithmique Numérique Distribuée

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