-misc fixes to DHT tests

11 files changed, 362 insertions, 134 deletions

diff --git a/src/dht/dht_api.c b/src/dht/dht_api.c
index de1043ca9..f67c1075f 100644
--- a/src/dht/dht_api.c
+++ b/src/dht/dht_api.c
@@ -993,6 +993,10 @@ GNUNET_DHT_put (struct GNUNET_DHT_Handle *handle,
   }
   if (NULL == handle->mq)
     return NULL;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Sending PUT for %s to DHT via %p\n",
+       GNUNET_h2s (key),
+       handle);
   ph = GNUNET_new (struct GNUNET_DHT_PutHandle);
   ph->dht_handle = handle;
   ph->cont = cont;
diff --git a/src/dht/dht_test_lib.c b/src/dht/dht_test_lib.c
index a395ce58b..4c1bd3057 100644
--- a/src/dht/dht_test_lib.c
+++ b/src/dht/dht_test_lib.c
@@ -118,7 +118,8 @@ dht_connect_cb (void *cls,
 
   if (NULL != emsg)
   {
-    fprintf (stderr, "Failed to connect to DHT service: %s\n",
+    fprintf (stderr,
+             "Failed to connect to DHT service: %s\n",
              emsg);
     GNUNET_SCHEDULER_shutdown ();
     return;
@@ -134,7 +135,7 @@ dht_connect_cb (void *cls,
                  ctx,
                  ctx->num_peers,
                  ctx->peers,
-                 ctx->dhts);            
+                 ctx->dhts);
 }
 
 
@@ -203,8 +204,10 @@ GNUNET_DHT_TEST_run (const char *testname,
 
   ctx = GNUNET_new (struct GNUNET_DHT_TEST_Context);
   ctx->num_peers = num_peers;
-  ctx->ops = GNUNET_malloc (num_peers * sizeof (struct GNUNET_TESTBED_Operation *));
-  ctx->dhts = GNUNET_malloc (num_peers * sizeof (struct GNUNET_DHT_Handle *));
+  ctx->ops = GNUNET_new_array (num_peers,
+                               struct GNUNET_TESTBED_Operation *);
+  ctx->dhts = GNUNET_new_array (num_peers,
+                                struct GNUNET_DHT_Handle *);
   ctx->app_main = tmain;
   ctx->app_main_cls = tmain_cls;
   (void) GNUNET_TESTBED_test_run (testname,
diff --git a/src/dht/gnunet-service-dht.c b/src/dht/gnunet-service-dht.c
index a44be2dfe..abdd77548 100644
--- a/src/dht/gnunet-service-dht.c
+++ b/src/dht/gnunet-service-dht.c
@@ -57,6 +57,11 @@ struct GNUNET_BLOCK_Context *GDS_block_context;
 const struct GNUNET_CONFIGURATION_Handle *GDS_cfg;
 
 /**
+ * Handle to our server.
+ */
+struct GNUNET_SERVER_Handle *GDS_server;
+
+/**
  * Our HELLO
  */
 struct GNUNET_MessageHeader *GDS_my_hello;
@@ -140,10 +145,13 @@ shutdown_task (void *cls)
  * @param c configuration to use
  */
 static void
-run (void *cls, struct GNUNET_SERVER_Handle *server,
+run (void *cls,
+     struct GNUNET_SERVER_Handle *server,
      const struct GNUNET_CONFIGURATION_Handle *c)
 {
   GDS_cfg = c;
+  GDS_server = server;
+  GNUNET_SERVER_suspend (server);
   if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_time (c, "transport", "HELLO_EXPIRATION", &hello_expiration))
   {
@@ -155,7 +163,6 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
   GDS_NSE_init ();
   GDS_DATACACHE_init ();
   GDS_HELLO_init ();
-  GDS_CLIENTS_init (server);
   if (GNUNET_OK != GDS_NEIGHBOURS_init ())
   {
     shutdown_task (NULL);
diff --git a/src/dht/gnunet-service-dht.h b/src/dht/gnunet-service-dht.h
index 86e866666..6f641cb96 100644
--- a/src/dht/gnunet-service-dht.h
+++ b/src/dht/gnunet-service-dht.h
@@ -48,6 +48,11 @@ extern struct GNUNET_BLOCK_Context *GDS_block_context;
 extern struct GNUNET_STATISTICS_Handle *GDS_stats;
 
 /**
+ * Handle to our server.
+ */
+extern struct GNUNET_SERVER_Handle *GDS_server;
+
+/**
  * Our HELLO
  */
 extern struct GNUNET_MessageHeader *GDS_my_hello;
diff --git a/src/dht/gnunet-service-dht_clients.c b/src/dht/gnunet-service-dht_clients.c
index 305826aec..b508df68a 100644
--- a/src/dht/gnunet-service-dht_clients.c
+++ b/src/dht/gnunet-service-dht_clients.c
@@ -974,7 +974,7 @@ process_pending_messages (struct ClientList *client)
 
 
 /**
- * Closure for 'forward_reply'
+ * Closure for #forward_reply()
  */
 struct ForwardReplyContext
 {
@@ -1000,7 +1000,7 @@ struct ForwardReplyContext
   size_t data_size;
 
   /**
-   * Do we need to copy 'pm' because it was already used?
+   * Do we need to copy @a pm because it was already used?
    */
   int do_copy;
 
@@ -1015,11 +1015,13 @@ struct ForwardReplyContext
  * @param cls the 'struct ForwardReplyContext'
  * @param key current key
  * @param value value in the hash map, a ClientQueryRecord
- * @return GNUNET_YES (we should continue to iterate),
- *         if the result is mal-formed, GNUNET_NO
+ * @return #GNUNET_YES (we should continue to iterate),
+ *         if the result is mal-formed, #GNUNET_NO
  */
 static int
-forward_reply (void *cls, const struct GNUNET_HashCode * key, void *value)
+forward_reply (void *cls,
+               const struct GNUNET_HashCode *key,
+               void *value)
 {
   struct ForwardReplyContext *frc = cls;
   struct ClientQueryRecord *record = value;
@@ -1167,12 +1169,11 @@ GDS_CLIENTS_handle_reply (struct GNUNET_TIME_Absolute expiration,
   struct GNUNET_PeerIdentity *paths;
   size_t msize;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "reply for key %s\n",
-       GNUNET_h2s (key));
-
   if (NULL == GNUNET_CONTAINER_multihashmap_get (forward_map, key))
   {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "No matching client for reply for key %s\n",
+         GNUNET_h2s (key));
     GNUNET_STATISTICS_update (GDS_stats,
                               gettext_noop
                               ("# REPLIES ignored for CLIENTS (no match)"), 1,
@@ -1184,10 +1185,13 @@ GDS_CLIENTS_handle_reply (struct GNUNET_TIME_Absolute expiration,
       (get_path_length + put_path_length) * sizeof (struct GNUNET_PeerIdentity);
   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                _("Could not pass reply to client, message too big!\n"));
+    GNUNET_break (0);
     return;
   }
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Forwarding reply for key %s to client\n",
+       GNUNET_h2s (key));
+
   pm = GNUNET_malloc (msize + sizeof (struct PendingMessage));
   reply = (struct GNUNET_DHT_ClientResultMessage *) &pm[1];
   pm->msg = &reply->header;
@@ -1210,7 +1214,9 @@ GDS_CLIENTS_handle_reply (struct GNUNET_TIME_Absolute expiration,
   frc.data = data;
   frc.data_size = data_size;
   frc.type = type;
-  GNUNET_CONTAINER_multihashmap_get_multiple (forward_map, key, &forward_reply,
+  GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
+                                              key,
+                                              &forward_reply,
                                               &frc);
 
   if (GNUNET_NO == frc.do_copy)
@@ -1309,7 +1315,7 @@ GDS_CLIENTS_process_get (uint32_t options,
  * @param exp Expiration time of the data.
  * @param key Key of the data.
  * @param data Pointer to the result data.
- * @param size Number of bytes in data.
+ * @param size Number of bytes in @a data.
  */
 void
 GDS_CLIENTS_process_get_resp (enum GNUNET_BLOCK_Type type,
@@ -1452,7 +1458,8 @@ GDS_CLIENTS_process_put (uint32_t options,
       msg_path = (struct GNUNET_PeerIdentity *) &mmsg[1];
       if (path_length > 0)
       {
-        memcpy (msg_path, path,
+        memcpy (msg_path,
+                path,
                 path_length * sizeof (struct GNUNET_PeerIdentity));
       }
       mmsg->expiration_time = GNUNET_TIME_absolute_hton(exp);
@@ -1472,7 +1479,7 @@ GDS_CLIENTS_process_put (uint32_t options,
  * @param server the initialized server
  */
 void
-GDS_CLIENTS_init (struct GNUNET_SERVER_Handle *server)
+GDS_CLIENTS_init ()
 {
   static struct GNUNET_SERVER_MessageHandler plugin_handlers[] = {
     {&handle_dht_local_put, NULL,
@@ -1492,10 +1499,15 @@ GDS_CLIENTS_init (struct GNUNET_SERVER_Handle *server)
      GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_RESULTS_KNOWN, 0},
     {NULL, NULL, 0, 0}
   };
+
   forward_map = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_NO);
   retry_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
-  GNUNET_SERVER_add_handlers (server, plugin_handlers);
-  GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
+  GNUNET_SERVER_resume (GDS_server);
+  GNUNET_SERVER_add_handlers (GDS_server,
+                              plugin_handlers);
+  GNUNET_SERVER_disconnect_notify (GDS_server,
+                                   &handle_client_disconnect,
+                                   NULL);
 }
 
 
diff --git a/src/dht/gnunet-service-dht_clients.h b/src/dht/gnunet-service-dht_clients.h
index 2e1b1b89a..d8e1c184d 100644
--- a/src/dht/gnunet-service-dht_clients.h
+++ b/src/dht/gnunet-service-dht_clients.h
@@ -133,11 +133,9 @@ GDS_CLIENTS_process_put (uint32_t options,
 
 /**
  * Initialize client subsystem.
- *
- * @param server the initialized server
  */
 void
-GDS_CLIENTS_init (struct GNUNET_SERVER_Handle *server);
+GDS_CLIENTS_init (void);
 
 
 /**
diff --git a/src/dht/gnunet-service-dht_neighbours.c b/src/dht/gnunet-service-dht_neighbours.c
index 773ad8983..7b1efff1f 100644
--- a/src/dht/gnunet-service-dht_neighbours.c
+++ b/src/dht/gnunet-service-dht_neighbours.c
@@ -870,6 +870,11 @@ handle_core_disconnect (void *cls,
                  GNUNET_CONTAINER_multipeermap_remove (all_connected_peers,
                                                        peer,
                                                        to_remove));
+  if (0 == GNUNET_CONTAINER_multipeermap_size (all_connected_peers))
+  {
+    GNUNET_SCHEDULER_cancel (find_peer_task);
+    find_peer_task = NULL;
+  }
   GNUNET_CRYPTO_hash (peer,
                       sizeof (struct GNUNET_PeerIdentity),
                       &phash);
@@ -894,6 +899,9 @@ handle_core_disconnect (void *cls,
     GNUNET_CONTAINER_DLL_remove (to_remove->head,
                                  to_remove->tail,
                                  pos);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Dropping message of type %u due to disconnect\n",
+                ntohs (pos->msg->type));
     discarded++;
     GNUNET_free (pos);
   }
@@ -927,6 +935,9 @@ core_transmit_notify (void *cls,
   size_t off;
   size_t msize;
 
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "DHT ctn called with buffer of %u bytes\n",
+              (unsigned int) size);
   peer->th = NULL;
   while ((NULL != (pending = peer->head)) &&
          (0 == GNUNET_TIME_absolute_get_remaining (pending->timeout).rel_value_us))
@@ -936,24 +947,35 @@ core_transmit_notify (void *cls,
                               ("# Messages dropped (CORE timeout)"),
                               1,
                               GNUNET_NO);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Dropping message of type %u due to timeout\n",
+                ntohs (pending->msg->type));
     peer->pending_count--;
-    GNUNET_CONTAINER_DLL_remove (peer->head, peer->tail, pending);
+    GNUNET_CONTAINER_DLL_remove (peer->head,
+                                 peer->tail,
+                                 pending);
     GNUNET_free (pending);
   }
   if (NULL == pending)
   {
     /* no messages pending */
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "No messages pending\n");
     return 0;
   }
   if (NULL == buf)
   {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Got NULL buffer, trying again\n");
     peer->th =
         GNUNET_CORE_notify_transmit_ready (core_api, GNUNET_NO,
                                            GNUNET_CORE_PRIO_BEST_EFFORT,
                                            GNUNET_TIME_absolute_get_remaining
-                                           (pending->timeout), &peer->id,
+                                           (pending->timeout),
+                                           &peer->id,
                                            ntohs (pending->msg->size),
-                                           &core_transmit_notify, peer);
+                                           &core_transmit_notify,
+                                           peer);
     GNUNET_break (NULL != peer->th);
     return 0;
   }
@@ -965,7 +987,13 @@ core_transmit_notify (void *cls,
                               gettext_noop
                               ("# Bytes transmitted to other peers"), msize,
                               GNUNET_NO);
-    memcpy (&cbuf[off], pending->msg, msize);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Transmitting message of type %u to %s\n",
+                ntohs (pending->msg->type),
+                GNUNET_i2s (&peer->id));
+    memcpy (&cbuf[off],
+            pending->msg,
+            msize);
     off += msize;
     peer->pending_count--;
     GNUNET_CONTAINER_DLL_remove (peer->head,
@@ -973,6 +1001,11 @@ core_transmit_notify (void *cls,
                                  pending);
     GNUNET_free (pending);
   }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%u bytes fit in %u bytes available, next message is %u bytes\n",
+              (unsigned int) off,
+              (unsigned int) size,
+              (NULL != peer->head) ? ntohs (peer->head->msg->size) : 0);
   if (NULL != (pending = peer->head))
   {
     /* technically redundant, but easier to read and
@@ -1006,7 +1039,10 @@ process_peer_queue (struct PeerInfo *peer)
   if (NULL == (pending = peer->head))
     return;
   if (NULL != peer->th)
-    return;
+  {
+    GNUNET_CORE_notify_transmit_ready_cancel (peer->th);
+    peer->th = NULL;
+  }
   GNUNET_STATISTICS_update (GDS_stats,
                             gettext_noop
                             ("# Bytes of bandwidth requested from core"),
@@ -1445,7 +1481,8 @@ GDS_NEIGHBOURS_handle_put (enum GNUNET_BLOCK_Type type,
   GNUNET_assert (NULL != bf);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Adding myself (%s) to PUT bloomfilter for %s\n",
-              GNUNET_i2s (&my_identity), GNUNET_h2s (key));
+              GNUNET_i2s (&my_identity),
+              GNUNET_h2s (key));
   GNUNET_CONTAINER_bloomfilter_add (bf, &my_identity_hash);
   GNUNET_STATISTICS_update (GDS_stats, gettext_noop ("# PUT requests routed"),
                             1, GNUNET_NO);
@@ -1492,8 +1529,10 @@ GDS_NEIGHBOURS_handle_put (enum GNUNET_BLOCK_Type type,
       continue;
     }
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Routing PUT for %s after %u hops to %s\n", GNUNET_h2s (key),
-                (unsigned int) hop_count, GNUNET_i2s (&target->id));
+                "Routing PUT for %s after %u hops to %s\n",
+                GNUNET_h2s (key),
+                (unsigned int) hop_count,
+                GNUNET_i2s (&target->id));
     pending = GNUNET_malloc (sizeof (struct P2PPendingMessage) + msize);
     pending->importance = 0;    /* FIXME */
     pending->timeout = expiration_time;
@@ -1521,10 +1560,15 @@ GDS_NEIGHBOURS_handle_put (enum GNUNET_BLOCK_Type type,
     pp = (struct GNUNET_PeerIdentity *) &ppm[1];
     memcpy (pp, put_path,
             sizeof (struct GNUNET_PeerIdentity) * put_path_length);
-    memcpy (&pp[put_path_length], data, data_size);
-    GNUNET_CONTAINER_DLL_insert_tail (target->head, target->tail, pending);
+    memcpy (&pp[put_path_length],
+            data,
+            data_size);
+    GNUNET_CONTAINER_DLL_insert_tail (target->head,
+                                      target->tail,
+                                      pending);
     target->pending_count++;
-    process_peer_queue (target);
+    if (pending == target->head)
+      process_peer_queue (target);
   }
   GNUNET_free (targets);
   return (skip_count < target_count) ? GNUNET_OK : GNUNET_NO;
@@ -1579,8 +1623,10 @@ GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
                         &targets);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Adding myself (%s) to GET bloomfilter for %s\n",
-              GNUNET_i2s (&my_identity), GNUNET_h2s (key));
-  GNUNET_CONTAINER_bloomfilter_add (peer_bf, &my_identity_hash);
+              GNUNET_i2s (&my_identity),
+              GNUNET_h2s (key));
+  GNUNET_CONTAINER_bloomfilter_add (peer_bf,
+                                    &my_identity_hash);
   if (0 == target_count)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1616,8 +1662,10 @@ GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
       continue;
     }
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Routing GET for %s after %u hops to %s\n", GNUNET_h2s (key),
-                (unsigned int) hop_count, GNUNET_i2s (&target->id));
+                "Routing GET for %s after %u hops to %s\n",
+                GNUNET_h2s (key),
+                (unsigned int) hop_count,
+                GNUNET_i2s (&target->id));
     pending = GNUNET_malloc (sizeof (struct P2PPendingMessage) + msize);
     pending->importance = 0;    /* FIXME */
     pending->timeout = GNUNET_TIME_relative_to_absolute (GET_TIMEOUT);
@@ -1650,9 +1698,12 @@ GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
                                                                 &xq
                                                                 [xquery_size],
                                                                 reply_bf_size));
-    GNUNET_CONTAINER_DLL_insert_tail (target->head, target->tail, pending);
+    GNUNET_CONTAINER_DLL_insert_tail (target->head,
+                                      target->tail,
+                                      pending);
     target->pending_count++;
-    process_peer_queue (target);
+    if (pending == target->head)
+      process_peer_queue (target);
   }
   GNUNET_free (targets);
   return (skip_count < target_count) ? GNUNET_OK : GNUNET_NO;
