examples/msg/chord/chord.c

   1
   2 /* Copyright (c) 2010. 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 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
  13                              "Messages specific for this msg example");
  14
  15 #define NB_BITS 6
  16 #define NB_KEYS 64
  17
  18 /**
  19  * Finger element.
  20  */
  21 typedef struct finger {
  22   int id;
  23   const char* mailbox;
  24 } s_finger_t, *finger_t;
  25
  26 /**
  27  * Node data.
  28  */
  29 typedef struct node {
  30   int id;                                 // my id
  31   const char* mailbox;
  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
  36 } s_node_t, *node_t;
  37
  38 /**
  39  * Task data
  40  */
  41 typedef struct task_data {
  42   int request_id;
  43   int request_finger;
  44   int answer_id;
  45   const char* answer_to;
  46   const char* issuer_host_name; // used for logging
  47 } s_task_data_t, *task_data_t;
  48
  49 // utility functions
  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);
  54
  55 // process functions
  56 static int node(int argc, char *argv[]);
  57
  58 // initialization
  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);
  62
  63 // Chord core
  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);
  77
  78 /**
  79  * \brief Turns an id into an equivalent id in [0, NB_KEYS[
  80  * \param id an id
  81  * \return the corresponding normalized id
  82  */
  83 static int normalize(int id) {
  84
  85   // make sure id >= 0
  86   while (id < 0) {
  87     id += NB_KEYS;
  88   }
  89
  90   // make sure id < NB_KEYS
  91   id = id % NB_KEYS;
  92
  93   return id;
  94 }
  95
  96 /**
  97  * \brief Returns whether a id belongs to the interval [start, end].
  98  *
  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]
 106  *
 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]
 111  */
 112 static int is_in_interval(int id, int start, int end) {
 113
 114   id = normalize(id);
 115   start = normalize(start);
 116   end = normalize(end);
 117
 118   // make sure end >= start and id >= start
 119   if (end < start) {
 120     end += NB_KEYS;
 121   }
 122
 123   if (id < start) {
 124     id += NB_KEYS;
 125   }
 126
 127   return id <= end;
 128 }
 129
 130 /**
 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
 135  */
 136 static char* get_mailbox(int node_id) {
 137
 138   return bprintf("mailbox%d", node_id);
 139 }
 140
 141 /**
 142  * \brief Displays the finger table of a node.
 143  * \param node a node
 144  */
 145 static void print_finger_table(node_t node) {
 146
 147   int i;
 148   int pow = 1;
 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);
 153     pow = pow << 1;
 154   }
 155   INFO1("Predecessor: %d", node->pred_id);
 156 }
 157
 158 /**
 159  * \brief Node Function
 160  * Arguments:
 161  * - my id
 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)
 164  */
 165 int node(int argc, char *argv[])
 166 {
 167   msg_comm_t comm = NULL;
 168
 169   xbt_assert0(argc == 2 || argc == 4, "Wrong number of arguments for this node");
 170
 171   // initialize my node
 172   s_node_t node = {0};
 173   node.id = atoi(argv[1]);
 174   node.mailbox = get_mailbox(node.id);
 175   node.comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
 176
 177   if (argc == 2) { // first ring
 178     initialize_first_node(&node);
 179   }
 180   else {
 181     int known_id = atoi(argv[2]);
 182     double sleep_time = atof(argv[3]);
 183
 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.");
 188
 189     join(&node, known_id);
 190   }
 191
 192   while (1) {
 193
 194     unsigned int cursor;
 195     comm = NULL;
 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);
 199       }
 200     }
 201
 202     m_task_t task = NULL;
 203     MSG_error_t res = MSG_task_receive(&task, node.mailbox);
 204
 205     xbt_assert0(res == MSG_OK, "MSG_task_receive failed");
 206
 207     // get data
 208     const char* task_name = MSG_task_get_name(task);
 209     task_data_t task_data = (task_data_t) MSG_task_get_data(task);
 210
 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);
 220       }
 221       else {
 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);
 227       }
 228     }
 229     /*
 230     else if (!strcmp(task_name, "Find Successor Answer")) {
 231
 232     }
 233     */
 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);
 243       }
 244       else {
 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);
 250       }
 251     }
 252     /*
 253     else if (!strcmp(task_name, "Find Predecessor Answer")) {
 254
 255     }
 256     */
 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);
 261     }
 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);
 266     }
 267     /*
 268     else if (!strcmp(task_name, "Fix Fingers"))
 269     {
 270       int i;
 271       for (i = KEY_BITS - 1 ; i >= 0; i--)
 272       {
 273         data->fingers[i] = find_finger_elem(data,(data->id)+pow(2,i-1));
 274       }
 275     }
 276     */
 277   }
 278
 279   xbt_dynar_free(&node.comms);
 280 }
 281
 282 /**
 283  * \brief Initializes the current node as the first one of the system.
 284  * \param node the current node
 285  */
 286 static void initialize_first_node(node_t node)
 287 {
 288   INFO0("Create a new Chord ring...");
 289
 290   // I am my own successor and predecessor
 291   int i;
 292   for (i = 0; i < NB_BITS; i++) {
 293     node->fingers[i].id = node->id;
 294     node->fingers[i].mailbox = node->mailbox;
 295   }
 296   node->pred_id = node->id;
 297   node->pred_mailbox = node->mailbox;
 298   print_finger_table(node);
 299 }
 300
 301 /**
 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
 305  */
 306 static void join(node_t node, int known_id)
 307 {
 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
 312 }
 313
 314 /*
 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
 318  */
 319 static void initialize_finger_table(node_t node, int known_id)
 320 {
 321   int my_id = node->id;
 322   int i;
 323   int pow = 1; // 2^i
 324
 325   INFO0("Initializing my finger table...");
 326
 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);
 330
 331   // find all other fingers
 332   for (i = 0; i < NB_BITS - 1; i++) {
 333
 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;
 338     }
 339     else {
 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);
 342     }
 343     node->fingers[i + 1].mailbox = get_mailbox(node->fingers[i + 1].id);
 344   }
 345
 346   node->pred_id = find_predecessor(node, node->id);
 347   node->pred_mailbox = get_mailbox(node->pred_id);
 348
 349   INFO0("Finger table initialized!");
 350   print_finger_table(node);
 351 }
 352
 353 /**
 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
 356  */
 357 static void notify_predecessors(node_t node)
 358 {
 359   int i, pred_id;
 360   int pow = 1;
 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);
 366     }
 367     pow = pow << 1; // pow = pow * 2
 368   }
 369 }
 370
 371 /**
 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
 376  */
 377 static void update_finger_table(node_t node, int candidate_id, int finger_index)
 378 {
 379   int pow = 1;
 380   int i;
 381   for (i = 0; i < finger_index; i++) {
 382     pow = pow << 1;
 383   }
 384
 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);
 393
 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);
 397     }
 398   }
 399 }
 400
 401 /**
 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
 406  */
 407 static void remote_update_finger_table(node_t node, int ask_to_id, int candidate_id, int finger_index)
 408 {
 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());
 413
 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);
 419 }
 420
 421 /**
 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
 426  */
 427 static int find_successor(node_t node, int id)
 428 {
 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;
 432   }
 433
 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);
 437 }
 438
 439 /**
 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
 445  */
 446 static int remote_find_successor(node_t node, int ask_to, int id)
 447 {
 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());
 453
 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));
 458
 459   // receive the answer
 460   task = NULL;
 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;
 465   xbt_free(mailbox);
 466   INFO2("Received the answer to my Find Successor request: the successor of key %d is %d", id, successor);
 467
 468   return successor;
 469 }
 470
 471 /**
 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
 476  */
 477 static int find_predecessor(node_t node, int id)
 478 {
 479   if (node->id == node->fingers[0].id) {
 480     // I am the only node in the system
 481     return node->id;
 482   }
 483
 484   if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
 485     return node->id;
 486   }
 487   int ask_to = closest_preceding_finger(node, id);
 488   return remote_find_predecessor(node, ask_to, id);
 489 }
 490
 491 /**
 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
 497  */
 498 static int remote_find_predecessor(node_t node, int ask_to, int id)
 499 {
 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());
 505
 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));
 510
 511   // receive the answer
 512   task = NULL;
 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;
 517   xbt_free(mailbox);
 518   INFO2("Received the answer to my 'Find Predecessor' request: the predecessor of key %d is %d", id, predecessor);
 519
 520   return predecessor;
 521 }
 522
 523 /**
 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
 529  */
 530 int closest_preceding_finger(node_t node, int id)
 531 {
 532   int i;
 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;
 536     }
 537   }
 538   return node->id;
 539 }
 540
 541 /**
 542  * \brief This function is called periodically. It checks the immediate
 543  * successor of the current node.
 544  * \param node the current node
 545  */
 546 static void stabilize(node_t node) {
 547
 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);
 552   }
 553   remote_notify(node, node->fingers[0].id, node->id);
 554 }
 555
 556 /**
 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
 560  */
 561 static void notify(node_t node, int predecessor_candidate_id) {
 562
 563   if (node->pred_id == node->id
 564     || is_in_interval(predecessor_candidate_id, node->pred_id, node->id)) {
 565
 566     node->pred_id = predecessor_candidate_id;
 567     node->pred_mailbox = get_mailbox(predecessor_candidate_id);
 568
 569     INFO1("My new predecessor is %d", predecessor_candidate_id);
 570     print_finger_table(node);
 571   }
 572   else {
 573     INFO1("I don't have to change my predecessor to %d", predecessor_candidate_id);
 574   }
 575 }
 576
 577 /**
 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
 582  */
 583 static void remote_notify(node_t node, int notify_id, int predecessor_candidate_id) {
 584
 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());
 588
 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);
 594 }
 595
 596 /**
 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
 600  */
 601 static void remote_move_keys(node_t node, int take_from_id) {
 602   // TODO
 603 }
 604
 605 /**
 606  * \brief Main function.
 607  */
 608 int main(int argc, char *argv[])
 609 {
 610   if (argc < 3) {
 611     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 612     printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
 613     exit(1);
 614   }
 615
 616   MSG_global_init(&argc, argv);
 617
 618   const char* platform_file = argv[1];
 619   const char* application_file = argv[2];
 620
 621   /* MSG_config("workstation/model","KCCFLN05"); */
 622   MSG_set_channel_number(0);
 623   MSG_create_environment(platform_file);
 624
 625   MSG_function_register("node", node);
 626   MSG_launch_application(application_file);
 627
 628   MSG_error_t res = MSG_main();
 629   INFO1("Simulation time: %g", MSG_get_clock());
 630
 631   MSG_clean();
 632
 633   if (res == MSG_OK)
 634     return 0;
 635   else
 636     return 1;
 637 }