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

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