5 #include "xbt/asserts.h"
7 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry,
8 "Messages specific for this msg example");
10 /***************************************
14 * - handle node departure *
15 * - handle objects on the network *
16 * - handle neighborood in the update *
18 ***************************************/
22 #define MAILBOX_NAME_SIZE 10
25 #define LEVELS_COUNT 8 // sizeof(int)*8/DOMAIN_SIZE
26 #define LEVEL_SIZE 16 // 2^DOMAIN_SIZE
27 #define NEIGHBORHOOD_SIZE 6
28 #define NAMESPACE_SIZE 6
29 #define MAILBOX_NAME_SIZE 10
31 static int nb_bits = 16;
32 static int timeout = 50;
33 static int max_simulation_time = 1000;
35 extern long int smx_total_comms;
38 typedef struct s_node {
39 int id; //128bits generated random(2^128 -1)
41 char mailbox[MAILBOX_NAME_SIZE]; // my mailbox name (string representation of the id)
42 int namespace_set[NAMESPACE_SIZE];
43 int neighborhood_set[NEIGHBORHOOD_SIZE];
44 int routing_table[LEVELS_COUNT][LEVEL_SIZE];
46 msg_comm_t comm_receive; // current communication to receive
47 xbt_fifo_t pending_tasks;
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];
55 } s_state_t, *state_t;
58 * Types of tasks exchanged between nodes.
67 typedef struct s_task_data {
68 e_task_type_t type; // type of task
69 int sender_id; // id paramater (used by some types of tasks)
70 //int request_finger; // finger parameter (used by some types of tasks)
71 int answer_id; // answer (used by some types of tasks)
72 char answer_to[MAILBOX_NAME_SIZE]; // mailbox to send an answer to (if any)
73 //const char* issuer_host_name; // used for logging
76 } s_task_data_t, *task_data_t;
79 static void print_node(node_t node);
80 static void print_node_id(node_t node);
81 static void print_node_neighborood_set(node_t node);
82 static void print_node_routing_table(node_t node);
83 static void print_node_namespace_set(node_t node);
84 static state_t node_get_state(node_t node);
85 static void get_mailbox(int node_id, char* mailbox);
86 static int domain(int a, int level);
87 static int shl(int a, int b);
88 static int closest_in_namespace_set(node_t node, int dest);
89 static int routing_next(node_t node, int dest);
90 static void create(node_t node);
91 static int join(node_t node);
95 * \brief Gets the mailbox name of a host given its chord id.
96 * \param node_id id of a node
97 * \param mailbox pointer to where the mailbox name should be written
98 * (there must be enough space)
100 static void get_mailbox(int node_id, char* mailbox)
102 snprintf(mailbox, MAILBOX_NAME_SIZE - 1, "%d", node_id);
106 * Get the specific level of a node id
109 static int domain(int a, int level) {
110 if (domain_mask == 0)
111 domain_mask = pow(2, DOMAIN_SIZE) - 1;
112 int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
113 return (a >> shift) & domain_mask;
117 * Get the shared domains between the two givens ids
119 static int shl(int a, int b) {
121 while(l<LEVELS_COUNT && domain(a,l) == domain(b,l))
127 * Get the cloest id to the dest in the node namespace_set
129 static int closest_in_namespace_set(node_t node, int dest) {
132 if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) & (dest <= node->namespace_set[0])) {
133 best_dist = abs(node->id - dest);
136 for (i=0; i<NAMESPACE_SIZE; i++) {
137 if (node->namespace_set[i]!=-1) {
138 dist = abs(node->namespace_set[i] - dest);
139 if (dist<best_dist) {
141 res = node->namespace_set[i];
150 * Find the next node to forward a meassage to
152 static int routing_next(node_t node, int dest) {
153 int closest = closest_in_namespace_set(node, dest);
158 int l = shl(node->id, dest);
159 res = node->routing_table[l][domain(dest, l)];
164 int dist = abs(node->id - dest);
166 for (i=l; i<LEVELS_COUNT; i++) {
167 for (j=0; j<LEVEL_SIZE; j++) {
168 res = node->routing_table[i][j];
169 if (res!=-1 && abs(res - dest)<dist)
174 for (i=0; i<NEIGHBORHOOD_SIZE; i++) {
175 res = node->neighborhood_set[i];
176 if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
180 for (i=0; i<NAMESPACE_SIZE; i++) {
181 res = node->namespace_set[i];
182 if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
190 * Handle a given task
192 static void handle_task(node_t node, msg_task_t task) {
193 XBT_DEBUG("Handling task %p", task);
194 char mailbox[MAILBOX_NAME_SIZE];
195 int i, j, min, max, d;
196 msg_task_t task_sent;
197 task_data_t req_data;
198 task_data_t task_data = (task_data_t) MSG_task_get_data(task);
199 e_task_type_t type = task_data->type;
200 // If the node is not ready keep the task for later
201 if (node->ready != 0 && !(type==TASK_JOIN_LAST_REPLY || type==TASK_JOIN_REPLY)) {
202 XBT_DEBUG("Task pending %i", type);
203 xbt_fifo_push(node->pending_tasks, task);
208 * Try to join the ring
211 int next = routing_next(node, task_data->answer_id);
212 XBT_DEBUG("Join request from %08x forwarding to %08x", task_data->answer_id, next);
213 type = TASK_JOIN_LAST_REPLY;
215 req_data = xbt_new0(s_task_data_t,1);
216 req_data->answer_id = task_data->sender_id;
217 req_data->steps = task_data->steps + 1;
219 // if next different from current node forward the join
220 if (next!=node->id) {
221 get_mailbox(next, mailbox);
222 task_data->sender_id = node->id;
224 task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, task_data);
225 MSG_task_send_with_timeout(task_sent, mailbox, timeout);
226 type = TASK_JOIN_REPLY;
229 // send back the current node state to the joining node
230 req_data->type = type;
231 req_data->sender_id = node->id;
232 get_mailbox(node->id, req_data->answer_to);
233 req_data->state = node_get_state(node);
234 task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
235 MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout);
239 * Join reply from all the node touched by the join
241 case TASK_JOIN_LAST_REPLY:
242 // if last node touched reply, copy its namespace set
243 // TODO: it's work only if the two nodes are side to side (is it really the case ?)
244 j = (task_data->sender_id < node->id) ? -1 : 0;
245 for (i=0; i<NAMESPACE_SIZE/2; i++) {
246 node->namespace_set[i] = task_data->state->namespace_set[i-j];
247 node->namespace_set[NAMESPACE_SIZE-1-i] = task_data->state->namespace_set[NAMESPACE_SIZE-1-i-j-1];
249 node->namespace_set[NAMESPACE_SIZE/2+j] = task_data->sender_id;
250 node->ready += task_data->steps + 1;
251 case TASK_JOIN_REPLY:
252 XBT_DEBUG("Joining Reply");
254 // if first node touched reply, copy its neighborood set
255 if (task_data->sender_id == node->known_id) {
256 node->neighborhood_set[0] = task_data->sender_id;
257 for (i=1; i<NEIGHBORHOOD_SIZE; i++)
258 node->neighborhood_set[i] = task_data->state->neighborhood_set[i-1];
261 // copy the corresponding routing table levels
262 min = (node->id==task_data->answer_id) ? 0 : shl(node->id, task_data->answer_id);
263 max = shl(node->id, task_data->sender_id)+1;
264 for (i=min;i<max;i++) {
265 d = domain(node->id, i);
266 for (j=0; j<LEVEL_SIZE; j++)
268 node->routing_table[i][j] = task_data->state->routing_table[i][j];
272 // if the node is ready, do all the pending tasks and send update to known nodes
273 if (node->ready==0) {
274 XBT_DEBUG("Node %i is ready!!!", node->id);
276 while(xbt_fifo_size(node->pending_tasks))
277 handle_task(node, xbt_fifo_pop(node->pending_tasks));
279 for (i=0; i<NAMESPACE_SIZE; i++) {
280 j = node->namespace_set[i];
282 XBT_DEBUG("Send update to %i", j);
283 get_mailbox(j, mailbox);
285 req_data = xbt_new0(s_task_data_t,1);
286 req_data->answer_id = node->id;
288 req_data->type = TASK_UPDATE;
289 req_data->sender_id = node->id;
290 get_mailbox(node->id, req_data->answer_to);
291 req_data->state = node_get_state(node);
292 task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
293 MSG_task_send_with_timeout(task_sent, mailbox, timeout);
300 * Recieved an update of state
303 XBT_DEBUG("Task update %i !!!", node->id);
305 /* Update namespace ses */
306 printf("Task update from %i !!!\n", task_data->sender_id);
308 print_node_namespace_set(node);
309 int curr_namespace_set[NAMESPACE_SIZE];
310 int task_namespace_set[NAMESPACE_SIZE+1];
312 // Copy the current namedspace
313 // and the task state namespace with state->id in the middle
315 for (; i<NAMESPACE_SIZE/2; i++){
316 curr_namespace_set[i] = node->namespace_set[i];
317 task_namespace_set[i] = task_data->state->namespace_set[i];
319 task_namespace_set[i] = task_data->state->id;
320 for (; i<NAMESPACE_SIZE; i++){
321 curr_namespace_set[i] = node->namespace_set[i];
322 task_namespace_set[i+1] = task_data->state->namespace_set[i];
325 // get the index of values before and after node->id in task_namespace
328 for (i=0; i<=NAMESPACE_SIZE; i++) {
329 j = task_namespace_set[i];
330 if (i<NAMESPACE_SIZE)
331 printf("%08x %08x | ", j, curr_namespace_set[i]);
332 if (j != -1 && j < node->id) min = i;
333 if (j != -1 && max == -1 && j > node->id) max = i;
337 // add lower elements
338 j = NAMESPACE_SIZE/2-1;
339 for (i=NAMESPACE_SIZE/2-1; i>=0; i--) {
340 printf("i:%i, j:%i, min:%i, currj:%08x, taskmin:%08x\n", i, j, min, curr_namespace_set[j], task_namespace_set[min]);
342 node->namespace_set[i] = curr_namespace_set[j];
344 } else if (curr_namespace_set[j] == task_namespace_set[min]) {
345 node->namespace_set[i] = curr_namespace_set[j];
347 } else if (curr_namespace_set[j] > task_namespace_set[min]) {
348 node->namespace_set[i] = curr_namespace_set[j];
351 node->namespace_set[i] = task_namespace_set[min];
356 // add greater elements
357 j = NAMESPACE_SIZE/2;
358 for (i=NAMESPACE_SIZE/2; i<NAMESPACE_SIZE; i++) {
359 printf("i:%i, j:%i, max:%i, currj:%08x, taskmax:%08x\n", i, j, max, curr_namespace_set[j], task_namespace_set[max]);
360 if (min<0 || max>=NAMESPACE_SIZE) {
361 node->namespace_set[i] = curr_namespace_set[j];
363 } else if (curr_namespace_set[j] == -1) {
364 node->namespace_set[i] = task_namespace_set[max];
366 } else if (curr_namespace_set[j] == task_namespace_set[max]) {
367 node->namespace_set[i] = curr_namespace_set[j];
369 } else if (curr_namespace_set[j] < task_namespace_set[max]) {
370 node->namespace_set[i] = curr_namespace_set[j];
373 node->namespace_set[i] = task_namespace_set[max];
377 print_node_namespace_set(node);
379 /* Update routing table */
380 for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
381 for (j=0; j<LEVEL_SIZE; j++) {
382 if (node->routing_table[i][j]==-1 && task_data->state->routing_table[i][j]==-1)
383 node->routing_table[i][j] = task_data->state->routing_table[i][j];
390 * \brief Initializes the current node as the first one of the system.
391 * \param node the current node
393 static void create(node_t node){
395 XBT_DEBUG("Create a new Pastry ring...");
401 static int join(node_t node){
402 task_data_t req_data = xbt_new0(s_task_data_t,1);
403 req_data->type = TASK_JOIN;
404 req_data->sender_id = node->id;
405 req_data->answer_id = node->id;
407 get_mailbox(node->id, req_data->answer_to);
409 char mailbox[MAILBOX_NAME_SIZE];
410 get_mailbox(node->known_id, mailbox);
412 msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
413 XBT_DEBUG("Trying to join Pastry ring... (with node %s)", mailbox);
414 MSG_task_send_with_timeout(task_sent, mailbox, timeout);
420 * Print the node infomations
422 static void print_node(node_t node) {
425 print_node_neighborood_set(node);
426 print_node_routing_table(node);
427 print_node_namespace_set(node);
433 static void print_node_id(node_t node) {
435 printf(" id: %i '%08x' ", node->id, node->id);
436 for (i=0;i<LEVELS_COUNT;i++)
437 printf(" %x", domain(node->id, i));
442 * Print the node neighborood set
444 static void print_node_neighborood_set(node_t node) {
446 printf(" neighborood:\n");
447 for (i=0; i<NEIGHBORHOOD_SIZE; i++)
448 printf(" %08x\n", node->neighborhood_set[i]);
452 * Print the routing table
454 static void print_node_routing_table(node_t node) {
456 printf(" routing table:\n");
457 for (i=0; i<LEVELS_COUNT; i++){
459 for (j=0; j<LEVEL_SIZE; j++)
460 printf("%08x ", node->routing_table[i][j]);
466 * Print the node namespace set
468 static void print_node_namespace_set(node_t node) {
470 printf(" namespace:\n");
471 for (i=0; i<NAMESPACE_SIZE; i++)
472 printf(" %08x\n", node->namespace_set[i]);
478 * Get the corresponding state of a node
480 static state_t node_get_state(node_t node) {
482 state_t state = xbt_new0(s_state_t,1);
483 state->id = node->id;
484 for (i=0; i<NEIGHBORHOOD_SIZE; i++)
485 state->neighborhood_set[i] = node->neighborhood_set[i];
487 for (i=0; i<LEVELS_COUNT; i++)
488 for (j=0; j<LEVEL_SIZE; j++)
489 state->routing_table[i][j] = node->routing_table[i][j];
491 for (i=0; i<NAMESPACE_SIZE; i++)
492 state->namespace_set[i] = node->namespace_set[i];
499 * \brief Node Function
502 * - the id of a guy I know in the system (except for the first node)
503 * - the time to sleep before I join (except for the first node)
504 * - the deadline time
506 static int node(int argc, char *argv[])
508 double init_time = MSG_get_clock();
509 msg_task_t task_received = NULL;
510 int join_success = 0;
512 xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
514 node.id = atoi(argv[1]);
517 node.pending_tasks = xbt_fifo_new();
518 get_mailbox(node.id, node.mailbox);
519 XBT_DEBUG("New node with id %s (%08x)", node.mailbox, node.id);
522 for (i=0; i<LEVELS_COUNT; i++){
523 d = domain(node.id, i);
524 for (j=0; j<LEVEL_SIZE; j++)
525 node.routing_table[i][j] = (d==j) ? node.id : -1;
528 for (i=0; i<NEIGHBORHOOD_SIZE; i++)
529 node.neighborhood_set[i] = -1;
531 for (i=0; i<NAMESPACE_SIZE; i++)
532 node.namespace_set[i] = -1;
534 if (argc == 3) { // first ring
535 XBT_DEBUG("Hey! Let's create the system.");
536 deadline = atof(argv[2]);
541 node.known_id = atoi(argv[2]);
542 double sleep_time = atof(argv[3]);
543 deadline = atof(argv[4]);
545 // sleep before starting
546 XBT_DEBUG("Let's sleep during %f", sleep_time);
547 MSG_process_sleep(sleep_time);
548 XBT_DEBUG("Hey! Let's join the system.");
550 join_success = join(&node);
554 XBT_DEBUG("Waiting ….");
556 while (MSG_get_clock() < init_time + deadline
557 // && MSG_get_clock() < node.last_change_date + 1000
558 && MSG_get_clock() < max_simulation_time) {
559 if (node.comm_receive == NULL) {
560 task_received = NULL;
561 node.comm_receive = MSG_task_irecv(&task_received, node.mailbox);
562 // FIXME: do not make MSG_task_irecv() calls from several functions
564 if (!MSG_comm_test(node.comm_receive)) {
565 MSG_process_sleep(5);
567 // a transfer has occurred
569 msg_error_t status = MSG_comm_get_status(node.comm_receive);
571 if (status != MSG_OK) {
572 XBT_DEBUG("Failed to receive a task. Nevermind.");
573 MSG_comm_destroy(node.comm_receive);
574 node.comm_receive = NULL;
577 // the task was successfully received
578 MSG_comm_destroy(node.comm_receive);
579 node.comm_receive = NULL;
580 handle_task(&node, task_received);
593 /*typedef struct s_node {
595 char mailbox[MAILBOX_NAME_SIZE]; // my mailbox name (string representation of the id)
596 s_finger_t *fingers; // finger table, of size nb_bits (fingers[0] is my successor)
597 int pred_id; // predecessor id
598 char pred_mailbox[MAILBOX_NAME_SIZE]; // predecessor's mailbox name
599 int next_finger_to_fix; // index of the next finger to fix in fix_fingers()
600 msg_comm_t comm_receive; // current communication to receive
601 double last_change_date; // last time I changed a finger or my predecessor
602 } s_node_t, *node_t;*/
605 * \brief Main function.
607 int main(int argc, char *argv[])
609 MSG_init(&argc, argv);
611 printf("Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n", argv[0]);
612 printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
616 char **options = &argv[1];
617 while (!strncmp(options[0], "-", 1)) {
619 int length = strlen("-nb_bits=");
620 if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
621 nb_bits = atoi(options[0] + length);
622 XBT_DEBUG("Set nb_bits to %d", nb_bits);
626 length = strlen("-timeout=");
627 if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
628 timeout = atoi(options[0] + length);
629 XBT_DEBUG("Set timeout to %d", timeout);
632 xbt_die("Invalid chord option '%s'", options[0]);
638 const char* platform_file = options[0];
639 const char* application_file = options[1];
641 MSG_create_environment(platform_file);
643 MSG_function_register("node", node);
644 MSG_launch_application(application_file);
646 msg_error_t res = MSG_main();
647 XBT_CRITICAL("Messages created: %ld", smx_total_comms);
648 XBT_INFO("Simulated time: %g", MSG_get_clock());