indentation fixes
[oweals/gnunet.git] / src / dht / gnunet-service-wdht_neighbours.c
index b74f7be39ec5436599a0af40ed6d5b509103a533..78a04d66d0564f95eef09b6aa28afa5c6fbb8415 100644 (file)
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     Copyright (C) 2009-2015 Christian Grothoff (and other contributing authors)
+     Copyright (C) 2009-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
 
      You should have received a copy of the GNU General Public License
      along with GNUnet; see the file COPYING.  If not, write to the
-     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
-     Boston, MA 02111-1307, USA.
+     Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+     Boston, MA 02110-1301, USA.
 */
 /**
  * @file dht/gnunet-service-wdht_neighbours.c
  * @brief GNUnet DHT service's finger and friend table management code
  * @author Supriti Singh
+ * @author Christian Grothoff
+ * @author Arthur Dewarumez
+ *
+ * TODO:
+ * - initiate finding of successors
  */
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_transport_service.h"
 #include "gnunet_dht_service.h"
 #include "gnunet_statistics_service.h"
-#include "gnunet-service-wdht.h"
-#include "gnunet-service-wdht_clients.h"
-#include "gnunet-service-wdht_datacache.h"
-#include "gnunet-service-wdht_neighbours.h"
-#include "gnunet-service-wdht_nse.h"
-#include <fenv.h>
-#include <stdlib.h>
-#include <string.h>
+#include "gnunet-service-dht.h"
+#include "gnunet-service-dht_datacache.h"
+#include "gnunet-service-dht_neighbours.h"
+#include "gnunet-service-dht_nse.h"
 #include "dht.h"
 
 #define DEBUG(...)                                           \
  */
 struct FriendInfo;
 
+/**
+ *
+ */
+struct FingerTable;
 
 /**
  * Information we keep per trail.
@@ -83,6 +88,21 @@ struct FriendInfo;
 struct Trail
 {
 
+  /**
+   * Identifier of the trail with the predecessor.
+   */
+  struct GNUNET_HashCode pred_id;
+
+  /**
+   * Identifier of the trail with the successor.
+   */
+  struct GNUNET_HashCode succ_id;
+
+  /**
+   * When does this trail expire.
+   */
+  struct GNUNET_TIME_Absolute expiration_time;
+
   /**
    * MDLL entry in the list of all trails with the same predecessor.
    */
@@ -113,21 +133,6 @@ struct Trail
    */
   struct FriendInfo *succ;
 
-  /**
-   * Identifier of the trail with the predecessor.
-   */
-  struct GNUNET_HashCode pred_id;
-
-  /**
-   * Identifier of the trail with the successor.
-   */
-  struct GNUNET_HashCode succ_id;
-
-  /**
-   * When does this trail expire.
-   */
-  struct GNUNET_TIME_Absolute expiration_time;
-
   /**
    * Location of this trail in the heap.
    */
@@ -135,9 +140,17 @@ struct Trail
 
   /**
    * If this peer started the to create a Finger (and thus @e pred is
-   * NULL), this is the Finger we are trying to intialize.
+   * NULL), this is the finger table of the finger we are trying to
+   * intialize.
+   */
+  struct FingerTable *ft;
+
+  /**
+   * If this peer started the trail to create a Finger (and thus @e
+   * pred is NULL), this is the offset of the finger we are trying to
+   * intialize in the unsorted array.
    */
-  struct Finger **finger;
+  unsigned int finger_off;
 
 };
 
@@ -150,7 +163,7 @@ struct FriendInfo
   /**
    * Friend Identity
    */
-  struct GNUNET_PeerIdentity id;
+  const struct GNUNET_PeerIdentity *id;
 
   /**
    *
@@ -180,12 +193,6 @@ struct FriendInfo
 };
 
 
-/**
- *
- */
-struct FingerTable;
-
-
 /**
  *
  */
@@ -220,18 +227,13 @@ struct FingerTable
    */
   struct Finger **fingers;
 
-  /**
-   * Array of sorted fingers (sorted by destination, valid fingers first).
-   */
-  struct Finger **sorted_fingers;
-
   /**
    * Size of the finger array.
    */
   unsigned int finger_array_size;
 
   /**
-   * Number of valid entries in @e sorted_fingers (contiguous from offset 0)
+   * Number of valid entries in @e fingers
    */
   unsigned int number_valid_fingers;
 
@@ -341,7 +343,7 @@ struct TrailDestroyMessage
 struct FindSuccessorMessage
 {
   /**
-   * Type: #GNUNET_MESSAGE_TYPE_WDHT_FIND_SUCCESSOR
+   * Type: #GNUNET_MESSAGE_TYPE_WDHT_SUCCESSOR_FIND
    */
   struct GNUNET_MessageHeader header;
 
@@ -350,12 +352,6 @@ struct FindSuccessorMessage
    */
   uint32_t reserved GNUNET_PACKED;
 
-  /**
-   * Unique (random) identifier this peer will use to
-   * identify the finger (in future messages).
-   */
-  struct GNUNET_HashCode trail_id;
-
   /**
    * Key for which we would like close values returned.
    * identify the finger (in future messages).
@@ -510,7 +506,7 @@ struct PeerGetResultMessage
   struct GNUNET_MessageHeader header;
 
   /**
-   * The type for the data.
+   * The type for the data in NBO.
    */
   uint32_t type GNUNET_PACKED;
 
@@ -520,20 +516,10 @@ struct PeerGetResultMessage
    */
   uint32_t put_path_length GNUNET_PACKED;
 
-  /**
-   * Length of the GET path that follows (if tracked).
-   */
-  uint32_t get_path_length GNUNET_PACKED;
-
-  /**
-   * Peer which queried for get and should get the result.
-   */
-  struct GNUNET_PeerIdentity querying_peer;
-
   /**
    * When does the content expire?
    */
-  struct GNUNET_TIME_Absolute expiration_time;
+  struct GNUNET_TIME_AbsoluteNBO expiration_time;
 
   /**
    * The key of the corresponding GET request.
@@ -542,8 +528,6 @@ struct PeerGetResultMessage
 
   /* put path (if tracked) */
 
-  /* get path (if tracked) */
-
   /* Payload */
 
 };
