Logo AND Algorithmique Numérique Distribuée

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