convert fs publish to MQ
[oweals/gnunet.git] / src / dht / gnunet-service-wdht_neighbours.c
index 68b05d2eb13ff8434565258e53a1b416cf85cd92..54c52507e1714a5553734ad97f324a2167ae4ddc 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-2015 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"
@@ -39,9 +44,6 @@
 #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 "dht.h"
 
 #define DEBUG(...)                                           \
  */
 struct FriendInfo;
 
+/**
+ *
+ */
+struct FingerTable;
 
 /**
  * Information we keep per trail.
@@ -83,6 +89,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 +134,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 +141,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;
 
 };
 
@@ -152,12 +166,24 @@ struct FriendInfo
    */
   struct GNUNET_PeerIdentity id;
 
+  /**
+   *
+   */
   struct Trail *pred_head;
 
+  /**
+   *
+   */
   struct Trail *pred_tail;
 
+  /**
+   *
+   */
   struct Trail *succ_head;
 
+  /**
+   *
+   */
   struct Trail *succ_tail;
 
   /**
@@ -168,15 +194,24 @@ struct FriendInfo
 };
 
 
-struct FingerTable;
-
-
+/**
+ *
+ */
 struct Finger
 {
+  /**
+   *
+   */
   struct Trail *trail;
 
+  /**
+   *
+   */
   struct FingerTable *ft;
 
+  /**
+   *
+   */
   struct GNUNET_HashCode destination;
 
   /**
@@ -193,18 +228,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;
 
@@ -314,7 +344,7 @@ struct TrailDestroyMessage
 struct FindSuccessorMessage
 {
   /**
-   * Type: #GNUNET_MESSAGE_TYPE_WDHT_FIND_SUCCESSOR
+   * Type: #GNUNET_MESSAGE_TYPE_WDHT_SUCCESSOR_FIND
    */
   struct GNUNET_MessageHeader header;
 
@@ -323,12 +353,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).
@@ -349,9 +373,14 @@ struct TrailRouteMessage
   struct GNUNET_MessageHeader header;
 
   /**
-   * Zero, for alignment.
+   * #GNUNET_YES if the path should be recorded, #GNUNET_NO if not; in NBO.
    */
-  uint32_t reserved GNUNET_PACKED;
+  uint16_t record_path GNUNET_PACKED;
+
+  /**
+   * Length of the recorded trail, 0 if @e record_path is #GNUNET_NO; in NBO.
+   */
+  uint16_t path_length GNUNET_PACKED;
 
   /**
    * Unique (random) identifier this peer will use to
@@ -359,7 +388,13 @@ struct TrailRouteMessage
    */
   struct GNUNET_HashCode trail_id;
 
-  /* followed by payload to send along the trail */
+  /**
+   * Path the message has taken so far (excluding sender).
+   */
+  /* struct GNUNET_PeerIdentity path[path_length]; */
+
+  /* followed by payload (another `struct GNUNET_MessageHeader`) to
+     send along the trail */
 };
 
 
@@ -472,7 +507,7 @@ struct PeerGetResultMessage
   struct GNUNET_MessageHeader header;
 
   /**
-   * The type for the data.
+   * The type for the data in NBO.
    */
   uint32_t type GNUNET_PACKED;
 
@@ -482,20 +517,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.
@@ -504,8 +529,6 @@ struct PeerGetResultMessage
 
   /* put path (if tracked) */
 
-  /* get path (if tracked) */
-
   /* Payload */
 
 };
@@ -576,14 +599,24 @@ GDS_NEIGHBOURS_handle_put (const struct GNUNET_HashCode *key,
                            enum GNUNET_DHT_RouteOption options,
                            uint32_t desired_replication_level,
                            struct GNUNET_TIME_Absolute expiration_time,
-                           const void *data, size_t data_size)
+                           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,
+                           expiration_time,
+                           key,
+                           data,
+                           data_size);
 }
 
 