@@ -600,59 +584,85 @@ static struct GNUNET_CORE_Handle *core_api;
 /**
  * Handle the put request from the client.
  *
- * @param key Key for the content
  * @param block_type Type of the block
- * @param options Routing options
- * @param desired_replication_level Desired replication count
- * @param expiration_time When does the content expire
- * @param data Content to store
- * @param data_size Size of content @a data in bytes
+ * @param options routing options
+ * @param desired_replication_level desired replication level
+ * @param expiration_time when does the content expire
+ * @param hop_count how many hops has this message traversed so far
+ * @param bf Bloom filter of peers this PUT has already traversed
+ * @param key key for the content
+ * @param put_path_length number of entries in put_path
+ * @param put_path peers this request has traversed so far (if tracked)
+ * @param data payload to store
+ * @param data_size number of bytes in data
+ * @return #GNUNET_OK if the request was forwarded, #GNUNET_NO if not
  */
-void
-GDS_NEIGHBOURS_handle_put (const struct GNUNET_HashCode *key,
-                           enum GNUNET_BLOCK_Type block_type,
+int
+GDS_NEIGHBOURS_handle_put (enum GNUNET_BLOCK_Type block_type,
                            enum GNUNET_DHT_RouteOption options,
                            uint32_t desired_replication_level,
                            struct GNUNET_TIME_Absolute expiration_time,
+                           uint32_t hop_count,
+                           struct GNUNET_CONTAINER_BloomFilter *bf,
+                           const struct GNUNET_HashCode *key,
+                           unsigned int put_path_length,
+                           struct GNUNET_PeerIdentity *put_path,
                            const void *data,
                            size_t data_size)
 {
   GDS_DATACACHE_handle_put (expiration_time,
                             key,
                             0, NULL,
-                            0, NULL,
                             block_type,
                             data_size,
                             data);
   GDS_CLIENTS_process_put (options,
                            block_type,
-                           0, 0,
-                           0, NULL,
+                           hop_count,
+                           desired_replication_level,
+                           put_path_length, put_path,
                            expiration_time,
                            key,
                            data,
                            data_size);
+  return GNUNET_OK; /* FIXME... */
 }
 
 
 /**
- * Handle the get request from the client file. If I am destination do
- * datacache put and return. Else find the target friend and forward message
- * to it.
+ * Perform a GET operation.  Forwards the given request to other
+ * peers.  Does not lookup the key locally.  May do nothing if this is
+ * the only peer in the network (or if we are the closest peer in the
+ * network).
  *
- * @param key Key for the content
- * @param block_type Type of the block
- * @param options Routing options
- * @param desired_replication_level Desired replication count
+ * @param type type of the block
+ * @param options routing options
+ * @param desired_replication_level desired replication count
+ * @param hop_count how many hops did this request traverse so far?
+ * @param key key for the content
+ * @param xquery extended query
+ * @param xquery_size number of bytes in @a xquery
+ * @param reply_bf bloomfilter to filter duplicates
+ * @param reply_bf_mutator mutator for @a reply_bf
+ * @param peer_bf filter for peers not to select (again, updated)
+ * @return #GNUNET_OK if the request was forwarded, #GNUNET_NO if not
  */
-void
-GDS_NEIGHBOURS_handle_get (const struct GNUNET_HashCode *key,
-                           enum GNUNET_BLOCK_Type block_type,
+int
+GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
                            enum GNUNET_DHT_RouteOption options,
-                           uint32_t desired_replication_level)
+                           uint32_t desired_replication_level,
+                           uint32_t hop_count,
+                           const struct GNUNET_HashCode *key,
+                           const void *xquery, size_t xquery_size,
+                           const struct GNUNET_CONTAINER_BloomFilter *reply_bf,
+                           uint32_t reply_bf_mutator,
+                           struct GNUNET_CONTAINER_BloomFilter *peer_bf)
 {
   // find closest finger(s) on all layers
   // use TrailRoute with PeerGetMessage embedded to contact peer
+  // NOTE: actually more complicated, see paper!
+  GNUNET_break (0); // not implemented!
+  return GNUNET_SYSERR;
 }
 
 
@@ -707,20 +717,92 @@ delete_trail (struct Trail *trail,
   }
   GNUNET_break (trail ==
                 GNUNET_CONTAINER_heap_remove_node (trail->hn));
-  finger = *trail->finger;
+  finger = trail->ft->fingers[trail->finger_off];
   if (NULL != finger)
   {
-    *trail->finger = NULL;
+    trail->ft->fingers[trail->finger_off] = NULL;
+    trail->ft->number_valid_fingers--;
     GNUNET_free (finger);
   }
   GNUNET_free (trail);
 }
 
 
