Logo AND Algorithmique Numérique Distribuée

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