Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
remove leaks from pastry example
[simgrid.git] / examples / msg / dht-pastry / dht-pastry.c
index d0afa78..214e509 100644 (file)
@@ -12,7 +12,7 @@
 /* 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
@@ -101,6 +101,18 @@ 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;
@@ -180,43 +192,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);
@@ -255,7 +259,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 +272,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  */
@@ -321,7 +331,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 +344,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,18 +368,13 @@ 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;
       }
-      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--;
@@ -385,8 +393,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++;
@@ -404,7 +410,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++) {
@@ -414,6 +419,7 @@ 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.
@@ -438,12 +444,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:
@@ -531,8 +539,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;
 }