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

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