examples/msg/chord/chord.c

   1 /* Copyright (c) 2010. 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 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
  12                              "Messages specific for this msg example");
  13
  14 #define NB_BITS 6
  15 #define NB_KEYS 64
  16
  17 /**
  18  * Finger element.
  19  */
  20 typedef struct finger {
  21   int id;
  22   const char* mailbox;
  23 } s_finger_t, *finger_t;
  24
  25 /**
  26  * Node data.
  27  */
  28 typedef struct node {
  29   int id;                                 // my id
  30   const char* mailbox;
  31   s_finger_t fingers[NB_BITS];            // finger table (fingers[0] is my successor)
  32   int pred_id;                            // predecessor id
  33   const char* pred_mailbox;
  34 } s_node_t, *node_t;
  35
  36 /**
  37  * Task data
  38  */
  39 typedef struct task_data {
  40   int request_id;
  41   int request_finger;
  42   int answer_id;
  43   const char* answer_to;
  44 } s_task_data_t, *task_data_t;
  45
  46 // utility functions
  47 static int normalize(int id);
  48 static int is_in_interval(int id, int start, int end);
  49 static char* get_mailbox(int host_id);
  50
  51 // process functions
  52 static int node(int argc, char *argv[]);
  53 static int sender(int argc,char *argv[]);
  54
  55 // initialization
  56 static void initialize_first_node(node_t node);
  57 static void initialize_finger_table(node_t data, int known_id);
  58 static void join(node_t node, int known_id);
  59
  60 // Chord core
  61 static int find_successor(node_t node, int id);
  62 static int remote_find_successor(node_t node, int ask_to_id, int id);
  63 static int find_predecessor(node_t node, int id);
  64 static int remote_find_predecessor(node_t node, int ask_to_id, int id);
  65 static int closest_preceding_finger(node_t node, int id);
  66 static int remote_closest_preceding_finger(int ask_to_id, int id);
  67 static void notify(node_t);
  68 static void remote_move_keys(node_t node, int take_from_id);
  69 static void update_finger_table(node_t node, int candidate_id, int finger_index);
  70 static void remote_update_finger_table(int ask_to_id, int candidate_id, int finger_index);
  71
  72 //static void find_successor_node(node_t my_data, m_task_t join_task);
  73
  74 /**
  75  * \brief Turns an id into an equivalent id in [0, NB_KEYS[
  76  * \param id an id
  77  * \return the corresponding normalized id
  78  */
  79 static int normalize(int id) {
  80
  81   // make sure id >= 0
  82   while (id < 0) {
  83     id += NB_KEYS;
  84   }
  85
  86   // make sure id < NB_KEYS
  87   id = id % NB_KEYS;
  88
  89   return id;
  90 }
  91
  92 /**
  93  * \brief Returns whether a id belongs to the interval [start, end].
  94  *
  95  * The parameters are noramlized to make sure they are between 0 and CHORD_NB_KEYS - 1).
  96  * 1 belongs to [62, 3]
  97  * 1 does not belong to [3, 62]
  98  * 63 belongs to [62, 3]
  99  * 63 does not belong to [3, 62]
 100  * 24 belongs to [21, 29]
 101  * 24 does not belong to [29, 21]
 102  *
 103  * \param id id to check
 104  * \param start lower bound
 105  * \param end upper bound
 106  * \return a non-zero value if id in in [start, end]
 107  */
 108 static int is_in_interval(int id, int start, int end) {
 109
 110   id = normalize(id);
 111   start = normalize(start);
 112   end = normalize(end);
 113
 114   // make sure end >= start and id >= start
 115   if (end < start) {
 116     end += NB_KEYS;
 117   }
 118
 119   if (id < start) {
 120     id += NB_KEYS;
 121   }
 122
 123   return id <= end;
 124 }
 125
 126 /**
 127  * \brief Gets the mailbox name of a host given its chord id.
 128  * \param node_id id of a node
 129  * \return the name of its mailbox
 130  * FIXME: free the memory
 131  */
 132 static char* get_mailbox(int node_id) {
 133
 134   return bprintf("mailbox%d", node_id);
 135 }
 136
 137 /**
 138  * \brief Node Function
 139  * Arguments:
 140  * - my id
 141  * - the id of a guy I know in the system (except for the first node)
 142  * - the time to sleep before I join (except for the first node)
 143  */
 144 int node(int argc, char *argv[])
 145 {
 146   xbt_assert0(argc == 2 || argc == 4, "Wrong number of arguments for this node");
 147
 148   // initialize my node
 149   s_node_t node = {0};
 150   node.id = atoi(argv[1]);
 151   node.mailbox = get_mailbox(node.id);
 152
 153   if (argc == 2) { // first ring
 154     initialize_first_node(&node);
 155   }
 156   else {
 157     int known_id = atoi(argv[2]);
 158     double sleep_time = atof(argv[3]);
 159
 160     // sleep before starting
 161     INFO1("Let's sleep >>%f", sleep_time);
 162     MSG_process_sleep(sleep_time);
 163     INFO0("Hey! Let's join the system.");
 164
 165     join(&node, known_id);
 166   }
 167
 168   while (1) {
 169
 170     m_task_t task = NULL;
 171     MSG_error_t res = MSG_task_receive(&task, node.mailbox);
 172
 173     xbt_assert0(res == MSG_OK, "MSG_task_receive failed");
 174
 175     // get data
 176     const char* task_name = MSG_task_get_name(task);
 177     task_data_t task_data = (task_data_t) MSG_task_get_data(task);
 178
 179     if (!strcmp(task_name, "Find Successor")) {
 180
 181       // is my successor the successor?
 182       if (is_in_interval(task_data->request_id, node.id + 1, node.fingers[0].id)) {
 183         task_data->answer_id = node.fingers[0].id;
 184         MSG_task_set_name(task, "Find Successor Answer");
 185         MSG_task_send(task, task_data->answer_to);
 186       }
 187       else {
 188         // otherwise, forward the request to the closest preceding finger in my table
 189         int closest = closest_preceding_finger(&node, task_data->request_id);
 190         MSG_task_send(task, get_mailbox(closest));
 191       }
 192     }
 193     /*
 194     else if (!strcmp(task_name, "Find Successor Answer")) {
 195
 196     }
 197     */
 198     else if (!strcmp(task_name, "Find Predecessor")) {
 199
 200       // am I the predecessor?
 201       if (is_in_interval(task_data->request_id, node.id + 1, node.fingers[0].id)) {
 202         task_data->answer_id = node.id;
 203         MSG_task_set_name(task, "Find Predecessor Answer");
 204         MSG_task_send(task, task_data->answer_to);
 205       }
 206       else {
 207         // otherwise, forward the request to the closest preceding finger in my table
 208         int closest = closest_preceding_finger(&node, task_data->request_id);
 209         MSG_task_send(task, get_mailbox(closest));
 210       }
 211     }
 212     /*
 213     else if (!strcmp(task_name, "Find Predecessor Answer")) {
 214
 215     }
 216     */
 217     else if (!strcmp(task_name, "Update Finger")) {
 218       update_finger_table(&node, task_data->request_id, task_data->request_finger);
 219     }
 220     /*
 221     else if (!strcmp(task_name, "Fix Fingers"))
 222     {
 223       int i;
 224       for (i = KEY_BITS - 1 ; i >= 0; i--)
 225       {
 226         data->fingers[i] = find_finger_elem(data,(data->id)+pow(2,i-1));
 227       }
 228     }
 229     */
 230   }
 231 }
 232
 233 /**
 234  * \brief Initializes the current node as the first one of the system.
 235  * \param node the current node
 236  */
 237 static void initialize_first_node(node_t node)
 238 {
 239   INFO0("Create a new Chord ring...");
 240
 241   // I am my own successor and predecessor
 242   int i;
 243   for (i = 0; i < NB_BITS; i++) {
 244     node->fingers[i].id = node->id;
 245     node->fingers[i].mailbox = node->mailbox;
 246   }
 247   node->pred_id = node->id;
 248   node->pred_mailbox = node->mailbox;
 249 }
 250
 251 /**
 252  * \brief Makes the current node join the system, knowing the id of a node already in the system
 253  * \param node the current node
 254  * \param known_id id of a node already in the system
 255  */
 256 static void join(node_t node, int known_id)
 257 {
 258   initialize_finger_table(node, known_id); // determine my fingers, asking to known_id
 259   notify(node); // tell others that I may have became their finger
 260   remote_move_keys(node, node->fingers[0].id); // take some key-value pairs from my sucessor
 261 }
 262
 263 /*
 264  * \brief Initializes my finger table, knowing the id of a node already in the system.
 265  * \param node the current node
 266  * \param known_id id of a node already in the system
 267  */
 268 static void initialize_finger_table(node_t node, int known_id)
 269 {
 270  int my_id = node->id;
 271  int i;
 272  int pow = 1; // 2^i
 273
 274   // ask known_id who is my immediate successor
 275   node->fingers[0].id = remote_find_successor(node, known_id, my_id + 1);
 276   node->fingers[0].mailbox = get_mailbox(node->fingers[0].id);
 277
 278   // find all other fingers
 279   for (i = 0; i < NB_BITS - 1; i++) {
 280
 281     pow = pow << 1; // equivalent to pow = pow * 2
 282     if (is_in_interval(my_id + pow, my_id, node->fingers[i].id - 1)) {
 283       // I already have the info for this finger
 284       node->fingers[i + 1].id = node->fingers[i].id;
 285     }
 286     else {
 287       // I don't have the info, ask the only guy I know
 288       node->fingers[i + 1].id = remote_find_successor(node, known_id, my_id + pow);
 289     }
 290     node->fingers[i + 1].mailbox = get_mailbox(node->fingers[i + 1].id);
 291   }
 292 }
 293
 294 /**
 295  * \brief Notifies some nodes that the current node may have became their finger.
 296  * \param node the current node, which has just joined the system
 297  */
 298 static void notify(node_t node)
 299 {
 300   int i, pred_id;
 301   int pow = 1;
 302   for (i = 0; i < NB_KEYS; i++) {
 303     // find the closest node whose finger #i can be me
 304     pred_id = find_predecessor(node, node->id - pow);
 305     remote_update_finger_table(pred_id, node->id, i);
 306     pow = pow << 1; // pow = pow * 2
 307   }
 308 }
 309
 310 /**
 311  * \brief Tells the current node that a node may have became its new finger.
 312  * \param node the current node
 313  * \param candidate_id id of the node that may be a new finger of the current node
 314  * \param finger_index index of the finger to update
 315  */
 316 static void update_finger_table(node_t node, int candidate_id, int finger_index)
 317 {
 318   if (is_in_interval(candidate_id, node->id, node->fingers[finger_index].id - 1)) {
 319
 320     // candidate_id is my new finger
 321     node->fingers[finger_index].id = candidate_id;
 322     node->fingers[finger_index].mailbox = get_mailbox(candidate_id);
 323
 324     // my predecessor may be concerned too
 325     remote_update_finger_table(node->pred_id, candidate_id, finger_index);
 326   }
 327 }
 328
 329 /**
 330  * \brief Tells a remote node that a node may have became its new finger.
 331  * \param ask_to_id id of the remote node to update
 332  * \param candidate_id id of the node that may be a new finger of the remote node
 333  * \param finger_index index of the finger to update
 334  */
 335 static void remote_update_finger_table(int ask_to_id, int candidate_id, int finger_index)
 336 {
 337   s_task_data_t req_data;
 338   req_data.request_id = candidate_id;
 339   req_data.request_finger = finger_index;
 340
 341   // send a "Update Finger" request to ask_to_id
 342   m_task_t task = MSG_task_create("Update Finger", 1000, 5000, &req_data);
 343   MSG_task_send(task, get_mailbox(ask_to_id));
 344 }
 345
 346 /* deprecated version where the remote host modifies the issuer's node data
 347 static void find_successor_node(node_t n_data,m_task_t join_task)  //use all data
 348 {
 349         //get recv data
 350         node_t recv_data = (node_t)MSG_task_get_data(join_task);
 351         INFO3("recv_data->id : %i , n_data->id :%i , successor->id :%i",recv_data->id,n_data->id,n_data->fingers[0].id);
 352         //if ((recv_data->id >= n_data->id) && (recv_data->id <= n_data->fingers[0].id))
 353         if (is_in_interval(recv_data->id,n_data->id,n_data->fingers[0].id))
 354         {
 355                 INFO1("Successor Is %s",n_data->fingers[0].host_name);
 356                 //predecessor(recv_data) >>>> n_data
 357                 recv_data->pred_host_name = n_data->host_name;
 358                 recv_data->pred_id = n_data->id;
 359                 recv_data->pred_mailbox = n_data->pred_mailbox;
 360                 // successor(recv_data) >>>> n_data.finger[0]
 361                 recv_data->fingers_nb = 1;
 362                 recv_data->fingers[0].host_name = n_data->fingers[0].host_name;
 363                 recv_data->fingers[0].id = n_data->fingers[0].id;
 364                 recv_data->fingers[0].mailbox = n_data->fingers[0].mailbox;
 365                 // successor(n_data) >>>> recv_sucessor
 366                 n_data->fingers[0].id = recv_data->id;
 367                 n_data->fingers[0].host_name = recv_data->host_name;
 368                 n_data->fingers[0].mailbox = recv_data->mailbox;
 369                 // Logs
 370                 INFO1("Sending back a Join Request to %s",recv_data->host_name);
 371                 MSG_task_set_name(join_task,"Join Response");
 372                 MSG_task_send(join_task,recv_data->mailbox);
 373         }
 374
 375         else
 376         {
 377                 const char* closest_preceding_mailbox = find_closest_preceding(n_data,recv_data->id);
 378                 INFO1("Forwarding Join Call to mailbox %s",closest_preceding_mailbox);
 379                 MSG_task_send(join_task,closest_preceding_mailbox);
 380         }
 381 }
 382 */
 383
 384 /**
 385  * \brief Makes the current node find the successor node of an id.
 386  * \param node the current node
 387  * \param id the id to find
 388  * \return the id of the successor node
 389  */
 390 static int find_successor(node_t node, int id)
 391 {
 392   // is my successor the successor?
 393   if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
 394     return node->fingers[0].id;
 395   }
 396
 397   // otherwise, ask the closest preceding finger in my table
 398   int closest = closest_preceding_finger(node, id);
 399   return remote_find_successor(node, closest, id);
 400 }
 401
 402 /**
 403  * \brief Asks another node the successor node of an id.
 404  * \param node the current node
 405  * \param ask_to the node to ask to
 406  * \param id the id to find
 407  * \return the id of the successor node
 408  */
 409 static int remote_find_successor(node_t node, int ask_to, int id)
 410 {
 411   s_task_data_t req_data;
 412   req_data.request_id = id;
 413   req_data.answer_to = node->mailbox;
 414
 415   // send a "Find Successor" request to ask_to_id
 416   m_task_t task = MSG_task_create("Find Successor", 1000, 5000, &req_data);
 417   MSG_task_send(task, get_mailbox(ask_to));
 418
 419   // receive the answer
 420   task = NULL;
 421   MSG_task_receive(&task, node->mailbox); // FIXME: don't receive here other messages than the excepted one
 422   task_data_t ans_data;
 423   ans_data = MSG_task_get_data(task);
 424   int successor = ans_data->answer_id;
 425
 426   return successor;
 427 }
 428
 429 /**
 430  * \brief Makes the current node find the predecessor node of an id.
 431  * \param node the current node
 432  * \param id the id to find
 433  * \return the id of the predecessor node
 434  */
 435 static int find_predecessor(node_t node, int id)
 436 {
 437   if (node->id == node->fingers[0].id) {
 438     // I am the only node in the system
 439     return node->id;
 440   }
 441
 442   if (is_in_interval(id, node->id + 1, node->fingers[0].id)) {
 443     return node->id;
 444   }
 445   int ask_to = closest_preceding_finger(node, id);
 446   return remote_find_predecessor(node, ask_to, id);
 447 }
 448
 449 /**
 450  * \brief Asks another node the predecessor node of an id.
 451  * \param node the current node
 452  * \param ask_to the node to ask to
 453  * \param id the id to find
 454  * \return the id of the predecessor node
 455  */
 456 static int remote_find_predecessor(node_t node, int ask_to, int id)
 457 {
 458   s_task_data_t req_data;
 459   req_data.request_id = id;
 460   req_data.answer_to = node->mailbox;
 461
 462   // send a "Find Predecessor" request to ask_to
 463   m_task_t task = MSG_task_create("Find Predecessor", 1000, 5000, &req_data);
 464   MSG_task_send(task, get_mailbox(ask_to));
 465
 466   // receive the answer
 467   task = NULL;
 468   MSG_task_receive(&task, node->mailbox); // FIXME: don't receive here other messages than the excepted one
 469   task_data_t ans_data;
 470   ans_data = MSG_task_get_data(task);
 471   int predecessor = ans_data->answer_id;
 472
 473   return predecessor;
 474 }
 475
 476 /**
 477  * \brief Returns the closest preceding finger of an id
 478  * with respect to the finger table of the current node.
 479  * \param node the current node
 480  * \param id the id to find
 481  * \return the closest preceding finger of that id
 482  */
 483 int closest_preceding_finger(node_t node, int id)
 484 {
 485   int i;
 486   for (i = NB_BITS - 1; i >= 0; i--) {
 487     if (is_in_interval(node->fingers[i].id, node->id + 1, id - 1)) {
 488       return node->fingers[i].id;
 489     }
 490   }
 491   return node->id;
 492 }
 493
 494 /**
 495  * \brief Asks a node to take some of its keys.
 496  * \param node the current node, which has just joined the system
 497  * \param take_from_id id of a node who may have keys to give to the current node
 498  */
 499 static void remote_move_keys(node_t node, int take_from_id) {
 500   // TODO
 501 }
 502
 503 /**
 504  * \brief Main function.
 505  */
 506 int main(int argc, char *argv[])
 507 {
 508   if (argc < 3) {
 509     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 510     printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
 511     exit(1);
 512   }
 513
 514   MSG_global_init(&argc, argv);
 515
 516   const char* platform_file = argv[1];
 517   const char* application_file = argv[2];
 518
 519   /* MSG_config("workstation/model","KCCFLN05"); */
 520   MSG_set_channel_number(0);
 521   MSG_create_environment(platform_file);
 522
 523   MSG_function_register("node", node);
 524   MSG_launch_application(application_file);
 525
 526   MSG_error_t res = MSG_main();
 527   INFO1("Simulation time: %g", MSG_get_clock());
 528
 529   MSG_clean();
 530
 531   if (res == MSG_OK)
 532     return 0;
 533   else
 534     return 1;
 535 }