1 /* Copyright (c) 2013-2015. The SimGrid Team.
2 * All rights reserved. */
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. */
8 #include "simgrid/msg.h"
10 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example");
13 * - handle node departure *
14 * - handle objects on the network *
15 * - handle neighborood in the update */
19 #define MAILBOX_NAME_SIZE 10
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
28 static int nb_bits = 16;
29 static int timeout = 50;
30 static int max_simulation_time = 1000;
32 typedef struct s_node {
33 int id; //128bits generated random(2^128 -1)
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];
40 msg_comm_t comm_receive; // current communication to receive
41 xbt_fifo_t pending_tasks;
44 typedef struct s_state {
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;
51 /** Types of tasks exchanged between nodes. */
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
68 } s_task_data_t, *task_data_t;
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);
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)
82 static void get_mailbox(int node_id, char* mailbox)
84 snprintf(mailbox, MAILBOX_NAME_SIZE - 1, "%d", node_id);
87 /** Get the specific level of a node id */
89 static int domain(int a, int level) {
91 domain_mask = pow(2, DOMAIN_SIZE) - 1;
92 int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
93 return (a >> shift) & domain_mask;
96 /* Get the shared domains between the two givens ids */
97 static int shl(int a, int b) {
99 while(l<LEVELS_COUNT && domain(a,l) == domain(b,l))
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) {
108 if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) & (dest <= node->namespace_set[0])) {
109 best_dist = abs(node->id - dest);
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) {
117 res = node->namespace_set[i];
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);
132 int l = shl(node->id, dest);
133 res = node->routing_table[l][domain(dest, l)];
138 int dist = abs(node->id - dest);
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)
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)
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)
163 /* Get the corresponding state of a node */
164 static state_t node_get_state(node_t node) {
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];
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];
175 for (i=0; i<NAMESPACE_SIZE; i++)
176 state->namespace_set[i] = node->namespace_set[i];
181 /* Print the node id */
182 static void print_node_id(node_t node) {
184 printf(" id: %i '%08x' ", node->id, node->id);
185 for (i=0;i<LEVELS_COUNT;i++)
186 printf(" %x", domain(node->id, i));
190 /* * Print the node neighborhood set */
191 static void print_node_neighborood_set(node_t node) {
193 printf(" Neighborhood:\n");
194 for (i=0; i<NEIGHBORHOOD_SIZE; i++)
195 printf(" %08x\n", node->neighborhood_set[i]);
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++){
203 for (int j=0; j<LEVEL_SIZE; j++)
204 printf("%08x ", node->routing_table[i][j]);
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]);
217 /* Print the node information */
218 static void print_node(node_t node) {
221 print_node_neighborood_set(node);
222 print_node_routing_table(node);
223 print_node_namespace_set(node);
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);
242 /* Try to join the ring */
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;
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;
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;
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;
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);
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];
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");
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];
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++)
299 node->routing_table[i][j] = task_data->state->routing_table[i][j];
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);
307 while(xbt_fifo_size(node->pending_tasks))
308 handle_task(node, xbt_fifo_pop(node->pending_tasks));
310 for (i=0; i<NAMESPACE_SIZE; i++) {
311 j = node->namespace_set[i];
313 XBT_DEBUG("Send update to %i", j);
314 get_mailbox(j, mailbox);
316 req_data = xbt_new0(s_task_data_t,1);
317 req_data->answer_id = node->id;
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);
329 /* Received an update of state */
331 XBT_DEBUG("Task update %i !!!", node->id);
333 /* Update namespace ses */
334 printf("Task update from %i !!!\n", task_data->sender_id);
336 print_node_namespace_set(node);
337 int curr_namespace_set[NAMESPACE_SIZE];
338 int task_namespace_set[NAMESPACE_SIZE+1];
340 // Copy the current namedspace
341 // and the task state namespace with state->id in the middle
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];
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];
353 // get the index of values before and after node->id in task_namespace
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;
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]);
371 node->namespace_set[i] = curr_namespace_set[j];
373 } else if (curr_namespace_set[j] == task_namespace_set[min]) {
374 node->namespace_set[i] = curr_namespace_set[j];
376 } else if (curr_namespace_set[j] > task_namespace_set[min]) {
377 node->namespace_set[i] = curr_namespace_set[j];
380 node->namespace_set[i] = task_namespace_set[min];
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];
393 } else if (curr_namespace_set[j] == -1) {
394 node->namespace_set[i] = task_namespace_set[max];
396 } else if (curr_namespace_set[j] == task_namespace_set[max]) {
397 node->namespace_set[i] = curr_namespace_set[j];
399 } else if (curr_namespace_set[j] < task_namespace_set[max]) {
400 node->namespace_set[i] = curr_namespace_set[j];
403 node->namespace_set[i] = task_namespace_set[max];
407 print_node_namespace_set(node);
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];
419 /** \brief Initializes the current node as the first one of the system.
420 * \param node the current node
422 static void create(node_t node){
424 XBT_DEBUG("Create a new Pastry 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;
434 get_mailbox(node->id, req_data->answer_to);
436 char mailbox[MAILBOX_NAME_SIZE];
437 get_mailbox(node->known_id, mailbox);
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);
448 * \brief Node Function
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
455 static int node(int argc, char *argv[])
457 double init_time = MSG_get_clock();
458 msg_task_t task_received = NULL;
459 int join_success = 0;
461 xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
463 node.id = xbt_str_parse_int(argv[1], "Invalid ID: %s");
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);
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;
477 for (i=0; i<NEIGHBORHOOD_SIZE; i++)
478 node.neighborhood_set[i] = -1;
480 for (i=0; i<NAMESPACE_SIZE; i++)
481 node.namespace_set[i] = -1;
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");
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");
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.");
499 join_success = join(&node);
503 XBT_DEBUG("Waiting ….");
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
513 if (!MSG_comm_test(node.comm_receive)) {
514 MSG_process_sleep(5);
516 // a transfer has occurred
518 msg_error_t status = MSG_comm_get_status(node.comm_receive);
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;
526 // the task was successfully received
527 MSG_comm_destroy(node.comm_receive);
528 node.comm_receive = NULL;
529 handle_task(&node, task_received);
539 /** \brief Main function. */
540 int main(int argc, char *argv[])
542 MSG_init(&argc, argv);
544 "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
545 "\tExample: %s ../msg_platform.xml pastry10.xml\n",
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);
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);
560 xbt_die("Invalid chord option '%s'", options[0]);
566 MSG_create_environment(options[0]);
568 MSG_function_register("node", node);
569 MSG_launch_application(options[1]);
571 msg_error_t res = MSG_main();
572 XBT_INFO("Simulated time: %g", MSG_get_clock());
574 return res != MSG_OK;