examples/msg/dht-pastry/dht-pastry.c

   1 /* Copyright (c) 2013-2015. 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 "simgrid/msg.h"
   8 #include "xbt/dynar.h"
   9 #include <math.h>
  10
  11
  12 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example");
  13
  14 /* TODO:                               *
  15  *  - handle node departure            *
  16  *  - handle objects on the network    *
  17  *  - handle neighborhood in the update */
  18
  19 #define COMM_SIZE 10
  20 #define COMP_SIZE 0
  21 #define MAILBOX_NAME_SIZE 10
  22
  23 #define DOMAIN_SIZE 4
  24 #define LEVELS_COUNT 8 // sizeof(int)*8/DOMAIN_SIZE
  25 #define LEVEL_SIZE 16 // 2^DOMAIN_SIZE
  26 #define NEIGHBORHOOD_SIZE 6
  27 #define NAMESPACE_SIZE 6
  28 #define MAILBOX_NAME_SIZE 10
  29
  30 static int nb_bits = 16;
  31 static int timeout = 50;
  32 static int max_simulation_time = 1000;
  33
  34 typedef struct s_node {
  35   int id;                                 //128bits generated random(2^128 -1)
  36   int known_id;
  37   char mailbox[MAILBOX_NAME_SIZE];        // my mailbox name (string representation of the id)
  38   int namespace_set[NAMESPACE_SIZE];
  39   int neighborhood_set[NEIGHBORHOOD_SIZE];
  40   int routing_table[LEVELS_COUNT][LEVEL_SIZE];
  41   int ready;
  42   msg_comm_t comm_receive;                // current communication to receive
  43   xbt_dynar_t pending_tasks;
  44 } s_node_t, *node_t;
  45
  46 typedef struct s_state {
  47   int id;
  48   int namespace_set[NAMESPACE_SIZE];
  49   int neighborhood_set[NEIGHBORHOOD_SIZE];
  50   int routing_table[LEVELS_COUNT][LEVEL_SIZE];
  51 } s_state_t, *state_t;
  52
  53 /** Types of tasks exchanged between nodes. */
  54 typedef enum {
  55   TASK_JOIN,
  56   TASK_JOIN_REPLY,
  57   TASK_JOIN_LAST_REPLY,
  58   TASK_UPDATE
  59 } e_task_type_t;
  60
  61 typedef struct s_task_data {
  62   e_task_type_t type;                     // type of task
  63   int sender_id;                          // id parameter (used by some types of tasks)
  64   //int request_finger;                     // finger parameter (used by some types of tasks)
  65   int answer_id;                          // answer (used by some types of tasks)
  66   char answer_to[MAILBOX_NAME_SIZE];      // mailbox to send an answer to (if any)
  67   //const char* issuer_host_name;           // used for logging
  68   int steps;
  69   state_t state;
  70 } s_task_data_t, *task_data_t;
  71
  72 static void get_mailbox(int node_id, char* mailbox);
  73 static int domain(unsigned int a, unsigned int level);
  74 static int shl(int a, int b);
  75 static int closest_in_namespace_set(node_t node, int dest);
  76 static int routing_next(node_t node, int dest);
  77
  78 /**
  79  * \brief Gets the mailbox name of a host given its chord id.
  80  * \param node_id id of a node
  81  * \param mailbox pointer to where the mailbox name should be written
  82  * (there must be enough space)
  83  */
  84 static void get_mailbox(int node_id, char* mailbox)
  85 {
  86   snprintf(mailbox, MAILBOX_NAME_SIZE - 1, "%d", node_id);
  87 }
  88
  89 /** Get the specific level of a node id */
  90 unsigned int domain_mask = 0;
  91 static int domain(unsigned int a, unsigned int level)
  92 {
  93   if (domain_mask == 0)
  94     domain_mask = pow(2, DOMAIN_SIZE) - 1;
  95   unsigned int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
  96   return (a >> shift) & domain_mask;
  97 }
  98
  99 /* Get the shared domains between the two givens ids */
 100 static int shl(int a, int b) {
 101   int l = 0;
 102   while(l<LEVELS_COUNT && domain(a,l) == domain(b,l))
 103     l++;
 104   return l;
 105 }
 106
 107 /* Frees the memory used by a task and destroy it */
 108 static void task_free(void* task)
 109 {
 110   if(task != NULL){
 111     s_task_data_t* data = (s_task_data_t*)MSG_task_get_data(task);
 112     xbt_free(data->state);
 113     xbt_free(data);
 114     MSG_task_destroy(task);
 115   }
 116 }
 117
 118 /* Get the closest id to the dest in the node namespace_set */
 119 static int closest_in_namespace_set(node_t node, int dest) {
 120   int res = -1;
 121   if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) && (dest <= node->namespace_set[0])) {
 122     int best_dist = abs(node->id - dest);
 123     res = node->id;
 124     for (int i=0; i<NAMESPACE_SIZE; i++) {
 125       if (node->namespace_set[i]!=-1) {
 126         int dist = abs(node->namespace_set[i] - dest);
 127         if (dist<best_dist) {
 128           best_dist = dist;
 129           res = node->namespace_set[i];
 130         }
 131       }
 132     }
 133   }
 134   return res;
 135 }
 136
 137 /* Find the next node to forward a message to */
 138 static int routing_next(node_t node, int dest) {
 139   int closest = closest_in_namespace_set(node, dest);
 140   if (closest!=-1)
 141     return closest;
 142
 143   int l = shl(node->id, dest);
 144   int res = node->routing_table[l][domain(dest, l)];
 145   if (res != -1)
 146     return res;
 147
 148   //rare case
 149   int dist = abs(node->id - dest);
 150   for (int i=l; i<LEVELS_COUNT; i++) {
 151     for (int j=0; j<LEVEL_SIZE; j++) {
 152       res = node->routing_table[i][j];
 153       if (res!=-1 && abs(res - dest)<dist)
 154         return res;
 155     }
 156   }
 157
 158   for (int i=0; i<NEIGHBORHOOD_SIZE; i++) {
 159     res = node->neighborhood_set[i];
 160     if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
 161         return res;
 162   }
 163
 164   for (int i=0; i<NAMESPACE_SIZE; i++) {
 165     res = node->namespace_set[i];
 166     if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
 167         return res;
 168   }
 169
 170   return node->id;
 171 }
 172
 173 /* Get the corresponding state of a node */
 174 static state_t node_get_state(node_t node) {
 175   state_t state = xbt_new0(s_state_t,1);
 176   state->id = node->id;
 177   for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
 178     state->neighborhood_set[i] = node->neighborhood_set[i];
 179
 180   for (int i=0; i<LEVELS_COUNT; i++)
 181     for (int j=0; j<LEVEL_SIZE; j++)
 182       state->routing_table[i][j] = node->routing_table[i][j];
 183
 184   for (int i=0; i<NAMESPACE_SIZE; i++)
 185     state->namespace_set[i] = node->namespace_set[i];
 186
 187   return state;
 188 }
 189
 190 static void print_node_id(node_t node) {
 191   XBT_INFO(" Id: %i '%08x' ", node->id, node->id);
 192 }
 193
 194 static void print_node_neighborood_set(node_t node) {
 195   XBT_INFO(" Neighborhood:");
 196   for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
 197     XBT_INFO("  %08x", node->neighborhood_set[i]);
 198 }
 199
 200 static void print_node_routing_table(node_t node) {
 201   XBT_INFO(" Routing table:");
 202   for (int i=0; i<LEVELS_COUNT; i++){
 203     for (int j=0; j<LEVEL_SIZE; j++)
 204       XBT_INFO("  %08x ", node->routing_table[i][j]);
 205   }
 206 }
 207 /* Print the node namespace set */
 208 static void print_node_namespace_set(node_t node) {
 209   XBT_INFO(" Namespace:");
 210   for (int i=0; i<NAMESPACE_SIZE; i++)
 211     XBT_INFO("  %08x", node->namespace_set[i]);
 212 }
 213
 214 /* Print the node information */
 215 static void print_node(node_t node) {
 216   XBT_INFO("Node:");
 217   print_node_id(node);
 218   print_node_neighborood_set(node);
 219   print_node_routing_table(node);
 220   print_node_namespace_set(node);
 221 }
 222
 223 /** Handle a given task */
 224 static void handle_task(node_t node, msg_task_t task) {
 225   XBT_DEBUG("Handling task %p", task);
 226   char mailbox[MAILBOX_NAME_SIZE];
 227   int i;
 228   int j;
 229   int min;
 230   int max;
 231   int d;
 232   int next;
 233   msg_task_t task_sent;
 234   task_data_t req_data;
 235   task_data_t task_data = (task_data_t) MSG_task_get_data(task);
 236   e_task_type_t type = task_data->type;
 237   // If the node is not ready keep the task for later
 238   if (node->ready != 0 && !(type==TASK_JOIN_LAST_REPLY || type==TASK_JOIN_REPLY)) {
 239     XBT_DEBUG("Task pending %i", type);
 240     xbt_dynar_push(node->pending_tasks, &task);
 241     return;
 242   }
 243   switch (type) {
 244     /* Try to join the ring */
 245     case TASK_JOIN:
 246       next = routing_next(node, task_data->answer_id);
 247       XBT_DEBUG("Join request from %08x forwarding to %08x", task_data->answer_id, next);
 248       type = TASK_JOIN_LAST_REPLY;
 249
 250       req_data = xbt_new0(s_task_data_t,1);
 251       req_data->answer_id = task_data->sender_id;
 252       req_data->steps = task_data->steps + 1;
 253
 254       // if next different from current node forward the join
 255       if (next!=node->id) {
 256         get_mailbox(next, mailbox);
 257         task_data->sender_id = node->id;
 258         task_data->steps++;
 259         task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, task_data);
 260         if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) {
 261           XBT_DEBUG("Timeout expired when forwarding join to next %d", next);
 262           task_free(task_sent);
 263         }
 264         type = TASK_JOIN_REPLY;
 265       }
 266
 267       // send back the current node state to the joining node
 268       req_data->type = type;
 269       req_data->sender_id = node->id;
 270       get_mailbox(node->id, req_data->answer_to);
 271       req_data->state = node_get_state(node);
 272       task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 273       if (MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout)== MSG_TIMEOUT) {
 274         XBT_DEBUG("Timeout expired when sending back the current node state to the joining node to %d", node->id);
 275         task_free(task_sent);
 276       }
 277       break;
 278     /* Join reply from all the node touched by the join  */
 279     case TASK_JOIN_LAST_REPLY:
 280       // if last node touched reply, copy its namespace set
 281       // TODO: it works only if the two nodes are side to side (is it really the case ?)
 282       j = (task_data->sender_id < node->id) ? -1 : 0;
 283       for (i=0; i<NAMESPACE_SIZE/2; i++) {
 284         node->namespace_set[i] = task_data->state->namespace_set[i-j];
 285         node->namespace_set[NAMESPACE_SIZE-1-i] = task_data->state->namespace_set[NAMESPACE_SIZE-1-i-j-1];
 286       }
 287       node->namespace_set[NAMESPACE_SIZE/2+j] = task_data->sender_id;
 288       node->ready += task_data->steps + 1;
 289       /* no break */
 290     case TASK_JOIN_REPLY:
 291       XBT_DEBUG("Joining Reply");
 292
 293       // if first node touched reply, copy its neighborhood set
 294       if (task_data->sender_id == node->known_id) {
 295         node->neighborhood_set[0] = task_data->sender_id;
 296         for (i=1; i<NEIGHBORHOOD_SIZE; i++)
 297           node->neighborhood_set[i] = task_data->state->neighborhood_set[i-1];
 298       }
 299
 300       // copy the corresponding routing table levels
 301       min = (node->id==task_data->answer_id) ? 0 : shl(node->id, task_data->answer_id);
 302       max = shl(node->id, task_data->sender_id)+1;
 303       for (i=min;i<max;i++) {
 304         d = domain(node->id, i);
 305         for (j=0; j<LEVEL_SIZE; j++)
 306           if (d!=j)
 307             node->routing_table[i][j] =  task_data->state->routing_table[i][j];
 308           }
 309
 310       node->ready--;
 311       // if the node is ready, do all the pending tasks and send update to known nodes
 312       if (node->ready==0) {
 313         XBT_DEBUG("Node %i is ready!!!", node->id);
 314         while(xbt_dynar_length(node->pending_tasks)){
 315           msg_task_t task;
 316           xbt_dynar_shift(node->pending_tasks, &task);
 317           handle_task(node, task);
 318         }
 319
 320         for (i=0; i<NAMESPACE_SIZE; i++) {
 321           j = node->namespace_set[i];
 322           if (j!=-1) {
 323             XBT_DEBUG("Send update to %i", j);
 324             get_mailbox(j, mailbox);
 325
 326             req_data = xbt_new0(s_task_data_t,1);
 327             req_data->answer_id = node->id;
 328             req_data->steps = 0;
 329             req_data->type = TASK_UPDATE;
 330             req_data->sender_id = node->id;
 331             get_mailbox(node->id, req_data->answer_to);
 332             req_data->state = node_get_state(node);
 333             task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 334             if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) {
 335               XBT_DEBUG("Timeout expired when sending update to %d", j);
 336               task_free(task_sent);
 337             }
 338           }
 339         }
 340         }
 341       break;
 342     /* Received an update of state */
 343     case TASK_UPDATE:
 344       XBT_DEBUG("Task update %i !!!", node->id);
 345
 346       /* Update namespace ses */
 347       XBT_INFO("Task update from %i !!!", task_data->sender_id);
 348       XBT_INFO("Node:");
 349       print_node_id(node);
 350       print_node_namespace_set(node);
 351       int curr_namespace_set[NAMESPACE_SIZE];
 352       int task_namespace_set[NAMESPACE_SIZE+1];
 353
 354       // Copy the current namespace and the task state namespace with state->id in the middle
 355       i=0;
 356       for (; i<NAMESPACE_SIZE/2; i++){
 357         curr_namespace_set[i] = node->namespace_set[i];
 358         task_namespace_set[i] = task_data->state->namespace_set[i];
 359       }
 360       task_namespace_set[i] = task_data->state->id;
 361       for (; i<NAMESPACE_SIZE; i++){
 362         curr_namespace_set[i] = node->namespace_set[i];
 363         task_namespace_set[i+1] = task_data->state->namespace_set[i];
 364       }
 365
 366       // get the index of values before and after node->id in task_namespace
 367       min = -1;
 368       max = -1;
 369       for (i=0; i<=NAMESPACE_SIZE; i++) {
 370         j = task_namespace_set[i];
 371         if (j != -1 && j < node->id)
 372           min = i;
 373         if (j != -1 && max == -1 && j > node->id)
 374           max = i;
 375       }
 376
 377       // add lower elements
 378       j = NAMESPACE_SIZE/2-1;
 379       for (i=NAMESPACE_SIZE/2-1; i>=0; i--) {
 380         if (min<0) {
 381           node->namespace_set[i] = curr_namespace_set[j];
 382           j--;
 383         } else if (curr_namespace_set[j] == task_namespace_set[min]) {
 384           node->namespace_set[i] = curr_namespace_set[j];
 385           j--;
 386           min--;
 387         } else if (curr_namespace_set[j] > task_namespace_set[min]) {
 388           node->namespace_set[i] = curr_namespace_set[j];
 389           j--;
 390         } else {
 391           node->namespace_set[i] = task_namespace_set[min];
 392           min--;
 393         }
 394       }
 395
 396       // add greater elements
 397       j = NAMESPACE_SIZE/2;
 398       for (i=NAMESPACE_SIZE/2; i<NAMESPACE_SIZE; i++) {
 399         if (min<0 || max>=NAMESPACE_SIZE) {
 400          node->namespace_set[i] = curr_namespace_set[j];
 401          j++;
 402         } else if (curr_namespace_set[j] == -1) {
 403           node->namespace_set[i] = task_namespace_set[max];
 404           max++;
 405         } else if (curr_namespace_set[j] == task_namespace_set[max]) {
 406           node->namespace_set[i] = curr_namespace_set[j];
 407           j++;
 408           max++;
 409         } else if (curr_namespace_set[j] < task_namespace_set[max]) {
 410           node->namespace_set[i] = curr_namespace_set[j];
 411           j++;
 412         } else {
 413           node->namespace_set[i] = task_namespace_set[max];
 414           max++;
 415         }
 416       }
 417
 418       /* Update routing table */
 419       for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
 420         for (j=0; j<LEVEL_SIZE; j++) {
 421           if (node->routing_table[i][j]==-1 && task_data->state->routing_table[i][j]==-1)
 422             node->routing_table[i][j] = task_data->state->routing_table[i][j];
 423         }
 424       }
 425       break;
 426     default:
 427       THROW_IMPOSSIBLE;
 428   }
 429   task_free(task);
 430 }
 431
 432 /* Join the ring */
 433 static int join(node_t node){
 434   task_data_t req_data = xbt_new0(s_task_data_t,1);
 435   req_data->type = TASK_JOIN;
 436   req_data->sender_id = node->id;
 437   req_data->answer_id = node->id;
 438   req_data->steps = 0;
 439   get_mailbox(node->id, req_data->answer_to);
 440
 441   char mailbox[MAILBOX_NAME_SIZE];
 442   get_mailbox(node->known_id, mailbox);
 443
 444   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 445   XBT_DEBUG("Trying to join Pastry ring... (with node %s)", mailbox);
 446   if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) {
 447     XBT_DEBUG("Timeout expired when joining ring with node %d", node->known_id);
 448     task_free(task_sent);
 449   }
 450
 451   return 1;
 452 }
 453
 454 /**
 455  * \brief Node Function
 456  * Arguments:
 457  * - my id
 458  * - the id of a guy I know in the system (except for the first node)
 459  * - the time to sleep before I join (except for the first node)
 460  * - the deadline time
 461  */
 462 static int node(int argc, char *argv[])
 463 {
 464   double init_time = MSG_get_clock();
 465   msg_task_t task_received = NULL;
 466   int join_success = 0;
 467   double deadline;
 468   xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
 469   s_node_t node = {0};
 470   node.id = xbt_str_parse_int(argv[1], "Invalid ID: %s");
 471   node.known_id = -1;
 472   node.ready = -1;
 473   node.pending_tasks = xbt_dynar_new(sizeof(msg_task_t), NULL);
 474   get_mailbox(node.id, node.mailbox);
 475   XBT_DEBUG("New node with id %s (%08x)", node.mailbox, node.id);
 476
 477   for (int i=0; i<LEVELS_COUNT; i++){
 478     int d = domain(node.id, i);
 479     for (int j=0; j<LEVEL_SIZE; j++)
 480       node.routing_table[i][j] = (d==j) ? node.id : -1;
 481   }
 482
 483   for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
 484     node.neighborhood_set[i] = -1;
 485
 486   for (int i=0; i<NAMESPACE_SIZE; i++)
 487     node.namespace_set[i] = -1;
 488
 489   if (argc == 3) { // first ring
 490     XBT_DEBUG("Hey! Let's create the system.");
 491     deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
 492     node.ready = 0;
 493     XBT_DEBUG("Create a new Pastry ring...");
 494     join_success = 1;
 495   } else {
 496     node.known_id = xbt_str_parse_int(argv[2], "Invalid known ID: %s");
 497     double sleep_time = xbt_str_parse_double(argv[3], "Invalid sleep time: %s");
 498     deadline = xbt_str_parse_double(argv[4], "Invalid deadline: %s");
 499
 500     // sleep before starting
 501     XBT_DEBUG("Let's sleep during %f", sleep_time);
 502     MSG_process_sleep(sleep_time);
 503     XBT_DEBUG("Hey! Let's join the system.");
 504
 505     join_success = join(&node);
 506   }
 507
 508   if (join_success) {
 509     XBT_DEBUG("Waiting ….");
 510
 511     while (MSG_get_clock() < init_time + deadline
 512 //      && MSG_get_clock() < node.last_change_date + 1000
 513         && MSG_get_clock() < max_simulation_time) {
 514       if (node.comm_receive == NULL) {
 515         task_received = NULL;
 516         node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
 517         // FIXME: do not make MSG_task_irecv() calls from several functions
 518       }
 519       if (!MSG_comm_test(node.comm_receive)) {
 520         MSG_process_sleep(5);
 521       } else {
 522         // a transfer has occurred
 523
 524         msg_error_t status = MSG_comm_get_status(node.comm_receive);
 525
 526         if (status != MSG_OK) {
 527           XBT_DEBUG("Failed to receive a task. Nevermind.");
 528           MSG_comm_destroy(node.comm_receive);
 529           node.comm_receive = NULL;
 530         } else {
 531           // the task was successfully received
 532           MSG_comm_destroy(node.comm_receive);
 533           node.comm_receive = NULL;
 534           handle_task(&node, task_received);
 535         }
 536       }
 537
 538     }
 539   //Cleanup the receiving communication.
 540   if (node.comm_receive != NULL) {
 541     if (MSG_comm_test(node.comm_receive) && MSG_comm_get_status(node.comm_receive) == MSG_OK) {
 542       task_free(MSG_comm_get_task(node.comm_receive));
 543     }
 544     MSG_comm_destroy(node.comm_receive);
 545   }
 546
 547   }
 548   xbt_dynar_free(&node.pending_tasks);
 549   return 1;
 550 }
 551
 552 /** \brief Main function. */
 553 int main(int argc, char *argv[])
 554 {
 555   MSG_init(&argc, argv);
 556   xbt_assert(argc > 2,
 557        "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
 558        "\tExample: %s ../msg_platform.xml pastry10.xml\n",
 559        argv[0], argv[0]);
 560
 561   char **options = &argv[1];
 562   while (!strncmp(options[0], "-", 1)) {
 563     int length = strlen("-nb_bits=");
 564     if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
 565       nb_bits = xbt_str_parse_int(options[0] + length, "Invalid nb_bits parameter: %s");
 566       XBT_DEBUG("Set nb_bits to %d", nb_bits);
 567     } else {
 568       length = strlen("-timeout=");
 569       if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
 570         timeout = xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
 571         XBT_DEBUG("Set timeout to %d", timeout);
 572       } else {
 573         xbt_die("Invalid pastry option '%s'", options[0]);
 574       }
 575     }
 576     options++;
 577   }
 578
 579   MSG_create_environment(options[0]);
 580
 581   MSG_function_register("node", node);
 582   MSG_launch_application(options[1]);
 583
 584   msg_error_t res = MSG_main();
 585   XBT_INFO("Simulated time: %g", MSG_get_clock());
 586
 587   return res != MSG_OK;
 588 }