@@ -1712,16 +1763,28 @@ GDS_NEIGHBOURS_handle_reply (const struct GNUNET_PeerIdentity *target,
   if (NULL == pi)
   {
     /* peer disconnected in the meantime, drop reply */
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "No matching peer for reply for key %s\n",
+                GNUNET_h2s (key));
     return;
   }
   if (pi->pending_count >= MAXIMUM_PENDING_PER_PEER)
   {
     /* skip */
-    GNUNET_STATISTICS_update (GDS_stats, gettext_noop ("# P2P messages dropped due to full queue"),
-                              1, GNUNET_NO);
+    GNUNET_STATISTICS_update (GDS_stats,
+                              gettext_noop ("# P2P messages dropped due to full queue"),
+                              1,
+                              GNUNET_NO);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Peer queue full, ignoring reply for key %s\n",
+                GNUNET_h2s (key));
     return;
   }
 
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Forwarding reply for key %s to peer %s\n",
+              GNUNET_h2s (key),
+              GNUNET_i2s (target));
   GNUNET_STATISTICS_update (GDS_stats,
                             gettext_noop
                             ("# RESULT messages queued for transmission"), 1,
@@ -1739,12 +1802,18 @@ GDS_NEIGHBOURS_handle_reply (const struct GNUNET_PeerIdentity *target,
   prm->expiration_time = GNUNET_TIME_absolute_hton (expiration_time);
   prm->key = *key;
   paths = (struct GNUNET_PeerIdentity *) &prm[1];
-  memcpy (paths, put_path,
+  memcpy (paths,
+          put_path,
           put_path_length * sizeof (struct GNUNET_PeerIdentity));
-  memcpy (&paths[put_path_length], get_path,
+  memcpy (&paths[put_path_length],
+          get_path,
           get_path_length * sizeof (struct GNUNET_PeerIdentity));
-  memcpy (&paths[put_path_length + get_path_length], data, data_size);
-  GNUNET_CONTAINER_DLL_insert (pi->head, pi->tail, pending);
+  memcpy (&paths[put_path_length + get_path_length],
+          data,
+          data_size);
+  GNUNET_CONTAINER_DLL_insert (pi->head,
+                               pi->tail,
+                               pending);
   pi->pending_count++;
   process_peer_queue (pi);
 }
@@ -1760,10 +1829,14 @@ static void
 core_init (void *cls,
            const struct GNUNET_PeerIdentity *identity)
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "CORE called, I am %s\n",
+              GNUNET_i2s (identity));
   my_identity = *identity;
   GNUNET_CRYPTO_hash (identity,
                       sizeof (struct GNUNET_PeerIdentity),
                       &my_identity_hash);
+  GDS_CLIENTS_init ();
 }
 
 
@@ -1778,7 +1851,8 @@ core_init (void *cls,
  *         #GNUNET_SYSERR to close it (signal serious error)
  */
 static int
-handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
+handle_dht_p2p_put (void *cls,
+                    const struct GNUNET_PeerIdentity *peer,
                     const struct GNUNET_MessageHeader *message)
 {
   const struct PeerPutMessage *put;
@@ -1811,10 +1885,12 @@ handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
     return GNUNET_YES;
   }
   GNUNET_STATISTICS_update (GDS_stats,
-                            gettext_noop ("# P2P PUT requests received"), 1,
+                            gettext_noop ("# P2P PUT requests received"),
+                            1,
                             GNUNET_NO);
   GNUNET_STATISTICS_update (GDS_stats,
-                            gettext_noop ("# P2P PUT bytes received"), msize,
+                            gettext_noop ("# P2P PUT bytes received"),
+                            msize,
                             GNUNET_NO);
   put_path = (const struct GNUNET_PeerIdentity *) &put[1];
   payload = &put_path[putlen];
@@ -1833,10 +1909,14 @@ handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
     tmp = GNUNET_strdup (GNUNET_i2s (&my_identity));
     LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
                  "R5N PUT %s: %s->%s (%u, %u=>%u)\n",
-                 GNUNET_h2s (&put->key), GNUNET_i2s (peer), tmp,
+                 GNUNET_h2s (&put->key),
+                 GNUNET_i2s (peer),
+                 tmp,
                  ntohl(put->hop_count),
-                 GNUNET_CRYPTO_hash_matching_bits (&phash, &put->key),
-                 GNUNET_CRYPTO_hash_matching_bits (&my_identity_hash, &put->key)
+                 GNUNET_CRYPTO_hash_matching_bits (&phash,
+                                                   &put->key),
+                 GNUNET_CRYPTO_hash_matching_bits (&my_identity_hash,
+                                                   &put->key)
                 );
     GNUNET_free (tmp);
   }
