/* 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 <math.h>
#include "simgrid/msg.h"
+#include "xbt/dynar.h"
+#include <math.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 *
- * - handle neighborood in the update */
+ * - handle neighborhood in the update */
#define COMM_SIZE 10
#define COMP_SIZE 0
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 {
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(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);
}
/** 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;
}
return l;
}
+/* Frees the memory used by a task and destroy it */
+static void task_free(void* task)
+{
+ 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; i<NAMESPACE_SIZE; i++) {
+ for (int i=0; i<NAMESPACE_SIZE; i++) {
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;
- res = node->namespace_set[i];
+ res = node->namespace_set[i];
}
}
}
/* 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;
- for (i=l; i<LEVELS_COUNT; i++) {
- for (j=0; j<LEVEL_SIZE; j++) {
+ for (int i=l; i<LEVELS_COUNT; i++) {
+ for (int j=0; j<LEVEL_SIZE; j++) {
res = node->routing_table[i][j];
if (res!=-1 && abs(res - dest)<dist)
return res;
}
}
- for (i=0; i<NEIGHBORHOOD_SIZE; i++) {
+ for (int i=0; i<NEIGHBORHOOD_SIZE; i++) {
res = node->neighborhood_set[i];
if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
return res;
}
- for (i=0; i<NAMESPACE_SIZE; i++) {
+ for (int i=0; i<NAMESPACE_SIZE; i++) {
res = node->namespace_set[i];
if (res!=-1 && shl(res, dest)>=l && abs(res - dest)<dist)
return res;
/* Get the corresponding state of a node */
static state_t node_get_state(node_t node) {
- int i,j;
state_t state = xbt_new0(s_state_t,1);
state->id = node->id;
- for (i=0; i<NEIGHBORHOOD_SIZE; i++)
+ for (int 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 i=0; i<LEVELS_COUNT; i++)
+ 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++)
+ for (int i=0; i<NAMESPACE_SIZE; i++)
state->namespace_set[i] = node->namespace_set[i];
return state;
}
-/* 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) {
- 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) {
- printf(" routing table:\n");
+ XBT_INFO(" Routing table:");
for (int i=0; i<LEVELS_COUNT; i++){
- printf(" ");
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) {
- printf(" namespace:\n");
+ XBT_INFO(" Namespace:");
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) {
- printf("Node:\n");
+ XBT_INFO("Node:");
print_node_id(node);
print_node_neighborood_set(node);
print_node_routing_table(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);
// 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;
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;
}
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; i<NAMESPACE_SIZE/2; i++) {
node->namespace_set[i] = task_data->state->namespace_set[i-j];
}
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");
// 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; i<NAMESPACE_SIZE; i++) {
j = node->namespace_set[i];
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);
+ }
}
}
}
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 (; i<NAMESPACE_SIZE/2; i++){
curr_namespace_set[i] = node->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--) {
- 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--;
// 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++;
- } 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++;
+ }
}
}
- print_node_namespace_set(node);
/* Update routing table */
for (i=shl(node->id, task_data->state->id); i<LEVELS_COUNT; i++) {
node->routing_table[i][j] = task_data->state->routing_table[i][j];
}
}
+ break;
+ default:
+ THROW_IMPOSSIBLE;
}
-}
-
-/** \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...");
+ task_free(task);
}
/* Join the ring */
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:
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,j,d;
- for (i=0; i<LEVELS_COUNT; i++){
- d = domain(node.id, i);
- for (j=0; j<LEVEL_SIZE; j++)
+
+ for (int i=0; i<LEVELS_COUNT; i++){
+ int d = domain(node.id, i);
+ for (int j=0; j<LEVEL_SIZE; j++)
node.routing_table[i][j] = (d==j) ? node.id : -1;
}
- for (i=0; i<NEIGHBORHOOD_SIZE; i++)
+ for (int i=0; i<NEIGHBORHOOD_SIZE; i++)
node.neighborhood_set[i] = -1;
- for (i=0; i<NAMESPACE_SIZE; i++)
+ for (int i=0; i<NAMESPACE_SIZE; i++)
node.namespace_set[i] = -1;
if (argc == 3) { // first ring
XBT_DEBUG("Hey! Let's create the system.");
deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
- create(&node);
+ node.ready = 0;
+ XBT_DEBUG("Create a new Pastry ring...");
join_success = 1;
- }
- else {
+ } else {
node.known_id = xbt_str_parse_int(argv[2], "Invalid known ID: %s");
double sleep_time = xbt_str_parse_double(argv[3], "Invalid sleep time: %s");
deadline = xbt_str_parse_double(argv[4], "Invalid deadline: %s");
XBT_DEBUG("Failed to receive a task. Nevermind.");
MSG_comm_destroy(node.comm_receive);
node.comm_receive = NULL;
- }
- else {
+ } else {
// the task was successfully received
MSG_comm_destroy(node.comm_receive);
node.comm_receive = NULL;
}
}
- 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_dynar_free(&node.pending_tasks);
return 1;
}
timeout = xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
XBT_DEBUG("Set timeout to %d", timeout);
} else {
- xbt_die("Invalid chord option '%s'", options[0]);
+ xbt_die("Invalid pastry option '%s'", options[0]);
}
}
options++;
