X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/48eccb2c1532e35819830ca56fad7cf89887359f..da6da59e6a2770e25be5d88661e8f8e32810e0ff:/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..270cb94aed 100644
--- 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"
 
-	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
@@ -85,11 +85,11 @@ static void get_mailbox(int node_id, char* mailbox)
 }
 
 /** 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;
-  int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
+  unsigned int shift = (LEVELS_COUNT-level-1)*DOMAIN_SIZE;
   return (a >> shift) & domain_mask;
 }
 
@@ -101,20 +101,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; 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];
         }
       }
     }
@@ -136,9 +146,9 @@ static int routing_next(node_t node, int dest) {
 
   //rare case
   int dist = abs(node->id - dest);
-  int i,j;
+  int 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;
@@ -162,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) {
-  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++)
-    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++)
@@ -180,43 +190,35 @@ static state_t node_get_state(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) {
-  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);
@@ -227,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];
-  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);
@@ -255,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);
-        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,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);
-      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  */
@@ -279,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;
+      /* no break */
     case TASK_JOIN_REPLY:
       XBT_DEBUG("Joining Reply");
 
@@ -321,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);
-            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 +347,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 (; i<NAMESPACE_SIZE/2; i++){
         curr_namespace_set[i] = node->namespace_set[i];
@@ -355,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];
-        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--;
@@ -385,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++) {
-        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++;
@@ -395,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];
-          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 +417,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); i<LEVELS_COUNT; i++) {
@@ -413,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];
         }
       }
+      break;
+    default:
+      THROW_IMPOSSIBLE;
   }
+  task_free(task);
 }
 
 /** \brief Initializes the current node as the first one of the system.
@@ -438,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_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 +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);
   
-  int i,j,d;
+  int 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;
   }
 
@@ -531,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;
 }