1 /* Copyright (c) 2013-2021. 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/actor.h"
7 #include "simgrid/comm.h"
8 #include "simgrid/engine.h"
9 #include "simgrid/mailbox.h"
11 #include "xbt/dynar.h"
15 #include "xbt/sysdep.h"
19 XBT_LOG_NEW_DEFAULT_CATEGORY(pastry, "Messages specific for this example");
23 #define MAILBOX_NAME_SIZE 10
26 #define LEVELS_COUNT 8 // sizeof(int)*8/DOMAIN_SIZE
27 #define LEVEL_SIZE 16 // 2^DOMAIN_SIZE
28 #define NEIGHBORHOOD_SIZE 6
29 #define NAMESPACE_SIZE 6
30 #define MAILBOX_NAME_SIZE 10
32 static int nb_bits = 16;
33 static int timeout = 50;
34 static int max_simulation_time = 1000;
36 typedef struct s_node {
37 int id; // 128bits generated random(2^128 -1)
40 int namespace_set[NAMESPACE_SIZE];
41 int neighborhood_set[NEIGHBORHOOD_SIZE];
42 int routing_table[LEVELS_COUNT][LEVEL_SIZE];
44 sg_comm_t comm_receive; // current communication to receive
45 xbt_dynar_t pending_messages;
47 typedef s_node_t* node_t;
48 typedef const s_node_t* const_node_t;
50 typedef struct s_state {
52 int namespace_set[NAMESPACE_SIZE];
53 int neighborhood_set[NEIGHBORHOOD_SIZE];
54 int routing_table[LEVELS_COUNT][LEVEL_SIZE];
56 typedef s_state_t* state_t;
58 /** Types of tasks exchanged between nodes. */
59 typedef enum { JOIN, JOIN_REPLY, JOIN_LAST_REPLY, UPDATE } e_message_type_t;
61 typedef struct s_pastry_message {
62 e_message_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 sg_mailbox_t answer_to; // mailbox to send an answer to (if any)
70 typedef s_pastry_message_t* pastry_message_t;
71 typedef const s_pastry_message_t* const_pastry_message_t;
73 /** Get the mailbox of a host given its pastry id. */
74 static sg_mailbox_t get_mailbox(int node_id)
76 char mailbox_name[MAILBOX_NAME_SIZE];
77 snprintf(mailbox_name, MAILBOX_NAME_SIZE - 1, "%d", node_id);
78 return sg_mailbox_by_name(mailbox_name);
81 /** Get the specific level of a node id */
82 unsigned int domain_mask = 0;
83 static int domain(unsigned int a, unsigned int level)
86 domain_mask = (1U << DOMAIN_SIZE) - 1;
87 unsigned int shift = (LEVELS_COUNT - level - 1) * DOMAIN_SIZE;
88 return (a >> shift) & domain_mask;
91 /* Get the shared domains between the two givens ids */
92 static int shl(int a, int b)
95 while (l < LEVELS_COUNT && domain(a, l) == domain(b, l))
100 /* Frees the memory used by a task and destroy it */
101 static void message_free(pastry_message_t message)
103 if (message != NULL) {
104 xbt_free(message->state);
109 /* Get the closest id to the dest in the node namespace_set */
110 static int closest_in_namespace_set(const_node_t node, int dest)
113 if ((node->namespace_set[NAMESPACE_SIZE - 1] <= dest) && (dest <= node->namespace_set[0])) {
114 int best_dist = abs(node->id - dest);
116 for (int i = 0; i < NAMESPACE_SIZE; i++) {
117 if (node->namespace_set[i] != -1) {
118 int dist = abs(node->namespace_set[i] - dest);
119 if (dist < best_dist) {
121 res = node->namespace_set[i];
129 /* Find the next node to forward a message to */
130 static int routing_next(const_node_t node, int dest)
132 int closest = closest_in_namespace_set(node, dest);
136 int l = shl(node->id, dest);
137 int res = node->routing_table[l][domain(dest, l)];
142 int dist = abs(node->id - dest);
143 for (int i = l; i < LEVELS_COUNT; i++) {
144 for (int j = 0; j < LEVEL_SIZE; j++) {
145 res = node->routing_table[i][j];
146 if (res != -1 && abs(res - dest) < dist)
151 for (int i = 0; i < NEIGHBORHOOD_SIZE; i++) {
152 res = node->neighborhood_set[i];
153 if (res != -1 && shl(res, dest) >= l && abs(res - dest) < dist)
157 for (int i = 0; i < NAMESPACE_SIZE; i++) {
158 res = node->namespace_set[i];
159 if (res != -1 && shl(res, dest) >= l && abs(res - dest) < dist)
166 /* Get the corresponding state of a node */
167 static state_t node_get_state(const_node_t node)
169 state_t state = xbt_new0(s_state_t, 1);
170 state->id = node->id;
171 for (int i = 0; i < NEIGHBORHOOD_SIZE; i++)
172 state->neighborhood_set[i] = node->neighborhood_set[i];
174 for (int i = 0; i < LEVELS_COUNT; i++)
175 for (int j = 0; j < LEVEL_SIZE; j++)
176 state->routing_table[i][j] = node->routing_table[i][j];
178 for (int i = 0; i < NAMESPACE_SIZE; i++)
179 state->namespace_set[i] = node->namespace_set[i];
184 static void print_node_id(const_node_t node)
186 XBT_INFO(" Id: %i '%08x' ", node->id, (unsigned)node->id);
189 /* Print the node namespace set */
190 static void print_node_namespace_set(const_node_t node)
192 XBT_INFO(" Namespace:");
193 for (int i = 0; i < NAMESPACE_SIZE; i++)
194 XBT_INFO(" %08x", (unsigned)node->namespace_set[i]);
197 /** Handle a given task */
198 static void handle_message(node_t node, pastry_message_t message)
200 XBT_DEBUG("Handling task %p", message);
206 sg_mailbox_t mailbox;
207 sg_comm_t comm = NULL;
208 sg_error_t err = SG_OK;
209 pastry_message_t request;
210 e_message_type_t type = message->type;
212 // If the node is not ready keep the task for later
213 if (node->ready != 0 && !(type == JOIN_LAST_REPLY || type == JOIN_REPLY)) {
214 XBT_DEBUG("Task pending %u", type);
215 xbt_dynar_push(node->pending_messages, &message);
220 /* Try to join the ring */
222 next = routing_next(node, message->answer_id);
223 XBT_DEBUG("Join request from %08x forwarding to %08x", (unsigned)message->answer_id, (unsigned)next);
224 type = JOIN_LAST_REPLY;
226 request = xbt_new0(s_pastry_message_t, 1);
227 request->answer_id = message->sender_id;
228 request->steps = message->steps + 1;
230 // if next different from current node forward the join
231 if (next != node->id) {
232 mailbox = get_mailbox(next);
233 message->sender_id = node->id;
235 comm = sg_mailbox_put_async(mailbox, message, COMM_SIZE);
236 err = sg_comm_wait_for(comm, timeout);
237 if (err == SG_ERROR_TIMEOUT) {
238 XBT_DEBUG("Timeout expired when forwarding join to next %d", next);
245 // send back the current node state to the joining node
246 request->type = type;
247 request->sender_id = node->id;
248 request->answer_to = get_mailbox(node->id);
249 request->state = node_get_state(node);
250 comm = sg_mailbox_put_async(message->answer_to, request, COMM_SIZE);
251 err = sg_comm_wait_for(comm, timeout);
252 if (err == SG_ERROR_TIMEOUT) {
253 XBT_DEBUG("Timeout expired when sending back the current node state to the joining node to %d", node->id);
254 message_free(request);
257 /* Join reply from all the node touched by the join */
258 case JOIN_LAST_REPLY:
259 // if last node touched reply, copy its namespace set
260 // TODO: it works only if the two nodes are side to side (is it really the case ?)
261 j = (message->sender_id < node->id) ? -1 : 0;
262 for (i = 0; i < NAMESPACE_SIZE / 2; i++) {
263 node->namespace_set[i] = message->state->namespace_set[i - j];
264 node->namespace_set[NAMESPACE_SIZE - 1 - i] = message->state->namespace_set[NAMESPACE_SIZE - 1 - i - j - 1];
266 node->namespace_set[NAMESPACE_SIZE / 2 + j] = message->sender_id;
267 node->ready += message->steps + 1;
270 XBT_DEBUG("Joining Reply");
272 // if first node touched reply, copy its neighborhood set
273 if (message->sender_id == node->known_id) {
274 node->neighborhood_set[0] = message->sender_id;
275 for (i = 1; i < NEIGHBORHOOD_SIZE; i++)
276 node->neighborhood_set[i] = message->state->neighborhood_set[i - 1];
279 // copy the corresponding routing table levels
280 min = (node->id == message->answer_id) ? 0 : shl(node->id, message->answer_id);
281 max = shl(node->id, message->sender_id) + 1;
282 for (i = min; i < max; i++) {
283 int d = domain(node->id, i);
284 for (j = 0; j < LEVEL_SIZE; j++)
286 node->routing_table[i][j] = message->state->routing_table[i][j];
290 // if the node is ready, do all the pending tasks and send update to known nodes
291 if (node->ready == 0) {
292 XBT_DEBUG("Node %i is ready!!!", node->id);
293 while (!xbt_dynar_is_empty(node->pending_messages)) {
295 xbt_dynar_shift(node->pending_messages, &m);
296 handle_message(node, m);
299 for (i = 0; i < NAMESPACE_SIZE; i++) {
300 j = node->namespace_set[i];
302 XBT_DEBUG("Send update to %i", j);
303 mailbox = get_mailbox(j);
305 request = xbt_new0(s_pastry_message_t, 1);
306 request->answer_id = node->id;
308 request->type = UPDATE;
309 request->sender_id = node->id;
310 request->answer_to = get_mailbox(node->id);
311 request->state = node_get_state(node);
312 comm = sg_mailbox_put_async(mailbox, request, COMM_SIZE);
313 err = sg_comm_wait_for(comm, timeout);
314 if (err == SG_ERROR_TIMEOUT) {
315 XBT_DEBUG("Timeout expired when sending update to %d", j);
316 message_free(request);
323 /* Received an update of state */
325 XBT_DEBUG("Task update %i !!!", node->id);
327 /* Update namespace ses */
328 XBT_INFO("Task update from %i !!!", message->sender_id);
331 print_node_namespace_set(node);
332 int curr_namespace_set[NAMESPACE_SIZE];
333 int task_namespace_set[NAMESPACE_SIZE + 1];
335 // Copy the current namespace and the task state namespace with state->id in the middle
337 for (; i < NAMESPACE_SIZE / 2; i++) {
338 curr_namespace_set[i] = node->namespace_set[i];
339 task_namespace_set[i] = message->state->namespace_set[i];
341 task_namespace_set[i] = message->state->id;
342 for (; i < NAMESPACE_SIZE; i++) {
343 curr_namespace_set[i] = node->namespace_set[i];
344 task_namespace_set[i + 1] = message->state->namespace_set[i];
347 // get the index of values before and after node->id in task_namespace
350 for (i = 0; i <= NAMESPACE_SIZE; i++) {
351 j = task_namespace_set[i];
352 if (j != -1 && j < node->id)
354 if (j != -1 && max == -1 && j > node->id)
358 // add lower elements
359 j = NAMESPACE_SIZE / 2 - 1;
360 for (i = NAMESPACE_SIZE / 2 - 1; i >= 0; i--) {
361 if (min < 0 || curr_namespace_set[j] > task_namespace_set[min]) {
362 node->namespace_set[i] = curr_namespace_set[j];
364 } else if (curr_namespace_set[j] == task_namespace_set[min]) {
365 node->namespace_set[i] = curr_namespace_set[j];
369 node->namespace_set[i] = task_namespace_set[min];
374 // add greater elements
375 j = NAMESPACE_SIZE / 2;
376 for (i = NAMESPACE_SIZE / 2; i < NAMESPACE_SIZE; i++) {
377 if (min < 0 || max >= NAMESPACE_SIZE) {
378 node->namespace_set[i] = curr_namespace_set[j];
380 } else if (max >= 0) {
381 if (curr_namespace_set[j] == -1 || curr_namespace_set[j] > task_namespace_set[max]) {
382 node->namespace_set[i] = task_namespace_set[max];
384 } else if (curr_namespace_set[j] == task_namespace_set[max]) {
385 node->namespace_set[i] = curr_namespace_set[j];
389 node->namespace_set[i] = curr_namespace_set[j];
395 /* Update routing table */
396 for (i = shl(node->id, message->state->id); i < LEVELS_COUNT; i++) {
397 for (j = 0; j < LEVEL_SIZE; j++) {
398 // FIXME: this is a no-op!
399 if (node->routing_table[i][j] == -1 && message->state->routing_table[i][j] == -1)
400 node->routing_table[i][j] = message->state->routing_table[i][j];
407 message_free(message);
411 static int join(const_node_t node)
413 pastry_message_t request = xbt_new0(s_pastry_message_t, 1);
414 request->type = JOIN;
415 request->sender_id = node->id;
416 request->answer_id = node->id;
418 request->answer_to = get_mailbox(node->id);
420 sg_mailbox_t mailbox = get_mailbox(node->known_id);
422 XBT_DEBUG("Trying to join Pastry ring... (with node %s)", sg_mailbox_get_name(mailbox));
423 sg_comm_t comm = sg_mailbox_put_async(mailbox, request, COMM_SIZE);
424 sg_error_t err = sg_comm_wait_for(comm, timeout);
425 if (err == SG_ERROR_TIMEOUT) {
426 XBT_DEBUG("Timeout expired when joining ring with node %d", node->known_id);
427 message_free(request);
435 * @brief Node Function
438 * - the id of a guy I know in the system (except for the first node)
439 * - the time to sleep before I join (except for the first node)
440 * - the deadline time
442 static void node(int argc, char* argv[])
444 double init_time = simgrid_get_clock();
445 void* received = NULL;
446 int join_success = 0;
448 xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
450 node.id = (int)xbt_str_parse_int(argv[1], "Invalid ID: %s");
453 node.pending_messages = xbt_dynar_new(sizeof(pastry_message_t), NULL);
454 node.mailbox = get_mailbox(node.id);
456 XBT_DEBUG("New node with id %s (%08x)", sg_mailbox_get_name(node.mailbox), (unsigned)node.id);
458 for (int i = 0; i < LEVELS_COUNT; i++) {
459 int d = domain(node.id, i);
460 for (int j = 0; j < LEVEL_SIZE; j++)
461 node.routing_table[i][j] = (d == j) ? node.id : -1;
464 for (int i = 0; i < NEIGHBORHOOD_SIZE; i++)
465 node.neighborhood_set[i] = -1;
467 for (int i = 0; i < NAMESPACE_SIZE; i++)
468 node.namespace_set[i] = -1;
470 if (argc == 3) { // first ring
471 XBT_DEBUG("Hey! Let's create the system.");
472 deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
474 XBT_DEBUG("Create a new Pastry ring...");
477 node.known_id = (int)xbt_str_parse_int(argv[2], "Invalid known ID: %s");
478 double sleep_time = xbt_str_parse_double(argv[3], "Invalid sleep time: %s");
479 deadline = xbt_str_parse_double(argv[4], "Invalid deadline: %s");
481 // sleep before starting
482 XBT_DEBUG("Let's sleep during %f", sleep_time);
483 sg_actor_sleep_for(sleep_time);
484 XBT_DEBUG("Hey! Let's join the system.");
486 join_success = join(&node);
490 XBT_DEBUG("Waiting ….");
492 while (simgrid_get_clock() < init_time + deadline && simgrid_get_clock() < max_simulation_time) {
493 if (node.comm_receive == NULL) {
495 node.comm_receive = sg_mailbox_get_async(node.mailbox, &received);
497 if (!sg_comm_test(node.comm_receive)) {
498 sg_actor_sleep_for(5);
500 // the task was successfully received
501 handle_message(&node, received);
502 node.comm_receive = NULL;
505 // Cleanup the receiving communication.
506 if (node.comm_receive != NULL)
507 sg_comm_unref(node.comm_receive);
509 xbt_dynar_free(&node.pending_messages);
512 /** @brief Main function. */
513 int main(int argc, char* argv[])
515 simgrid_init(&argc, argv);
517 "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
518 "\tExample: %s ../msg_platform.xml pastry10.xml\n",
521 char** options = &argv[1];
522 while (!strncmp(options[0], "-", 1)) {
523 int length = strlen("-nb_bits=");
524 if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
525 nb_bits = (int)xbt_str_parse_int(options[0] + length, "Invalid nb_bits parameter: %s");
526 XBT_DEBUG("Set nb_bits to %d", nb_bits);
528 length = strlen("-timeout=");
529 xbt_assert(strncmp(options[0], "-timeout=", length) == 0 && strlen(options[0]) > length,
530 "Invalid pastry option '%s'", options[0]);
531 timeout = (int)xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
532 XBT_DEBUG("Set timeout to %d", timeout);
537 simgrid_load_platform(options[0]);
539 simgrid_register_function("node", node);
540 simgrid_load_deployment(options[1]);
543 XBT_INFO("Simulated time: %g", simgrid_get_clock());