+/**
+ * Forward the given payload message along the trail.
+ *
+ * @param next_target which direction along the trail should we forward
+ * @param trail_id which trail should we forward along
+ * @param have_path do we track the forwarding path?
+ * @param predecessor which peer do we tack on to the path?
+ * @param path path the message has taken so far along the trail
+ * @param path_length number of entries in @a path
+ * @param payload payload of the message
+ */
+static void
+forward_message_on_trail (struct FriendInfo *next_target,
+                          const struct GNUNET_HashCode *trail_id,
+                          int have_path,
+                          const struct GNUNET_PeerIdentity *predecessor,
+                          const struct GNUNET_PeerIdentity *path,
+                          uint16_t path_length,
+                          const struct GNUNET_MessageHeader *payload)
+{
+  struct GNUNET_MQ_Envelope *env;
+  struct TrailRouteMessage *trm;
+  struct GNUNET_PeerIdentity *new_path;
+  unsigned int plen;
+  uint16_t payload_len;
+
+  payload_len = ntohs (payload->size);
+  if (have_path)
+  {
+    plen = path_length + 1;
+    if (plen >= (GNUNET_SERVER_MAX_MESSAGE_SIZE
+                 - payload_len
+                 - sizeof (struct TrailRouteMessage))
+        / sizeof (struct GNUNET_PeerIdentity))
+    {
+      /* Should really not have paths this long... */
+      GNUNET_break_op (0);
+      plen = 0;
+      have_path = 0;
+    }
+  }
+  else
+  {
+    GNUNET_break_op (0 == path_length);
+    path_length = 0;
+    plen = 0;
+  }
+  env = GNUNET_MQ_msg_extra (trm,
+                             payload_len +
+                             plen * sizeof (struct GNUNET_PeerIdentity),
+                             GNUNET_MESSAGE_TYPE_WDHT_TRAIL_ROUTE);
+  trm->record_path = htons (have_path);
+  trm->path_length = htons (plen);
+  trm->trail_id = *trail_id;
+  new_path = (struct GNUNET_PeerIdentity *) &trm[1];
+  if (have_path)
+  {
+    GNUNET_memcpy (new_path,
+            path,
+            path_length * sizeof (struct GNUNET_PeerIdentity));
+    new_path[path_length] = *predecessor;
+  }
+  GNUNET_memcpy (&new_path[plen],
+          payload,
+          payload_len);
+  GNUNET_MQ_send (next_target->mq,
+                  env);
+}
+
+
 /**
  * Send the get result to requesting client.
  *
- * @param trail_id trail identifying where to send the result to, NULL for us
+ * @param cls trail identifying where to send the result to, NULL for us
+ * @param options routing options (from GET request)
  * @param key Key of the requested data.
  * @param type Block type
  * @param put_path_length Number of peers in @a put_path
@@ -730,7 +812,8 @@ delete_trail (struct Trail *trail,
  * @param data_size Size of the @a data
  */
 void
-GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *trail_id,
+GDS_NEIGHBOURS_send_get_result (void *cls,
+                                enum GNUNET_DHT_RouteOption options,
                                 const struct GNUNET_HashCode *key,
                                 enum GNUNET_BLOCK_Type type,
                                 unsigned int put_path_length,
@@ -739,8 +822,41 @@ GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *trail_id,
                                 const void *data,
                                 size_t data_size)
 {
-  // TRICKY: need to introduce some context to remember trail from
-  // the lookup...
+  const struct GNUNET_HashCode *trail_id = cls;
+  struct GNUNET_MessageHeader *payload;
+  struct Trail *trail;
+
+  trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+                                             trail_id);
+  if (NULL == trail)
+  {
+    /* TODO: inform statistics */
+    return;
+  }
+  if (NULL == trail->pred)
+  {
+    /* result is for *us* (local client) */
+    GDS_CLIENTS_handle_reply (expiration,
+                              key,
+                              0, NULL,
+                              put_path_length, put_path,
+                              type,
+                              data_size,
+                              data);
+    return;
+  }
+
+  payload = GNUNET_malloc(sizeof(struct GNUNET_MessageHeader) + data_size);
+  payload->size = data_size;
+  payload->type = GNUNET_MESSAGE_TYPE_WDHT_GET_RESULT;
+
+  forward_message_on_trail (trail->pred,
+                            trail_id,
+                            0 != (options & GNUNET_DHT_RO_RECORD_ROUTE),
+                            &my_identity,
+                            NULL, 0,
+                            payload);
+  GNUNET_free (payload);
 }
 
 
@@ -749,28 +865,19 @@ GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *trail_id,
  *
  * @param cls closure
  * @param peer peer identity this notification is about