@@ -1848,6 +1928,7 @@ handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
     if (0 != memcmp (&test_key, &put->key, sizeof (struct GNUNET_HashCode)))
     {
       char *put_s = GNUNET_strdup (GNUNET_h2s_full (&put->key));
+
       GNUNET_break_op (0);
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                   "PUT with key `%s' for block with key %s\n",
@@ -1889,7 +1970,8 @@ handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer,
     }
   }
 
-  bf = GNUNET_CONTAINER_bloomfilter_init (put->bloomfilter, DHT_BLOOM_SIZE,
+  bf = GNUNET_CONTAINER_bloomfilter_init (put->bloomfilter,
+                                          DHT_BLOOM_SIZE,
                                           GNUNET_CONSTANTS_BLOOMFILTER_K);
   GNUNET_break_op (GNUNET_YES ==
                    GNUNET_CONTAINER_bloomfilter_test (bf, &phash));
@@ -2025,8 +2107,12 @@ handle_find_peer (const struct GNUNET_PeerIdentity *sender,
       return;                   /* no non-masked peer available */
     if (peer == NULL)
       peer = bucket->head;
-    GNUNET_CRYPTO_hash (&peer->id, sizeof (struct GNUNET_PeerIdentity), &phash);
-    GNUNET_BLOCK_mingle_hash (&phash, bf_mutator, &mhash);
+    GNUNET_CRYPTO_hash (&peer->id,
+                        sizeof (struct GNUNET_PeerIdentity),
+                        &phash);
+    GNUNET_BLOCK_mingle_hash (&phash,
+                              bf_mutator,
+                              &mhash);
     hello = GDS_HELLO_get (&peer->id);
   }
   while ((hello == NULL) ||
@@ -2067,7 +2153,8 @@ handle_dht_p2p_get (void *cls,
   int forwarded;
 
   GNUNET_break (0 !=
-                memcmp (peer, &my_identity,
+                memcmp (peer,
+                        &my_identity,
                         sizeof (struct GNUNET_PeerIdentity)));
   /* parse and validate message */
   msize = ntohs (message->size);
diff --git a/src/dht/test_dht_2dtorus.conf b/src/dht/test_dht_2dtorus.conf
index 2a7f75b9f..844f20431 100644
--- a/src/dht/test_dht_2dtorus.conf
+++ b/src/dht/test_dht_2dtorus.conf
@@ -1,6 +1,11 @@
+@INLINE@ ../../contrib/no_forcestart.conf
+@INLINE@ ../../contrib/no_autostart_above_core.conf
 [PATHS]
 GNUNET_TEST_HOME = /tmp/test_dht_2dtorus/
 
+[dht]
+AUTOSTART = YES
+
 [dhtcache]
 QUOTA = 1 MB
 DATABASE = heap
@@ -22,7 +27,10 @@ WAN_QUOTA_OUT = 1 GB
 OVERLAY_TOPOLOGY = 2D_TORUS
 
 [nse]
+AUTOSTART = YES
 WORKDELAY = 500 ms
 INTERVAL = 60 s
 WORKBITS = 0
 
+[transport]
+PLUGINS = unix
diff --git a/src/dht/test_dht_line.conf b/src/dht/test_dht_line.conf
index 024ff2770..45badfae0 100644
--- a/src/dht/test_dht_line.conf
+++ b/src/dht/test_dht_line.conf
@@ -1,5 +1,11 @@
+@INLINE@ ../../contrib/no_forcestart.conf
+@INLINE@ ../../contrib/no_autostart_above_core.conf
 [PATHS]
 GNUNET_TEST_HOME = /tmp/test_dht_line/
+
+[dht]
+AUTOSTART = YES
+
 [dhtcache]
 QUOTA = 1 MB
 DATABASE = heap
@@ -22,10 +28,10 @@ WAN_QUOTA_OUT = 1 GB
 OVERLAY_TOPOLOGY = LINE
 
 [transport]
-plugins = tcp
+plugins = unix
 
 [nse]
+AUTOSTART = YES
 WORKDELAY = 500 ms
 INTERVAL = 60 s
 WORKBITS = 0
-
diff --git a/src/dht/test_dht_multipeer.conf b/src/dht/test_dht_multipeer.conf
index b6e1646d3..f113aa0f8 100644
--- a/src/dht/test_dht_multipeer.conf
+++ b/src/dht/test_dht_multipeer.conf
@@ -1,9 +1,15 @@
+@INLINE@ ../../contrib/no_forcestart.conf
+@INLINE@ ../../contrib/no_autostart_above_core.conf
+
+[dht]
+AUTOSTART = YES
+
 [dhtcache]
 QUOTA = 1 MB
 DATABASE = heap
 
 [transport]
-PLUGINS = tcp
+PLUGINS = unix
 
 [ats]
 WAN_QUOTA_IN = 1 GB
@@ -26,3 +32,8 @@ INTERNAL_ADDRESS = 127.0.0.1
 EXTERNAL_ADDRESS = 127.0.0.1
 USE_LOCALADDR = YES
 
+[nse]
+AUTOSTART = YES
+WORKDELAY = 500 ms
+INTERVAL = 60 s
+WORKBITS = 0
diff --git a/src/dht/test_dht_topo.c b/src/dht/test_dht_topo.c
index 205f6b99a..8be3064f7 100644
--- a/src/dht/test_dht_topo.c
+++ b/src/dht/test_dht_topo.c
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     Copyright (C) 2012 GNUnet e.V.
+     Copyright (C) 2012, 2016 GNUnet e.V.
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -71,12 +71,17 @@ static int ok = 1;
 /**
  * Task to do DHT_puts
  */
-static struct GNUNET_SCHEDULER_Task * put_task;
+static struct GNUNET_SCHEDULER_Task *put_task;
+
+/**
+ * Task to do DHT_gets
+ */
+static struct GNUNET_SCHEDULER_Task *get_task;
 
 /**
  * Task to time out / regular shutdown.
  */
-static struct GNUNET_SCHEDULER_Task * timeout_task;
+static struct GNUNET_SCHEDULER_Task *timeout_task;
 
 /**
  * Head of list of active GET operations.
@@ -147,6 +152,39 @@ static struct
 };
 
 
+static struct GNUNET_DHT_TEST_Context *
+stop_ops ()
+{
+  struct GetOperation *get_op;
+  struct GNUNET_DHT_TEST_Context *ctx = NULL;
+
+  if (NULL != timeout_task)
+  {
+    ctx = GNUNET_SCHEDULER_cancel (timeout_task);
+    timeout_task = NULL;
+  }
+  if (NULL != put_task)
+  {
+    GNUNET_SCHEDULER_cancel (put_task);
+    put_task = NULL;
+  }
+  if (NULL != get_task)
+  {
+    GNUNET_SCHEDULER_cancel (get_task);
+    get_task = NULL;
+  }
+  while (NULL != (get_op = get_tail))
+  {
+    GNUNET_DHT_get_stop (get_op->get);
+    GNUNET_CONTAINER_DLL_remove (get_head,
+                                 get_tail,
+                                 get_op);
+    GNUNET_free (get_op);
+  }
+  return ctx;
+}
+
+
 /**
  * Function called once we're done processing stats.
  *
@@ -163,10 +201,11 @@ stats_finished (void *cls,
   unsigned int i;
 
   if (NULL != op)
-    GNUNET_TESTBED_operation_done (op); // needed?
+    GNUNET_TESTBED_operation_done (op);
   if (NULL != emsg)
   {
-    fprintf (stderr, _("Gathering statistics failed: %s\n"),
+    fprintf (stderr,
+             _("Gathering statistics failed: %s\n"),
              emsg);
     GNUNET_SCHEDULER_cancel (put_task);
     GNUNET_DHT_TEST_cleanup (ctx);
@@ -178,8 +217,8 @@ stats_finished (void *cls,
              stats[i].subsystem,
              stats[i].name,
              stats[i].total);
-  GNUNET_SCHEDULER_cancel (put_task);
   GNUNET_DHT_TEST_cleanup (ctx);
+  GNUNET_SCHEDULER_shutdown ();
 }
 
 
@@ -191,8 +230,8 @@ stats_finished (void *cls,
  * @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
+ * @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
 handle_stats (void *cls,
@@ -215,32 +254,31 @@ handle_stats (void *cls,
 
 
 /**
- * Task run on success or timeout to clean up.
- * Terminates active get operations and shuts down
- * the testbed.
+ * Task run on shutdown to clean up.  Terminates active get operations
+ * and shuts down the testbed.
  *
  * @param cls the 'struct GNUNET_DHT_TestContext'
  */
 static void
 shutdown_task (void *cls)
 {
-  struct GNUNET_DHT_TEST_Context *ctx = cls;
-  struct GetOperation *get_op;
+  (void) stop_ops ();
+}
 
-  while (NULL != (get_op = get_tail))
-  {
-    GNUNET_DHT_get_stop (get_op->get);
-    GNUNET_CONTAINER_DLL_remove (get_head,
-                                 get_tail,
-                                 get_op);
-    GNUNET_free (get_op);
-  }
-  (void) GNUNET_TESTBED_get_statistics (NUM_PEERS,
-                                        my_peers,
-                                        NULL, NULL,
-                                        &handle_stats,
-                                        &stats_finished,
-                                        ctx);
+
+/**
+ * Task run on timeout to clean up.  Terminates active get operations
+ * and shuts down the testbed.
+ *
+ * @param cls the `struct GNUNET_DHT_TestContext`
+ */
+static void
+timeout_cb (void *cls)
+{
+  timeout_task = NULL;
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+              "Timeout\n");
+  GNUNET_SCHEDULER_shutdown ();
 }
 
 
@@ -252,33 +290,40 @@ shutdown_task (void *cls)
  * @param exp when will this value expire
  * @param key key of the result
  * @param get_path peers on reply path (or NULL if not recorded)
- * @param get_path_length number of entries in get_path
+ * @param get_path_length number of entries in @a get_path
  * @param put_path peers on the PUT path (or NULL if not recorded)
- * @param put_path_length number of entries in get_path
+ * @param put_path_length number of entries in @a put_path
  * @param type type of the result
- * @param size number of bytes in data
+ * @param size number of bytes in @a data
  * @param data pointer to the result data
  */
 static void
-dht_get_handler (void *cls, struct GNUNET_TIME_Absolute exp,
-                 const struct GNUNET_HashCode * key,
+dht_get_handler (void *cls,
+                 struct GNUNET_TIME_Absolute exp,
+                 const struct GNUNET_HashCode *key,
                  const struct GNUNET_PeerIdentity *get_path,
                  unsigned int get_path_length,
                  const struct GNUNET_PeerIdentity *put_path,
-                 unsigned int put_path_length, enum GNUNET_BLOCK_Type type,
-                 size_t size, const void *data)
+                 unsigned int put_path_length,
+                 enum GNUNET_BLOCK_Type type,
+                 size_t size,
+                 const void *data)
 {
   struct GetOperation *get_op = cls;
   struct GNUNET_HashCode want;
-  struct GNUNET_DHT_TestContext *ctx;
+  struct GNUNET_DHT_TEST_Context *ctx;
 
   if (sizeof (struct GNUNET_HashCode) != size)
   {
     GNUNET_break (0);
     return;
   }
-  GNUNET_CRYPTO_hash (key, sizeof (*key), &want);
-  if (0 != memcmp (&want, data, sizeof (want)))
+  GNUNET_CRYPTO_hash (key,
+                      sizeof (*key),
+                      &want);
+  if (0 != memcmp (&want,
+                   data,
+                   sizeof (want)))
   {
     GNUNET_break (0);
     return;
@@ -289,14 +334,19 @@ dht_get_handler (void *cls, struct GNUNET_TIME_Absolute exp,
   {
     int i;
 
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "PATH: (get %u, put %u)\n",
-                get_path_length, put_path_length);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  LOCAL\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "PATH: (get %u, put %u)\n",
+                get_path_length,
+                put_path_length);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "  LOCAL\n");
     for (i = get_path_length - 1; i >= 0; i--)
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  %s\n",
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "  %s\n",
                   GNUNET_i2s (&get_path[i]));
     for (i = put_path_length - 1; i >= 0; i--)
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  %s\n",
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "  %s\n",
                   GNUNET_i2s (&put_path[i]));
   }
 #endif
@@ -307,10 +357,18 @@ dht_get_handler (void *cls, struct GNUNET_TIME_Absolute exp,
   GNUNET_free (get_op);
   if (NULL != get_head)
     return;
-  /* all DHT GET operations successful; terminate! */
+  /* all DHT GET operations successful; get stats! */
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "All DHT operations successful. Obtaining stats!\n");
   ok = 0;
-  ctx = GNUNET_SCHEDULER_cancel (timeout_task);
-  timeout_task = GNUNET_SCHEDULER_add_now (&shutdown_task, ctx);
+  ctx = stop_ops ();
+  GNUNET_assert (NULL != ctx);
+  (void) GNUNET_TESTBED_get_statistics (NUM_PEERS,
+                                        my_peers,
+                                        NULL, NULL,
+                                        &handle_stats,
+                                        &stats_finished,
+                                        ctx);
 }
 
 
