X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/479a2412c9a12c28c80e1ee08fb2b5d04c2e88fc..ed26e84f533a3dc510abdc384aee02506ab9798b:/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 e6c50ff29a..12d13daf11 100644 --- a/examples/msg/dht-pastry/dht-pastry.c +++ b/examples/msg/dht-pastry/dht-pastry.c @@ -4,8 +4,10 @@ /* 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/dynar.h" +#include + XBT_LOG_NEW_DEFAULT_CATEGORY(msg_pastry, "Messages specific for this msg example"); @@ -38,15 +40,17 @@ typedef struct s_node { int routing_table[LEVELS_COUNT][LEVEL_SIZE]; int ready; msg_comm_t comm_receive; // current communication to receive - xbt_fifo_t pending_tasks; -} s_node_t, *node_t; + xbt_dynar_t pending_tasks; +} s_node_t; +typedef s_node_t* node_t; typedef struct s_state { int id; int namespace_set[NAMESPACE_SIZE]; int neighborhood_set[NEIGHBORHOOD_SIZE]; int routing_table[LEVELS_COUNT][LEVEL_SIZE]; -} s_state_t, *state_t; +} s_state_t; +typedef s_state_t* state_t; /** Types of tasks exchanged between nodes. */ typedef enum { @@ -58,16 +62,16 @@ typedef enum { typedef struct s_task_data { e_task_type_t type; // type of task - int sender_id; // id paramater (used by some types of tasks) + int sender_id; // id parameter (used by some types of tasks) //int request_finger; // finger parameter (used by some types of tasks) int answer_id; // answer (used by some types of tasks) char answer_to[MAILBOX_NAME_SIZE]; // mailbox to send an answer to (if any) //const char* issuer_host_name; // used for logging int steps; state_t state; -} s_task_data_t, *task_data_t; +} s_task_data_t; +typedef s_task_data_t* task_data_t; -static void get_mailbox(int node_id, char* mailbox); 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); @@ -105,7 +109,6 @@ static int shl(int a, int b) { /* 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); @@ -136,19 +139,17 @@ 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; - for (i=l; irouting_table[i][j]; if (res!=-1 && abs(res - dest)neighborhood_set[i]; if (res!=-1 && shl(res, dest)>=l && abs(res - dest)namespace_set[i]; if (res!=-1 && shl(res, dest)>=l && 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; inamespace_set[i] = node->namespace_set[i]; return state; } -/* Print the node id */ static void print_node_id(node_t node) { XBT_INFO(" Id: %i '%08x' ", node->id, node->id); } -/* * Print the node neighborhood set */ static void print_node_neighborood_set(node_t node) { XBT_INFO(" Neighborhood:"); for (int i=0; ineighborhood_set[i]); } -/* Print the routing table */ static void print_node_routing_table(node_t node) { XBT_INFO(" Routing table:"); for (int i=0; irouting_table[i][j]); } } - /* Print the node namespace set */ static void print_node_namespace_set(node_t node) { XBT_INFO(" Namespace:"); @@ -235,6 +231,7 @@ static void handle_task(node_t node, msg_task_t task) { 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); @@ -242,13 +239,13 @@ static void handle_task(node_t node, msg_task_t task) { // If the node is not ready keep the task for later if (node->ready != 0 && !(type==TASK_JOIN_LAST_REPLY || type==TASK_JOIN_REPLY)) { XBT_DEBUG("Task pending %i", type); - xbt_fifo_push(node->pending_tasks, task); + xbt_dynar_push(node->pending_tasks, &task); return; } 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; @@ -280,11 +277,10 @@ static void handle_task(node_t node, msg_task_t task) { 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]; @@ -317,9 +313,11 @@ static void handle_task(node_t node, msg_task_t task) { // if the node is ready, do all the pending tasks and send update to known nodes if (node->ready==0) { XBT_DEBUG("Node %i is ready!!!", node->id); - - while(xbt_fifo_size(node->pending_tasks)) - handle_task(node, xbt_fifo_pop(node->pending_tasks)); + while(xbt_dynar_length(node->pending_tasks)){ + msg_task_t task; + xbt_dynar_shift(node->pending_tasks, &task); + handle_task(node, task); + } for (i=0; inamespace_set[i]; @@ -403,19 +401,21 @@ static void handle_task(node_t node, msg_task_t task) { if (min<0 || max>=NAMESPACE_SIZE) { node->namespace_set[i] = curr_namespace_set[j]; j++; - } else if (curr_namespace_set[j] == -1) { - node->namespace_set[i] = task_namespace_set[max]; - max++; - } else if (curr_namespace_set[j] == task_namespace_set[max]) { - node->namespace_set[i] = curr_namespace_set[j]; - j++; - max++; - } else if (curr_namespace_set[j] < task_namespace_set[max]) { - node->namespace_set[i] = curr_namespace_set[j]; - j++; - } else { - node->namespace_set[i] = task_namespace_set[max]; - max++; + } else if (max >= 0){ + if (curr_namespace_set[j] == -1) { + node->namespace_set[i] = task_namespace_set[max]; + max++; + } else if (curr_namespace_set[j] == task_namespace_set[max]) { + node->namespace_set[i] = curr_namespace_set[j]; + j++; + max++; + } else if (curr_namespace_set[j] < task_namespace_set[max]) { + node->namespace_set[i] = curr_namespace_set[j]; + j++; + } else { + node->namespace_set[i] = task_namespace_set[max]; + max++; + } } } @@ -433,14 +433,6 @@ static void handle_task(node_t node, msg_task_t task) { task_free(task); } -/** \brief Initializes the current node as the first one of the system. - * \param node the current node - */ -static void create(node_t node){ - node->ready = 0; - XBT_DEBUG("Create a new Pastry ring..."); -} - /* Join the ring */ static int join(node_t node){ task_data_t req_data = xbt_new0(s_task_data_t,1); @@ -482,30 +474,29 @@ static int node(int argc, char *argv[]) node.id = xbt_str_parse_int(argv[1], "Invalid ID: %s"); node.known_id = -1; node.ready = -1; - node.pending_tasks = xbt_fifo_new(); + node.pending_tasks = xbt_dynar_new(sizeof(msg_task_t), NULL); get_mailbox(node.id, node.mailbox); XBT_DEBUG("New node with id %s (%08x)", node.mailbox, node.id); - - int i; - for (i=0; i