2 /* Copyright (c) 2010. The SimGrid Team.
3 * All rights reserved. */
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. */
11 #include "xbt/asserts.h"
12 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
13 "Messages specific for this msg example");
21 typedef struct finger {
24 } s_finger_t, *finger_t;
32 s_finger_t fingers[NB_BITS]; // finger table (fingers[0] is my successor)
33 int pred_id; // predecessor id
34 const char* pred_mailbox;
35 xbt_dynar_t comms; // current communications to finish
41 typedef struct task_data {
45 const char* answer_to;
46 const char* issuer_host_name; // used for logging
47 } s_task_data_t, *task_data_t;
50 static int normalize(int id);
51 static int is_in_interval(int id, int start, int end);
52 static char* get_mailbox(int host_id);
53 static void print_finger_table(node_t node);
56 static int node(int argc, char *argv[]);
59 static void initialize_first_node(node_t node);
60 static void initialize_finger_table(node_t data, int known_id);
61 static void join(node_t node, int known_id);
64 static int find_successor(node_t node, int id);
65 static int remote_find_successor(node_t node, int ask_to_id, int id);
66 static int find_predecessor(node_t node, int id);
67 static int remote_find_predecessor(node_t node, int ask_to_id, int id);
68 static int closest_preceding_finger(node_t node, int id);
69 static int remote_closest_preceding_finger(int ask_to_id, int id);
70 static void notify_predecessors(node_t node);
71 static void remote_move_keys(node_t node, int take_from_id);
72 static void update_finger_table(node_t node, int candidate_id, int finger_index);
73 static void remote_update_finger_table(node_t node, int ask_to_id, int candidate_id, int finger_index);
74 static void notify(node_t node, int predecessor_candidate_id);
75 static void remote_notify(node_t node, int notify_to, int predecessor_candidate_id);
76 static void stabilize(node_t node);
79 * \brief Turns an id into an equivalent id in [0, NB_KEYS[
81 * \return the corresponding normalized id
83 static int normalize(int id) {
90 // make sure id < NB_KEYS
97 * \brief Returns whether a id belongs to the interval [start, end].
99 * The parameters are noramlized to make sure they are between 0 and CHORD_NB_KEYS - 1).
100 * 1 belongs to [62, 3]
101 * 1 does not belong to [3, 62]
102 * 63 belongs to [62, 3]
103 * 63 does not belong to [3, 62]
104 * 24 belongs to [21, 29]
105 * 24 does not belong to [29, 21]
107 * \param id id to check
108 * \param start lower bound
109 * \param end upper bound
110 * \return a non-zero value if id in in [start, end]
112 static int is_in_interval(int id, int start, int end) {
115 start = normalize(start);
116 end = normalize(end);
118 // make sure end >= start and id >= start
131 * \brief Gets the mailbox name of a host given its chord id.
132 * \param node_id id of a node
133 * \return the name of its mailbox
134 * FIXME: free the memory
136 static char* get_mailbox(int node_id) {
138 return bprintf("mailbox%d", node_id);
142 * \brief Displays the finger table of a node.
145 static void print_finger_table(node_t node) {
149 INFO0("My finger table:");
150 INFO0("Start | Succ ");
151 for (i = 0; i < NB_BITS; i++) {
152 INFO2(" %4d | %4d ", (node->id + pow) % NB_KEYS, node->fingers[i].id);
155 INFO1("Predecessor: %d", node->pred_id);
159 * \brief Node Function
162 * - the id of a guy I know in the system (except for the first node)
163 * - the time to sleep before I join (except for the first node)
165 int node(int argc, char *argv[])
167 msg_comm_t comm = NULL;
169 xbt_assert0(argc == 2 || argc == 4, "Wrong number of arguments for this node");
171 // initialize my node
173 node.id = atoi(argv[1]);
174 node.mailbox = get_mailbox(node.id);
175 node.comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
177 if (argc == 2) { // first ring
178 initialize_first_node(&node);
181 int known_id = atoi(argv[2]);
182 double sleep_time = atof(argv[3]);
184 // sleep before starting
185 INFO1("Let's sleep >>%f", sleep_time);
186 MSG_process_sleep(sleep_time);
187 INFO0("Hey! Let's join the system.");
189 join(&node, known_id);
196 xbt_dynar_foreach(node.comms, cursor, comm) {
197 if (MSG_comm_test(comm)) { // FIXME: try with MSG_comm_testany instead
198 xbt_dynar_cursor_rm(node.comms, &cursor);
202 m_task_t task = NULL;
203 MSG_error_t res = MSG_task_receive(&task, node.mailbox);
205 xbt_assert0(res == MSG_OK, "MSG_task_receive failed");
208 const char* task_name = MSG_task_get_name(task);
209 task_data_t task_data = (task_data_t) MSG_task_get_data(task);
211 if (!strcmp(task_name, "Find Successor")) {
212 INFO2("Receiving a 'Find Successor' Request from %s for id %d", task_data->issuer_host_name, task_data->request_id);
213 // is my successor the successor?
214 if (is_in_interval(task_data->request_id, node.id + 1, node.fingers[0].id)) {
215 task_data->answer_id = node.fingers[0].id;
216 MSG_task_set_name(task, "Find Successor Answer");
217 INFO3("Sending back a 'Find Successor' Answer to %s: the successor of %d is %d", task_data->issuer_host_name, task_data->request_id, task_data->answer_id);
218 comm = MSG_task_isend(task, task_data->answer_to);
219 xbt_dynar_push(node.comms, &comm);
222 // otherwise, forward the request to the closest preceding finger in my table
223 int closest = closest_preceding_finger(&node, task_data->request_id);
224 INFO2("Forwarding 'Find Successor' request for id %d to my closest preceding finger %d", task_data->request_id, closest);
225 comm = MSG_task_isend(task, get_mailbox(closest));
226 xbt_dynar_push(node.comms, &comm);
230 else if (!strcmp(task_name, "Find Successor Answer")) {
234 else if (!strcmp(task_name, "Find Predecessor")) {
235 INFO2("Receiving a 'Find Predecessor' Request from %s for id %d", task_data->issuer_host_name, task_data->request_id);
236 // am I the predecessor?
237 if (is_in_interval(task_data->request_id, node.id + 1, node.fingers[0].id)) {
238 task_data->answer_id = node.id;
239 MSG_task_set_name(task, "Find Predecessor Answer");
240 INFO3("Sending back a 'Find Predecessor' Answer to %s: the predecessor of %d is %d", task_data->issuer_host_name, task_data->request_id, task_data->answer_id);
241 comm = MSG_task_isend(task, task_data->answer_to);
242 xbt_dynar_push(node.comms, &comm);
245 // otherwise, forward the request to the closest preceding finger in my table
246 int closest = closest_preceding_finger(&node, task_data->request_id);
247 INFO2("Forwarding 'Find Predecessor' request for id %d to my closest preceding finger %d", task_data->request_id, closest);
248 comm = MSG_task_isend(task, get_mailbox(closest));
249 xbt_dynar_push(node.comms, &comm);
253 else if (!strcmp(task_name, "Find Predecessor Answer")) {
257 else if (!strcmp(task_name, "Update Finger")) {
258 // someone is telling me that he may be my new finger
259 INFO1("Receving an 'Update Finger' from %s", task_data->issuer_host_name);
260 update_finger_table(&node, task_data->request_id, task_data->request_finger);
262 else if (!strcmp(task_name, "Notify")) {
263 // someone is telling me that he may be my new predecessor
264 INFO1("Receving an 'Notify' from %s", task_data->issuer_host_name);
265 notify(&node, task_data->request_id);
268 else if (!strcmp(task_name, "Fix Fingers"))
271 for (i = KEY_BITS - 1 ; i >= 0; i--)
273 data->fingers[i] = find_finger_elem(data,(data->id)+pow(2,i-1));
279 xbt_dynar_free(&node.comms);
283 * \brief Initializes the current node as the first one of the system.
284 * \param node the current node
286 static void initialize_first_node(node_t node)
288 INFO0("Create a new Chord ring...");
290 // I am my own successor and predecessor
292 for (i = 0; i < NB_BITS; i++) {
293 node->fingers[i].id = node->id;
294 node->fingers[i].mailbox = node->mailbox;
296 node->pred_id = node->id;
297 node->pred_mailbox = node->mailbox;
298 print_finger_table(node);
302 * \brief Makes the current node join the system, knowing the id of a node already in the system
303 * \param node the current node
304 * \param known_id id of a node already in the system
306 static void join(node_t node, int known_id)
308 initialize_finger_table(node, known_id); // determine my fingers, asking to known_id
309 remote_notify(node, node->fingers[0].id, node->id); // tell my successor that I'm his new predecessor
310 notify_predecessors(node); // tell others that I may have became their finger
311 remote_move_keys(node, node->fingers[0].id); // take some key-value pairs from my sucessor
315 * \brief Initializes my finger table, knowing the id of a node already in the system.
316 * \param node the current node
317 * \param known_id id of a node already in the system
319 static void initialize_finger_table(node_t node, int known_id)
321 int my_id = node->id;
325 INFO0("Initializing my finger table...");
327 // ask known_id who is my immediate successor
328 node->fingers[0].id = remote_find_successor(node, known_id, my_id + 1);
329 node->fingers[0].mailbox = get_mailbox(node->fingers[0].id);
331 // find all other fingers
332 for (i = 0; i < NB_BITS - 1; i++) {
334 pow = pow << 1; // equivalent to pow = pow * 2
335 if (is_in_interval(my_id + pow, my_id, node->fingers[i].id - 1)) {
336 // I already have the info for this finger
337 node->fingers[i + 1].id = node->fingers[i].id;
340 // I don't have the info, ask the only guy I know
341 node->fingers[i + 1].id = remote_find_successor(node, known_id, my_id + pow);
343 node->fingers[i + 1].mailbox = get_mailbox(node->fingers[i + 1].id);
346 node->pred_id = find_predecessor(node, node->id);
347 node->pred_mailbox = get_mailbox(node->pred_id);
349 INFO0("Finger table initialized!");
350 print_finger_table(node);
354 * \brief Notifies some nodes that the current node may have became their finger.
355 * \param node the current node, which has just joined the system
357 static void notify_predecessors(node_t node)
361 for (i = 0; i < NB_BITS; i++) {
362 // find the closest node whose finger #i can be me
363 pred_id = find_predecessor(node, node->id - pow + 1); // note: no "+1" in the article!
364 if (pred_id != node->id) {
365 remote_update_finger_table(node, pred_id, node->id, i);
367 pow = pow << 1; // pow = pow * 2
372 * \brief Tells the current node that a node may have became its new finger.
373 * \param node the current node
374 * \param candidate_id id of the node that may be a new finger of the current node
375 * \param finger_index index of the finger to update
377 static void update_finger_table(node_t node, int candidate_id, int finger_index)
381 for (i = 0; i < finger_index; i++) {
385 // if (is_in_interval(candidate_id, node->id + pow, node->fingers[finger_index].id - 1)) {
386 if (is_in_interval(candidate_id, node->id, node->fingers[finger_index].id - 1)) {
387 // INFO3("Candidate %d is between %d and %d!", candidate_id, node->id + pow, node->fingers[finger_index].id - 1);
388 // candidate_id is my new finger
389 node->fingers[finger_index].id = candidate_id;
390 node->fingers[finger_index].mailbox = get_mailbox(candidate_id);
391 INFO2("My new finger #%d is %d", finger_index, candidate_id);
392 print_finger_table(node);
394 if (node->pred_id != node->id) { // FIXME: is this necessary?
395 // my predecessor may be concerned too
396 remote_update_finger_table(node, node->pred_id, candidate_id, finger_index);
402 * \brief Tells a remote node that a node may have became its new finger.
403 * \param ask_to_id id of the remote node to update
404 * \param candidate_id id of the node that may be a new finger of the remote node
405 * \param finger_index index of the finger to update
407 static void remote_update_finger_table(node_t node, int ask_to_id, int candidate_id, int finger_index)
409 task_data_t req_data = xbt_new0(s_task_data_t, 1);
410 req_data->request_id = candidate_id;
411 req_data->request_finger = finger_index;
412 req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
414 // send a "Update Finger" request to ask_to_id
415 INFO3("Sending an 'Update Finger' request to %d: his finger #%d may be %d now", ask_to_id, finger_index, candidate_id);
416 m_task_t task = MSG_task_create("Update Finger", 1000, 5000, req_data);
417 msg_comm_t comm = MSG_task_isend(task, get_mailbox(ask_to_id));
418 xbt_dynar_push(node->comms, &comm);
422 * \brief Makes the current node find the successor node of an id.
423 * \param node the current node
424 * \param id the id to find
425 * \return the id of the successor node
427 static int find_successor(node_t node, int id)
429 // is my successor the successor?
430 if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
431 return node->fingers[0].id;
434 // otherwise, ask the closest preceding finger in my table
435 int closest = closest_preceding_finger(node, id);
436 return remote_find_successor(node, closest, id);
440 * \brief Asks another node the successor node of an id.
441 * \param node the current node
442 * \param ask_to the node to ask to
443 * \param id the id to find
444 * \return the id of the successor node
446 static int remote_find_successor(node_t node, int ask_to, int id)
448 s_task_data_t req_data;
449 char *mailbox = bprintf("%s Find Successor", node->mailbox);
450 req_data.request_id = id;
451 req_data.answer_to = mailbox;
452 req_data.issuer_host_name = MSG_host_get_name(MSG_host_self());
454 // send a "Find Successor" request to ask_to_id
455 INFO2("Sending a 'Find Successor' request to %d for key %d", ask_to, id);
456 m_task_t task = MSG_task_create("Find Successor", 1000, 5000, &req_data);
457 MSG_task_send(task, get_mailbox(ask_to));
459 // receive the answer
461 MSG_task_receive(&task, req_data.answer_to);
462 task_data_t ans_data;
463 ans_data = MSG_task_get_data(task);
464 int successor = ans_data->answer_id;
466 INFO2("Received the answer to my Find Successor request: the successor of key %d is %d", id, successor);
472 * \brief Makes the current node find the predecessor node of an id.
473 * \param node the current node
474 * \param id the id to find
475 * \return the id of the predecessor node
477 static int find_predecessor(node_t node, int id)
479 if (node->id == node->fingers[0].id) {
480 // I am the only node in the system
484 if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
487 int ask_to = closest_preceding_finger(node, id);
488 return remote_find_predecessor(node, ask_to, id);
492 * \brief Asks another node the predecessor node of an id.
493 * \param node the current node
494 * \param ask_to the node to ask to
495 * \param id the id to find
496 * \return the id of the predecessor node
498 static int remote_find_predecessor(node_t node, int ask_to, int id)
500 s_task_data_t req_data;
501 char *mailbox = bprintf("%s Find Predecessor", node->mailbox);
502 req_data.request_id = id;
503 req_data.answer_to = mailbox;
504 req_data.issuer_host_name = MSG_host_get_name(MSG_host_self());
506 // send a "Find Predecessor" request to ask_to
507 INFO2("Sending a 'Find Predecessor' request to %d for key %d", ask_to, id);
508 m_task_t task = MSG_task_create("Find Predecessor", 1000, 5000, &req_data);
509 MSG_task_send(task, get_mailbox(ask_to));
511 // receive the answer
513 MSG_task_receive(&task, req_data.answer_to);
514 task_data_t ans_data;
515 ans_data = MSG_task_get_data(task);
516 int predecessor = ans_data->answer_id;
518 INFO2("Received the answer to my 'Find Predecessor' request: the predecessor of key %d is %d", id, predecessor);
524 * \brief Returns the closest preceding finger of an id
525 * with respect to the finger table of the current node.
526 * \param node the current node
527 * \param id the id to find
528 * \return the closest preceding finger of that id
530 int closest_preceding_finger(node_t node, int id)
533 for (i = NB_BITS - 1; i >= 0; i--) {
534 if (is_in_interval(node->fingers[i].id, node->id + 1, id - 1)) {
535 return node->fingers[i].id;
542 * \brief This function is called periodically. It checks the immediate
543 * successor of the current node.
544 * \param node the current node
546 static void stabilize(node_t node) {
548 int x = find_predecessor(node, node->fingers[0].id);
549 if (is_in_interval(x, node->id + 1, node->fingers[0].id)) {
550 node->fingers[0].id = x;
551 node->fingers[0].mailbox = get_mailbox(x);
553 remote_notify(node, node->fingers[0].id, node->id);
557 * \brief Notifies the current node that its predecessor may have changed.
558 * \param node the current node
559 * \param candidate_id the possible new predecessor
561 static void notify(node_t node, int predecessor_candidate_id) {
563 if (node->pred_id == node->id
564 || is_in_interval(predecessor_candidate_id, node->pred_id, node->id)) {
566 node->pred_id = predecessor_candidate_id;
567 node->pred_mailbox = get_mailbox(predecessor_candidate_id);
569 INFO1("My new predecessor is %d", predecessor_candidate_id);
570 print_finger_table(node);
573 INFO1("I don't have to change my predecessor to %d", predecessor_candidate_id);
578 * \brief Notifies a remote node that its predecessor may have changed.
579 * \param node the current node
580 * \param notify_id id of the node to notify
581 * \param candidate_id the possible new predecessor
583 static void remote_notify(node_t node, int notify_id, int predecessor_candidate_id) {
585 task_data_t req_data = xbt_new0(s_task_data_t, 1);
586 req_data->request_id = predecessor_candidate_id;
587 req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
589 // send a "Notify" request to notify_id
590 INFO1("Sending a 'Notify' request to %d", notify_id);
591 m_task_t task = MSG_task_create("Notify", 1000, 5000, req_data);
592 msg_comm_t comm = MSG_task_isend(task, get_mailbox(notify_id));
593 xbt_dynar_push(node->comms, &comm);
597 * \brief Asks a node to take some of its keys.
598 * \param node the current node, which has just joined the system
599 * \param take_from_id id of a node who may have keys to give to the current node
601 static void remote_move_keys(node_t node, int take_from_id) {
606 * \brief Main function.
608 int main(int argc, char *argv[])
611 printf("Usage: %s platform_file deployment_file\n", argv[0]);
612 printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
616 MSG_global_init(&argc, argv);
618 const char* platform_file = argv[1];
619 const char* application_file = argv[2];
621 /* MSG_config("workstation/model","KCCFLN05"); */
622 MSG_set_channel_number(0);
623 MSG_create_environment(platform_file);
625 MSG_function_register("node", node);
626 MSG_launch_application(application_file);
628 MSG_error_t res = MSG_main();
629 INFO1("Simulation time: %g", MSG_get_clock());