@@ -328,55 +386,52 @@ do_puts (void *cls)
   struct GNUNET_HashCode value;
   unsigned int i;
 
+  put_task = NULL;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Putting values into DHT\n");
   for (i = 0; i < NUM_PEERS; i++)
   {
-    GNUNET_CRYPTO_hash (&i, sizeof (i), &key);
-    GNUNET_CRYPTO_hash (&key, sizeof (key), &value);
-    GNUNET_DHT_put (hs[i], &key, 10U,
+    GNUNET_CRYPTO_hash (&i,
+                        sizeof (i),
+                        &key);
+    GNUNET_CRYPTO_hash (&key,
+                        sizeof (key),
+                        &value);
+    GNUNET_DHT_put (hs[i],
+                    &key,
+                    10U,
                     GNUNET_DHT_RO_RECORD_ROUTE |
                     GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                     GNUNET_BLOCK_TYPE_TEST,
-                    sizeof (value), &value,
+                    sizeof (value),
+                    &value,
                     GNUNET_TIME_UNIT_FOREVER_ABS,
-                    NULL, NULL);
+                    NULL,
+                    NULL);
   }
   put_task = GNUNET_SCHEDULER_add_delayed (PUT_FREQUENCY,
-                                           &do_puts, hs);
+                                           &do_puts,
+                                           hs);
 }
 
 
 /**
- * Main function of the test.
- *
- * @param cls closure (NULL)
- * @param ctx argument to give to GNUNET_DHT_TEST_cleanup on test end
- * @param num_peers number of peers that are running
- * @param peers array of peers
- * @param dhts handle to each of the DHTs of the peers
+ * Start GET operations.
  */
 static void
