examples/msg/dht-pastry/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 <math.h>
   8 #include "simgrid/msg.h"
   9
  10         XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example");
  11
  12 /* TODO:                               *
  13  *  - handle node departure            *
  14  *  - handle objects on the network    *
  15  *  - handle neighborood in the update */
  16
  17 #define COMM_SIZE 10
  18 #define COMP_SIZE 0
  19 #define MAILBOX_NAME_SIZE 10
  20
  21 #define DOMAIN_SIZE 4
  22 #define LEVELS_COUNT 8 // sizeof(int)*8/DOMAIN_SIZE
  23 #define LEVEL_SIZE 16 // 2^DOMAIN_SIZE
  24 #define NEIGHBORHOOD_SIZE 6
  25 #define NAMESPACE_SIZE 6
  26 #define MAILBOX_NAME_SIZE 10
  27
  28 static int nb_bits = 16;
  29 static int timeout = 50;
  30 static int max_simulation_time = 1000;
  31
  32 extern long int smx_total_comms;
  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_fifo_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 paramater (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(int a, 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 int domain_mask = 0;
  91 static int domain(int a, int level) {
  92   if (domain_mask == 0)
  93     domain_mask = pow(2, DOMAIN_SIZE) - 1;
  94   int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
  95   return (a >> shift) & domain_mask;
  96 }
  97
  98 /* Get the shared domains between the two givens ids */
  99 static int shl(int a, int b) {
 100   int l = 0;
 101   while(l<LEVELS_COUNT && domain(a,l) == domain(b,l))
 102     l++;
 103   return l;
 104 }
 105
 106 /* Get the closest id to the dest in the node namespace_set */
 107 static int closest_in_namespace_set(node_t node, int dest) {
 108   int best_dist;
 109   int res = -1;
 110   if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) & (dest <= node->namespace_set[0])) {
 111     best_dist = abs(node->id - dest);
 112     res = node->id;
 113     int i, dist;
 114     for (i=0; i<NAMESPACE_SIZE; i++) {
 115       if (node->namespace_set[i]!=-1) {
 116         dist = abs(node->namespace_set[i] - dest);
 117         if (dist<best_dist) {
 118           best_dist = dist;
 119           res = node->namespace_set[i];
 120         }
 121       }
 122     }
 123   }
 124   return res;
 125 }
 126
 127 /* Find the next node to forward a message to */
 128 static int routing_next(node_t node, int dest) {
 129   int closest = closest_in_namespace_set(node, dest);
 130   int res = -1;
 131   if (closest!=-1)
 132     return closest;
 133
 134   int l = shl(node->id, dest);
 135   res = node->routing_table[l][domain(dest, l)];
 136   if (res!=-1)
 137     return res;
 138
 139   //rare case
 140   int dist = abs(node->id - dest);
 141   int i,j;
 142   for (i=l; i<LEVELS_COUNT; i++) {
 143     for (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 (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 (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(node_t node) {
 167   int i,j;
 168   state_t state = xbt_new0(s_state_t,1);
 169   state->id = node->id;
 170   for (i=0; i<NEIGHBORHOOD_SIZE; i++)
 171     state->neighborhood_set[i] = node->neighborhood_set[i];
 172
 173   for (i=0; i<LEVELS_COUNT; i++)
 174     for (j=0; j<LEVEL_SIZE; j++)
 175       state->routing_table[i][j] = node->routing_table[i][j];
 176
 177   for (i=0; i<NAMESPACE_SIZE; i++)
 178     state->namespace_set[i] = node->namespace_set[i];
 179
 180   return state;
 181 }
 182
 183 /* Print the node id */
 184 static void print_node_id(node_t node) {
 185   int i;
 186   printf(" id: %i '%08x' ", node->id, node->id);
 187   for (i=0;i<LEVELS_COUNT;i++)
 188     printf(" %x", domain(node->id, i));
 189   printf("\n");
 190 }
 191
 192 /* * Print the node neighborhood set */
 193 static void print_node_neighborood_set(node_t node) {
 194   int i;
 195   printf(" Neighborhood:\n");
 196   for (i=0; i<NEIGHBORHOOD_SIZE; i++)
 197     printf("  %08x\n", node->neighborhood_set[i]);
 198 }
 199
 200 /* Print the routing table */
 201 static void print_node_routing_table(node_t node) {
 202   printf(" routing table:\n");
 203   for (int i=0; i<LEVELS_COUNT; i++){
 204     printf("  ");
 205     for (int j=0; j<LEVEL_SIZE; j++)
 206       printf("%08x ", node->routing_table[i][j]);
 207     printf("\n");
 208   }
 209 }
 210
 211 /* Print the node namespace set */
 212 static void print_node_namespace_set(node_t node) {
 213   printf(" namespace:\n");
 214   for (int i=0; i<NAMESPACE_SIZE; i++)
 215     printf("  %08x\n", node->namespace_set[i]);
 216   printf("\n");
 217 }
 218
 219 /* Print the node information */
 220 static void print_node(node_t node) {
 221   printf("Node:\n");
 222   print_node_id(node);
 223   print_node_neighborood_set(node);
 224   print_node_routing_table(node);
 225   print_node_namespace_set(node);
 226 }
 227
 228 /** Handle a given task */
 229 static void handle_task(node_t node, msg_task_t task) {
 230   XBT_DEBUG("Handling task %p", task);
 231   char mailbox[MAILBOX_NAME_SIZE];
 232   int i, j, min, max, d;
 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_fifo_push(node->pending_tasks, task);
 241     return;
 242   }
 243   switch (type) {
 244     /* Try to join the ring */
 245     case TASK_JOIN: {
 246       int 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         MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 261         type = TASK_JOIN_REPLY;
 262       }
 263
 264       // send back the current node state to the joining node
 265       req_data->type = type;
 266       req_data->sender_id = node->id;
 267       get_mailbox(node->id, req_data->answer_to);
 268       req_data->state = node_get_state(node);
 269       task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 270       MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout);
 271       break;
 272     }
 273     /* Join reply from all the node touched by the join  */
 274     case TASK_JOIN_LAST_REPLY:
 275       // if last node touched reply, copy its namespace set
 276       // TODO: it's work only if the two nodes are side to side (is it really the case ?)
 277       j = (task_data->sender_id < node->id) ? -1 : 0;
 278       for (i=0; i<NAMESPACE_SIZE/2; i++) {
 279         node->namespace_set[i] = task_data->state->namespace_set[i-j];
 280         node->namespace_set[NAMESPACE_SIZE-1-i] = task_data->state->namespace_set[NAMESPACE_SIZE-1-i-j-1];
 281       }
 282       node->namespace_set[NAMESPACE_SIZE/2+j] = task_data->sender_id;
 283       node->ready += task_data->steps + 1;
 284     case TASK_JOIN_REPLY:
 285       XBT_DEBUG("Joining Reply");
 286
 287       // if first node touched reply, copy its neighborhood set
 288       if (task_data->sender_id == node->known_id) {
 289         node->neighborhood_set[0] = task_data->sender_id;
 290         for (i=1; i<NEIGHBORHOOD_SIZE; i++)
 291           node->neighborhood_set[i] = task_data->state->neighborhood_set[i-1];
 292       }
 293
 294       // copy the corresponding routing table levels
 295       min = (node->id==task_data->answer_id) ? 0 : shl(node->id, task_data->answer_id);
 296       max = shl(node->id, task_data->sender_id)+1;
 297       for (i=min;i<max;i++) {
 298         d = domain(node->id, i);
 299         for (j=0; j<LEVEL_SIZE; j++)
 300           if (d!=j)
 301             node->routing_table[i][j] =  task_data->state->routing_table[i][j];
 302           }
 303
 304       node->ready--;
 305       // if the node is ready, do all the pending tasks and send update to known nodes
 306       if (node->ready==0) {
 307         XBT_DEBUG("Node %i is ready!!!", node->id);
 308
 309         while(xbt_fifo_size(node->pending_tasks))
 310           handle_task(node, xbt_fifo_pop(node->pending_tasks));
 311
 312         for (i=0; i<NAMESPACE_SIZE; i++) {
 313           j = node->namespace_set[i];
 314           if (j!=-1) {
 315             XBT_DEBUG("Send update to %i", j);
 316             get_mailbox(j, mailbox);
 317
 318             req_data = xbt_new0(s_task_data_t,1);
 319             req_data->answer_id = node->id;
 320             req_data->steps = 0;
 321             req_data->type = TASK_UPDATE;
 322             req_data->sender_id = node->id;
 323             get_mailbox(node->id, req_data->answer_to);
 324             req_data->state = node_get_state(node);
 325             task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
 326             MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 327           }
 328         }
 329         }
 330       break;
 331     /* Received an update of state */
 332     case TASK_UPDATE:
 333       XBT_DEBUG("Task update %i !!!", node->id);
 334
 335       /* Update namespace ses */
 336       printf("Task update from %i !!!\n", task_data->sender_id);
 337       print_node_id(node);
 338       print_node_namespace_set(node);
 339       int curr_namespace_set[NAMESPACE_SIZE];
 340       int task_namespace_set[NAMESPACE_SIZE+1];
 341
 342       // Copy the current namedspace
 343       // and the task state namespace with state->id in the middle
 344       i=0;
 345       for (; i<NAMESPACE_SIZE/2; i++){
 346         curr_namespace_set[i] = node->namespace_set[i];
 347         task_namespace_set[i] = task_data->state->namespace_set[i];
 348       }
 349       task_namespace_set[i] = task_data->state->id;
 350       for (; i<NAMESPACE_SIZE; i++){
 351         curr_namespace_set[i] = node->namespace_set[i];
 352         task_namespace_set[i+1] = task_data->state->namespace_set[i];
 353       }
 354
 355       // get the index of values before and after node->id in task_namespace
 356       min = -1;
 357       max = -1;
 358       for (i=0; i<=NAMESPACE_SIZE; i++) {
 359         j = task_namespace_set[i];
 360         if (i<NAMESPACE_SIZE)
 361           printf("%08x %08x | ", j, curr_namespace_set[i]);
 362         if (j != -1 && j < node->id) min = i;
 363         if (j != -1 && max == -1 && j > node->id) max = i;
 364       }
 365       printf("\n");
 366
 367       // add lower elements
 368       j = NAMESPACE_SIZE/2-1;
 369       for (i=NAMESPACE_SIZE/2-1; i>=0; i--) {
 370         printf("i:%i, j:%i, min:%i, currj:%08x, taskmin:%08x\n", i, j, min, curr_namespace_set[j],
 371                task_namespace_set[min]);
 372         if (min<0) {
 373           node->namespace_set[i] = curr_namespace_set[j];
 374           j--;
 375         } else if (curr_namespace_set[j] == task_namespace_set[min]) {
 376           node->namespace_set[i] = curr_namespace_set[j];
 377           j--; min--;
 378         } else if (curr_namespace_set[j] > task_namespace_set[min]) {
 379           node->namespace_set[i] = curr_namespace_set[j];
 380           j--;
 381         } else {
 382           node->namespace_set[i] = task_namespace_set[min];
 383           min--;
 384         }
 385       }
 386
 387       // add greater elements
 388       j = NAMESPACE_SIZE/2;
 389       for (i=NAMESPACE_SIZE/2; i<NAMESPACE_SIZE; i++) {
 390         printf("i:%i, j:%i, max:%i, currj:%08x, taskmax:%08x\n", i, j, max, curr_namespace_set[j],
 391                task_namespace_set[max]);
 392         if (min<0 || max>=NAMESPACE_SIZE) {
 393          node->namespace_set[i] = curr_namespace_set[j];
 394          j++;
 395         } else if (curr_namespace_set[j] == -1) {
 396           node->namespace_set[i] = task_namespace_set[max];
 397           max++;
 398         } else if (curr_namespace_set[j] == task_namespace_set[max]) {
 399           node->namespace_set[i] = curr_namespace_set[j];
 400           j++; max++;
 401         } else if (curr_namespace_set[j] < task_namespace_set[max]) {
 402           node->namespace_set[i] = curr_namespace_set[j];
 403           j++;
 404         } else {
 405           node->namespace_set[i] = task_namespace_set[max];
 406           max++;
 407         }
 408       }
 409       print_node_namespace_set(node);
 410
 411       /* Update routing table */
 412       for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
 413         for (j=0; j<LEVEL_SIZE; j++) {
 414           if (node->routing_table[i][j]==-1 && task_data->state->routing_table[i][j]==-1)
 415             node->routing_table[i][j] = task_data->state->routing_table[i][j];
 416         }
 417       }
 418   }
 419 }
 420
 421 /** \brief Initializes the current node as the first one of the system.
 422  *  \param node the current node
 423  */
 424 static void create(node_t node){
 425   node->ready = 0;
 426   XBT_DEBUG("Create a new Pastry ring...");
 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   MSG_task_send_with_timeout(task_sent, mailbox, timeout);
 444
 445   return 1;
 446 }
 447
 448
 449 /**
 450  * \brief Node Function
 451  * Arguments:
 452  * - my id
 453  * - the id of a guy I know in the system (except for the first node)
 454  * - the time to sleep before I join (except for the first node)
 455  * - the deadline time
 456  */
 457 static int node(int argc, char *argv[])
 458 {
 459   double init_time = MSG_get_clock();
 460   msg_task_t task_received = NULL;
 461   int join_success = 0;
 462   double deadline;
 463   xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
 464   s_node_t node = {0};
 465   node.id = xbt_str_parse_int(argv[1], "Invalid ID: %s");
 466   node.known_id = -1;
 467   node.ready = -1;
 468   node.pending_tasks = xbt_fifo_new();
 469   get_mailbox(node.id, node.mailbox);
 470   XBT_DEBUG("New node with id %s (%08x)", node.mailbox, node.id);
 471
 472   int i,j,d;
 473   for (i=0; i<LEVELS_COUNT; i++){
 474     d = domain(node.id, i);
 475     for (j=0; j<LEVEL_SIZE; j++)
 476       node.routing_table[i][j] = (d==j) ? node.id : -1;
 477   }
 478
 479   for (i=0; i<NEIGHBORHOOD_SIZE; i++)
 480     node.neighborhood_set[i] = -1;
 481
 482   for (i=0; i<NAMESPACE_SIZE; i++)
 483     node.namespace_set[i] = -1;
 484
 485   if (argc == 3) { // first ring
 486     XBT_DEBUG("Hey! Let's create the system.");
 487     deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
 488     create(&node);
 489     join_success = 1;
 490   }
 491   else {
 492     node.known_id = xbt_str_parse_int(argv[2], "Invalid known ID: %s");
 493     double sleep_time = xbt_str_parse_double(argv[3], "Invalid sleep time: %s");
 494     deadline = xbt_str_parse_double(argv[4], "Invalid deadline: %s");
 495
 496     // sleep before starting
 497     XBT_DEBUG("Let's sleep during %f", sleep_time);
 498     MSG_process_sleep(sleep_time);
 499     XBT_DEBUG("Hey! Let's join the system.");
 500
 501     join_success = join(&node);
 502   }
 503
 504   if (join_success) {
 505     XBT_DEBUG("Waiting ….");
 506
 507     while (MSG_get_clock() < init_time + deadline
 508 //      && MSG_get_clock() < node.last_change_date + 1000
 509         && MSG_get_clock() < max_simulation_time) {
 510       if (node.comm_receive == NULL) {
 511         task_received = NULL;
 512         node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
 513         // FIXME: do not make MSG_task_irecv() calls from several functions
 514       }
 515       if (!MSG_comm_test(node.comm_receive)) {
 516         MSG_process_sleep(5);
 517       } else {
 518         // a transfer has occurred
 519
 520         msg_error_t status = MSG_comm_get_status(node.comm_receive);
 521
 522         if (status != MSG_OK) {
 523           XBT_DEBUG("Failed to receive a task. Nevermind.");
 524           MSG_comm_destroy(node.comm_receive);
 525           node.comm_receive = NULL;
 526         }
 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     print_node(&node);
 537   }
 538   return 1;
 539 }
 540
 541 /** \brief Main function. */
 542 int main(int argc, char *argv[])
 543 {
 544   MSG_init(&argc, argv);
 545   xbt_assert(argc > 2,
 546        "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
 547        "\tExample: %s ../msg_platform.xml pastry10.xml\n",
 548        argv[0], argv[0]);
 549
 550   char **options = &argv[1];
 551   while (!strncmp(options[0], "-", 1)) {
 552     int length = strlen("-nb_bits=");
 553     if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
 554       nb_bits = xbt_str_parse_int(options[0] + length, "Invalid nb_bits parameter: %s");
 555       XBT_DEBUG("Set nb_bits to %d", nb_bits);
 556     } else {
 557       length = strlen("-timeout=");
 558       if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
 559         timeout = xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
 560         XBT_DEBUG("Set timeout to %d", timeout);
 561       } else {
 562         xbt_die("Invalid chord option '%s'", options[0]);
 563       }
 564     }
 565     options++;
 566   }
 567
 568   MSG_create_environment(options[0]);
 569
 570   MSG_function_register("node", node);
 571   MSG_launch_application(options[1]);
 572
 573   msg_error_t res = MSG_main();
 574   XBT_CRITICAL("Messages created: %ld", smx_total_comms);
 575   XBT_INFO("Simulated time: %g", MSG_get_clock());
 576
 577   return res != MSG_OK;
 578 }