Checking if circle is formed.

4 files changed, 303 insertions, 178 deletions

diff --git a/src/dht/gnunet-service-xdht_clients.c b/src/dht/gnunet-service-xdht_clients.c
index 8d877654a..901b9c321 100644
--- a/src/dht/gnunet-service-xdht_clients.c
+++ b/src/dht/gnunet-service-xdht_clients.c
@@ -43,6 +43,9 @@
 
 #define LOG(kind,...) GNUNET_log_from (kind, "dht-clients",__VA_ARGS__)
 
+#define DEBUG(...)                                           \
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
+
 #if ENABLE_MALICIOUS
 /**
  * Should this peer act malicious?
@@ -645,6 +648,7 @@ GDS_CLIENTS_handle_reply (struct GNUNET_TIME_Absolute expiration,
                 _("Could not pass reply to client, message too big!\n"));
     return;
   }
+  DEBUG("reply FOR DATA_SIZE = %lu\n",msize);
   pm = GNUNET_malloc (msize + sizeof (struct PendingMessage));
   reply = (struct GNUNET_DHT_ClientResultMessage *) &pm[1];
   pm->msg = &reply->header;
@@ -949,7 +953,7 @@ handle_dht_local_put (void *cls, struct GNUNET_SERVER_Client *client,
        "Handling local PUT of %u-bytes for query %s\n",
        size - sizeof (struct GNUNET_DHT_ClientPutMessage),
        GNUNET_h2s (&put_msg->key));
-  
+  DEBUG("PUT doing put i = %s\n",GNUNET_h2s(&(put_msg->key)));
   GDS_CLIENTS_handle_reply (GNUNET_TIME_absolute_ntoh (put_msg->expiration),
                             &put_msg->key, 0, NULL, 0, NULL,
                             ntohl (put_msg->type),
diff --git a/src/dht/gnunet-service-xdht_datacache.c b/src/dht/gnunet-service-xdht_datacache.c
index a7952cc0c..5f054c730 100644
--- a/src/dht/gnunet-service-xdht_datacache.c
+++ b/src/dht/gnunet-service-xdht_datacache.c
@@ -34,6 +34,8 @@
 
 #define LOG(kind,...) GNUNET_log_from (kind, "dht-dtcache",__VA_ARGS__)
 
+#define DEBUG(...)                                           \
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
 
 /**
  * Handle to the datacache service (for inserting/retrieving data)
@@ -79,7 +81,7 @@ GDS_DATACACHE_handle_put (struct GNUNET_TIME_Absolute expiration,
   GNUNET_STATISTICS_update (GDS_stats,
                             gettext_noop ("# ITEMS stored in datacache"), 1,
                             GNUNET_NO);
-
+   DEBUG("PUT doing put key = %s\n",GNUNET_h2s((key)));
   r = GNUNET_DATACACHE_put (datacache, key, data_size, data, type, expiration,
                             put_path_length, put_path);
   LOG (GNUNET_ERROR_TYPE_DEBUG,
diff --git a/src/dht/gnunet-service-xdht_neighbours.c b/src/dht/gnunet-service-xdht_neighbours.c
index dfd210795..c5a89208f 100644
--- a/src/dht/gnunet-service-xdht_neighbours.c
+++ b/src/dht/gnunet-service-xdht_neighbours.c
@@ -2117,8 +2117,10 @@ GDS_NEIGHBOURS_send_put (const struct GNUNET_HashCode *key,
     msize = data_size + sizeof (struct PeerPutMessage);
   }
 
+  /* Should it be GNUNET_SERVER_MAX_MESSAGE_SIZE? */
   if (msize >= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
   {
+    DEBUG("msize = %lu\n",msize);
     GNUNET_break (0);
     return;
   }
@@ -2141,6 +2143,7 @@ GDS_NEIGHBOURS_send_put (const struct GNUNET_HashCode *key,
     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&best_known_dest, &my_identity))
     {
       /* I am the destination. */
+      DEBUG("PUT destination is me = %s,key =%s\n",GNUNET_i2s(&my_identity),GNUNET_h2s(key));
       GDS_DATACACHE_handle_put (expiration_time, key, 0, NULL,
                                 block_type,data_size,data);
       return;
@@ -2233,7 +2236,9 @@ GDS_NEIGHBOURS_send_get (const struct GNUNET_HashCode *key,
     GNUNET_break (0);
     return;
   }
-
+  
+  DEBUG("GET FOR DATA_SIZE = %lu\n",msize);
+  
   /* This is the first time we got request from our own client file. */
   if (NULL == target_peer)
   {
@@ -2251,6 +2256,7 @@ GDS_NEIGHBOURS_send_get (const struct GNUNET_HashCode *key,
     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&my_identity,
                                               &best_known_dest))
     {
+      DEBUG("GET destination is me = %s,KEY = %s\n",GNUNET_i2s(&my_identity),GNUNET_h2s(key));
       GDS_DATACACHE_handle_get (key,block_type, NULL, 0,
                                 NULL, 0, 1, &my_identity, NULL,&my_identity);
 
@@ -2346,7 +2352,7 @@ GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *key,
     GNUNET_break(0);
     return;
   }
-  
+  DEBUG("GET RESULT  FOR DATA_SIZE = %lu\n",msize);
   current_path_index = 0;
   if(get_path_length > 0)
   {
@@ -3107,23 +3113,6 @@ send_verify_successor_message (void *cls,
   /* Trail stored at this index. */
   GNUNET_assert (GNUNET_YES == trail->is_present);
   
-  /* Code for testing ONLY: Store the successor for path tracking */
-  if (track_topology &&  (NULL != GDS_stats))
-  {
-    char *my_id_str;
-    char *succ_id_str;
-    char *key;
-    
-    my_id_str = GNUNET_strdup (GNUNET_i2s (&my_identity));
-    succ_id_str = GNUNET_strdup (GNUNET_i2s
-                                 (&successor->finger_identity));
-    GNUNET_asprintf (&key, "XDHT:%s:%s", my_id_str, succ_id_str);
-    GNUNET_free (my_id_str);
-    GNUNET_free (succ_id_str);
-    GNUNET_STATISTICS_update (GDS_stats, key, 1, 0);
-    GNUNET_free (key);
-  }
-  
   trail_id = trail->trail_id;
   trail_length = trail->trail_length;
   
@@ -3524,7 +3513,7 @@ handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
     GNUNET_break_op (0);
     return GNUNET_OK;
   }
-
+  DEBUG("GET FOR DATA_SIZE = %lu\n",msize);
   GNUNET_STATISTICS_update (GDS_stats,
                             gettext_noop
                             ("# Bytes received from other peers"), (int64_t) msize,
@@ -3636,6 +3625,7 @@ handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
   /* I am the final destination */
   if (0 == GNUNET_CRYPTO_cmp_peer_identity (&my_identity, &best_known_dest))
   {
+    DEBUG("PUT destination is me = %s,KEY = %s\n",GNUNET_i2s(&my_identity),GNUNET_h2s(&(put->key)));
     GDS_DATACACHE_handle_put (GNUNET_TIME_absolute_ntoh (put->expiration_time),
                               &(put->key),putlen, pp, ntohl (put->block_type),
                               payload_size, payload);
@@ -3698,7 +3688,7 @@ handle_dht_p2p_get (void *cls, const struct GNUNET_PeerIdentity *peer,
     GNUNET_break_op (0);
     return GNUNET_YES;
   }
-
+  DEBUG("PUT FOR DATA_SIZE = %lu\n",msize);
   GNUNET_STATISTICS_update (GDS_stats,
                             gettext_noop
                             ("# Bytes received from other peers"), msize,
@@ -3745,6 +3735,7 @@ handle_dht_p2p_get (void *cls, const struct GNUNET_PeerIdentity *peer,
   /* I am the final destination. */
   if (0 == GNUNET_CRYPTO_cmp_peer_identity(&my_identity, &best_known_dest))
   {
+    DEBUG("GET destination is me = %s,KEY = %s\n",GNUNET_i2s(&my_identity),GNUNET_h2s(&(get->key)));
     struct GNUNET_PeerIdentity final_get_path[get_length+1];
 
     memcpy (final_get_path, gp, get_length * sizeof (struct GNUNET_PeerIdentity));
@@ -3811,7 +3802,7 @@ handle_dht_p2p_get_result (void *cls, const struct GNUNET_PeerIdentity *peer,
     GNUNET_break_op (0);
     return GNUNET_YES;
   }
-
+  DEBUG("GET_RESULT  FOR DATA_SIZE = %lu\n",msize);
   GNUNET_STATISTICS_update (GDS_stats,
                             gettext_noop
                             ("# Bytes received from other peers"), msize,
@@ -4956,8 +4947,24 @@ compare_and_update_successor (struct GNUNET_PeerIdentity curr_succ,
 
   /* If the current_successor in the finger table is closest, then do nothing. */
   if (closest_peer == &current_successor->finger_identity)
+  {
+    /* Code for testing ONLY: Store the successor for path tracking */
+    track_topology = 1;
+    if (track_topology &&  (NULL != GDS_stats))
+    {
+      char *my_id_str;
+      uint64_t succ;
+      char *key;
+    
+      my_id_str = GNUNET_strdup (GNUNET_i2s_full (&my_identity));
+      memcpy(&succ, &current_successor->finger_identity, sizeof(uint64_t));
+      GNUNET_asprintf (&key, "XDHT:%s:", my_id_str);
+      GNUNET_free (my_id_str);
+      GNUNET_STATISTICS_set (GDS_stats, key, succ, 0);
+      GNUNET_free (key);
+    }
     return;
-
+  }
   /* Probable successor is the closest peer.*/
   if(trail_length > 0)
   {
@@ -5603,7 +5610,7 @@ remove_matching_trails (const struct GNUNET_PeerIdentity *disconnected_friend,
   matching_trails_count = 0;
 
   /* Iterate over all the trails of finger. */
-  for (i = 0; i < remove_finger->trails_count; i++)
+  for (i = 0; i < MAXIMUM_TRAILS_PER_FINGER; i++)
   {
     struct Trail *trail;
     trail = &remove_finger->trail_list[i];
diff --git a/src/dht/gnunet_dht_profiler.c b/src/dht/gnunet_dht_profiler.c
index fcec257f0..0fb7ab7df 100644
--- a/src/dht/gnunet_dht_profiler.c
+++ b/src/dht/gnunet_dht_profiler.c
@@ -38,7 +38,7 @@
 /**
  * Number of peers which should perform a PUT out of 100 peers
  */
-#define PUT_PROBABILITY 100
+#define PUT_PROBABILITY 50
 
 /**
  * Configuration
@@ -246,31 +246,71 @@ static unsigned int average_put_path_length;
 static unsigned int average_get_path_length;
 
 /**
- * 
+ * Total put path length across all peers. 
  */
 static unsigned int total_put_path_length;
 
 /**
- *
+ * Total get path length across all peers. 
  */
 static unsigned int total_get_path_length;
 
 /**
- *
+ * Hashmap to store pair of peer and its corresponding successor. 
  */
 static struct GNUNET_CONTAINER_MultiHashMap *successor_peer_hashmap;
 
 /**
- *
+ * Key to start the lookup on successor_peer_hashmap. 
  */
 static struct GNUNET_HashCode *start_key;
 
 /**
- *
+ * Flag used to get the start_key.
  */
 static int flag = 0;
 
 /**
+ * Task to collect peer and its current successor statistics.
+ */
+static GNUNET_SCHEDULER_TaskIdentifier successor_stats_task;
+
+/**
+ * Closure for successor_stats_task.
+ */
+struct Collect_Stat_Context
+{
+  /**
+   * Current Peer Context. 
+   */
+  struct Context *service_connect_ctx;
+  
+  /**
+   * Testbed operation acting on this peer
+   */
+  struct GNUNET_TESTBED_Operation *op;
+};
+
+/**
+ * List of all the peers contexts.
+ */
+struct Context **peer_contexts = NULL;
+
+/**
+ * Counter to keep track of peers added to peer_context lists. 
+ */
+static int peers_started = 0;
+
+/**
+ * Task that collects successor statistics from all the peers. 
+ * @param cls
+ * @param tc
+ */
+static void
+collect_stats (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+/**
  * Shutdown task.  Cleanup all resources and operations.
  *
  * @param cls NULL
@@ -536,121 +576,6 @@ put_cont (void *cls, int success)
 
 
 /**
- * Stats callback. Finish the stats testbed operation and when all stats have
- * been iterated, shutdown the test.
- *
- * @param cls closure
- * @param op the operation that has been finished
- * @param emsg error message in case the operation has failed; will be NULL if
- *          operation has executed successfully.
- */
-static void
-successor_stats_cont (void *cls, 
-                      struct GNUNET_TESTBED_Operation *op, 
-                      const char *emsg)
-{
-  /* Check if ring is formed. If yes then schedule put. */
-  struct GNUNET_HashCode *val;
-  struct GNUNET_HashCode *start_val;
-  int count = 0;
-  struct GNUNET_HashCode *key;
-  
-  start_val = GNUNET_CONTAINER_multihashmap_get(successor_peer_hashmap,
-                                                start_key);
-  
-  val = GNUNET_new(struct GNUNET_HashCode);
-  val = start_val;
-  while (count < n_active)
-  {
-    key = val;
-    val = GNUNET_CONTAINER_multihashmap_get (successor_peer_hashmap,
-                                             key);
-    count++;
-  }
-  
-  if (start_val == val)
-  {
-    DEBUG("Circle complete\n");
-    /* FIXME: Schedule the delayed PUT task */
-  }
-  else
-  {
-    static unsigned int tries;
-
-    DEBUG("Circle not complete\n");
-    if (max_searches == ++tries)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Maximum tries %u exceeded while checking successor"
-                  " cirle formation.  Exiting\n",
-                  max_searches);
-      GNUNET_SCHEDULER_shutdown ();
-      return;
-    }
-    /* FIXME: Re-schedule the successor stats gathering task to run after some
-       delay */
-  }
-}
-
-
-/**
- * Process successor statistic values.
- *
- * @param cls closure
- * @param peer the peer the statistic belong to
- * @param subsystem name of subsystem that created the statistic
- * @param name the name of the datum
- * @param value the current value
- * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
- * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
- */
-static int
-successor_stats_iterator (void *cls, 
-                          const struct GNUNET_TESTBED_Peer *peer,
-                          const char *subsystem, 
-                          const char *name,
-                          uint64_t value, 
-                          int is_persistent)
-{
-  static const char *key_string = "XDHT";
-  
-  DEBUG (" Inside successor stats,name = %s\n",name);
-  if (0 == strncmp (key_string, name, strlen (key_string)))
-  {
-    char *my_id_str;
-    char *successor_id_str;
-    struct GNUNET_HashCode *my_id;
-    struct GNUNET_HashCode *successor_id;
-    
-    /* Parse the string to get the peer and its successor. */
-    strtok((char *)name,":");
-    my_id_str = strtok(NULL,":");
-    successor_id_str = strtok(NULL,":");
-    
-    /* Get Hash of my_id_str and successor_id_str */
-    my_id = GNUNET_new(struct GNUNET_HashCode);
-    successor_id = GNUNET_new(struct GNUNET_HashCode);
-    GNUNET_CRYPTO_hash (my_id_str, sizeof(my_id_str),my_id);
-    GNUNET_CRYPTO_hash (successor_id_str, sizeof(successor_id_str),successor_id);
-    
-    if (0 == flag)
-    {
-      start_key = my_id;
-      flag = 1;
-    }
-    
-    GNUNET_CONTAINER_multihashmap_put (successor_peer_hashmap,
-                                       my_id, (void *)successor_id,
-                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
-   
-
-  }
-  
-  return GNUNET_OK;
-}
-
-
-/**
  * Task to do DHT PUTS
  *
  * @param cls the active context
@@ -660,17 +585,7 @@ static void
 delayed_put (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct ActiveContext *ac = cls;
-  
-  successor_peer_hashmap = GNUNET_CONTAINER_multihashmap_create (n_active, 
-                                                                 GNUNET_NO);
-  /* Check for successor pointer, don't start put till the virtual ring topology
-   is not created. */
-  successor_stats_op = 
-          GNUNET_TESTBED_get_statistics (n_active, testbed_handles,
-                                         "dht", NULL,
-                                          successor_stats_iterator, 
-                                          successor_stats_cont, NULL);
-  
+  DEBUG("PUT SUPU \n");
   ac->delay_task = GNUNET_SCHEDULER_NO_TASK;
   /* Generate and DHT PUT some random data */
   ac->put_data_size = 16;       /* minimum */
@@ -694,7 +609,6 @@ delayed_put (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 }
 
 
-
 /**
  * Connection to DHT has been established.  Call the delay task.
  *
@@ -724,11 +638,7 @@ dht_connected (void *cls,
     ctx->op = NULL;
     return;
   }
-  
-  DEBUG (" Call stats \n");
- 
-  /* FIXME: move this to happen after the successor circle formation is
-     complete */
+   
   ac->delay_task = GNUNET_SCHEDULER_add_delayed (delay, &delayed_put, ac);
 }
 
@@ -772,6 +682,197 @@ dht_disconnect (void *cls, void *op_result)
 
 
 /**
+ * FIXME:Verify where is n_active used. Should this service be started only
+ * for n_active peers?
+ * Start testbed service for all the peers. 
+ */
+static void
+start_testbed_service_on_all_peers()
+{
+  unsigned int i;
+  for(i = 0; i < peers_started; i++)
+  {
+      struct Context *ctx = peer_contexts[i];
+      DEBUG("GNUNET_TESTBED_service_connect \n");
+      ctx->op = 
+              GNUNET_TESTBED_service_connect (ctx, 
+                                              ctx->peer,
+                                              "dht",
+                                              &dht_connected, ctx->ac,
+                                              &dht_connect,
+                                              &dht_disconnect,
+                                              ctx->ac);
+     
+  }
+}
+
+
+/**
+ * Stats callback. Iterate over the hashmap and check if all th peers form
+ * a virtual ring topology.
+ *
+ * @param cls closure
+ * @param op the operation that has been finished
+ * @param emsg error message in case the operation has failed; will be NULL if
+ *          operation has executed successfully.
+ */
+static void
+successor_stats_cont (void *cls, 
+                      struct GNUNET_TESTBED_Operation *op, 
+                      const char *emsg)
+{
+  struct GNUNET_HashCode *val;
+  struct GNUNET_HashCode *start_val;
+  int count = 0;
+  struct GNUNET_HashCode *key;
+  
+  start_val =(struct GNUNET_HashCode *) GNUNET_CONTAINER_multihashmap_get(successor_peer_hashmap,
+                                                start_key);
+
+  val = GNUNET_new(struct GNUNET_HashCode);
+  val = start_val;
+  while (count < num_peers)
+  {
+    key = GNUNET_new(struct GNUNET_HashCode);
+    key = val;
+    val = GNUNET_CONTAINER_multihashmap_get (successor_peer_hashmap,
+                                             key);
+    GNUNET_assert(NULL != val);
+    count++;
+  }
+  
+  if (start_val == val)
+  {
+    DEBUG("Circle completed\n");
+    if (GNUNET_SCHEDULER_NO_TASK != successor_stats_task)
+    {
+      successor_stats_task = GNUNET_SCHEDULER_NO_TASK;
+      //FIXME: free hashmap. 
+    }
+    successor_stats_op = NULL;
+    
+    if(GNUNET_SCHEDULER_NO_TASK == successor_stats_task)
+    {
+      start_testbed_service_on_all_peers();
+    }
+    
+    return;
+  }
+  else
+  {
+    static unsigned int tries;
+
+    DEBUG("Circle not complete\n");
+    if (max_searches == ++tries)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Maximum tries %u exceeded while checking successor TOTAL TRIES %u"
+                  " cirle formation.  Exiting\n",
+                  max_searches,tries);
+      if (GNUNET_SCHEDULER_NO_TASK != successor_stats_task)
+      {
+        successor_stats_task = GNUNET_SCHEDULER_NO_TASK;
+        //FIXME: free hashmap
+      }
+      successor_stats_op = NULL;
+      
+      if(GNUNET_SCHEDULER_NO_TASK == successor_stats_task)
+      {
+        start_testbed_service_on_all_peers();
+      }
+      
+      return;
+    }
+    
+    //FIXME: change delay use exponential back off. 
+    successor_stats_task = GNUNET_SCHEDULER_add_delayed (delay, &collect_stats, cls);
+  }
+}
+
+
+/**
+ * Process successor statistic values.
+ *
+ * @param cls closure
+ * @param peer the peer the statistic belong to
+ * @param subsystem name of subsystem that created the statistic
+ * @param name the name of the datum
+ * @param value the current value
+ * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
+ * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
+ */
+static int
+successor_stats_iterator (void *cls, 
+                          const struct GNUNET_TESTBED_Peer *peer,
+                          const char *subsystem, 
+                          const char *name,
+                          uint64_t value, 
+                          int is_persistent)
+{
+  static const char *key_string = "XDHT";
+  
+  if (0 == strncmp (key_string, name, strlen (key_string)))
+  {
+    char *my_id_str;
+    char successor_str[13];
+    char truncated_my_id_str[13];
+    char truncated_successor_str[13];
+    struct GNUNET_HashCode *my_id_key;
+    struct GNUNET_HashCode *succ_key;
+    
+    strtok((char *)name,":");
+    my_id_str = strtok(NULL,":");
+    
+    strncpy(truncated_my_id_str, my_id_str, 12);
+    truncated_my_id_str[12] = '\0';
+    
+    my_id_key = GNUNET_new(struct GNUNET_HashCode);
+    GNUNET_CRYPTO_hash (truncated_my_id_str, sizeof(truncated_my_id_str),my_id_key);
+    
+    GNUNET_STRINGS_data_to_string(&value, sizeof(uint64_t), successor_str, 13);
+    strncpy(truncated_successor_str, successor_str, 12);
+    truncated_successor_str[12] ='\0';
+   
+    succ_key = GNUNET_new(struct GNUNET_HashCode);
+    GNUNET_CRYPTO_hash (truncated_successor_str, sizeof(truncated_successor_str),succ_key);
+    
+    if (0 == flag)
+    {
+      start_key = my_id_key;
+      flag = 1;
+    }
+    GNUNET_CONTAINER_multihashmap_put (successor_peer_hashmap,
+                                       my_id_key, (void *)succ_key,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
+  }
+  return GNUNET_OK;
+}
+
+
+/* 
+ * Task that collects peer and its corresponding successors. 
+ * 
+ * @param cls Closure (NULL).
+ * @param tc Task Context.
+ */
+static void
+collect_stats (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  if ((GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason) != 0)
+    return;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Start collecting statistics...\n");
+  
+ /* Check for successor pointer, don't start put till the virtual ring topology
+   is not created. */
+  successor_stats_op = 
+          GNUNET_TESTBED_get_statistics (num_peers, testbed_handles,
+                                         "dht", NULL,
+                                          successor_stats_iterator, 
+                                          successor_stats_cont, cls);
+}
+
+/**
  * Callback called when DHT service on the peer is started
  *
  * @param cls the context
@@ -785,24 +886,35 @@ service_started (void *cls,
                  const char *emsg)
 {
   struct Context *ctx = cls;
-  static unsigned int nstarted;
-
+  
   GNUNET_assert (NULL != ctx);
   GNUNET_assert (NULL != ctx->op);
   GNUNET_TESTBED_operation_done (ctx->op);
   ctx->op = NULL;
   if (NULL == ctx->ac)
     return;
-  ctx->op = GNUNET_TESTBED_service_connect (ctx, ctx->peer,
-                                            "dht",
-                                            &dht_connected, ctx->ac,
-                                            &dht_connect,
-                                            &dht_disconnect,
-                                            ctx->ac);
-  if (num_peers == ++nstarted)
+  
+  if (NULL == peer_contexts)
+  {
+    peer_contexts = GNUNET_malloc(num_peers * sizeof(struct Context *));
+  }
+  
+  peer_contexts[peers_started] = ctx;
+  peers_started++;
+  DEBUG("Peers Started = %d \n", peers_started);
+    
+  if (GNUNET_SCHEDULER_NO_TASK == successor_stats_task)
   {
-    /* FIXME: schedule a delayed task to scan the successors from statistics of
-       all peers */
+     DEBUG("successor_stats_task \n");
+     struct Collect_Stat_Context *collect_stat_cls = GNUNET_new(struct Collect_Stat_Context);
+     collect_stat_cls->service_connect_ctx = cls;
+     collect_stat_cls->op = op;
+     
+     successor_peer_hashmap = GNUNET_CONTAINER_multihashmap_create (num_peers, 
+                                                                    GNUNET_NO);
+     successor_stats_task = GNUNET_SCHEDULER_add_delayed (delay, 
+                                                          &collect_stats, 
+                                                          collect_stat_cls);
   }
 }
 
@@ -936,11 +1048,11 @@ main (int argc, char *const *argv)
     GNUNET_GETOPT_OPTION_END
   };
 
-  max_searches = 1;
+  max_searches = 10;
   if (GNUNET_OK != GNUNET_STRINGS_get_utf8_args (argc, argv, &argc, &argv))
     return 2;
-  delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1); /* default delay */
-  timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1); /* default timeout */
+  delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2); /* default delay */
+  timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1); /* default timeout */
   replication = 1;      /* default replication */
   rc = 0;
   if (GNUNET_OK !=