A
lgorithmique
N
umérique
D
istribuée
Public GIT Repository
projects
/
simgrid.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
a few smell in pastry, including a nasty one
[simgrid.git]
/
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
df37565
..
270cb94
100644
(file)
--- a/
examples/msg/dht-pastry/dht-pastry.c
+++ b/
examples/msg/dht-pastry/dht-pastry.c
@@
-7,12
+7,12
@@
#include <math.h>
#include "simgrid/msg.h"
#include <math.h>
#include "simgrid/msg.h"
-
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 *
/* TODO: *
* - handle node departure *
* - handle objects on the network *
- * - handle neighborood in the update */
+ * - handle neighbor
h
ood in the update */
#define COMM_SIZE 10
#define COMP_SIZE 0
#define COMM_SIZE 10
#define COMP_SIZE 0
@@
-29,8
+29,6
@@
static int nb_bits = 16;
static int timeout = 50;
static int max_simulation_time = 1000;
static int timeout = 50;
static int max_simulation_time = 1000;
-extern long int smx_total_comms;
-
typedef struct s_node {
int id; //128bits generated random(2^128 -1)
int known_id;
typedef struct s_node {
int id; //128bits generated random(2^128 -1)
int known_id;
@@
-87,11
+85,11
@@
static void get_mailbox(int node_id, char* mailbox)
}
/** Get the specific level of a node id */
}
/** Get the specific level of a node id */
-int domain_mask = 0;
+
unsigned
int domain_mask = 0;
static int domain(int a, int level) {
if (domain_mask == 0)
domain_mask = pow(2, DOMAIN_SIZE) - 1;
static int domain(int a, 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;
}
return (a >> shift) & domain_mask;
}
@@
-103,20
+101,30
@@
static int shl(int a, int b) {
return l;
}
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) {
/* 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;
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;
res = node->id;
- int i, dist;
- for (i=0; i<NAMESPACE_SIZE; i++) {
+ for (int i=0; i<NAMESPACE_SIZE; i++) {
if (node->namespace_set[i]!=-1) {
if (node->namespace_set[i]!=-1) {
- dist = abs(node->namespace_set[i] - dest);
+
int
dist = abs(node->namespace_set[i] - dest);
if (dist<best_dist) {
best_dist = dist;
if (dist<best_dist) {
best_dist = dist;
- res = node->namespace_set[i];
+ res = node->namespace_set[i];
}
}
}
}
}
}
@@
-138,9
+146,9
@@
static int routing_next(node_t node, int dest) {
//rare case
int dist = abs(node->id - dest);
//rare case
int dist = abs(node->id - dest);
- int i
,j
;
+ int i;
for (i=l; i<LEVELS_COUNT; i++) {
for (i=l; i<LEVELS_COUNT; i++) {
- for (j=0; j<LEVEL_SIZE; j++) {
+ for (
int
j=0; j<LEVEL_SIZE; j++) {
res = node->routing_table[i][j];
if (res!=-1 && abs(res - dest)<dist)
return res;
res = node->routing_table[i][j];
if (res!=-1 && abs(res - dest)<dist)
return res;
@@
-164,14
+172,14
@@
static int routing_next(node_t node, int dest) {
/* Get the corresponding state of a node */
static state_t node_get_state(node_t node) {
/* Get the corresponding state of a node */
static state_t node_get_state(node_t node) {
- int i
,j
;
+ int i;
state_t state = xbt_new0(s_state_t,1);
state->id = node->id;
for (i=0; i<NEIGHBORHOOD_SIZE; i++)
state->neighborhood_set[i] = node->neighborhood_set[i];
for (i=0; i<LEVELS_COUNT; i++)
state_t state = xbt_new0(s_state_t,1);
state->id = node->id;
for (i=0; i<NEIGHBORHOOD_SIZE; i++)
state->neighborhood_set[i] = node->neighborhood_set[i];
for (i=0; i<LEVELS_COUNT; i++)
- for (j=0; j<LEVEL_SIZE; j++)
+ for (
int
j=0; j<LEVEL_SIZE; j++)
state->routing_table[i][j] = node->routing_table[i][j];
for (i=0; i<NAMESPACE_SIZE; i++)
state->routing_table[i][j] = node->routing_table[i][j];
for (i=0; i<NAMESPACE_SIZE; i++)
@@
-182,43
+190,35
@@
static state_t node_get_state(node_t node) {
/* Print the node id */
static void print_node_id(node_t node) {
/* Print the node id */
static void print_node_id(node_t node) {
- int i;
- printf(" id: %i '%08x' ", node->id, node->id);
- for (i=0;i<LEVELS_COUNT;i++)
- printf(" %x", domain(node->id, 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) {
}
/* * Print the node neighborhood set */
static void print_node_neighborood_set(node_t node) {
- int i;
- printf(" Neighborhood:\n");
- for (i=0; i<NEIGHBORHOOD_SIZE; i++)
- printf(" %08x\n", node->neighborhood_set[i]);
+ XBT_INFO(" Neighborhood:");
+ for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
+ XBT_INFO(" %08x", node->neighborhood_set[i]);
}
/* Print the routing table */
static void print_node_routing_table(node_t node) {
}
/* 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; i<LEVELS_COUNT; i++){
for (int i=0; i<LEVELS_COUNT; i++){
- printf(" ");
for (int j=0; j<LEVEL_SIZE; j++)
for (int j=0; j<LEVEL_SIZE; j++)
- printf("%08x ", node->routing_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) {
}
}
/* Print the node namespace set */
static void print_node_namespace_set(node_t node) {
-
printf(" namespace:\n
");
+
XBT_INFO(" Namespace:
");
for (int i=0; i<NAMESPACE_SIZE; i++)
for (int i=0; i<NAMESPACE_SIZE; i++)
- printf(" %08x\n", node->namespace_set[i]);
- printf("\n");
+ XBT_INFO(" %08x", node->namespace_set[i]);
}
/* Print the node information */
static void print_node(node_t node) {
}
/* 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);
print_node_id(node);
print_node_neighborood_set(node);
print_node_routing_table(node);
@@
-229,7
+229,11
@@
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];
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;
msg_task_t task_sent;
task_data_t req_data;
task_data_t task_data = (task_data_t) MSG_task_get_data(task);
msg_task_t task_sent;
task_data_t req_data;
task_data_t task_data = (task_data_t) MSG_task_get_data(task);
@@
-257,7
+261,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);
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;
}
type = TASK_JOIN_REPLY;
}
@@
-267,7
+274,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);
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 */
break;
}
/* Join reply from all the node touched by the join */
@@
-281,6
+291,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;
}
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");
case TASK_JOIN_REPLY:
XBT_DEBUG("Joining Reply");
@@
-323,7
+334,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);
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);
+ }
}
}
}
}
}
}
@@
-333,14
+347,14
@@
static void handle_task(node_t node, msg_task_t task) {
XBT_DEBUG("Task update %i !!!", node->id);
/* Update namespace ses */
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];
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 (; i<NAMESPACE_SIZE/2; i++){
curr_namespace_set[i] = node->namespace_set[i];
i=0;
for (; i<NAMESPACE_SIZE/2; i++){
curr_namespace_set[i] = node->namespace_set[i];
@@
-357,24
+371,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];
max = -1;
for (i=0; i<=NAMESPACE_SIZE; i++) {
j = task_namespace_set[i];
- if (
i<NAMESPACE_SIZE
)
-
printf("%08x %08x | ", j, curr_namespace_set[i])
;
- if (j != -1 &&
j < node->id) 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--) {
// 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];
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--;
} else if (curr_namespace_set[j] > task_namespace_set[min]) {
node->namespace_set[i] = curr_namespace_set[j];
j--;
@@
-387,8
+399,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; i++) {
// add greater elements
j = NAMESPACE_SIZE/2;
for (i=NAMESPACE_SIZE/2; i<NAMESPACE_SIZE; i++) {
- printf("i:%i, j:%i, max:%i, currj:%08x, taskmax:%08x\n", i, j, max, curr_namespace_set[j],
- task_namespace_set[max]);
if (min<0 || max>=NAMESPACE_SIZE) {
node->namespace_set[i] = curr_namespace_set[j];
j++;
if (min<0 || max>=NAMESPACE_SIZE) {
node->namespace_set[i] = curr_namespace_set[j];
j++;
@@
-397,7
+407,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];
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++;
} else if (curr_namespace_set[j] < task_namespace_set[max]) {
node->namespace_set[i] = curr_namespace_set[j];
j++;
@@
-406,7
+417,6
@@
static void handle_task(node_t node, msg_task_t task) {
max++;
}
}
max++;
}
}
- print_node_namespace_set(node);
/* Update routing table */
for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
/* Update routing table */
for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
@@
-415,7
+425,11
@@
static void handle_task(node_t node, msg_task_t task) {
node->routing_table[i][j] = task_data->state->routing_table[i][j];
}
}
node->routing_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.
}
/** \brief Initializes the current node as the first one of the system.
@@
-440,12
+454,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_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;
}
return 1;
}
-
/**
* \brief Node Function
* Arguments:
/**
* \brief Node Function
* Arguments:
@@
-469,10
+485,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);
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<LEVELS_COUNT; i++){
for (i=0; i<LEVELS_COUNT; i++){
- d = domain(node.id, i);
- for (j=0; j<LEVEL_SIZE; j++)
+
int
d = domain(node.id, i);
+ for (
int
j=0; j<LEVEL_SIZE; j++)
node.routing_table[i][j] = (d==j) ? node.id : -1;
}
node.routing_table[i][j] = (d==j) ? node.id : -1;
}
@@
-533,8
+549,16
@@
static int node(int argc, char *argv[])
}
}
}
}
- print_node(&node);
+ //Cleanup the receiving communication.
+ if (node.comm_receive != NULL) {
+ if (MSG_comm_test(node.comm_receive) && MSG_comm_get_status(node.comm_receive) == MSG_OK) {
+ task_free(MSG_comm_get_task(node.comm_receive));
+ }
+ MSG_comm_destroy(node.comm_receive);
+ }
+
}
}
+ xbt_free(node.pending_tasks);
return 1;
}
return 1;
}
@@
-571,7
+595,6
@@
int main(int argc, char *argv[])
MSG_launch_application(options[1]);
msg_error_t res = MSG_main();
MSG_launch_application(options[1]);
msg_error_t res = MSG_main();
- XBT_CRITICAL("Messages created: %ld", smx_total_comms);
XBT_INFO("Simulated time: %g", MSG_get_clock());
return res != MSG_OK;
XBT_INFO("Simulated time: %g", MSG_get_clock());
return res != MSG_OK;