Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Merge branch 'master' of scm.gforge.inria.fr:/gitroot/simgrid/simgrid
[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 <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 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   int i;
184   printf(" id: %i '%08x' ", node->id, node->id);
185   for (i=0;i<LEVELS_COUNT;i++)
186     printf(" %x", domain(node->id, i));
187   printf("\n");
188 }
189
190 /* * Print the node neighborhood set */
191 static void print_node_neighborood_set(node_t node) {
192   int i;
193   printf(" Neighborhood:\n");
194   for (i=0; i<NEIGHBORHOOD_SIZE; i++)
195     printf("  %08x\n", node->neighborhood_set[i]);
196 }
197
198 /* Print the routing table */
199 static void print_node_routing_table(node_t node) {
200   printf(" routing table:\n");
201   for (int i=0; i<LEVELS_COUNT; i++){
202     printf("  ");
203     for (int j=0; j<LEVEL_SIZE; j++)
204       printf("%08x ", node->routing_table[i][j]);
205     printf("\n");
206   }
207 }
208
209 /* Print the node namespace set */
210 static void print_node_namespace_set(node_t node) {
211   printf(" namespace:\n");
212   for (int i=0; i<NAMESPACE_SIZE; i++)
213     printf("  %08x\n", node->namespace_set[i]);
214   printf("\n");
215 }
216
217 /* Print the node information */
218 static void print_node(node_t node) {
219   printf("Node:\n");
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, j, min, max, d;
231   msg_task_t task_sent;
232   task_data_t req_data;
233   task_data_t task_data = (task_data_t) MSG_task_get_data(task);
234   e_task_type_t type = task_data->type;
235   // If the node is not ready keep the task for later
236   if (node->ready != 0 && !(type==TASK_JOIN_LAST_REPLY || type==TASK_JOIN_REPLY)) {
237     XBT_DEBUG("Task pending %i", type);
238     xbt_fifo_push(node->pending_tasks, task);
239     return;
240   }
241   switch (type) {
242     /* Try to join the ring */
243     case TASK_JOIN: {
244       int next = routing_next(node, task_data->answer_id);
245       XBT_DEBUG("Join request from %08x forwarding to %08x", task_data->answer_id, next);      
246       type = TASK_JOIN_LAST_REPLY;
247
248       req_data = xbt_new0(s_task_data_t,1);
249       req_data->answer_id = task_data->sender_id;
250       req_data->steps = task_data->steps + 1;
251       
252       // if next different from current node forward the join
253       if (next!=node->id) {
254         get_mailbox(next, mailbox);
255         task_data->sender_id = node->id;
256         task_data->steps++;
257         task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, task_data);
258         MSG_task_send_with_timeout(task_sent, mailbox, timeout);
259         type = TASK_JOIN_REPLY;
260       } 
261       
262       // send back the current node state to the joining node
263       req_data->type = type;
264       req_data->sender_id = node->id;
265       get_mailbox(node->id, req_data->answer_to);
266       req_data->state = node_get_state(node);
267       task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
268       MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout);
269       break;
270     }
271     /* Join reply from all the node touched by the join  */
272     case TASK_JOIN_LAST_REPLY:
273       // if last node touched reply, copy its namespace set
274       // TODO: it's work only if the two nodes are side to side (is it really the case ?)
275       j = (task_data->sender_id < node->id) ? -1 : 0;
276       for (i=0; i<NAMESPACE_SIZE/2; i++) {
277         node->namespace_set[i] = task_data->state->namespace_set[i-j];
278         node->namespace_set[NAMESPACE_SIZE-1-i] = task_data->state->namespace_set[NAMESPACE_SIZE-1-i-j-1];
279       }
280       node->namespace_set[NAMESPACE_SIZE/2+j] = task_data->sender_id;
281       node->ready += task_data->steps + 1;
282     case TASK_JOIN_REPLY:
283       XBT_DEBUG("Joining Reply");
284
285       // if first node touched reply, copy its neighborhood set
286       if (task_data->sender_id == node->known_id) {
287         node->neighborhood_set[0] = task_data->sender_id;
288         for (i=1; i<NEIGHBORHOOD_SIZE; i++)
289           node->neighborhood_set[i] = task_data->state->neighborhood_set[i-1];
290       }
291
292       // copy the corresponding routing table levels
293       min = (node->id==task_data->answer_id) ? 0 : shl(node->id, task_data->answer_id);
294       max = shl(node->id, task_data->sender_id)+1;
295       for (i=min;i<max;i++) {
296         d = domain(node->id, i); 
297         for (j=0; j<LEVEL_SIZE; j++)
298           if (d!=j)
299             node->routing_table[i][j] =  task_data->state->routing_table[i][j];
300           }
301
302       node->ready--;
303       // if the node is ready, do all the pending tasks and send update to known nodes
304       if (node->ready==0) {
305         XBT_DEBUG("Node %i is ready!!!", node->id);
306
307         while(xbt_fifo_size(node->pending_tasks))
308           handle_task(node, xbt_fifo_pop(node->pending_tasks));
309
310         for (i=0; i<NAMESPACE_SIZE; i++) {
311           j = node->namespace_set[i];
312           if (j!=-1) {
313             XBT_DEBUG("Send update to %i", j);
314             get_mailbox(j, mailbox);
315
316             req_data = xbt_new0(s_task_data_t,1);
317             req_data->answer_id = node->id;
318             req_data->steps = 0;
319             req_data->type = TASK_UPDATE;
320             req_data->sender_id = node->id;
321             get_mailbox(node->id, req_data->answer_to);
322             req_data->state = node_get_state(node);
323             task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
324             MSG_task_send_with_timeout(task_sent, mailbox, timeout);
325           }
326         }
327         }
328       break;
329     /* Received an update of state */
330     case TASK_UPDATE:
331       XBT_DEBUG("Task update %i !!!", node->id);
332
333       /* Update namespace ses */
334       printf("Task update from %i !!!\n", task_data->sender_id);
335       print_node_id(node);
336       print_node_namespace_set(node);
337       int curr_namespace_set[NAMESPACE_SIZE];
338       int task_namespace_set[NAMESPACE_SIZE+1];
339       
340       // Copy the current namedspace
341       // and the task state namespace with state->id in the middle
342       i=0;
343       for (; i<NAMESPACE_SIZE/2; i++){
344         curr_namespace_set[i] = node->namespace_set[i];
345         task_namespace_set[i] = task_data->state->namespace_set[i];
346       }
347       task_namespace_set[i] = task_data->state->id;
348       for (; i<NAMESPACE_SIZE; i++){
349         curr_namespace_set[i] = node->namespace_set[i];  
350         task_namespace_set[i+1] = task_data->state->namespace_set[i];
351       }
352
353       // get the index of values before and after node->id in task_namespace
354       min = -1;
355       max = -1;
356       for (i=0; i<=NAMESPACE_SIZE; i++) {
357         j = task_namespace_set[i];
358         if (i<NAMESPACE_SIZE)
359           printf("%08x %08x | ", j, curr_namespace_set[i]);
360         if (j != -1 && j < node->id) min = i;
361         if (j != -1 && max == -1 && j > node->id) max = i;
362       }
363       printf("\n");
364
365       // add lower elements
366       j = NAMESPACE_SIZE/2-1;
367       for (i=NAMESPACE_SIZE/2-1; i>=0; i--) {
368         printf("i:%i, j:%i, min:%i, currj:%08x, taskmin:%08x\n", i, j, min, curr_namespace_set[j],
369                task_namespace_set[min]);
370         if (min<0) {
371           node->namespace_set[i] = curr_namespace_set[j];
372           j--;
373         } else if (curr_namespace_set[j] == task_namespace_set[min]) {
374           node->namespace_set[i] = curr_namespace_set[j];
375           j--; min--;
376         } else if (curr_namespace_set[j] > task_namespace_set[min]) {
377           node->namespace_set[i] = curr_namespace_set[j];
378           j--;
379         } else {
380           node->namespace_set[i] = task_namespace_set[min];
381           min--;
382         }
383       }
384
385       // add greater elements
386       j = NAMESPACE_SIZE/2;
387       for (i=NAMESPACE_SIZE/2; i<NAMESPACE_SIZE; i++) {
388         printf("i:%i, j:%i, max:%i, currj:%08x, taskmax:%08x\n", i, j, max, curr_namespace_set[j],
389                task_namespace_set[max]);
390         if (min<0 || max>=NAMESPACE_SIZE) {
391          node->namespace_set[i] = curr_namespace_set[j];
392          j++;
393         } else if (curr_namespace_set[j] == -1) {
394           node->namespace_set[i] = task_namespace_set[max];
395           max++;
396         } else if (curr_namespace_set[j] == task_namespace_set[max]) {
397           node->namespace_set[i] = curr_namespace_set[j];
398           j++; max++;
399         } else if (curr_namespace_set[j] < task_namespace_set[max]) {
400           node->namespace_set[i] = curr_namespace_set[j];
401           j++;
402         } else {
403           node->namespace_set[i] = task_namespace_set[max];
404           max++;
405         }
406       }
407       print_node_namespace_set(node);
408
409       /* Update routing table */
410       for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
411         for (j=0; j<LEVEL_SIZE; j++) {
412           if (node->routing_table[i][j]==-1 && task_data->state->routing_table[i][j]==-1)
413             node->routing_table[i][j] = task_data->state->routing_table[i][j];
414         }
415       }
416   }
417 }
418
419 /** \brief Initializes the current node as the first one of the system.
420  *  \param node the current node
421  */
422 static void create(node_t node){
423   node->ready = 0;
424   XBT_DEBUG("Create a new Pastry ring...");
425 }
426
427 /* Join the ring */
428 static int join(node_t node){
429   task_data_t req_data = xbt_new0(s_task_data_t,1);
430   req_data->type = TASK_JOIN;
431   req_data->sender_id = node->id;
432   req_data->answer_id = node->id;
433   req_data->steps = 0;
434   get_mailbox(node->id, req_data->answer_to);
435
436   char mailbox[MAILBOX_NAME_SIZE];
437   get_mailbox(node->known_id, mailbox);
438
439   msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
440   XBT_DEBUG("Trying to join Pastry ring... (with node %s)", mailbox);
441   MSG_task_send_with_timeout(task_sent, mailbox, timeout);
442
443   return 1;
444 }
445
446
447 /**
448  * \brief Node Function
449  * Arguments:
450  * - my id
451  * - the id of a guy I know in the system (except for the first node)
452  * - the time to sleep before I join (except for the first node)
453  * - the deadline time
454  */
455 static int node(int argc, char *argv[])
456 {
457   double init_time = MSG_get_clock();
458   msg_task_t task_received = NULL;  
459   int join_success = 0;  
460   double deadline;
461   xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
462   s_node_t node = {0};
463   node.id = xbt_str_parse_int(argv[1], "Invalid ID: %s");
464   node.known_id = -1;
465   node.ready = -1;
466   node.pending_tasks = xbt_fifo_new();
467   get_mailbox(node.id, node.mailbox);
468   XBT_DEBUG("New node with id %s (%08x)", node.mailbox, node.id);
469   
470   int i,j,d;
471   for (i=0; i<LEVELS_COUNT; i++){
472     d = domain(node.id, i);
473     for (j=0; j<LEVEL_SIZE; j++)
474       node.routing_table[i][j] = (d==j) ? node.id : -1;
475   }
476
477   for (i=0; i<NEIGHBORHOOD_SIZE; i++)
478     node.neighborhood_set[i] = -1;
479
480   for (i=0; i<NAMESPACE_SIZE; i++)
481     node.namespace_set[i] = -1;
482
483   if (argc == 3) { // first ring
484     XBT_DEBUG("Hey! Let's create the system.");
485     deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
486     create(&node);
487     join_success = 1;
488   }
489   else {
490     node.known_id = xbt_str_parse_int(argv[2], "Invalid known ID: %s");
491     double sleep_time = xbt_str_parse_double(argv[3], "Invalid sleep time: %s");
492     deadline = xbt_str_parse_double(argv[4], "Invalid deadline: %s");
493
494     // sleep before starting
495     XBT_DEBUG("Let's sleep during %f", sleep_time);
496     MSG_process_sleep(sleep_time);
497     XBT_DEBUG("Hey! Let's join the system.");
498
499     join_success = join(&node);
500   }
501
502   if (join_success) {
503     XBT_DEBUG("Waiting ….");
504
505     while (MSG_get_clock() < init_time + deadline
506 //      && MSG_get_clock() < node.last_change_date + 1000
507         && MSG_get_clock() < max_simulation_time) {
508       if (node.comm_receive == NULL) {
509         task_received = NULL;
510         node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
511         // FIXME: do not make MSG_task_irecv() calls from several functions
512       }
513       if (!MSG_comm_test(node.comm_receive)) {
514         MSG_process_sleep(5);
515       } else {
516         // a transfer has occurred
517
518         msg_error_t status = MSG_comm_get_status(node.comm_receive);
519
520         if (status != MSG_OK) {
521           XBT_DEBUG("Failed to receive a task. Nevermind.");
522           MSG_comm_destroy(node.comm_receive);
523           node.comm_receive = NULL;
524         }
525         else {
526           // the task was successfully received
527           MSG_comm_destroy(node.comm_receive);
528           node.comm_receive = NULL;
529           handle_task(&node, task_received);
530         }
531       }
532
533     }
534     print_node(&node);
535   }
536   return 1;
537 }
538
539 /** \brief Main function. */
540 int main(int argc, char *argv[])
541 {
542   MSG_init(&argc, argv);
543   xbt_assert(argc > 2, 
544        "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
545        "\tExample: %s ../msg_platform.xml pastry10.xml\n", 
546        argv[0], argv[0]);
547
548   char **options = &argv[1];
549   while (!strncmp(options[0], "-", 1)) {
550     int length = strlen("-nb_bits=");
551     if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
552       nb_bits = xbt_str_parse_int(options[0] + length, "Invalid nb_bits parameter: %s");
553       XBT_DEBUG("Set nb_bits to %d", nb_bits);
554     } else {
555       length = strlen("-timeout=");
556       if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
557         timeout = xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
558         XBT_DEBUG("Set timeout to %d", timeout);
559       } else {
560         xbt_die("Invalid chord option '%s'", options[0]);
561       }
562     }
563     options++;
564   }
565
566   MSG_create_environment(options[0]);
567
568   MSG_function_register("node", node);
569   MSG_launch_application(options[1]);
570
571   msg_error_t res = MSG_main();
572   XBT_INFO("Simulated time: %g", MSG_get_clock());
573
574   return res != MSG_OK;
575 }