1 /* Copyright (c) 2013-2019. The SimGrid Team. All rights reserved. */
3 /* This program is free software; you can redistribute it and/or modify it
4 * under the terms of the license (GNU LGPL) which comes with this package. */
6 #include "simgrid/msg.h"
12 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example");
15 * - handle node departure *
16 * - handle objects on the network *
17 * - handle neighborhood in the update */
21 #define MAILBOX_NAME_SIZE 10
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
30 static int nb_bits = 16;
31 static int timeout = 50;
32 static int max_simulation_time = 1000;
34 typedef struct s_node {
35 int id; //128bits generated random(2^128 -1)
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];
42 msg_comm_t comm_receive; // current communication to receive
43 xbt_dynar_t pending_tasks;
45 typedef s_node_t* node_t;
46 typedef const s_node_t* const_node_t;
48 typedef struct s_state {
50 int namespace_set[NAMESPACE_SIZE];
51 int neighborhood_set[NEIGHBORHOOD_SIZE];
52 int routing_table[LEVELS_COUNT][LEVEL_SIZE];
54 typedef s_state_t* state_t;
56 /** Types of tasks exchanged between nodes. */
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
74 typedef s_task_data_t* task_data_t;
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 */
88 unsigned int domain_mask = 0;
89 static int domain(unsigned int a, unsigned int level)
92 domain_mask = pow(2, DOMAIN_SIZE) - 1;
93 unsigned int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
94 return (a >> shift) & domain_mask;
97 /* Get the shared domains between the two givens ids */
98 static int shl(int a, int b) {
100 while(l<LEVELS_COUNT && domain(a,l) == domain(b,l))
105 /* Frees the memory used by a task and destroy it */
106 static void task_free(void* task)
109 s_task_data_t* data = (s_task_data_t*)MSG_task_get_data(task);
110 xbt_free(data->state);
112 MSG_task_destroy(task);
116 /* Get the closest id to the dest in the node namespace_set */
117 static int closest_in_namespace_set(const_node_t node, int dest)
120 if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) && (dest <= node->namespace_set[0])) {
121 int best_dist = abs(node->id - dest);
123 for (int i=0; i<NAMESPACE_SIZE; i++) {
124 if (node->namespace_set[i]!=-1) {
125 int dist = abs(node->namespace_set[i] - dest);
126 if (dist<best_dist) {
128 res = node->namespace_set[i];
136 /* Find the next node to forward a message to */
137 static int routing_next(node_t node, int dest) {
138 int closest = closest_in_namespace_set(node, dest);
142 int l = shl(node->id, dest);
143 int res = node->routing_table[l][domain(dest, l)];
148 int dist = abs(node->id - dest);
149 for (int i=l; i<LEVELS_COUNT; i++) {
150 for (int j=0; j<LEVEL_SIZE; j++) {
151 res = node->routing_table[i][j];
152 if (res!=-1 && abs(res - dest)<dist)
157 for (int i=0; i<NEIGHBORHOOD_SIZE; i++) {
158 res = node->neighborhood_set[i];
159 if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
163 for (int i=0; i<NAMESPACE_SIZE; i++) {
164 res = node->namespace_set[i];
165 if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
172 /* Get the corresponding state of a node */
173 static state_t node_get_state(const_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];
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];
184 for (int i=0; i<NAMESPACE_SIZE; i++)
185 state->namespace_set[i] = node->namespace_set[i];
190 static void print_node_id(const_node_t node)
192 XBT_INFO(" Id: %i '%08x' ", node->id, (unsigned)node->id);
195 static void print_node_neighborood_set(const_node_t node)
197 XBT_INFO(" Neighborhood:");
198 for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
199 XBT_INFO(" %08x", (unsigned)node->neighborhood_set[i]);
202 static void print_node_routing_table(const_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]);
210 /* Print the node namespace set */
211 static void print_node_namespace_set(const_node_t node)
213 XBT_INFO(" Namespace:");
214 for (int i=0; i<NAMESPACE_SIZE; i++)
215 XBT_INFO(" %08x", (unsigned)node->namespace_set[i]);
218 /* Print the node information */
219 static void print_node(const_node_t node)
223 print_node_neighborood_set(node);
224 print_node_routing_table(node);
225 print_node_namespace_set(node);
228 /** Handle a given task */
229 static void handle_task(node_t node, msg_task_t task) {
230 XBT_DEBUG("Handling task %p", task);
231 char mailbox[MAILBOX_NAME_SIZE];
237 msg_task_t task_sent;
238 task_data_t req_data;
239 task_data_t task_data = (task_data_t) MSG_task_get_data(task);
240 e_task_type_t type = task_data->type;
241 // If the node is not ready keep the task for later
242 if (node->ready != 0 && !(type==TASK_JOIN_LAST_REPLY || type==TASK_JOIN_REPLY)) {
243 XBT_DEBUG("Task pending %u", type);
244 xbt_dynar_push(node->pending_tasks, &task);
248 /* Try to join the ring */
250 next = routing_next(node, task_data->answer_id);
251 XBT_DEBUG("Join request from %08x forwarding to %08x", (unsigned)task_data->answer_id, (unsigned)next);
252 type = TASK_JOIN_LAST_REPLY;
254 req_data = xbt_new0(s_task_data_t,1);
255 req_data->answer_id = task_data->sender_id;
256 req_data->steps = task_data->steps + 1;
258 // if next different from current node forward the join
259 if (next!=node->id) {
260 get_mailbox(next, mailbox);
261 task_data->sender_id = node->id;
263 task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, task_data);
264 if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) {
265 XBT_DEBUG("Timeout expired when forwarding join to next %d", next);
266 task_free(task_sent);
268 type = TASK_JOIN_REPLY;
271 // send back the current node state to the joining node
272 req_data->type = type;
273 req_data->sender_id = node->id;
274 get_mailbox(node->id, req_data->answer_to);
275 req_data->state = node_get_state(node);
276 task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
277 if (MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout)== MSG_TIMEOUT) {
278 XBT_DEBUG("Timeout expired when sending back the current node state to the joining node to %d", node->id);
279 task_free(task_sent);
282 /* Join reply from all the node touched by the join */
283 case TASK_JOIN_LAST_REPLY:
284 // if last node touched reply, copy its namespace set
285 // TODO: it works only if the two nodes are side to side (is it really the case ?)
286 j = (task_data->sender_id < node->id) ? -1 : 0;
287 for (i=0; i<NAMESPACE_SIZE/2; i++) {
288 node->namespace_set[i] = task_data->state->namespace_set[i-j];
289 node->namespace_set[NAMESPACE_SIZE-1-i] = task_data->state->namespace_set[NAMESPACE_SIZE-1-i-j-1];
291 node->namespace_set[NAMESPACE_SIZE/2+j] = task_data->sender_id;
292 node->ready += task_data->steps + 1;
294 case TASK_JOIN_REPLY:
295 XBT_DEBUG("Joining Reply");
297 // if first node touched reply, copy its neighborhood set
298 if (task_data->sender_id == node->known_id) {
299 node->neighborhood_set[0] = task_data->sender_id;
300 for (i=1; i<NEIGHBORHOOD_SIZE; i++)
301 node->neighborhood_set[i] = task_data->state->neighborhood_set[i-1];
304 // copy the corresponding routing table levels
305 min = (node->id==task_data->answer_id) ? 0 : shl(node->id, task_data->answer_id);
306 max = shl(node->id, task_data->sender_id)+1;
307 for (i=min;i<max;i++) {
308 int d = domain(node->id, i);
309 for (j=0; j<LEVEL_SIZE; j++)
311 node->routing_table[i][j] = task_data->state->routing_table[i][j];
315 // if the node is ready, do all the pending tasks and send update to known nodes
316 if (node->ready==0) {
317 XBT_DEBUG("Node %i is ready!!!", node->id);
318 while (!xbt_dynar_is_empty(node->pending_tasks)) {
320 xbt_dynar_shift(node->pending_tasks, &task);
321 handle_task(node, task);
324 for (i=0; i<NAMESPACE_SIZE; i++) {
325 j = node->namespace_set[i];
327 XBT_DEBUG("Send update to %i", j);
328 get_mailbox(j, mailbox);
330 req_data = xbt_new0(s_task_data_t,1);
331 req_data->answer_id = node->id;
333 req_data->type = TASK_UPDATE;
334 req_data->sender_id = node->id;
335 get_mailbox(node->id, req_data->answer_to);
336 req_data->state = node_get_state(node);
337 task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
338 if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) {
339 XBT_DEBUG("Timeout expired when sending update to %d", j);
340 task_free(task_sent);
346 /* Received an update of state */
348 XBT_DEBUG("Task update %i !!!", node->id);
350 /* Update namespace ses */
351 XBT_INFO("Task update from %i !!!", task_data->sender_id);
354 print_node_namespace_set(node);
355 int curr_namespace_set[NAMESPACE_SIZE];
356 int task_namespace_set[NAMESPACE_SIZE+1];
358 // Copy the current namespace and the task state namespace with state->id in the middle
360 for (; i<NAMESPACE_SIZE/2; i++){
361 curr_namespace_set[i] = node->namespace_set[i];
362 task_namespace_set[i] = task_data->state->namespace_set[i];
364 task_namespace_set[i] = task_data->state->id;
365 for (; i<NAMESPACE_SIZE; i++){
366 curr_namespace_set[i] = node->namespace_set[i];
367 task_namespace_set[i+1] = task_data->state->namespace_set[i];
370 // get the index of values before and after node->id in task_namespace
373 for (i=0; i<=NAMESPACE_SIZE; i++) {
374 j = task_namespace_set[i];
375 if (j != -1 && j < node->id)
377 if (j != -1 && max == -1 && j > node->id)
381 // add lower elements
382 j = NAMESPACE_SIZE/2-1;
383 for (i=NAMESPACE_SIZE/2-1; i>=0; i--) {
384 if (min < 0 || curr_namespace_set[j] > task_namespace_set[min]) {
385 node->namespace_set[i] = curr_namespace_set[j];
387 } else if (curr_namespace_set[j] == task_namespace_set[min]) {
388 node->namespace_set[i] = curr_namespace_set[j];
392 node->namespace_set[i] = task_namespace_set[min];
397 // add greater elements
398 j = NAMESPACE_SIZE/2;
399 for (i=NAMESPACE_SIZE/2; i<NAMESPACE_SIZE; i++) {
400 if (min<0 || max>=NAMESPACE_SIZE) {
401 node->namespace_set[i] = curr_namespace_set[j];
403 } else if (max >= 0){
404 if (curr_namespace_set[j] == -1 || curr_namespace_set[j] > task_namespace_set[max]) {
405 node->namespace_set[i] = task_namespace_set[max];
407 } else if (curr_namespace_set[j] == task_namespace_set[max]) {
408 node->namespace_set[i] = curr_namespace_set[j];
412 node->namespace_set[i] = curr_namespace_set[j];
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];
433 static int join(const_node_t node)
435 task_data_t req_data = xbt_new0(s_task_data_t,1);
436 req_data->type = TASK_JOIN;
437 req_data->sender_id = node->id;
438 req_data->answer_id = node->id;
440 get_mailbox(node->id, req_data->answer_to);
442 char mailbox[MAILBOX_NAME_SIZE];
443 get_mailbox(node->known_id, mailbox);
445 msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
446 XBT_DEBUG("Trying to join Pastry ring... (with node %s)", mailbox);
447 if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) {
448 XBT_DEBUG("Timeout expired when joining ring with node %d", node->known_id);
449 task_free(task_sent);
456 * @brief Node Function
459 * - the id of a guy I know in the system (except for the first node)
460 * - the time to sleep before I join (except for the first node)
461 * - the deadline time
463 static int node(int argc, char *argv[])
465 double init_time = MSG_get_clock();
466 msg_task_t task_received = NULL;
467 int join_success = 0;
469 xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
471 node.id = xbt_str_parse_int(argv[1], "Invalid ID: %s");
474 node.pending_tasks = xbt_dynar_new(sizeof(msg_task_t), NULL);
475 get_mailbox(node.id, node.mailbox);
476 XBT_DEBUG("New node with id %s (%08x)", node.mailbox, (unsigned)node.id);
478 for (int i=0; i<LEVELS_COUNT; i++){
479 int d = domain(node.id, i);
480 for (int j=0; j<LEVEL_SIZE; j++)
481 node.routing_table[i][j] = (d==j) ? node.id : -1;
484 for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
485 node.neighborhood_set[i] = -1;
487 for (int i=0; i<NAMESPACE_SIZE; i++)
488 node.namespace_set[i] = -1;
490 if (argc == 3) { // first ring
491 XBT_DEBUG("Hey! Let's create the system.");
492 deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
494 XBT_DEBUG("Create a new Pastry ring...");
497 node.known_id = xbt_str_parse_int(argv[2], "Invalid known ID: %s");
498 double sleep_time = xbt_str_parse_double(argv[3], "Invalid sleep time: %s");
499 deadline = xbt_str_parse_double(argv[4], "Invalid deadline: %s");
501 // sleep before starting
502 XBT_DEBUG("Let's sleep during %f", sleep_time);
503 MSG_process_sleep(sleep_time);
504 XBT_DEBUG("Hey! Let's join the system.");
506 join_success = join(&node);
510 XBT_DEBUG("Waiting ….");
512 while (MSG_get_clock() < init_time + deadline
513 // && MSG_get_clock() < node.last_change_date + 1000
514 && MSG_get_clock() < max_simulation_time) {
515 if (node.comm_receive == NULL) {
516 task_received = NULL;
517 node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
519 if (!MSG_comm_test(node.comm_receive)) {
520 MSG_process_sleep(5);
522 // a transfer has occurred
524 msg_error_t status = MSG_comm_get_status(node.comm_receive);
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;
531 // the task was successfully received
532 MSG_comm_destroy(node.comm_receive);
533 node.comm_receive = NULL;
534 handle_task(&node, task_received);
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));
544 MSG_comm_destroy(node.comm_receive);
548 xbt_dynar_free(&node.pending_tasks);
552 /** @brief Main function. */
553 int main(int argc, char *argv[])
555 MSG_init(&argc, argv);
557 "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
558 "\tExample: %s ../msg_platform.xml pastry10.xml\n",
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);
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);
573 xbt_die("Invalid pastry option '%s'", options[0]);
579 MSG_create_environment(options[0]);
581 MSG_function_register("node", node);
582 MSG_launch_application(options[1]);
584 msg_error_t res = MSG_main();
585 XBT_INFO("Simulated time: %g", MSG_get_clock());
587 return res != MSG_OK;