@@ -605,6 +638,7 @@ GDS_NEIGHBOURS_handle_get (const struct GNUNET_HashCode *key,
 {
   // find closest finger(s) on all layers
   // use TrailRoute with PeerGetMessage embedded to contact peer
+  // NOTE: actually more complicated, see paper!
 }
 
 
@@ -623,160 +657,184 @@ delete_trail (struct Trail *trail,
   struct FriendInfo *friend;
   struct GNUNET_MQ_Envelope *env;
   struct TrailDestroyMessage *tdm;
-  struct Trail *current;
-
-  tdm = GNUNET_new(struct TrailDestroyMessage);
-  current = trail;
+  struct Finger *finger;
 
-  if (inform_pred)
+  friend = trail->pred;
+  if (NULL != friend)
   {
-    struct Trail *previous_pred;
-    struct Trail *next_pred;
-    tdm->trail_id = current->pred_id;
-    friend = current->pred;
-
-    while( NULL != current)
+    if (GNUNET_YES == inform_pred)
     {
-      if (NULL != friend)
-      {
-        env = GNUNET_MQ_msg (tdm,
-                             GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
-        GNUNET_MQ_send (friend->mq,
-                        env);
-      }
-
-      next_pred = current->next_pred;
-      previous_pred = current->prev_pred;
-
-      /* Freeing current trail memory */
-      GNUNET_free(current);
-
-      /* Destroying the next successor list */
-      while(NULL != previous_pred)
-      {
-        struct Trail *tmp;
-        if (NULL != friend)
-        {
-          env = GNUNET_MQ_msg (tdm,
-                               GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
-          GNUNET_MQ_send (friend->mq,
-                          env);
-        }
-
-        tmp = next_pred;
-        next_pred = next_pred->next_pred;
-        GNUNET_free(tmp);
-      }
-
-      /* Destroying the next predecessor list */
-      while(NULL != next_pred)
-      {
-        struct Trail *tmp;
-        if (NULL != friend)
-        {
-          env = GNUNET_MQ_msg (tdm,
-                               GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
-          GNUNET_MQ_send (friend->mq,
-                          env);
-        }
-
-        tmp = next_pred;
-        next_pred = next_pred->next_pred;
-        GNUNET_free(tmp);
-      }
+      env = GNUNET_MQ_msg (tdm,
+                           GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
+      tdm->trail_id = trail->pred_id;
+      GNUNET_MQ_send (friend->mq,
+                      env);
     }
+    GNUNET_CONTAINER_MDLL_remove (pred,
+                                  friend->pred_head,
+                                  friend->pred_tail,
+                                  trail);
   }
-
-  if (inform_succ)
+  friend = trail->succ;
+  if (NULL != friend)
   {
-    struct Trail *previous_succ;
-    struct Trail *next_succ;
-    tdm->trail_id = current->succ_id;
-    friend = current->succ;
-
-    while( NULL != current)
+    if (GNUNET_YES == inform_succ)
     {
-      if (NULL != friend)
-      {
-        env = GNUNET_MQ_msg (tdm,
-                             GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
-        GNUNET_MQ_send (friend->mq,
-                        env);
-      }
-
-      next_succ = current->next_succ;
-      previous_succ = current->prev_succ;
+      env = GNUNET_MQ_msg (tdm,
+                           GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
+      tdm->trail_id = trail->pred_id;
+      GNUNET_MQ_send (friend->mq,
+                      env);
+    }
+    GNUNET_CONTAINER_MDLL_remove (succ,
+                                  friend->pred_head,
+                                  friend->pred_tail,
+                                  trail);
+  }
+  GNUNET_break (trail ==
+                GNUNET_CONTAINER_heap_remove_node (trail->hn));
+  finger = trail->ft->fingers[trail->finger_off];
+  if (NULL != finger)
+  {
+    trail->ft->fingers[trail->finger_off] = NULL;
+    trail->ft->number_valid_fingers--;
+    GNUNET_free (finger);
+  }
+  GNUNET_free (trail);
+}
 
-      /* Freeing current trail memory */
-      GNUNET_free(current);
 
-      /* Destroying the next successor list */
-      while(NULL != previous_succ)
-      {
-        struct Trail *tmp;
-        if (NULL != friend)
-        {
-          env = GNUNET_MQ_msg (tdm,
-                               GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
-          GNUNET_MQ_send (friend->mq,
-                          env);
-        }
-        tmp = next_succ;
-        previous_succ = previous_succ->prev_succ;
-        GNUNET_free(tmp);
-      }
+/**
+ * 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;
 
-      /* Destroying the next successor list */
-      while(NULL != next_succ)
-      {
-        struct Trail *tmp;
-        if (NULL != friend)
-        {
-          env = GNUNET_MQ_msg (tdm,
-                               GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
-          GNUNET_MQ_send (friend->mq,
-                          env);
-        }
-        tmp = next_succ;
-        next_succ = next_succ->next_pred;
-        GNUNET_free(tmp);
-      }
+  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)
+  {
+    memcpy (new_path,
+            path,
+            path_length * sizeof (struct GNUNET_PeerIdentity));
+    new_path[path_length] = *predecessor;
+  }
+  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 options routing options (from GET request)
  * @param key Key of the requested data.
  * @param type Block type
- * @param target_peer Next peer to forward the message to.
- * @param source_peer Peer which has the data for the key.
  * @param put_path_length Number of peers in @a put_path
  * @param put_path Path taken to put the data at its stored location.
- * @param get_path_length Number of peers in @a get_path
- * @param get_path Path taken to reach to the location of the key.
  * @param expiration When will this result expire?
  * @param data Payload to store
  * @param data_size Size of the @a data
  */
 void
-GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *key,
+GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *trail_id,
+                                enum GNUNET_DHT_RouteOption options,
+                                const struct GNUNET_HashCode *key,
                                 enum GNUNET_BLOCK_Type type,
-                                const struct GNUNET_PeerIdentity *target_peer,
-                                const struct GNUNET_PeerIdentity *source_peer,
                                 unsigned int put_path_length,
                                 const struct GNUNET_PeerIdentity *put_path,
-                                unsigned int get_path_length,
-                                const struct GNUNET_PeerIdentity *get_path,
                                 struct GNUNET_TIME_Absolute expiration,
-                                const void *data, size_t data_size)
+                                const void *data,
+                                size_t data_size)
 {
-  // TRICKY: need to introduce some context to remember trail from
-  // the lookup...
+  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);
 }
 
+
 /**
  * Method called whenever a peer disconnects.
  *
@@ -827,15 +885,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;
 }
 
 
@@ -843,11 +972,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;
@@ -856,6 +983,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 ();
@@ -879,6 +1007,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);
@@ -893,21 +1029,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;
@@ -993,14 +1127,20 @@ handle_dht_p2p_random_walk (void *cls,
   const struct RandomWalkMessage *m;
   struct Trail *t;
   struct FriendInfo *pred;
+  uint16_t layer;
 
   m = (const struct RandomWalkMessage *) message;
+  layer = ntohs (m->layer);
+  if (layer > NUMBER_LAYERED_ID)
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
   pred = GNUNET_CONTAINER_multipeermap_get (friends_peermap,
                                             peer);
   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,
@@ -1015,30 +1155,31 @@ handle_dht_p2p_random_walk (void *cls,
                                 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)
       {
@@ -1048,9 +1189,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;
       }
     }
@@ -1099,8 +1244,7 @@ handle_dht_p2p_random_walk (void *cls,
 
 
 /**
- * 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
@@ -1113,9 +1257,57 @@ handle_dht_p2p_random_walk_response (void *cls,
                                      const struct GNUNET_MessageHeader *message)
 {
   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 GNUNET_OK;
+  }
+  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 GNUNET_OK;
+  }
+  /* 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 GNUNET_OK;
+  }
+  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 GNUNET_OK;
+  }
+
+
+  // 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
@@ -1142,46 +1334,20 @@ handle_dht_p2p_trail_destroy (void *cls,
                              const struct GNUNET_MessageHeader *message)
 {
   const struct TrailDestroyMessage *tdm;
+  struct Trail *trail;
 
   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;
-}
-
-
-/**
- * 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)
-{
-  const struct TrailRouteMessage *trm;
-
-  trm = (const struct TrailRouteMessage *) message;
-
-  /*
-   * 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
-   */
-
+  trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+                                             &tdm->trail_id);
+  delete_trail (trail,
+                ( (NULL != trail->succ) &&
+                  (0 == memcmp (peer,
+                                &trail->succ->id,
+                                sizeof (struct GNUNET_PeerIdentity))) ),
+                ( (NULL != trail->pred) &&
+                  (0 == memcmp (peer,
+                                &trail->pred->id,
+                                sizeof (struct GNUNET_PeerIdentity))) ));
   return GNUNET_OK;
 }
 
@@ -1191,22 +1357,27 @@ handle_dht_p2p_trail_route (void *cls,
  * message.
  *
  * @param cls closure (NULL)
- * @param peer sender identity
+ * @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_PeerIdentity *peer,
+                               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!)
-
+  GDS_DATACACHE_get_successors (trail_id,
+                                &fsm->key);
   return GNUNET_OK;
 }
 
@@ -1215,13 +1386,17 @@ handle_dht_p2p_successor_find (void *cls,
  * Handle a `struct PeerGetMessage`.
  *
  * @param cls closure (NULL)
- * @param peer sender identity
+ * @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_PeerIdentity *peer,
+                         const struct GNUNET_HashCode *trail_id,
+                         const struct GNUNET_PeerIdentity *trail_path,
+                         unsigned int trail_path_length,
                          const struct GNUNET_MessageHeader *message)
 {
   const struct PeerGetMessage *pgm;
@@ -1246,20 +1421,38 @@ handle_dht_p2p_peer_get (void *cls,
  * Handle a `struct PeerGetResultMessage`.
  *
  * @param cls closure (NULL)
- * @param peer sender identity
+ * @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_PeerIdentity *peer,
-                             const struct GNUNET_MessageHeader *message)
+                                const struct GNUNET_HashCode *trail_id,
+                                const struct GNUNET_PeerIdentity *trail_path,
+                                unsigned int trail_path_length,
+                                const struct GNUNET_MessageHeader *message)
 {
   const struct PeerGetResultMessage *pgrm;
 
   pgrm = (const struct PeerGetResultMessage *) message;
   // pretty much: parse, & pass to client (there is some call for that...)
 
+#if 0
+  GDS_CLIENTS_process_get (options,
+                           type,
+                           0, 0,
+                           path_length, path,
+                           key);
+  (void) GDS_DATACACHE_handle_get (trail_id,
+                                   key,
+                                   type,
+                                   xquery,
+                                   xquery_size,
+                                   &reply_bf,
+                                   reply_bf_mutator);
+#endif
   return GNUNET_OK;
 }
 
@@ -1268,14 +1461,18 @@ handle_dht_p2p_peer_get_result (void *cls,
  * Handle a `struct PeerPutMessage`.
  *
  * @param cls closure (NULL)
- * @param peer sender identity
+ * @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_PeerIdentity *peer,
-                             const struct GNUNET_MessageHeader *message)
+                         const struct GNUNET_HashCode *trail_id,
+                         const struct GNUNET_PeerIdentity *trail_path,
+                         unsigned int trail_path_length,
+                         const struct GNUNET_MessageHeader *message)
 {
   const struct PeerGetResultMessage *pgrm;
 
@@ -1287,12 +1484,202 @@ handle_dht_p2p_peer_put (void *cls,
    * 2 use the API to add the value in the "database". Check on the xdht file, how to do it.
    * 3 Did i a have to return a notification or did i have to return GNUNET_[OK|SYSERR]?
    */
+#if 0
+  GDS_DATACACHE_handle_put (expiration_time,
+                            key,
+                            combined_path_length, combined_path,
+                            block_type,
+                            data_size,
+                            data);
+  GDS_CLIENTS_process_put (options,
+                           block_type,
+                           0, 0,
+                           combined_path_length, combined_path,
+                           expiration_time,
+                           key,
+                           data,
+                           data_size);
+#endif
+  return GNUNET_OK;
+}
+
+
+/**
+ * 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;
+};
+
+
+/**
+ * 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 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 }
+  };
+  unsigned int i;
+  const struct TrailRouteMessage *trm;
+  const struct GNUNET_PeerIdentity *path;
+  uint16_t path_length;
+  const struct GNUNET_MessageHeader *payload;
+  const struct TrailHandler *th;
+  struct Trail *trail;
+  size_t msize;
+
+  /* Parse and check message is well-formed */
+  msize = ntohs (message->size);
+  if (msize < sizeof (struct TrailRouteMessage))
+  {
+    GNUNET_break_op (0);
+    return GNUNET_YES;
+  }
+  trm = (const struct TrailRouteMessage *) message;
+  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_YES;
+  }
+  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_YES;
+  }
+
+  /* Is this message for us? */
+  trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+                                             &trm->trail_id);
+  if ( (NULL != trail->pred) &&
+       (0 == memcmp (peer,
+                     &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),
+                                peer,
+                                path,
+                                path_length,
+                                payload);
+      return GNUNET_OK;
+    }
+  }
+  else
+  {
+    /* forward to 'predecessor' */
+    GNUNET_break_op ( (NULL != trail->succ) &&
+                      (0 == memcmp (peer,
+                                    &trail->succ->id,
+                                    sizeof (struct GNUNET_PeerIdentity))) );
+    if (NULL != trail->pred)
+    {
+      forward_message_on_trail (trail->pred,
+                                &trail->pred_id,
+                                ntohs (trm->record_path),
+                                peer,
+                                path,
+                                path_length,
+                                payload);
+      return GNUNET_OK;
+    }
+  }
+
+  /* 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);
   return GNUNET_OK;
 }
 
 
 /**
  * Initialize neighbours subsystem.
+ *
  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 int
@@ -1311,18 +1698,6 @@ GDS_NEIGHBOURS_init (void)
     { &handle_dht_p2p_trail_route,
       GNUNET_MESSAGE_TYPE_WDHT_TRAIL_ROUTE,
       0},
-    { &handle_dht_p2p_successor_find,
-      GNUNET_MESSAGE_TYPE_WDHT_SUCCESSOR_FIND,
-      sizeof (struct FindSuccessorMessage) },
-    { &handle_dht_p2p_peer_get,
-      GNUNET_MESSAGE_TYPE_WDHT_GET,
-      sizeof (struct PeerGetMessage) },
-    { &handle_dht_p2p_peer_get_result,
-      GNUNET_MESSAGE_TYPE_WDHT_GET_RESULT,
-      0},
-    { &handle_dht_p2p_peer_put,
-      GNUNET_MESSAGE_TYPE_WDHT_PUT,
-      0},
     {NULL, 0, 0}
   };
 
@@ -1362,6 +1737,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;
+  }
 }