X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/d2cd6190c7a6ba8f102134cf6ab3043f7df8f77e..056d48c9fa3d5fa5afd2be695f54edbc5fd5434d:/examples/msg/dht-pastry/dht-pastry.c diff --git a/examples/msg/dht-pastry/dht-pastry.c b/examples/msg/dht-pastry/dht-pastry.c index d0afa78f88..d912875ebd 100644 --- a/examples/msg/dht-pastry/dht-pastry.c +++ b/examples/msg/dht-pastry/dht-pastry.c @@ -4,15 +4,16 @@ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ -#include #include "simgrid/msg.h" +#include "xbt/fifo.h" +#include - XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example"); +XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example"); /* TODO: * * - handle node departure * * - handle objects on the network * - * - handle neighborood in the update */ + * - handle neighborhood in the update */ #define COMM_SIZE 10 #define COMP_SIZE 0 @@ -68,7 +69,7 @@ typedef struct s_task_data { } s_task_data_t, *task_data_t; static void get_mailbox(int node_id, char* mailbox); -static int domain(int a, int level); +static int domain(unsigned int a, unsigned int level); static int shl(int a, int b); static int closest_in_namespace_set(node_t node, int dest); static int routing_next(node_t node, int dest); @@ -85,11 +86,12 @@ static void get_mailbox(int node_id, char* mailbox) } /** Get the specific level of a node id */ -int domain_mask = 0; -static int domain(int a, int level) { +unsigned int domain_mask = 0; +static int domain(unsigned int a, unsigned int level) +{ if (domain_mask == 0) domain_mask = pow(2, DOMAIN_SIZE) - 1; - int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE; + unsigned int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE; return (a >> shift) & domain_mask; } @@ -101,20 +103,30 @@ static int shl(int a, int b) { return l; } +/* Frees the memory used by a task and destroy it */ +static void task_free(void* task) +{ + // TODO add a parameter data_free_function to MSG_task_create? + if(task != NULL){ + s_task_data_t* data = (s_task_data_t*)MSG_task_get_data(task); + xbt_free(data->state); + xbt_free(data); + MSG_task_destroy(task); + } +} + /* Get the closest id to the dest in the node namespace_set */ static int closest_in_namespace_set(node_t node, int dest) { - int best_dist; int res = -1; - if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) & (dest <= node->namespace_set[0])) { - best_dist = abs(node->id - dest); + if ((node->namespace_set[NAMESPACE_SIZE-1] <= dest) && (dest <= node->namespace_set[0])) { + int best_dist = abs(node->id - dest); res = node->id; - int i, dist; - for (i=0; inamespace_set[i]!=-1) { - dist = abs(node->namespace_set[i] - dest); + int dist = abs(node->namespace_set[i] - dest); if (distnamespace_set[i]; + res = node->namespace_set[i]; } } } @@ -125,20 +137,19 @@ static int closest_in_namespace_set(node_t node, int dest) { /* Find the next node to forward a message to */ static int routing_next(node_t node, int dest) { int closest = closest_in_namespace_set(node, dest); - int res = -1; if (closest!=-1) return closest; int l = shl(node->id, dest); - res = node->routing_table[l][domain(dest, l)]; - if (res!=-1) + int res = node->routing_table[l][domain(dest, l)]; + if (res != -1) return res; //rare case int dist = abs(node->id - dest); - int i,j; + int i; for (i=l; irouting_table[i][j]; if (res!=-1 && abs(res - dest)id = node->id; for (i=0; ineighborhood_set[i] = node->neighborhood_set[i]; for (i=0; irouting_table[i][j] = node->routing_table[i][j]; for (i=0; iid, node->id); - for (i=0;iid, i)); - printf("\n"); + XBT_INFO(" Id: %i '%08x' ", node->id, node->id); } /* * Print the node neighborhood set */ static void print_node_neighborood_set(node_t node) { - int i; - printf(" Neighborhood:\n"); - for (i=0; ineighborhood_set[i]); + XBT_INFO(" Neighborhood:"); + for (int i=0; ineighborhood_set[i]); } /* Print the routing table */ static void print_node_routing_table(node_t node) { - printf(" routing table:\n"); + XBT_INFO(" Routing table:"); for (int i=0; irouting_table[i][j]); - printf("\n"); + XBT_INFO(" %08x ", node->routing_table[i][j]); } } /* Print the node namespace set */ static void print_node_namespace_set(node_t node) { - printf(" namespace:\n"); + XBT_INFO(" Namespace:"); for (int i=0; inamespace_set[i]); - printf("\n"); + XBT_INFO(" %08x", node->namespace_set[i]); } /* Print the node information */ static void print_node(node_t node) { - printf("Node:\n"); + XBT_INFO("Node:"); print_node_id(node); print_node_neighborood_set(node); print_node_routing_table(node); @@ -227,7 +230,12 @@ static void print_node(node_t node) { static void handle_task(node_t node, msg_task_t task) { XBT_DEBUG("Handling task %p", task); char mailbox[MAILBOX_NAME_SIZE]; - int i, j, min, max, d; + int i; + int j; + int min; + int max; + int d; + int next; msg_task_t task_sent; task_data_t req_data; task_data_t task_data = (task_data_t) MSG_task_get_data(task); @@ -240,8 +248,8 @@ static void handle_task(node_t node, msg_task_t task) { } switch (type) { /* Try to join the ring */ - case TASK_JOIN: { - int next = routing_next(node, task_data->answer_id); + case TASK_JOIN: + next = routing_next(node, task_data->answer_id); XBT_DEBUG("Join request from %08x forwarding to %08x", task_data->answer_id, next); type = TASK_JOIN_LAST_REPLY; @@ -255,7 +263,10 @@ static void handle_task(node_t node, msg_task_t task) { task_data->sender_id = node->id; task_data->steps++; task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, task_data); - MSG_task_send_with_timeout(task_sent, mailbox, timeout); + if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) { + XBT_DEBUG("Timeout expired when forwarding join to next %d", next); + task_free(task_sent); + } type = TASK_JOIN_REPLY; } @@ -265,13 +276,15 @@ static void handle_task(node_t node, msg_task_t task) { get_mailbox(node->id, req_data->answer_to); req_data->state = node_get_state(node); task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data); - MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout); + if (MSG_task_send_with_timeout(task_sent, task_data->answer_to, timeout)== MSG_TIMEOUT) { + XBT_DEBUG("Timeout expired when sending back the current node state to the joining node to %d", node->id); + task_free(task_sent); + } break; - } /* Join reply from all the node touched by the join */ case TASK_JOIN_LAST_REPLY: // if last node touched reply, copy its namespace set - // TODO: it's work only if the two nodes are side to side (is it really the case ?) + // TODO: it works only if the two nodes are side to side (is it really the case ?) j = (task_data->sender_id < node->id) ? -1 : 0; for (i=0; inamespace_set[i] = task_data->state->namespace_set[i-j]; @@ -279,6 +292,7 @@ static void handle_task(node_t node, msg_task_t task) { } node->namespace_set[NAMESPACE_SIZE/2+j] = task_data->sender_id; node->ready += task_data->steps + 1; + /* no break */ case TASK_JOIN_REPLY: XBT_DEBUG("Joining Reply"); @@ -321,7 +335,10 @@ static void handle_task(node_t node, msg_task_t task) { get_mailbox(node->id, req_data->answer_to); req_data->state = node_get_state(node); task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data); - MSG_task_send_with_timeout(task_sent, mailbox, timeout); + if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) { + XBT_DEBUG("Timeout expired when sending update to %d", j); + task_free(task_sent); + } } } } @@ -331,14 +348,14 @@ static void handle_task(node_t node, msg_task_t task) { XBT_DEBUG("Task update %i !!!", node->id); /* Update namespace ses */ - printf("Task update from %i !!!\n", task_data->sender_id); + XBT_INFO("Task update from %i !!!", task_data->sender_id); + XBT_INFO("Node:"); print_node_id(node); print_node_namespace_set(node); int curr_namespace_set[NAMESPACE_SIZE]; int task_namespace_set[NAMESPACE_SIZE+1]; - // Copy the current namedspace - // and the task state namespace with state->id in the middle + // Copy the current namespace and the task state namespace with state->id in the middle i=0; for (; inamespace_set[i]; @@ -355,24 +372,22 @@ static void handle_task(node_t node, msg_task_t task) { max = -1; for (i=0; i<=NAMESPACE_SIZE; i++) { j = task_namespace_set[i]; - if (iid) min = i; - if (j != -1 && max == -1 && j > node->id) max = i; + if (j != -1 && j < node->id) + min = i; + if (j != -1 && max == -1 && j > node->id) + max = i; } - printf("\n"); // add lower elements j = NAMESPACE_SIZE/2-1; for (i=NAMESPACE_SIZE/2-1; i>=0; i--) { - printf("i:%i, j:%i, min:%i, currj:%08x, taskmin:%08x\n", i, j, min, curr_namespace_set[j], - task_namespace_set[min]); if (min<0) { node->namespace_set[i] = curr_namespace_set[j]; j--; } else if (curr_namespace_set[j] == task_namespace_set[min]) { node->namespace_set[i] = curr_namespace_set[j]; - j--; min--; + j--; + min--; } else if (curr_namespace_set[j] > task_namespace_set[min]) { node->namespace_set[i] = curr_namespace_set[j]; j--; @@ -385,8 +400,6 @@ static void handle_task(node_t node, msg_task_t task) { // add greater elements j = NAMESPACE_SIZE/2; for (i=NAMESPACE_SIZE/2; i=NAMESPACE_SIZE) { node->namespace_set[i] = curr_namespace_set[j]; j++; @@ -395,7 +408,8 @@ static void handle_task(node_t node, msg_task_t task) { max++; } else if (curr_namespace_set[j] == task_namespace_set[max]) { node->namespace_set[i] = curr_namespace_set[j]; - j++; max++; + j++; + max++; } else if (curr_namespace_set[j] < task_namespace_set[max]) { node->namespace_set[i] = curr_namespace_set[j]; j++; @@ -404,7 +418,6 @@ static void handle_task(node_t node, msg_task_t task) { max++; } } - print_node_namespace_set(node); /* Update routing table */ for (i=shl(node->id, task_data->state->id); irouting_table[i][j] = task_data->state->routing_table[i][j]; } } + break; + default: + THROW_IMPOSSIBLE; } + task_free(task); } /** \brief Initializes the current node as the first one of the system. @@ -438,12 +455,14 @@ static int join(node_t node){ msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data); XBT_DEBUG("Trying to join Pastry ring... (with node %s)", mailbox); - MSG_task_send_with_timeout(task_sent, mailbox, timeout); + if (MSG_task_send_with_timeout(task_sent, mailbox, timeout)== MSG_TIMEOUT) { + XBT_DEBUG("Timeout expired when joining ring with node %d", node->known_id); + task_free(task_sent); + } return 1; } - /** * \brief Node Function * Arguments: @@ -467,10 +486,10 @@ static int node(int argc, char *argv[]) get_mailbox(node.id, node.mailbox); XBT_DEBUG("New node with id %s (%08x)", node.mailbox, node.id); - int i,j,d; + int i; for (i=0; i