-run (void *cls,
-     struct GNUNET_DHT_TEST_Context *ctx,
-     unsigned int num_peers,
-     struct GNUNET_TESTBED_Peer **peers,
-     struct GNUNET_DHT_Handle **dhts)
+start_get (void *cls)
 {
+  struct GNUNET_DHT_Handle **dhts = cls;
   unsigned int i;
   unsigned int j;
   struct GNUNET_HashCode key;
   struct GetOperation *get_op;
 
-  GNUNET_assert (NUM_PEERS == num_peers);
-  my_peers = peers;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Peers setup, starting test\n");
-  put_task = GNUNET_SCHEDULER_add_now (&do_puts, dhts);
-  for (i=0;i<num_peers;i++)
+  get_task = NULL;
+  for (i=0;i<NUM_PEERS;i++)
   {
     GNUNET_CRYPTO_hash (&i, sizeof (i), &key);
-    for (j=0;j<num_peers;j++)
+    for (j=0;j<NUM_PEERS;j++)
     {
       get_op = GNUNET_new (struct GetOperation);
       GNUNET_CONTAINER_DLL_insert (get_head,
@@ -389,11 +444,43 @@ run (void *cls,
                                           GNUNET_DHT_RO_RECORD_ROUTE | GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                           NULL,        /* xquery */
                                           0,      /* xquery bits */
-                                          &dht_get_handler, get_op);
+                                          &dht_get_handler,
+                                          get_op);
     }
   }
+}
+
+
+/**
+ * Main function of the test.
+ *
+ * @param cls closure (NULL)
+ * @param ctx argument to give to #GNUNET_DHT_TEST_cleanup on test end
+ * @param num_peers number of @a peers that are running
+ * @param peers array of peers
+ * @param dhts handle to each of the DHTs of the peers
+ */
+static void
+run (void *cls,
+     struct GNUNET_DHT_TEST_Context *ctx,
+     unsigned int num_peers,
+     struct GNUNET_TESTBED_Peer **peers,
+     struct GNUNET_DHT_Handle **dhts)
+{
+  GNUNET_assert (NUM_PEERS == num_peers);
+  my_peers = peers;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Peers setup, starting test\n");
+  put_task = GNUNET_SCHEDULER_add_now (&do_puts,
+                                       dhts);
+  get_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
+                                           &start_get,
+                                           dhts);
   timeout_task = GNUNET_SCHEDULER_add_delayed (GET_TIMEOUT,
-                                               &shutdown_task, ctx);
+                                               &timeout_cb,
+                                               ctx);
+  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
+                                 ctx);
 }