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