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

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