+ * @param internal_cls our `struct FriendInfo` for @a peer
  */
 static void
 handle_core_disconnect (void *cls,
-                        const struct GNUNET_PeerIdentity *peer)
+                        const struct GNUNET_PeerIdentity *peer,
+                       void *internal_cls)
 {
-  struct FriendInfo *remove_friend;
+  struct FriendInfo *remove_friend = internal_cls;
   struct Trail *t;
 
   /* If disconnected to own identity, then return. */
-  if (0 == memcmp (&my_identity,
-                   peer,
-                   sizeof (struct GNUNET_PeerIdentity)))
+  if (NULL == remove_friend)
     return;
-
-  if (NULL == (remove_friend =
-               GNUNET_CONTAINER_multipeermap_get (friends_peermap,
-                                                  peer)))
-  {
-    GNUNET_break (0);
-    return;
-  }
-
   GNUNET_assert (GNUNET_YES ==
                  GNUNET_CONTAINER_multipeermap_remove (friends_peermap,
                                                        peer,
@@ -783,10 +890,8 @@ handle_core_disconnect (void *cls,
     delete_trail (t,
                   GNUNET_NO,
                   GNUNET_YES);
-  GNUNET_MQ_destroy (remove_friend->mq);
   GNUNET_free (remove_friend);
-  if (0 ==
-      GNUNET_CONTAINER_multipeermap_size (friends_peermap))
+  if (0 == GNUNET_CONTAINER_multipeermap_size (friends_peermap))
   {
     GNUNET_SCHEDULER_cancel (random_walk_task);
     random_walk_task = NULL;
@@ -794,15 +899,86 @@ handle_core_disconnect (void *cls,
 }
 
 
+/**
+ * Function called with a random friend to be returned.
+ *
+ * @param cls a `struct FriendInfo **` with where to store the result
+ * @param peer the peer identity of the friend (ignored)
+ * @param value the `struct FriendInfo *` that was selected at random
+ * @return #GNUNET_OK (all good)
+ */
+static int
+pick_random_helper (void *cls,
+                    const struct GNUNET_PeerIdentity *peer,
+                    void *value)
+{
+  struct FriendInfo **fi = cls;
+  struct FriendInfo *v = value;
+
+  *fi = v;
+  return GNUNET_OK;
+}
+
+
 /**
  * Pick random friend from friends for random walk.
+ *
+ * @return NULL if we have no friends
  */
 static struct FriendInfo *
 pick_random_friend ()
 {
-  // TODO: need to extend peermap API to return random entry...
-  // (Note: same extension exists for hashmap API).
-  return NULL; // FIXME...
+  struct FriendInfo *ret;
+
+  ret = NULL;
+  if (0 ==
+      GNUNET_CONTAINER_multipeermap_get_random (friends_peermap,
+                                                &pick_random_helper,
+                                                &ret))
+    return NULL;
+  return ret;
+}
+
+
+/**
+ * One of our trails might have timed out, check and
+ * possibly initiate cleanup.
+ *
+ * @param cls NULL
+ */
+static void
+trail_timeout_callback (void *cls)
+{
+  struct Trail *trail;
+  struct GNUNET_TIME_Relative left;
+
+  trail_timeout_task = NULL;
+  while (NULL != (trail = GNUNET_CONTAINER_heap_peek (trail_heap)))
+  {
+    left = GNUNET_TIME_absolute_get_remaining (trail->expiration_time);
+    if (0 != left.rel_value_us)
+      break;
+    delete_trail (trail,
+                  GNUNET_YES,
+                  GNUNET_YES);
+  }
+  if (NULL != trail)
+    trail_timeout_task = GNUNET_SCHEDULER_add_delayed (left,
+                                                       &trail_timeout_callback,
+                                                       NULL);
+}
+
+
+/**
+ * Compute how big our finger arrays should be (at least).
+ *
+ * @return size of the finger array, never 0
+ */
+static unsigned int
+get_desired_finger_array_size ()
+{
+  /* FIXME: This is just a stub... */
+  return 64;
 }
 
 
@@ -810,11 +986,9 @@ pick_random_friend ()
  * Initiate a random walk.
  *
  * @param cls NULL
- * @param tc unused
  */
 static void
-do_random_walk (void *cls,
-                const struct GNUNET_SCHEDULER_TaskContext *tc)
+do_random_walk (void *cls)
 {
   static unsigned int walk_layer;
   struct FriendInfo *friend;
@@ -823,6 +997,7 @@ do_random_walk (void *cls,
   struct FingerTable *ft;
   struct Finger *finger;
   struct Trail *trail;
+  unsigned int nsize;
 
   random_walk_task = NULL;
   friend = pick_random_friend ();
@@ -846,6 +1021,14 @@ do_random_walk (void *cls,
                                 friend->succ_head,
                                 friend->succ_tail,
                                 trail);
+  trail->expiration_time = GNUNET_TIME_relative_to_absolute (TRAIL_TIMEOUT);
+  trail->hn = GNUNET_CONTAINER_heap_insert (trail_heap,
+                                            trail,
+                                            trail->expiration_time.abs_value_us);
+  if (NULL == trail_timeout_task)
+    trail_timeout_task = GNUNET_SCHEDULER_add_delayed (TRAIL_TIMEOUT,
+                                                       &trail_timeout_callback,
+                                                       NULL);
   env = GNUNET_MQ_msg (rwm,
                        GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK);
   rwm->hops_taken = htonl (0);
@@ -860,21 +1043,19 @@ do_random_walk (void *cls,
     delete_trail (finger->trail,
                   GNUNET_NO,
                   GNUNET_YES);
-  if (ft->finger_array_size < 42)
-  {
-    // FIXME: must have finger array of the right size here,
-    // FIXME: growing / shrinking are tricy -- with pointers
-    // from Trails!!!
-  }
-
+  if (ft->finger_array_size < (nsize = get_desired_finger_array_size()) )
+    GNUNET_array_grow (ft->fingers,
+                       ft->finger_array_size,
+                       nsize);
   GNUNET_assert (NULL == ft->fingers[ft->walk_offset]);
-
+  trail->ft = ft;
+  trail->finger_off = ft->walk_offset;
   finger = GNUNET_new (struct Finger);
   finger->trail = trail;
-  trail->finger = &ft->fingers[ft->walk_offset];
   finger->ft = ft;
   ft->fingers[ft->walk_offset] = finger;
   ft->is_sorted = GNUNET_NO;
+  ft->number_valid_fingers++;
   ft->walk_offset = (ft->walk_offset + 1) % ft->finger_array_size;
 
   walk_layer = (walk_layer + 1) % NUMBER_LAYERED_ID;
@@ -889,10 +1070,13 @@ do_random_walk (void *cls,
  *
  * @param cls closure
  * @param peer_identity peer identity this notification is about
+ * @param mq message queue for transmission to @a peer_identity
+ * @return the `struct FriendInfo` for the @a peer_identity, NULL for us
  */
-static void
+static void *
 handle_core_connect (void *cls,
-                     const struct GNUNET_PeerIdentity *peer_identity)
+                     const struct GNUNET_PeerIdentity *peer_identity,
+                    struct GNUNET_MQ_Handle *mq)
 {
   struct FriendInfo *friend;
 
@@ -900,21 +1084,11 @@ handle_core_connect (void *cls,
   if (0 == memcmp (&my_identity,
                    peer_identity,
                    sizeof (struct GNUNET_PeerIdentity)))
-    return;
-
-  /* If peer already exists in our friend_peermap, then exit. */
-  if (GNUNET_YES ==
-      GNUNET_CONTAINER_multipeermap_contains (friends_peermap,
-                                              peer_identity))
-  {
-    GNUNET_break (0);
-    return;
-  }
+    return NULL;
 
   friend = GNUNET_new (struct FriendInfo);
-  friend->id = *peer_identity;
-  friend->mq = GNUNET_CORE_mq_create (core_api,
-                                      peer_identity);
+  friend->id = peer_identity;
+  friend->mq = mq;
   GNUNET_assert (GNUNET_OK ==
                  GNUNET_CONTAINER_multipeermap_put (friends_peermap,
                                                     peer_identity,
@@ -926,6 +1100,7 @@ handle_core_connect (void *cls,
     random_walk_task = GNUNET_SCHEDULER_add_now (&do_random_walk,
                                                  NULL);
   }
+  return friend;
 }
 
 
@@ -947,27 +1122,26 @@ core_init (void *cls,
  * Handle a `struct RandomWalkMessage` from a
  * #GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK message.
  *
- * @param cls closure (NULL)
- * @param peer sender identity
- * @param message the setup message
- * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
+ * @param cls the `struct FriendInfo` for the sender
+ * @param m the setup message
  */
-static int
+static void
 handle_dht_p2p_random_walk (void *cls,
-                            const struct GNUNET_PeerIdentity *peer,
-                            const struct GNUNET_MessageHeader *message)
+                            const struct RandomWalkMessage *m)
 {
-  const struct RandomWalkMessage *m;
+  struct FriendInfo *pred = cls;
   struct Trail *t;
-  struct FriendInfo *pred;
+  uint16_t layer;
 
-  m = (const struct RandomWalkMessage *) message;
-  pred = GNUNET_CONTAINER_multipeermap_get (friends_peermap,
-                                            peer);
+  layer = ntohs (m->layer);
+  if (layer > NUMBER_LAYERED_ID)
+  {
+    GNUNET_break_op (0);
+    return;
+  }
   t = GNUNET_new (struct Trail);
   t->pred_id = m->trail_id;
   t->pred = pred;
-  t->expiration_time = GNUNET_TIME_relative_to_absolute (TRAIL_TIMEOUT);
   if (GNUNET_OK !=
       GNUNET_CONTAINER_multihashmap_put (trail_map,
                                          &t->pred_id,
@@ -976,36 +1150,37 @@ handle_dht_p2p_random_walk (void *cls,
   {
     GNUNET_break_op (0);
     GNUNET_free (t);
-    return GNUNET_SYSERR;
+    return;
   }
   GNUNET_CONTAINER_MDLL_insert (pred,
                                 pred->pred_head,
                                 pred->pred_tail,
                                 t);
+  t->expiration_time = GNUNET_TIME_relative_to_absolute (TRAIL_TIMEOUT);
+  t->hn = GNUNET_CONTAINER_heap_insert (trail_heap,
+                                        t,
+                                        t->expiration_time.abs_value_us);
+  if (NULL == trail_timeout_task)
+    trail_timeout_task = GNUNET_SCHEDULER_add_delayed (TRAIL_TIMEOUT,
+                                                       &trail_timeout_callback,
+                                                       NULL);
+
   if (ntohl (m->hops_taken) > GDS_NSE_get ())
   {
     /* We are the last hop, generate response */
     struct GNUNET_MQ_Envelope *env;
     struct RandomWalkResponseMessage *rwrm;
-    uint16_t layer;
 
     env = GNUNET_MQ_msg (rwrm,
                          GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK_RESPONSE);
     rwrm->reserved = htonl (0);
     rwrm->trail_id = m->trail_id;
-    layer = ntohs (m->layer);
     if (0 == layer)
       (void) GDS_DATACACHE_get_random_key (&rwrm->location);
     else
     {
       struct FingerTable *ft;
 
-      if (layer > NUMBER_LAYERED_ID)
-      {
-        GNUNET_break_op (0);
-        // FIXME: clean up 't'...
-        return GNUNET_SYSERR;
-      }
       ft = &fingers[layer-1];
       if (0 == ft->number_valid_fingers)
       {
@@ -1015,9 +1190,13 @@ handle_dht_p2p_random_walk (void *cls,
       else
       {
         struct Finger *f;
+        unsigned int off;
+        unsigned int i;
 
-        f = ft->fingers[GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE,
-                                                  ft->number_valid_fingers)];
+        off = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE,
+                                        ft->number_valid_fingers);
+        for (i=0; (NULL == (f = ft->fingers[i])) || (off > 0); i++)
+          if (NULL != f) off--;
         rwrm->location = f->destination;
       }
     }
@@ -1047,7 +1226,7 @@ handle_dht_p2p_random_walk (void *cls,
                                     pred->pred_tail,
                                     t);
       GNUNET_free (t);
-      return GNUNET_OK;
+      return;
     }
     GNUNET_CONTAINER_MDLL_insert (succ,
                                   succ->succ_head,
@@ -1061,28 +1240,69 @@ handle_dht_p2p_random_walk (void *cls,
     GNUNET_MQ_send (succ->mq,
                     env);
   }
-  return GNUNET_OK;
 }
 
 
 /**
- * Handle a `struct RandomWalkResponseMessage` from a GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK_RESPONSE
- * message.
+ * Handle a `struct RandomWalkResponseMessage`.
  *
- * @param cls closure (NULL)
- * @param peer sender identity
- * @param message the setup response message
- * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
+ * @param cls closure
+ * @param rwrm the setup response message
  */
-static int
+static void
 handle_dht_p2p_random_walk_response (void *cls,
-                                     const struct GNUNET_PeerIdentity *peer,
-                                     const struct GNUNET_MessageHeader *message)
+                                     const struct RandomWalkResponseMessage *rwrm)
 {
-  const struct RandomWalkResponseMessage *rwrm;
+  struct Trail *trail;
+  struct FriendInfo *pred;
+  struct FingerTable *ft;
+  struct Finger *finger;
 
-  rwrm = (const struct RandomWalkResponseMessage *) message;
-  // 1) lookup trail => find Finger entry => fill in 'destination' and mark valid, move to end of sorted array, mark unsorted, update links from 'trails'
+  trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+                                             &rwrm->trail_id);
+  if (NULL == trail)
+  {
+    /* TODO: log/statistics: we didn't find the trail (can happen) */
+    return;
+  }
+  if (NULL != (pred = trail->pred))
+  {
+    /* We are not the first hop, keep forwarding */
+    struct GNUNET_MQ_Envelope *env;
+    struct RandomWalkResponseMessage *rwrm2;
+
+    env = GNUNET_MQ_msg (rwrm2,
+                         GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK_RESPONSE);
+    rwrm2->reserved = htonl (0);
+    rwrm2->location = rwrm->location;
+    rwrm2->trail_id = trail->pred_id;
+    GNUNET_MQ_send (pred->mq,
+                    env);
+    return;
+  }
+  /* We are the first hop, complete finger */
+  if (NULL == (ft = trail->ft))
+  {
+    /* Eh, why did we create the trail if we have no FT? */
+    GNUNET_break (0);
+    delete_trail (trail,
+                  GNUNET_NO,
+                  GNUNET_YES);
+    return;
+  }
+  if (NULL == (finger = ft->fingers[trail->finger_off]))
+  {
+    /* Eh, finger got deleted, but why not the trail as well? */
+    GNUNET_break (0);
+    delete_trail (trail,
+                  GNUNET_NO,
+                  GNUNET_YES);
+    return;
+  }
+
+
+  // 1) lookup trail => find Finger entry => fill in 'destination' and mark valid, move to end of sorted array,
+  //mark unsorted, update links from 'trails'
   /*
    * Steps :
    *  1 check if we are the correct layer
@@ -1091,105 +1311,33 @@ handle_dht_p2p_random_walk_response (void *cls,
    */
   /* FIXME: add the value in db structure 1.a */
 
-  return GNUNET_OK;
 }
 
 
 /**
  * Handle a `struct TrailDestroyMessage`.
  *
- * @param cls closure (NULL)
- * @param peer sender identity
- * @param message the finger destroy message
- * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
- */
-static int
-handle_dht_p2p_trail_destroy (void *cls,
-                             const struct GNUNET_PeerIdentity *peer,
-                             const struct GNUNET_MessageHeader *message)
-{
-  const struct TrailDestroyMessage *tdm;
-
-  tdm = (const struct TrailDestroyMessage *) message;
-
-  /*
-   * Steps :
-   *  1 check if message comme from a trail (that we still remember...)
-   *  1.a.1 if true: send the destroy message to the rest trail
-   *  1.a.2 clean the trail structure
-   *  1.a.3 did i have to remove the trail and ID from the db structure?
-   *  1.b if false: do nothing
-   */
-
-  return GNUNET_OK;
-}
-
-
-/**
- * Handler for a message we received along some trail.
- *
  * @param cls closure
- * @param trail_id trail identifier
- * @param message the message we got
- * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
- */
-typedef int
-(*TrailHandlerCallback)(void *cls,
-                        const struct GNUNET_HashCode *trail_id,
-                        const struct GNUNET_MessageHandler *message);
-
-
-/**
- * Definition of a handler for a message received along some trail.
+ * @param tdm the trail destroy message
  */
-struct TrailHandler
-{
-  /**
-   * NULL for end-of-list.
-   */
-  TrailHandlerCallback callback;
-
-  /**
-   * Closure for @e callback.
-   */
-  void *cls;
-
-  /**
-   * Message type this handler addresses.
-   */
-  uint16_t message_type;
-
-  /**
-   * Use 0 for variable-size.
-   */
-  uint16_t message_size;
-};
-
-
-/**
- * Handle a `struct TrailRouteMessage`.
- *
- * @param cls closure (NULL)
- * @param peer sender identity
- * @param message the finger destroy message
- * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
- */
-static int
-handle_dht_p2p_trail_route (void *cls,
-                            const struct GNUNET_PeerIdentity *peer,
-                            const struct GNUNET_MessageHeader *message)
+static void
+handle_dht_p2p_trail_destroy (void *cls,
+                             const struct TrailDestroyMessage *tdm)
 {
-  const struct TrailRouteMessage *trm;
-
-  trm = (const struct TrailRouteMessage *) message;
+  struct FriendInfo *sender = cls;
+  struct Trail *trail;
 
-  /*
-   * Steps :
-   *  1 check if message comme from a trail
-   *  1.a.1 if trail not finished with us, continue to forward
-   *  1.a.2 otherwise handle body message embedded in trail
-   */
-  return GNUNET_OK;
+  trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+                                             &tdm->trail_id);
+  delete_trail (trail,
+                ( (NULL != trail->succ) &&
+                  (0 == memcmp (sender->id,
+                                &trail->succ->id,
+                                sizeof (struct GNUNET_PeerIdentity))) ),
+                ( (NULL != trail->pred) &&
+                  (0 == memcmp (sender->id,
+                                &trail->pred->id,
+                                sizeof (struct GNUNET_PeerIdentity))) ));
 }
 
 
@@ -1199,24 +1347,27 @@ handle_dht_p2p_trail_route (void *cls,
  *
  * @param cls closure (NULL)
  * @param trail_id path to the originator
+ * @param trail_path path the message took on the trail, if available
+ * @param trail_path_length number of entries on the @a trail_path
  * @param message the finger setup message
  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 static int
 handle_dht_p2p_successor_find (void *cls,
                                const struct GNUNET_HashCode *trail_id,
+                               const struct GNUNET_PeerIdentity *trail_path,
+                               unsigned int trail_path_length,
                                const struct GNUNET_MessageHeader *message)
 {
   const struct FindSuccessorMessage *fsm;
 
+  /* We do not expect to track trails for the forward-direction
+     of successor finding... */
+  GNUNET_break_op (0 == trail_path_length);
   fsm = (const struct FindSuccessorMessage *) message;
-  // locate trail (for sending reply), if not exists, fail nicely.
-  // otherwise, go to datacache and return 'top k' elements closest to 'key'
-  // as "PUT" messages via the trail (need to extend DB API!)
-#if 0
-  GDS_DATACACHE_get_successors (trail_id,
-                                key);
-#endif
+  GDS_DATACACHE_get_successors (&fsm->key,
+                                &GDS_NEIGHBOURS_send_get_result,
+                                (void *) trail_id);
   return GNUNET_OK;
 }
 
@@ -1226,18 +1377,24 @@ handle_dht_p2p_successor_find (void *cls,
  *
  * @param cls closure (NULL)
  * @param trail_id path to the originator
+ * @param trail_path path the message took on the trail, if available
+ * @param trail_path_length number of entries on the @a trail_path
  * @param message the peer get message
  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 static int
 handle_dht_p2p_peer_get (void *cls,
                          const struct GNUNET_HashCode *trail_id,
+                         const struct GNUNET_PeerIdentity *trail_path,
+                         unsigned int trail_path_length,
                          const struct GNUNET_MessageHeader *message)
 {
+#if 0
   const struct PeerGetMessage *pgm;
 
   // FIXME: note: never called like this, message embedded with trail route!
   pgm = (const struct PeerGetMessage *) message;
+#endif
   // -> lookup in datacache (figure out way to remember trail!)
      /*
     * steps :
@@ -1257,17 +1414,23 @@ handle_dht_p2p_peer_get (void *cls,
  *
  * @param cls closure (NULL)
  * @param trail_id path to the originator
+ * @param trail_path path the message took on the trail, if available
+ * @param trail_path_length number of entries on the @a trail_path
  * @param message the peer get result message
  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 static int
 handle_dht_p2p_peer_get_result (void *cls,
                                 const struct GNUNET_HashCode *trail_id,
+                                const struct GNUNET_PeerIdentity *trail_path,
+                                unsigned int trail_path_length,
                                 const struct GNUNET_MessageHeader *message)
 {
+#if 0
   const struct PeerGetResultMessage *pgrm;
 
   pgrm = (const struct PeerGetResultMessage *) message;
+#endif
   // pretty much: parse, & pass to client (there is some call for that...)
 
 #if 0
@@ -1293,17 +1456,23 @@ handle_dht_p2p_peer_get_result (void *cls,
  *
  * @param cls closure (NULL)
  * @param trail_id path to the originator
+ * @param trail_path path the message took on the trail, if available
+ * @param trail_path_length number of entries on the @a trail_path
  * @param message the peer put message
  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 static int
 handle_dht_p2p_peer_put (void *cls,
                          const struct GNUNET_HashCode *trail_id,
+                         const struct GNUNET_PeerIdentity *trail_path,
+                         unsigned int trail_path_length,
                          const struct GNUNET_MessageHeader *message)
 {
+#if 0
   const struct PeerGetResultMessage *pgrm;
 
   pgrm = (const struct PeerGetResultMessage *) message;
+#endif
   // parse & store in datacache, this is in response to us asking for successors.
   /*
    * steps :
@@ -1314,14 +1483,14 @@ handle_dht_p2p_peer_put (void *cls,
 #if 0
   GDS_DATACACHE_handle_put (expiration_time,
                             key,
-                            path_length, path,
+                            combined_path_length, combined_path,
                             block_type,
                             data_size,
                             data);
   GDS_CLIENTS_process_put (options,
                            block_type,
                            0, 0,
-                           path_length, path,
+                           combined_path_length, combined_path,
                            expiration_time,
                            key,
                            data,
@@ -1331,42 +1500,229 @@ handle_dht_p2p_peer_put (void *cls,
 }
 
 
+/**
+ * Handler for a message we received along some trail.
+ *
+ * @param cls closure
+ * @param trail_id trail identifier
+ * @param trail_path path the message took on the trail, if available
+ * @param trail_path_length number of entries on the @a trail_path
+ * @param message the message we got
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
+ */
+typedef int
+(*TrailHandlerCallback)(void *cls,
+                        const struct GNUNET_HashCode *trail_id,
+                        const struct GNUNET_PeerIdentity *trail_path,
+                        unsigned int trail_path_length,
+                        const struct GNUNET_MessageHeader *message);
+
+
+/**
+ * Definition of a handler for a message received along some trail.
+ */
+struct TrailHandler
+{
+  /**
+   * NULL for end-of-list.
+   */
+  TrailHandlerCallback callback;
+
+  /**
+   * Closure for @e callback.
+   */
+  void *cls;
+
+  /**
+   * Message type this handler addresses.
+   */
+  uint16_t message_type;
+
+  /**
+   * Use 0 for variable-size.
+   */
+  uint16_t message_size;
+};
+
+
+/**
+ * Check that a `struct TrailRouteMessage` is well-formed.
+ *
+ * @param cls closure
+ * @param trm the finger destroy message
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
+ */
+static int
+check_dht_p2p_trail_route (void *cls,
+                          const struct TrailRouteMessage *trm)
+{
+  const struct GNUNET_PeerIdentity *path;
+  uint16_t path_length;
+  const struct GNUNET_MessageHeader *payload;
+  size_t msize;
+
+  msize = ntohs (trm->header.size);
+  path_length = ntohs (trm->path_length);
+  if (msize < sizeof (struct TrailRouteMessage) +
+      path_length * sizeof (struct GNUNET_PeerIdentity) +
+      sizeof (struct GNUNET_MessageHeader) )
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  path = (const struct GNUNET_PeerIdentity *) &trm[1];
+  payload = (const struct GNUNET_MessageHeader *) &path[path_length];
+  if (msize != (ntohs (payload->size) +
+                sizeof (struct TrailRouteMessage) +
+                path_length * sizeof (struct GNUNET_PeerIdentity)))
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  /* FIXME: verify payload is OK!? */
+  return GNUNET_OK;
+}
+
+
+/**
+ * Handle a `struct TrailRouteMessage`.
+ *
+ * @param cls closure
+ * @param trm the finger destroy message
+ */
+static void
+handle_dht_p2p_trail_route (void *cls,
+                            const struct TrailRouteMessage *trm)
+{
+  static const struct TrailHandler handlers[] = {
+    { &handle_dht_p2p_successor_find, NULL,
+      GNUNET_MESSAGE_TYPE_WDHT_SUCCESSOR_FIND,
+      sizeof (struct FindSuccessorMessage) },
+    { &handle_dht_p2p_peer_get, NULL,
+      GNUNET_MESSAGE_TYPE_WDHT_GET,
+      0 },
+    { &handle_dht_p2p_peer_get_result, NULL,
+      GNUNET_MESSAGE_TYPE_WDHT_GET_RESULT,
+      0 },
+    { &handle_dht_p2p_peer_put, NULL,
+      GNUNET_MESSAGE_TYPE_WDHT_PUT,
+      0 },
+    { NULL, NULL, 0, 0 }
+  };
+  struct FriendInfo *sender = cls;
+  unsigned int i;
+  const struct GNUNET_PeerIdentity *path;
+  uint16_t path_length;
+  const struct GNUNET_MessageHeader *payload;
+  const struct TrailHandler *th;
+  struct Trail *trail;
+
+  path_length = ntohs (trm->path_length);
+  path = (const struct GNUNET_PeerIdentity *) &trm[1];
+  payload = (const struct GNUNET_MessageHeader *) &path[path_length];
+  /* Is this message for us? */
+  trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+                                             &trm->trail_id);
+  if ( (NULL != trail->pred) &&
+       (0 == memcmp (sender->id,
+                     &trail->pred->id,
+                     sizeof (struct GNUNET_PeerIdentity))) )
+  {
+    /* forward to 'successor' */
+    if (NULL != trail->succ)
+    {
+      forward_message_on_trail (trail->succ,
+                                &trail->succ_id,
+                                ntohs (trm->record_path),
+                                sender->id,
+                                path,
+                                path_length,
+                                payload);
+      return;
+    }
+  }
+  else
+  {
+    /* forward to 'predecessor' */
+    GNUNET_break_op ( (NULL != trail->succ) &&
+                      (0 == memcmp (sender->id,
+                                    &trail->succ->id,
+                                    sizeof (struct GNUNET_PeerIdentity))) );
+    if (NULL != trail->pred)
+    {
+      forward_message_on_trail (trail->pred,
+                                &trail->pred_id,
+                                ntohs (trm->record_path),
+                                sender->id,
+                                path,
+                                path_length,
+                                payload);
+      return;
+    }
+  }
+
+  /* Message is for us, dispatch to handler */
+  th = NULL;
+  for (i=0; NULL != handlers[i].callback; i++)
+  {
+    th = &handlers[i];
+    if (ntohs (payload->type) == th->message_type)
+    {
+      if ( (0 == th->message_size) ||
+           (ntohs (payload->size) == th->message_size) )
+        th->callback (th->cls,
+                      &trm->trail_id,
+                      path,
+                      path_length,
+                      payload);
+      else
+        GNUNET_break_op (0);
+      break;
+    }
+  }
+  GNUNET_break_op (NULL != th);
+}
+
+
 /**
  * Initialize neighbours subsystem.
+ *
  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 int
 GDS_NEIGHBOURS_init (void)
 {
-  static const struct GNUNET_CORE_MessageHandler core_handlers[] = {
-    { &handle_dht_p2p_random_walk,
-      GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK,
-      sizeof (struct RandomWalkMessage) },
-    { &handle_dht_p2p_random_walk_response,
-      GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK_RESPONSE,
-      sizeof (struct RandomWalkResponseMessage) },
-    { &handle_dht_p2p_trail_destroy,
-      GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY,
-      sizeof (struct TrailDestroyMessage) },
-    { &handle_dht_p2p_trail_route,
-      GNUNET_MESSAGE_TYPE_WDHT_TRAIL_ROUTE,
-      0},
-    {NULL, 0, 0}
+  struct GNUNET_MQ_MessageHandler core_handlers[] = {
+    GNUNET_MQ_hd_fixed_size (dht_p2p_random_walk,
+                             GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK,
+                             struct RandomWalkMessage,
+                             NULL),
+    GNUNET_MQ_hd_fixed_size (dht_p2p_random_walk_response,
+                             GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK_RESPONSE,
+                             struct RandomWalkResponseMessage,
+                             NULL),
+    GNUNET_MQ_hd_fixed_size (dht_p2p_trail_destroy,
+                             GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY,
+                             struct TrailDestroyMessage,
+                             NULL),
+    GNUNET_MQ_hd_var_size (dht_p2p_trail_route,
+                           GNUNET_MESSAGE_TYPE_WDHT_TRAIL_ROUTE,
+                           struct TrailRouteMessage,
+                           NULL),
+    GNUNET_MQ_handler_end ()
   };
 
-  core_api =
-    GNUNET_CORE_connect (GDS_cfg, NULL,
-                         &core_init,
-                         &handle_core_connect,
-                         &handle_core_disconnect,
-                         NULL, GNUNET_NO,
-                         NULL, GNUNET_NO,
-                         core_handlers);
-
+  core_api = GNUNET_CORE_connect (GDS_cfg, NULL,
+                                 &core_init,
+                                 &handle_core_connect,
+                                 &handle_core_disconnect,
+                                 core_handlers);
   if (NULL == core_api)
     return GNUNET_SYSERR;
-  friends_peermap = GNUNET_CONTAINER_multipeermap_create (256, GNUNET_NO);
-  trail_map = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_YES);
+  friends_peermap = GNUNET_CONTAINER_multipeermap_create (256,
+                                                         GNUNET_NO);
+  trail_map = GNUNET_CONTAINER_multihashmap_create (1024,
+                                                   GNUNET_YES);
   trail_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
   return GNUNET_OK;
 }
@@ -1390,6 +1746,11 @@ GDS_NEIGHBOURS_done (void)
   trail_map = NULL;
   GNUNET_CONTAINER_heap_destroy (trail_heap);
   trail_heap = NULL;
+  if (NULL != trail_timeout_task)
+  {
+    GNUNET_SCHEDULER_cancel (trail_timeout_task);
+    trail_timeout_task = NULL;
+  }
 }
 
 
@@ -1398,10 +1759,10 @@ GDS_NEIGHBOURS_done (void)
  *
  * @return my identity
  */
-struct GNUNET_PeerIdentity
-GDS_NEIGHBOURS_get_my_id (void)
+struct GNUNET_PeerIdentity *
+GDS_NEIGHBOURS_get_id (void)
 {
-  return my_identity;
+  return &my_identity;
 }
 
 /* end of gnunet-service-wdht_neighbours.c */