gnunet-service-dht_neighbours.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2009-2017, 2021-2022, 2026 GNUnet e.V.
 
     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     or (at your option) any later version.
 
     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     Affero General Public License for more details.
 
     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
     SPDX-License-Identifier: AGPL3.0-or-later
 */
 
#include "gnunet_common.h"
#include "gnunet_constants.h"
#include "gnunet_datacache_lib.h"
#include "gnunet_dht_service.h"
#include "gnunet_protocols.h"
#include "gnunet_hello_uri_lib.h"
#include "gnunet_pils_service.h"
#include "gnunet-service-dht.h"
#include "gnunet-service-dht_neighbours.h"
#include "gnunet-service-dht_routing.h"
#include "dht.h"
#include "dht_helper.h"
#include "gnunet_scheduler_lib.h"
#include "gnunet_util_lib.h"
 
#define LOG_TRAFFIC(kind, ...) GNUNET_log_from (kind, "dht-traffic", \
                                                __VA_ARGS__)
 
#define SANITY_CHECKS 2
 
#define MAX_BUCKETS sizeof(struct GNUNET_HashCode) * 8
 
#define DEFAULT_BUCKET_SIZE 8
 
#define FIND_PEER_REPLICATION_LEVEL 4
 
#define MAXIMUM_PENDING_PER_PEER 64
 
#define DHT_MINIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_MINUTES, 2)
 
 
#define DHT_AVG_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply (    \
          GNUNET_TIME_UNIT_SECONDS, 6)
 
#define GET_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2)
 
 
GNUNET_NETWORK_STRUCT_BEGIN
 
 
struct PeerResultMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t type GNUNET_PACKED;
 
  uint16_t reserved GNUNET_PACKED;
 
  uint16_t options GNUNET_PACKED;
 
  uint16_t put_path_length GNUNET_PACKED;
 
  uint16_t get_path_length GNUNET_PACKED;
 
  struct GNUNET_TIME_AbsoluteNBO expiration_time;
 
  struct GNUNET_HashCode key;
 
  /* trunc_peer (if truncated) */
 
  /* put path (if tracked) */
 
  /* get path (if tracked) */
 
  /* sender_sig (if path tracking is on) */
 
  /* Payload */
};
 
 
struct PeerGetMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t type GNUNET_PACKED;
 
  uint16_t options GNUNET_PACKED;
 
  uint16_t hop_count GNUNET_PACKED;
 
  uint16_t desired_replication_level GNUNET_PACKED;
 
  uint16_t result_filter_size GNUNET_PACKED;
 
  char bloomfilter[DHT_BLOOM_SIZE];
 
  struct GNUNET_HashCode key;
 
  /* result bloomfilter */
 
  /* xquery */
 
};
GNUNET_NETWORK_STRUCT_END
 
 
struct PeerInfo;
 
 
struct Target
{
  struct Target *next;
 
  struct Target *prev;
 
  struct GNUNET_DHTU_Target *utarget;
 
  struct GDS_Underlay *u;
 
  struct PeerInfo *pi;
 
  struct GNUNET_DHTU_PreferenceHandle *ph;
 
  unsigned int load;
 
  bool dropped;
 
};
 
 
struct PeerInfo
{
  struct GNUNET_PeerIdentity id;
 
  struct GNUNET_HashCode phash;
 
  struct GNUNET_TIME_Absolute hello_expiration;
 
  struct PeerInfo *next;
 
  struct PeerInfo *prev;
 
  struct Target *t_head;
 
  struct Target *t_tail;
 
  void *hello;
 
  size_t hello_size;
 
  int peer_bucket;
};
 
 
struct PeerBucket
{
  struct PeerInfo *head;
 
  struct PeerInfo *tail;
 
  unsigned int peers_size;
};
 
 
static int cache_results;
 
static unsigned int closest_bucket;
 
static unsigned int newly_found_peers;
 
static int disable_try_connect;
 
static struct PeerBucket k_buckets[MAX_BUCKETS];
 
static struct GNUNET_CONTAINER_MultiPeerMap *all_connected_peers;
 
static unsigned int bucket_size = DEFAULT_BUCKET_SIZE;
 
static struct GNUNET_SCHEDULER_Task *find_peer_task;
 
 
static void
send_done_cb (void *cls)
{
  struct Target *t = cls;
  struct PeerInfo *pi = t->pi; /* NULL if t->dropped! */
 
  GNUNET_assert (t->load > 0);
  t->load--;
  if (0 < t->load)
    return;
  if (t->dropped)
  {
    GNUNET_free (t);
    return;
  }
  /* move target back to the front */
  GNUNET_CONTAINER_DLL_remove (pi->t_head,
                               pi->t_tail,
                               t);
  GNUNET_CONTAINER_DLL_insert (pi->t_head,
                               pi->t_tail,
                               t);
}
 
 
static bool
do_send (struct PeerInfo *pi,
         const struct GNUNET_MessageHeader *msg)
{
  struct Target *t;
 
  for (t = pi->t_head;
       NULL != t;
       t = t->next)
    if (t->load < MAXIMUM_PENDING_PER_PEER)
      break;
  if (NULL == t)
  {
    /* all targets busy, drop message */
    GNUNET_STATISTICS_update (GDS_stats,
                              "# messages dropped (underlays busy)",
                              1,
                              GNUNET_NO);
    return false;
  }
  t->load++;
  /* rotate busy targets to the end */
  if (MAXIMUM_PENDING_PER_PEER == t->load)
  {
    GNUNET_CONTAINER_DLL_remove (pi->t_head,
                                 pi->t_tail,
                                 t);
    GNUNET_CONTAINER_DLL_insert_tail (pi->t_head,
                                      pi->t_tail,
                                      t);
  }
  GDS_u_send (t->u,
              t->utarget,
              msg,
              ntohs (msg->size),
              &send_done_cb,
              t);
  return true;
}
 
 
static int
find_bucket (const struct GNUNET_HashCode *hc)
{
  const struct GNUNET_HashCode *my_identity_hash;
  struct GNUNET_HashCode xor;
  unsigned int bits;
 
  my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
  GNUNET_assert (NULL != my_identity_hash);
 
  GNUNET_CRYPTO_hash_xor (hc,
                          my_identity_hash,
                          &xor);
  bits = GNUNET_CRYPTO_hash_count_leading_zeros (&xor);
  if (bits == MAX_BUCKETS)
  {
    /* How can all bits match? Got my own ID? */
    GNUNET_break (0);
    return -1;
  }
  return bits;
}
 
 
static enum GNUNET_GenericReturnValue
add_known_to_bloom (void *cls,
                    const struct GNUNET_PeerIdentity *key,
                    void *value)
{
  struct GNUNET_BLOCK_Group *bg = cls;
  struct PeerInfo *pi = value;
 
  GNUNET_BLOCK_group_set_seen (bg,
                               &pi->phash,
                               1);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Adding known peer (%s) to Bloom filter for FIND PEER\n",
              GNUNET_i2s (key));
  return GNUNET_YES;
}
 
 
static void
send_find_peer_message (void *cls)
{
  (void) cls;
 
  /* Compute when to do this again (and if we should
     even send a message right now) */
  {
    struct GNUNET_TIME_Relative next_send_time;
    bool done_early;
 
    find_peer_task = NULL;
    done_early = (newly_found_peers > bucket_size);
    /* schedule next round, taking longer if we found more peers
       in the last round. */
    next_send_time.rel_value_us =
      DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value_us
      + GNUNET_CRYPTO_random_u64 (
        GNUNET_TIME_relative_multiply (
          DHT_AVG_FIND_PEER_INTERVAL,
          1 + 100 * (1 + newly_found_peers) / bucket_size).rel_value_us);
    newly_found_peers = 0;
    GNUNET_assert (NULL == find_peer_task);
    find_peer_task =
      GNUNET_SCHEDULER_add_delayed (next_send_time,
                                    &send_find_peer_message,
                                    NULL);
    if (done_early)
      return;
  }
 
  /* actually send 'find peer' request */
  {
    const struct GNUNET_HashCode *my_identity_hash;
    struct GNUNET_BLOCK_Group *bg;
    struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 
    my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
    GNUNET_assert (NULL != my_identity_hash);
 
    bg = GNUNET_BLOCK_group_create (GDS_block_context,
                                    GNUNET_BLOCK_TYPE_DHT_HELLO,
                                    NULL,
                                    0,
                                    "seen-set-size",
                                    GNUNET_CONTAINER_multipeermap_size (
                                      all_connected_peers),
                                    NULL);
    GNUNET_CONTAINER_multipeermap_iterate (all_connected_peers,
                                           &add_known_to_bloom,
                                           bg);
    peer_bf
      = GNUNET_CONTAINER_bloomfilter_init (NULL,
                                           DHT_BLOOM_SIZE,
                                           GNUNET_CONSTANTS_BLOOMFILTER_K);
    if (GNUNET_OK !=
        GDS_NEIGHBOURS_handle_get (GNUNET_BLOCK_TYPE_DHT_HELLO,
                                   GNUNET_DHT_RO_FIND_APPROXIMATE
                                   | GNUNET_DHT_RO_RECORD_ROUTE,
                                   FIND_PEER_REPLICATION_LEVEL,
                                   0, /* hop count */
                                   my_identity_hash,
                                   NULL, 0, /* xquery */
                                   bg,
                                   peer_bf))
    {
      GNUNET_STATISTICS_update (GDS_stats,
                                "# Failed to initiate FIND PEER lookup",
                                1,
                                GNUNET_NO);
    }
    else
    {
      GNUNET_STATISTICS_update (GDS_stats,
                                "# FIND PEER messages initiated",
                                1,
                                GNUNET_NO);
    }
    GNUNET_CONTAINER_bloomfilter_free (peer_bf);
    GNUNET_BLOCK_group_destroy (bg);
  }
}
 
 
static void
update_hold (struct PeerBucket *bucket)
{
  unsigned int off = 0;
 
  /* find the peer -- we just go over all of them, should
     be hardly any more expensive than just finding the 'right'
     one. */
  for (struct PeerInfo *pos = bucket->head;
       NULL != pos;
       pos = pos->next)
  {
    if (off > bucket_size)
      break;   /* We only hold up to #bucket_size peers per bucket */
    off++;
    for (struct Target *tp = pos->t_head;
         NULL != tp;
         tp = tp->next)
      if (NULL == tp->ph)
        tp->ph = GDS_u_hold (tp->u,
                             tp->utarget);
  }
}
 
 
void
GDS_u_connect (void *cls,
               struct GNUNET_DHTU_Target *target,
               const struct GNUNET_PeerIdentity *pid,
               void **ctx)
{
  const struct GNUNET_PeerIdentity *my_identity;
  struct GDS_Underlay *u = cls;
  struct PeerInfo *pi;
  struct PeerBucket *bucket;
  bool do_hold = false;
 
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
  GNUNET_assert (NULL != my_identity);
 
  /* Check for connect to self message */
  if (0 == GNUNET_memcmp (my_identity, pid))
    return;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Connected to peer %s\n",
              GNUNET_i2s (pid));
  pi = GNUNET_CONTAINER_multipeermap_get (all_connected_peers,
                                          pid);
  if (NULL == pi)
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# peers connected",
                              1,
                              GNUNET_NO);
    pi = GNUNET_new (struct PeerInfo);
    pi->id = *pid;
    GNUNET_CRYPTO_hash (pid,
                        sizeof(*pid),
                        &pi->phash);
    pi->peer_bucket = find_bucket (&pi->phash);
    GNUNET_assert ( (pi->peer_bucket >= 0) &&
                    ((unsigned int) pi->peer_bucket < MAX_BUCKETS));
    bucket = &k_buckets[pi->peer_bucket];
    GNUNET_CONTAINER_DLL_insert_tail (bucket->head,
                                      bucket->tail,
                                      pi);
    bucket->peers_size++;
    closest_bucket = GNUNET_MAX (closest_bucket,
                                 (unsigned int) pi->peer_bucket + 1);
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multipeermap_put (all_connected_peers,
                                                      &pi->id,
                                                      pi,
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    if (bucket->peers_size <= bucket_size)
    {
      newly_found_peers++;
      do_hold = true;
    }
    if ( (1 == GNUNET_CONTAINER_multipeermap_size (all_connected_peers)) &&
         (GNUNET_YES != disable_try_connect) )
    {
      /* got a first connection, good time to start with FIND PEER requests... */
      GNUNET_assert (NULL == find_peer_task);
      find_peer_task = GNUNET_SCHEDULER_add_now (&send_find_peer_message,
                                                 NULL);
    }
    /* A pending GET that found no route is sitting on an exponential
       back-off of up to 15 minutes; now that we have somewhere to send
       it, do so immediately.  A PUT that found no route was not sent at
       all and nothing else will ever send it. */
    GDS_CLIENTS_retry_requests ();
    GDS_CLIENTS_retry_puts ();
  }
  {
    struct Target *t;
 
    t = GNUNET_new (struct Target);
    t->u = u;
    t->utarget = target;
    t->pi = pi;
    GNUNET_CONTAINER_DLL_insert (pi->t_head,
                                 pi->t_tail,
                                 t);
    *ctx = t;
 
  }
  if (do_hold)
    update_hold (bucket);
}
 
 
void
GDS_u_disconnect (void *ctx)
{
  struct Target *t = ctx;
  struct PeerInfo *pi;
  struct PeerBucket *bucket;
  bool was_held = false;
 
  /* Check for disconnect from self message (on shutdown) */
  if (NULL == t)
    return;
  pi = t->pi;
  GNUNET_CONTAINER_DLL_remove (pi->t_head,
                               pi->t_tail,
                               t);
  if (NULL != t->ph)
  {
    GDS_u_drop (t->u,
                t->ph);
    t->ph = NULL;
    was_held = true;
  }
  if (t->load > 0)
  {
    t->dropped = true;
    t->pi = NULL;
  }
  else
  {
    GNUNET_free (t);
  }
  if (NULL != pi->t_head)
    return; /* got other connections still */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Disconnected from peer %s\n",
              GNUNET_i2s (&pi->id));
  GNUNET_STATISTICS_update (GDS_stats,
                            "# peers connected",
                            -1,
                            GNUNET_NO);
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_remove (all_connected_peers,
                                                       &pi->id,
                                                       pi));
  if ( (0 == GNUNET_CONTAINER_multipeermap_size (all_connected_peers)) &&
       (GNUNET_YES != disable_try_connect))
  {
    GNUNET_SCHEDULER_cancel (find_peer_task);
    find_peer_task = NULL;
  }
  GNUNET_assert (pi->peer_bucket >= 0);
  bucket = &k_buckets[pi->peer_bucket];
  GNUNET_CONTAINER_DLL_remove (bucket->head,
                               bucket->tail,
                               pi);
  GNUNET_assert (bucket->peers_size > 0);
  bucket->peers_size--;
  if ( (was_held) &&
       (bucket->peers_size >= bucket_size - 1) )
    update_hold (bucket);
  while ( (closest_bucket > 0) &&
          (0 == k_buckets[closest_bucket - 1].peers_size))
    closest_bucket--;
  GNUNET_free (pi->hello);
  GNUNET_free (pi);
}
 
 
static unsigned int
get_forward_count (uint16_t hop_count,
                   uint16_t target_replication)
{
  uint32_t random_value;
  uint32_t forward_count;
  float target_value;
  double rm1;
  double nse = GDS_NSE_get ();
 
  /* Guard the divisor below: an NSE of zero would make target_value
     infinite (or NaN), and the conversion to uint32_t undefined. */
  if (! (nse >= MINIMUM_LOG_NSE))
    nse = MINIMUM_LOG_NSE;
  if (hop_count > nse * 4.0)
  {
    /* forcefully terminate */
    GNUNET_STATISTICS_update (GDS_stats,
                              "# requests TTL-dropped",
                              1,
                              GNUNET_NO);
    return 0;
  }
  if (hop_count > nse * 2.0)
  {
    /* Once we have reached our ideal number of hops, only forward to 1 peer */
    return 1;
  }
  /* bound by system-wide maximum and minimum */
  if (0 == target_replication)
    target_replication = 1; /* 0 is verboten */
  target_replication =
    GNUNET_MIN (GNUNET_DHT_MAXIMUM_REPLICATION_LEVEL,
                target_replication);
  rm1 = target_replication - 1.0;
  target_value =
    1 + (rm1) / (nse + (rm1 * hop_count));
 
  /* Set forward count to floor of target_value */
  forward_count = (uint32_t) target_value;
  /* Subtract forward_count (floor) from target_value (yields value between 0 and 1) */
  target_value = target_value - forward_count;
  random_value = GNUNET_CRYPTO_random_u32 (UINT32_MAX);
  if (random_value < (target_value * UINT32_MAX))
    forward_count++;
  return GNUNET_MIN (forward_count,
                     GNUNET_DHT_MAXIMUM_REPLICATION_LEVEL);
}
 
 
enum GNUNET_GenericReturnValue
GDS_am_closest_peer (const struct GNUNET_HashCode *key,
                     const struct GNUNET_CONTAINER_BloomFilter *bloom)
{
  const struct GNUNET_HashCode *my_identity_hash;
  int delta;
  my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
  GNUNET_assert (NULL != my_identity_hash);
  if (0 == GNUNET_memcmp (my_identity_hash, key))
    return GNUNET_YES;
  for (int bucket_num = find_bucket (key);
       bucket_num < closest_bucket;
       bucket_num++)
  {
    unsigned int count = 0;
    GNUNET_assert (bucket_num >= 0);
    for (struct PeerInfo *pos = k_buckets[bucket_num].head;
         NULL != pos;
         pos = pos->next)
    {
      if (count >= bucket_size)
        break; /* we only consider first #bucket_size entries per bucket */
      count++;
      if ( (NULL != bloom) &&
           (GNUNET_YES ==
            GNUNET_CONTAINER_bloomfilter_test (bloom,
                                               &pos->phash)) )
        continue;               /* Ignore filtered peers */
      /* All peers in this bucket must be closer than us, as
         they mismatch with our PID on the pivotal bit. So
         because an unfiltered peer exists, we are not the
         closest. */
      delta = GNUNET_CRYPTO_hash_xorcmp (&pos->phash,
                                         my_identity_hash,
                                         key);
      switch (delta)
      {
      case -1: /* pos closer */
        return GNUNET_NO;
      case 0: /* identical, impossible! */
        GNUNET_assert (0);
        break;
      case 1: /* I am closer */
        break;
      }
    }
  }
  /* No closer (unfiltered) peers found; we must be the closest! */
  return GNUNET_YES;
}
 
 
static struct PeerInfo *
select_peer (const struct GNUNET_HashCode *key,
             const struct GNUNET_CONTAINER_BloomFilter *bloom,
             uint32_t hops)
{
  if (0 == closest_bucket)
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Peer selection failed",
                              1,
                              GNUNET_NO);
    return NULL; /* we have zero connections */
  }
  if (hops >= GDS_NSE_get ())
  {
    /* greedy selection (closest peer that is not in Bloom filter) */
    struct PeerInfo *chosen = NULL;
    int best_bucket;
    int bucket_offset;
 
    {
      const struct GNUNET_HashCode *my_identity_hash;
      struct GNUNET_HashCode xor;
      my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
      GNUNET_assert (NULL != my_identity_hash);
      GNUNET_CRYPTO_hash_xor (key,
                              my_identity_hash,
                              &xor);
      best_bucket = GNUNET_CRYPTO_hash_count_leading_zeros (&xor);
    }
    if (best_bucket >= closest_bucket)
      bucket_offset = closest_bucket - 1;
    else
      bucket_offset = best_bucket;
    while (-1 != bucket_offset)
    {
      struct PeerBucket *bucket = &k_buckets[bucket_offset];
      unsigned int count = 0;
 
      for (struct PeerInfo *pos = bucket->head;
           NULL != pos;
           pos = pos->next)
      {
        if (count >= bucket_size)
          break; /* we only consider first #bucket_size entries per bucket */
        count++;
        if ( (NULL != bloom) &&
             (GNUNET_YES ==
              GNUNET_CONTAINER_bloomfilter_test (bloom,
                                                 &pos->phash)) )
        {
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                      "Excluded peer `%s' due to BF match in greedy routing for %s\n",
                      GNUNET_i2s (&pos->id),
                      GNUNET_h2s (key));
          continue;
        }
        if (NULL == chosen)
        {
          /* First candidate */
          chosen = pos;
        }
        else
        {
          int delta = GNUNET_CRYPTO_hash_xorcmp (&pos->phash,
                                                 &chosen->phash,
                                                 key);
          switch (delta)
          {
          case -1: /* pos closer */
            chosen = pos;
            break;
          case 0: /* identical, impossible! */
            GNUNET_assert (0);
            break;
          case 1: /* chosen closer */
            break;
          }
        }
      } /* for all (#bucket_size) peers in bucket */
      if (NULL != chosen)
        break;
 
      /* If we chose nothing in first iteration, first go through deeper
         buckets (best chance to find a good match), and if we still found
         nothing, then to shallower buckets.  Terminate on any match in the
         current bucket, as this search order guarantees that it can only get
         worse as we keep going. */
      /* Both upward steps must be bounded by #closest_bucket, and the test
         has to be `>=': stepping from #best_bucket to #best_bucket + 1
         below can jump straight over an `==' test, after which nothing
         stops the offset from running off the end of #k_buckets. */
      if (bucket_offset > best_bucket)
      {
        /* Go through more deeper buckets */
        bucket_offset++;
        if (bucket_offset >= closest_bucket)
        {
          /* Can't go any deeper, if nothing selected,
             go for shallower buckets */
          bucket_offset = best_bucket - 1;
        }
      }
      else
      {
        /* We're either at the 'best_bucket' or already moving
           on to shallower buckets. */
        if (bucket_offset == best_bucket)
        {
          bucket_offset++; /* go for deeper buckets */
          if (bucket_offset >= closest_bucket)
            bucket_offset = best_bucket - 1;
        }
        else
          bucket_offset--; /* go for shallower buckets */
      }
    } /* for applicable buckets (starting at best match) */
    if (NULL == chosen)
    {
      GNUNET_STATISTICS_update (GDS_stats,
                                "# Peer selection failed",
                                1,
                                GNUNET_NO);
      return NULL;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Selected peer `%s' in greedy routing for %s\n",
                GNUNET_i2s (&chosen->id),
                GNUNET_h2s (key));
    return chosen;
  } /* end of 'greedy' peer selection */
 
  /* select "random" peer */
  /* count number of peers that are available and not filtered,
     but limit to at most #bucket_size peers, starting with
     those 'furthest' from us. */
  {
    unsigned int total = 0;
    unsigned int selected;
 
    for (unsigned int bc = 0; bc < closest_bucket; bc++)
    {
      struct PeerBucket *bucket = &k_buckets[bc];
      unsigned int count = 0;
 
      for (struct PeerInfo *pos = bucket->head;
           NULL != pos;
           pos = pos->next)
      {
        count++;
        if (count > bucket_size)
          break; /* limits search to #bucket_size peers per bucket */
        if ( (NULL != bloom) &&
             (GNUNET_YES ==
              GNUNET_CONTAINER_bloomfilter_test (bloom,
                                                 &pos->phash)) )
        {
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                      "Excluded peer `%s' due to BF match in random routing for %s\n",
                      GNUNET_i2s (&pos->id),
                      GNUNET_h2s (key));
          continue;             /* Ignore filtered peers */
        }
        total++;
      } /* for all peers in bucket */
    } /* for all buckets */
    if (0 == total)             /* No peers to select from! */
    {
      GNUNET_STATISTICS_update (GDS_stats,
                                "# Peer selection failed",
                                1,
                                GNUNET_NO);
      return NULL;
    }
 
    /* Now actually choose a peer */
    selected = GNUNET_CRYPTO_random_u32 (total);
    for (unsigned int bc = 0; bc < closest_bucket; bc++)
    {
      unsigned int count = 0;
 
      for (struct PeerInfo *pos = k_buckets[bc].head;
           pos != NULL;
           pos = pos->next)
      {
        count++;
        if (count > bucket_size)
          break; /* limits search to #bucket_size peers per bucket */
 
        if ( (NULL != bloom) &&
             (GNUNET_YES ==
              GNUNET_CONTAINER_bloomfilter_test (bloom,
                                                 &pos->phash)) )
          continue;             /* Ignore bloomfiltered peers */
        if (0 == selected--)
        {
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                      "Selected peer `%s' in random routing for %s\n",
                      GNUNET_i2s (&pos->id),
                      GNUNET_h2s (key));
          return pos;
        }
      } /* for peers in bucket */
    } /* for all buckets */
  } /* random peer selection scope */
  GNUNET_break (0);
  return NULL;
}
 
 
static unsigned int
get_target_peers (const struct GNUNET_HashCode *key,
                  struct GNUNET_CONTAINER_BloomFilter *bloom,
                  uint16_t hop_count,
                  uint16_t target_replication,
                  struct PeerInfo ***targets)
{
  unsigned int target;
  unsigned int off;
  struct PeerInfo **rtargets;
 
  GNUNET_assert (NULL != bloom);
  target = get_forward_count (hop_count,
                              target_replication);
  if (0 == target)
  {
    *targets = NULL;
    return 0;
  }
  rtargets = GNUNET_new_array (target,
                               struct PeerInfo *);
  for (off = 0; off < target; off++)
  {
    struct PeerInfo *nxt;
 
    nxt = select_peer (key,
                       bloom,
                       hop_count);
    if (NULL == nxt)
      break;
    rtargets[off] = nxt;
    /* Exclude the peer we just picked from the next round, or
       select_peer() would hand us the same peer again. */
    GNUNET_break (GNUNET_NO ==
                  GNUNET_CONTAINER_bloomfilter_test (bloom,
                                                     &nxt->phash));
    GNUNET_CONTAINER_bloomfilter_add (bloom,
                                      &nxt->phash);
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Selected %u/%u peers at hop %u for %s (target was %u)\n",
              off,
              GNUNET_CONTAINER_multipeermap_size (all_connected_peers),
              (unsigned int) hop_count,
              GNUNET_h2s (key),
              target);
  if (0 == off)
  {
    GNUNET_free (rtargets);
    *targets = NULL;
    return 0;
  }
  *targets = rtargets;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Forwarding query `%s' to %u peers (goal was %u peers)\n",
              GNUNET_h2s (key),
              off,
              target);
  return off;
}
 
 
static void
hello_check (const struct GNUNET_DATACACHE_Block *bd)
{
  struct GNUNET_HELLO_Parser *b;
 
  if (GNUNET_BLOCK_TYPE_DHT_HELLO != bd->type)
    return;
 
  b = GNUNET_HELLO_parser_from_block (bd->data,
                                      bd->data_size);
  if (GNUNET_YES != disable_try_connect)
  {
    GNUNET_HELLO_parser_iterate (b,
                                 &GDS_try_connect,
                                 NULL);
  }
  GNUNET_HELLO_parser_free (b);
}
 
 
struct GDS_RoutingPutCallbackData
{
  unsigned int hop_count;
  unsigned int target_count;
  struct PeerInfo **targets;
  struct GNUNET_HashCode key;
  unsigned int index;
  unsigned int *queued;
  bool *sent;
  GDS_PutOperationCallback cb;
  void *cb_cls;
};
 
 
static bool
cb_routing_put_message (void *cls,
                        size_t msize,
                        struct PeerPutMessage *ppm)
{
  struct GDS_RoutingPutCallbackData *gds_routing = cls;
  struct PeerInfo *target;
 
  if (NULL == ppm)
  {
    *(gds_routing->queued) = *(gds_routing->queued) + 1;
    if (*(gds_routing->queued) >= gds_routing->target_count)
    {
      if (gds_routing->cb)
        gds_routing->cb (gds_routing->cb_cls,
                         *(gds_routing->sent)
                         ? GNUNET_OK
                         : GNUNET_SYSERR);
 
      GNUNET_free (gds_routing->targets);
      GNUNET_free (gds_routing->queued);
      GNUNET_free (gds_routing->sent);
    }
 
    return true;
  }
 
  target = gds_routing->targets[gds_routing->index];
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Routing PUT for %s after %u hops to %s\n",
              GNUNET_h2s (&(gds_routing->key)),
              (unsigned int) gds_routing->hop_count,
              GNUNET_i2s (&target->id));
  if (do_send (target,
               &ppm->header))
    *(gds_routing->sent) = true;
  *(gds_routing->queued) = *(gds_routing->queued) + 1;
 
  if (*(gds_routing->queued) >= gds_routing->target_count)
  {
    /* Report success only if some underlay actually took the message:
       otherwise nothing stores this block and the caller has to keep
       it and try again. */
    if (gds_routing->cb)
      gds_routing->cb (gds_routing->cb_cls,
                       *(gds_routing->sent)
                       ? GNUNET_OK
                       : GNUNET_NO);
 
    GNUNET_free (gds_routing->targets);
    GNUNET_STATISTICS_update (GDS_stats,
                              "# PUT messages queued for transmission",
                              gds_routing->target_count,
                              GNUNET_NO);
    GNUNET_free (gds_routing->queued);
    GNUNET_free (gds_routing->sent);
  }
 
  return true;
}
 
 
void
GDS_NEIGHBOURS_handle_put (const struct GNUNET_DATACACHE_Block *bd,
                           uint16_t desired_replication_level,
                           uint16_t hop_count,
                           struct GNUNET_CONTAINER_BloomFilter *bf,
                           GDS_PutOperationCallback cb,
                           void *cb_cls)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_HashCode *my_identity_hash;
  struct GDS_RoutingPutCallbackData gds_routing;
  size_t msize;
  enum GNUNET_DHT_RouteOption ro = bd->ro;
  unsigned int put_path_length = bd->put_path_length;
  const struct GNUNET_DHT_PathElement *put_path = bd->put_path;
  bool truncated = (0 != (bd->ro & GNUNET_DHT_RO_TRUNCATED));
  const struct GNUNET_PeerIdentity *trunc_peer
    = truncated
    ? &bd->trunc_peer
    : NULL;
  struct GNUNET_PeerIdentity trunc_peer_out;
  enum GNUNET_GenericReturnValue ret;
 
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
  my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
  GNUNET_assert (NULL != my_identity);
 
  ret = GDS_helper_put_message_get_size (&msize,
                                         my_identity,
                                         bd->ro, &ro,
                                         bd->expiration_time,
                                         bd->data, bd->data_size,
                                         put_path, put_path_length,
                                         &put_path_length,
                                         trunc_peer,
                                         &trunc_peer_out,
                                         &truncated);
  if (truncated)
    trunc_peer = &trunc_peer_out;
  /* Path may have been truncated by the call above */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Adding myself (%s) to PUT bloomfilter for %s with RO(%s/%s)\n",
              GNUNET_i2s (my_identity),
              GNUNET_h2s (&bd->key),
              (bd->ro & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE) ? "x" : "-",
              (bd->ro & GNUNET_DHT_RO_RECORD_ROUTE) ? "R" : "-");
 
  /* if we got a HELLO, consider it for our own routing table */
  hello_check (bd);
  GNUNET_assert ((NULL != bf) && (NULL != my_identity_hash));
  GNUNET_CONTAINER_bloomfilter_add (bf, my_identity_hash);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# PUT requests routed",
                            1,
                            GNUNET_NO);
  if (GNUNET_OK != ret)
  {
    if (cb)
      cb (cb_cls, ret);
    return;
  }
  gds_routing.target_count
    = get_target_peers (&bd->key,
                        bf,
                        hop_count,
                        desired_replication_level,
                        &(gds_routing.targets));
  if (0 == gds_routing.target_count)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Routing PUT for %s terminates after %u hops at %s\n",
                GNUNET_h2s (&bd->key),
                (unsigned int) hop_count,
                GNUNET_i2s (my_identity));
    if (cb)
      cb (cb_cls, GNUNET_NO);
    if (gds_routing.targets)
      GNUNET_free (gds_routing.targets);
    return;
  }
  GNUNET_memcpy (&(gds_routing.key), &(bd->key),
                 sizeof (gds_routing.key));
  gds_routing.hop_count = hop_count;
  /* targets were added to @a bf by get_target_peers() */
 
  gds_routing.queued = GNUNET_new (unsigned int);
  *(gds_routing.queued) = 0;
  gds_routing.sent = GNUNET_new (bool);
  *(gds_routing.sent) = false;
 
  gds_routing.cb = cb;
  gds_routing.cb_cls = cb_cls;
 
  for (unsigned int i = 0; i < gds_routing.target_count; i++)
  {
    struct PeerInfo *target = gds_routing.targets[i];
    struct PeerPutMessage *ppm;
    char buf[msize] GNUNET_ALIGN;
 
    gds_routing.index = i;
 
    ppm = (struct PeerPutMessage *) buf;
    GDS_helper_make_put_message (ppm, msize,
                                 GNUNET_PILS_get_private_key (GDS_pils),
                                 &target->id,
                                 &target->phash,
                                 bf,
                                 &bd->key,
                                 ro,
                                 bd->type,
                                 bd->expiration_time,
                                 bd->data, bd->data_size,
                                 put_path, put_path_length,
                                 hop_count,
                                 desired_replication_level,
                                 trunc_peer,
                                 &cb_routing_put_message,
                                 sizeof (gds_routing),
                                 &gds_routing);
  }
}
 
 
enum GNUNET_GenericReturnValue
GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
                           enum GNUNET_DHT_RouteOption options,
                           uint16_t desired_replication_level,
                           uint16_t hop_count,
                           const struct GNUNET_HashCode *key,
                           const void *xquery,
                           size_t xquery_size,
                           struct GNUNET_BLOCK_Group *bg,
                           struct GNUNET_CONTAINER_BloomFilter *peer_bf)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_HashCode *my_identity_hash;
  unsigned int target_count;
  struct PeerInfo **targets;
  size_t msize;
  size_t result_filter_size;
  void *result_filter;
  bool sent = false;
 
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
  my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
 
  if (NULL == my_identity_hash)
    return GNUNET_NO;
 
  GNUNET_assert (NULL != peer_bf);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# GET requests routed",
                            1,
                            GNUNET_NO);
  target_count = get_target_peers (key,
                                   peer_bf,
                                   hop_count,
                                   desired_replication_level,
                                   &targets);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Adding myself (%s) to GET bloomfilter for %s with RO(%s/%s)\n",
              GNUNET_i2s (my_identity),
              GNUNET_h2s (key),
              (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE) ? "x" : "-",
              (options & GNUNET_DHT_RO_RECORD_ROUTE) ? "R" : "-");
  GNUNET_assert (NULL != my_identity_hash);
  GNUNET_CONTAINER_bloomfilter_add (peer_bf, my_identity_hash);
  if (0 == target_count)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Routing GET for %s terminates after %u hops at %s\n",
                GNUNET_h2s (key),
                (unsigned int) hop_count,
                GNUNET_i2s (my_identity));
    return GNUNET_NO;
  }
  if (GNUNET_OK !=
      GNUNET_BLOCK_group_serialize (bg,
                                    &result_filter,
                                    &result_filter_size))
  {
    result_filter = NULL;
    result_filter_size = 0;
  }
  msize = xquery_size + result_filter_size;
  if (msize + sizeof(struct PeerGetMessage) >= GNUNET_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    GNUNET_free (result_filter);
    GNUNET_free (targets);
    return GNUNET_NO;
  }
  /* targets were added to @a peer_bf by get_target_peers() */
  /* forward request */
  for (unsigned int i = 0; i < target_count; i++)
  {
    struct PeerInfo *target = targets[i];
    struct PeerGetMessage *pgm;
    char buf[sizeof (*pgm) + msize] GNUNET_ALIGN;
    char *rf;
 
    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));
    pgm = (struct PeerGetMessage *) buf;
    pgm->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_P2P_GET);
    pgm->header.size = htons (sizeof (buf));
    pgm->type = htonl (type);
    pgm->options = htons (options);
    pgm->hop_count = htons (hop_count + 1);
    pgm->desired_replication_level = htons (desired_replication_level);
    pgm->result_filter_size = htons ((uint16_t) result_filter_size);
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_bloomfilter_get_raw_data (peer_bf,
                                                              pgm->bloomfilter,
                                                              DHT_BLOOM_SIZE));
    pgm->key = *key;
    rf = (char *) &pgm[1];
    GNUNET_memcpy (rf,
                   result_filter,
                   result_filter_size);
    GNUNET_memcpy (&rf[result_filter_size],
                   xquery,
                   xquery_size);
    if (do_send (target,
                 &pgm->header))
      sent = true;
  }
  GNUNET_STATISTICS_update (GDS_stats,
                            "# GET messages queued for transmission",
                            target_count,
                            GNUNET_NO);
  GNUNET_free (targets);
  GNUNET_free (result_filter);
  /* Having picked a target is not the same as having sent to one: if
     every underlay dropped the message, the request was not routed and
     must go on the fast retry, not on the regular back-off. */
  return sent ? GNUNET_OK : GNUNET_NO;
}
 
 
struct PeerInfo *
GDS_NEIGHBOURS_lookup_peer (const struct GNUNET_PeerIdentity *target)
{
  return GNUNET_CONTAINER_multipeermap_get (all_connected_peers,
                                            target);
}
 
 
struct GDS_NeighboursReply
{
  struct PeerInfo *pi;
  struct PeerResultMessage *prm;
  struct GNUNET_DHT_PathElement *paths;
  struct GNUNET_DATACACHE_Block bd;
  void *block_data;
  struct GNUNET_DHT_PathElement *put_path;
  struct GNUNET_PeerIdentity trunc_peer_id;
  bool trunc_peer_is_null;
  char *buf;
 
  GNUNET_SCHEDULER_TaskCallback cb;
  void *cb_cls;
};
 
 
static void
cleanup_neighbours_reply (struct GDS_NeighboursReply *reply)
{
  if (reply->block_data)
    GNUNET_free (reply->block_data);
  if ((reply->bd.put_path_length > 0) && (reply->put_path))
    GNUNET_free (reply->put_path);
  if (reply->buf)
    GNUNET_free (reply->buf);
}
 
 
static void
safe_neighbours_callback (void *cls,
                          GNUNET_SCHEDULER_TaskCallback cb,
                          bool success)
{
  GNUNET_break (success);
  if (cb)
    cb (cls);
}
 
 
static bool
cb_path_signed (void *cls,
                const struct GNUNET_CRYPTO_EddsaSignature *sig)
{
  struct GDS_NeighboursReply *reply = cls;
  struct PeerResultMessage *prm = reply->prm;
  struct GNUNET_DHT_PathElement *paths = reply->paths;
  unsigned int ppl = ntohs (prm->put_path_length);
  unsigned int get_path_length = ntohs (prm->get_path_length);
  void *tgt = &paths[get_path_length + ppl];
  void *data;
 
  if (! sig)
  {
    cleanup_neighbours_reply (reply);
    safe_neighbours_callback (reply->cb_cls, reply->cb, false);
    return true;
  }
 
  memcpy (tgt,
          sig,
          sizeof (*sig));
  data = tgt + sizeof (*sig);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Signing GET PATH %u/%u of %s => %s\n",
              ppl,
              get_path_length,
              GNUNET_h2s (&prm->key),
              GNUNET_B2S (sig));
#if SANITY_CHECKS > 1
  {
    const struct GNUNET_PeerIdentity *my_identity;
    struct GNUNET_DHT_PathElement xpaths[get_path_length + 1];
    const struct GNUNET_PeerIdentity *trunc_peer = reply->trunc_peer_is_null?
                                                   NULL : &reply->trunc_peer_id;
 
    my_identity = GNUNET_PILS_get_identity (GDS_pils);
    GNUNET_assert (NULL != my_identity);
 
    memcpy (xpaths,
            &paths[ppl],
            get_path_length * sizeof (struct GNUNET_DHT_PathElement));
    xpaths[get_path_length].sig = *sig;
    xpaths[get_path_length].pred = *my_identity;
    if (0 !=
        GNUNET_DHT_verify_path (reply->bd.data,
                                reply->bd.data_size,
                                reply->bd.expiration_time,
                                trunc_peer,
                                paths,
                                ppl,
                                xpaths,
                                get_path_length + 1,
                                &reply->pi->id))
    {
      GNUNET_break (0);
      cleanup_neighbours_reply (reply);
      safe_neighbours_callback (reply->cb_cls, reply->cb, false);
      return true;
    }
  }
#endif
  GNUNET_memcpy (data,
                 reply->bd.data,
                 reply->bd.data_size);
  do_send (reply->pi,
           &prm->header);
  cleanup_neighbours_reply (reply);
  safe_neighbours_callback (reply->cb_cls, reply->cb, true);
  return true;
}
 
 
void
GDS_NEIGHBOURS_handle_reply (struct PeerInfo *pi,
                             const struct GNUNET_DATACACHE_Block *bd,
                             const struct GNUNET_HashCode *query_hash,
                             unsigned int get_path_length,
                             const struct GNUNET_DHT_PathElement *get_path,
                             GNUNET_SCHEDULER_TaskCallback cb,
                             void *cb_cls)
{
  struct GNUNET_DHT_PathElement *paths;
  size_t msize;
  unsigned int ppl = bd->put_path_length;
  const struct GNUNET_DHT_PathElement *put_path = bd->put_path;
  enum GNUNET_DHT_RouteOption ro = bd->ro;
  bool truncated = (0 != (ro & GNUNET_DHT_RO_TRUNCATED));
  const struct GNUNET_PeerIdentity *trunc_peer
    = truncated
    ? &bd->trunc_peer
    : NULL;
  bool tracking = (0 != (ro & GNUNET_DHT_RO_RECORD_ROUTE));
#if SANITY_CHECKS > 1
  const struct GNUNET_PeerIdentity *my_identity;
  unsigned int failure_offset;
 
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
  GNUNET_assert (NULL != my_identity);
 
  failure_offset
    = GNUNET_DHT_verify_path (bd->data,
                              bd->data_size,
                              bd->expiration_time,
                              trunc_peer,
                              put_path,
                              ppl,
                              get_path,
                              get_path_length,
                              my_identity);
  if (0 != failure_offset)
  {
    GNUNET_assert (failure_offset <= ppl + get_path_length);
    GNUNET_break_op (0);
    if (failure_offset < ppl)
    {
      trunc_peer = &put_path[failure_offset - 1].pred;
      put_path += failure_offset;
      ppl -= failure_offset;
      truncated = true;
      ro |= GNUNET_DHT_RO_TRUNCATED;
    }
    else
    {
      failure_offset -= ppl;
      if (0 == failure_offset)
        trunc_peer = &put_path[ppl - 1].pred;
      else
        trunc_peer = &get_path[failure_offset - 1].pred;
      ppl = 0;
      put_path = NULL;
      truncated = true;
      ro |= GNUNET_DHT_RO_TRUNCATED;
      get_path += failure_offset;
      get_path_length -= failure_offset;
    }
  }
#endif
  msize = bd->data_size + sizeof (struct PeerResultMessage);
  if (msize > GNUNET_MAX_MESSAGE_SIZE)
  {
    GNUNET_break_op (0);
    safe_neighbours_callback (cb_cls, cb, false);
    return;
  }
  if (truncated)
    msize += sizeof (struct GNUNET_PeerIdentity);
  if (tracking)
    msize += sizeof (struct GNUNET_CRYPTO_EddsaSignature);
  if (msize < bd->data_size)
  {
    GNUNET_break_op (0);
    safe_neighbours_callback (cb_cls, cb, false);
    return;
  }
  if ( (GNUNET_MAX_MESSAGE_SIZE - msize)
       / sizeof(struct GNUNET_DHT_PathElement)
       < (get_path_length + ppl) )
  {
    get_path_length = 0;
    ppl = 0;
  }
  if ( (get_path_length > UINT16_MAX) ||
       (ppl > UINT16_MAX) )
  {
    GNUNET_break (0);
    get_path_length = 0;
    ppl = 0;
  }
  msize += (get_path_length + ppl)
           * sizeof(struct GNUNET_DHT_PathElement);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Forwarding reply for key %s to peer %s\n",
              GNUNET_h2s (query_hash),
              GNUNET_i2s (&pi->id));
  GNUNET_STATISTICS_update (GDS_stats,
                            "# RESULT messages queued for transmission",
                            1,
                            GNUNET_NO);
  {
    struct PeerResultMessage *prm;
    char buf[msize] GNUNET_ALIGN;
 
    prm = (struct PeerResultMessage *) buf;
    prm->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT);
    prm->header.size = htons (sizeof (buf));
    prm->type = htonl ((uint32_t) bd->type);
    prm->reserved = htons (0);
    prm->options = htons ((uint16_t) ro);
    prm->put_path_length = htons ((uint16_t) ppl);
    prm->get_path_length = htons ((uint16_t) get_path_length);
    prm->expiration_time = GNUNET_TIME_absolute_hton (bd->expiration_time);
    prm->key = *query_hash;
    if (truncated)
    {
      void *tgt = &prm[1];
 
      GNUNET_memcpy (tgt,
                     trunc_peer,
                     sizeof (struct GNUNET_PeerIdentity));
      paths = (struct GNUNET_DHT_PathElement *)
              (tgt + sizeof (struct GNUNET_PeerIdentity));
    }
    else
    {
      paths = (struct GNUNET_DHT_PathElement *) &prm[1];
    }
    if (NULL != put_path)
    {
      GNUNET_memcpy (paths,
                     put_path,
                     ppl * sizeof(struct GNUNET_DHT_PathElement));
    }
    else
    {
      GNUNET_assert (0 == ppl);
    }
    if (NULL != get_path)
    {
      GNUNET_memcpy (&paths[ppl],
                     get_path,
                     get_path_length * sizeof(struct GNUNET_DHT_PathElement));
    }
    else
    {
      GNUNET_assert (0 == get_path_length);
    }
    if (tracking)
    {
      struct GDS_NeighboursReply reply;
      const struct GNUNET_PeerIdentity *pred;
 
      reply.pi = pi;
      GNUNET_memcpy (&reply.bd, bd, sizeof (reply.bd));
      reply.block_data = GNUNET_memdup (bd->data, bd->data_size);
      reply.put_path = GNUNET_memdup (bd->put_path,
                                      sizeof (struct GNUNET_DHT_PathElement)
                                      * bd->put_path_length);
 
      reply.bd.data = reply.block_data;
      reply.bd.put_path = reply.put_path;
 
      reply.buf = GNUNET_memdup (buf, msize);
      reply.prm = (struct PeerResultMessage*) reply.buf;
      reply.paths = (struct GNUNET_DHT_PathElement*)
                    (reply.buf + ((const char *) paths - buf));
 
      if (trunc_peer)
      {
        reply.trunc_peer_is_null = false;
        GNUNET_memcpy (&reply.trunc_peer_id, trunc_peer,
                       sizeof (reply.trunc_peer_id));
      }
      else
      {
        reply.trunc_peer_is_null = true;
      }
 
      reply.cb = cb;
      reply.cb_cls = cb_cls;
 
      if (ppl + get_path_length > 0)
        pred = &paths[ppl + get_path_length - 1].pred;
      else if (truncated)
        pred = trunc_peer;
      else
        pred = NULL; /* we are first! */
      /* Note that the last signature in 'paths' was not initialized before,
         so this is crucial to avoid sending garbage. */
      GDS_helper_sign_path (bd->data,
                            bd->data_size,
                            GNUNET_PILS_get_private_key (GDS_pils),
                            bd->expiration_time,
                            pred,
                            &pi->id,
                            &cb_path_signed,
                            sizeof (reply),
                            &reply);
    }
    else
    {
      void *data;
      data = &prm[1];
      GNUNET_memcpy (data,
                     bd->data,
                     bd->data_size);
      do_send (pi,
               &prm->header);
      safe_neighbours_callback (cb_cls, cb, true);
      return;
    }
  }
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_p2p_put (void *cls,
                   const struct PeerPutMessage *put)
{
  enum GNUNET_DHT_RouteOption ro = ntohs (put->options);
  bool truncated = (0 != (ro & GNUNET_DHT_RO_TRUNCATED));
  bool has_path = (0 != (ro & GNUNET_DHT_RO_RECORD_ROUTE));
  uint16_t msize = ntohs (put->header.size);
  uint16_t putlen = ntohs (put->put_path_length);
  size_t xsize = (has_path
                  ? sizeof (struct GNUNET_CRYPTO_EddsaSignature)
                  : 0)
                 + (truncated
                    ? sizeof (struct GNUNET_PeerIdentity)
                    : 0);
  size_t var_meta_size
    = putlen * sizeof(struct GNUNET_DHT_PathElement)
      + xsize;
 
  (void) cls;
  if ( (msize <
        sizeof (struct PeerPutMessage) + var_meta_size) ||
       (putlen >
        (GNUNET_MAX_MESSAGE_SIZE
         - sizeof (struct PeerPutMessage)
         - xsize)
        / sizeof(struct GNUNET_DHT_PathElement)) )
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  if (GNUNET_BLOCK_TYPE_ANY == htonl (put->type))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
struct ForwardedDHTPut
{
  struct GNUNET_DATACACHE_Block block;
 
  struct GNUNET_DHT_PathElement *put_path;
  void *data;
 
  uint32_t hop_count;
  uint32_t desired_replication_level;
};
 
 
static void
cb_forwarded_dht_p2p_put (void *cls,
                          enum GNUNET_GenericReturnValue forwarded)
{
  struct ForwardedDHTPut *put = cls;
 
  /* notify monitoring clients; we are the last hop exactly when we did
     not manage to forward, as in the GET case */
  put->block.ro |= ((GNUNET_OK == forwarded)
            ? 0
            : GNUNET_DHT_RO_LAST_HOP);
  GDS_CLIENTS_process_put (&put->block,
                           put->hop_count,
                           put->desired_replication_level);
 
  if (put->put_path)
    GNUNET_free (put->put_path);
  GNUNET_free (put->data);
  GNUNET_free (put);
}
 
 
static void
handle_dht_p2p_put (void *cls,
                    const struct PeerPutMessage *put)
{
  struct Target *t = cls;
  struct PeerInfo *peer = t->pi;
  enum GNUNET_DHT_RouteOption ro = ntohs (put->options);
  bool truncated = (0 != (ro & GNUNET_DHT_RO_TRUNCATED));
  bool has_path = (0 != (ro & GNUNET_DHT_RO_RECORD_ROUTE));
  uint16_t msize = ntohs (put->header.size);
  uint16_t putlen = ntohs (put->put_path_length);
  const struct GNUNET_PeerIdentity *trunc_peer
    = truncated
    ? (const struct GNUNET_PeerIdentity *) &put[1]
    : NULL;
  const struct GNUNET_DHT_PathElement *put_path
    = truncated
    ? (const struct GNUNET_DHT_PathElement *) &trunc_peer[1]
    : (const struct GNUNET_DHT_PathElement *) &put[1];
  const struct GNUNET_CRYPTO_EddsaSignature *last_sig
    = has_path
    ? (const struct GNUNET_CRYPTO_EddsaSignature *) &put_path[putlen]
    : NULL;
  const char *data
    = has_path
    ? (const char *) &last_sig[1]
    : (const char *) &put_path[putlen];
  size_t var_meta_size
    = putlen * sizeof(struct GNUNET_DHT_PathElement)
      + (has_path ? sizeof (*last_sig) : 0)
      + (truncated ? sizeof (*trunc_peer) : 0);
  struct GNUNET_DATACACHE_Block bd = {
    .key = put->key,
    .expiration_time = GNUNET_TIME_absolute_ntoh (put->expiration_time),
    .type = ntohl (put->type),
    .ro = ro,
    .data_size = msize - sizeof(*put) - var_meta_size,
    .data = data
  };
 
  if (NULL != trunc_peer)
    bd.trunc_peer = *trunc_peer;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "PUT for `%s' from %s with RO (%s/%s)\n",
              GNUNET_h2s (&put->key),
              GNUNET_i2s (&peer->id),
              (bd.ro & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE) ? "x" : "-",
              has_path ? "R" : "-");
  if (GNUNET_TIME_absolute_is_past (bd.expiration_time))
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Expired PUTs discarded",
                              1,
                              GNUNET_NO);
    return;
  }
  {
    /* Only call 'check_block' if that keeps our CPU load (from
       the cryptography) below 50% on average */
    static struct GNUNET_TIME_Relative avg_latency;
    static struct GNUNET_TIME_Absolute next_time;
 
    if (GNUNET_TIME_absolute_is_past (next_time))
    {
      struct GNUNET_TIME_Absolute now
        = GNUNET_TIME_absolute_get ();
      struct GNUNET_TIME_Relative latency;
      struct GNUNET_TIME_Relative delta;
 
      if (GNUNET_NO ==
          GNUNET_BLOCK_check_block (GDS_block_context,
                                    bd.type,
                                    bd.data,
                                    bd.data_size))
      {
        GNUNET_break_op (0);
        return;
      }
      latency = GNUNET_TIME_absolute_get_duration (now);
      /* Use *moving average* to estimate check_block latency */
      avg_latency
        = GNUNET_TIME_relative_divide (
            GNUNET_TIME_relative_add (
              GNUNET_TIME_relative_multiply (avg_latency,
                                             7),
              latency),
            8);
      /* average delay = 50% of avg_latency => 50% CPU load from crypto (at most) */
      delta.rel_value_us
        = GNUNET_CRYPTO_random_u64 (avg_latency.rel_value_us > 0
                                        ? avg_latency.rel_value_us
                                        : 1LLU);
      next_time = GNUNET_TIME_relative_to_absolute (delta);
    }
  }
  if (! has_path)
    putlen = 0;
  GNUNET_STATISTICS_update (GDS_stats,
                            "# P2P PUT requests received",
                            1,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# P2P PUT bytes received",
                            msize,
                            GNUNET_NO);
  {
    struct GNUNET_HashCode test_key;
    enum GNUNET_GenericReturnValue ret;
 
    ret = GNUNET_BLOCK_get_key (GDS_block_context,
                                bd.type,
                                bd.data,
                                bd.data_size,
                                &test_key);
    switch (ret)
    {
    case GNUNET_YES:
      if (0 != GNUNET_memcmp (&test_key,
                              &bd.key))
      {
        GNUNET_break_op (0);
        return;
      }
      break;
    case GNUNET_NO:
      /* cannot verify, good luck */
      break;
    case GNUNET_SYSERR:
      /* block type not supported, good luck */
      break;
    }
  }
 
  {
    struct GNUNET_CONTAINER_BloomFilter *bf;
    struct GNUNET_DHT_PathElement pp[putlen + 1];
 
    bf = GNUNET_CONTAINER_bloomfilter_init (put->bloomfilter,
                                            DHT_BLOOM_SIZE,
                                            GNUNET_CONSTANTS_BLOOMFILTER_K);
    GNUNET_break_op (GNUNET_YES ==
                     GNUNET_CONTAINER_bloomfilter_test (bf,
                                                        &peer->phash));
    /* extend 'put path' by sender */
    bd.put_path = pp;
    bd.put_path_length = putlen + 1;
    if (has_path)
    {
      unsigned int failure_offset;
 
      GNUNET_memcpy (pp,
                     put_path,
                     putlen * sizeof(struct GNUNET_DHT_PathElement));
      pp[putlen].pred = peer->id;
      pp[putlen].sig = *last_sig;
#if SANITY_CHECKS
      {
        const struct GNUNET_PeerIdentity *my_identity;
        my_identity = GNUNET_PILS_get_identity (GDS_pils);
        GNUNET_assert (NULL != my_identity);
        /* TODO: might want to eventually implement probabilistic
          load-based path verification, but for now it is all or nothing */
        failure_offset
          = GNUNET_DHT_verify_path (bd.data,
                                    bd.data_size,
                                    bd.expiration_time,
                                    trunc_peer,
                                    pp,
                                    putlen + 1,
                                    NULL, 0,   /* get_path */
                                    my_identity);
      }
#else
      failure_offset = 0;
#endif
      if (0 != failure_offset)
      {
        GNUNET_break_op (0);
        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                    "Recorded put path invalid at offset %u, truncating\n",
                    failure_offset);
        GNUNET_assert (failure_offset <= putlen + 1);
        bd.put_path = &pp[failure_offset];
        bd.put_path_length = (putlen + 1) - failure_offset;
        bd.ro |= GNUNET_DHT_RO_TRUNCATED;
        bd.trunc_peer = pp[failure_offset - 1].pred;
      }
    }
    else
    {
      bd.put_path_length = 0;
    }
 
    /* give to local clients */
    GNUNET_break (GDS_CLIENTS_handle_reply (&bd,
                                            &bd.key,
                                            0, NULL /* get path */));
 
    /* store locally */
    if ( (0 != (bd.ro & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
         (GDS_am_closest_peer (&put->key,
                               bf)) )
      GDS_DATACACHE_handle_put (&bd);
 
    {
      struct ForwardedDHTPut *forward = GNUNET_new (struct ForwardedDHTPut);
      GNUNET_memcpy (&forward->block, &bd, sizeof (bd));
 
      if (bd.put_path_length > 0)
      {
        forward->put_path = GNUNET_memdup (
          bd.put_path,
          sizeof (struct GNUNET_DHT_PathElement) * bd.put_path_length);
        forward->block.put_path = forward->put_path;
      }
 
      forward->data = GNUNET_memdup (bd.data, bd.data_size);
      forward->block.data = forward->data;
 
      forward->desired_replication_level = ntohs (put->desired_replication_level
                                                  );
      forward->hop_count = ntohs (put->hop_count);
 
      /* route to other peers */
      GDS_NEIGHBOURS_handle_put (&forward->block,
                                 forward->desired_replication_level,
                                 forward->hop_count,
                                 bf,
                                 &cb_forwarded_dht_p2p_put,
                                 forward);
    }
    GNUNET_CONTAINER_bloomfilter_free (bf);
  }
}
 
 
struct BlockCls
{
  struct PeerInfo *pi;
  const struct GNUNET_HashCode *query_hash;
  struct GNUNET_BLOCK_Group *bg;
};
 
 
static void
handle_find_my_hello (struct PeerInfo *pi,
                      const struct GNUNET_HashCode *query_hash,
                      struct GNUNET_BLOCK_Group *bg,
                      GNUNET_SCHEDULER_TaskCallback cb,
                      void *cb_cls)
{
  const struct GNUNET_HashCode *my_identity_hash;
  const struct GNUNET_PeerIdentity *my_identity;
  struct GNUNET_TIME_Absolute block_expiration;
  size_t block_size;
  void *block;
 
  my_identity_hash = GNUNET_PILS_get_identity_hash (GDS_pils);
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Handle finding my own HELLO %s\n",
              GNUNET_h2s (my_identity_hash));
  if (NULL == GDS_my_hello)
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# FIND PEER requests ignored due to lack of HELLO",
                              1,
                              GNUNET_NO);
    if (cb)
      cb (cb_cls);
    return;
  }
 
  if (GNUNET_SYSERR == GNUNET_HELLO_dht_msg_to_block (GDS_my_hello,
                                                      my_identity,
                                                      &block,
                                                      &block_size,
                                                      &block_expiration))
  {
    if (cb)
      cb (cb_cls);
    return;
  }
 
  if (GNUNET_BLOCK_REPLY_OK_MORE ==
      GNUNET_BLOCK_check_reply (GDS_block_context,
                                GNUNET_BLOCK_TYPE_DHT_HELLO,
                                bg,
                                my_identity_hash,
                                NULL, 0,
                                block,
                                block_size))
  {
    struct GNUNET_DATACACHE_Block bd = {
      .type = GNUNET_BLOCK_TYPE_DHT_HELLO,
      .expiration_time
        = GNUNET_TIME_relative_to_absolute (
            GNUNET_HELLO_ADDRESS_EXPIRATION),
      .key = *my_identity_hash,
      .data = block,
      .data_size = block_size
    };
 
    GDS_NEIGHBOURS_handle_reply (pi,
                                 &bd,
                                 query_hash,
                                 0, NULL /* get path */,
                                 cb,
                                 cb_cls);
  }
  else
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# FIND PEER requests ignored due to Bloomfilter",
                              1,
                              GNUNET_NO);
    if (cb)
      cb (cb_cls);
  }
 
  GNUNET_free (block);
}
 
 
static void
handle_find_local_hello (struct PeerInfo *pi,
                         const struct GNUNET_HashCode *query_hash,
                         struct GNUNET_BLOCK_Group *bg,
                         GNUNET_SCHEDULER_TaskCallback cb,
                         void *cb_cls)
{
  /* Force non-random selection by hop count */
  struct PeerInfo *peer;
 
  peer = select_peer (query_hash,
                      NULL,
                      GDS_NSE_get () + 1);
  if (NULL == peer)
  {
    /* no neighbour at all, nothing to answer with */
    if (cb)
      cb (cb_cls);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Handle finding local HELLO %s\n",
              GNUNET_h2s (&peer->phash));
  if ( (NULL != peer->hello) &&
       (! GNUNET_TIME_absolute_is_past (peer->hello_expiration)) &&
       (GNUNET_BLOCK_REPLY_OK_MORE ==
        GNUNET_BLOCK_check_reply (
          GDS_block_context,
          GNUNET_BLOCK_TYPE_DHT_HELLO,
          bg,
          &peer->phash,
          NULL, 0,        /* xquery */
          peer->hello,
          peer->hello_size)) )
  {
    struct GNUNET_DATACACHE_Block bd = {
      .type = GNUNET_BLOCK_TYPE_DHT_HELLO,
      .expiration_time = peer->hello_expiration,
      .key = peer->phash,
      .data = peer->hello,
      .data_size = peer->hello_size
    };
 
    GDS_NEIGHBOURS_handle_reply (pi,
                                 &bd,
                                 query_hash,
                                 0, NULL /* get path */,
                                 cb,
                                 cb_cls);
  }
  else if (cb)
    cb (cb_cls);
}
 
 
static void
handle_local_result (void *cls,
                     const struct GNUNET_DATACACHE_Block *bd)
{
  struct PeerInfo *peer = cls;
 
  GDS_NEIGHBOURS_handle_reply (peer,
                               bd,
                               &bd->key,
                               0, NULL /* get path */,
                               NULL,
                               NULL);
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_p2p_get (void *cls,
                   const struct PeerGetMessage *get)
{
  uint16_t msize = ntohs (get->header.size);
  uint16_t result_filter_size = ntohs (get->result_filter_size);
 
  (void) cls;
  if (msize < sizeof(*get) + result_filter_size)
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
struct HandleCallbackGet
{
  struct Target *t;
  struct PeerGetMessage *get;
  struct GNUNET_CONTAINER_BloomFilter *peer_bf;
  struct GNUNET_BLOCK_Group *bg;
  enum GNUNET_BLOCK_ReplyEvaluationResult eval;
};
 
 
static void
cb_handle_dht_p2p_get_local_result (void *cls)
{
  struct HandleCallbackGet *handle = cls;
  enum GNUNET_DHT_RouteOption options = ntohs (handle->get->options);
  enum GNUNET_BLOCK_Type type = ntohl (handle->get->type);
  const void *result_filter = (const void *) &handle->get[1];
  uint16_t msize = ntohs (handle->get->header.size);
  uint16_t result_filter_size = ntohs (handle->get->result_filter_size);
  const void *xquery = result_filter + result_filter_size;
  size_t xquery_size = msize - sizeof (*handle->get) - result_filter_size;
 
  /* remember request for routing replies
      TODO: why should we do this if GNUNET_BLOCK_REPLY_OK_LAST == eval?
  */
  GDS_ROUTING_add (&handle->t->pi->id,
                   type,
                   handle->bg, /* bg now owned by routing, but valid at least until end of this function! */
                   options,
                   &handle->get->key,
                   xquery,
                   xquery_size);
 
  /* P2P forwarding */
  {
    bool forwarded = false;
    uint16_t desired_replication_level = ntohs (
      handle->get->desired_replication_level);
    uint16_t hop_count = ntohs (handle->get->hop_count);
 
    if (handle->eval != GNUNET_BLOCK_REPLY_OK_LAST)
      forwarded = (GNUNET_OK ==
                   GDS_NEIGHBOURS_handle_get (type,
                                              options,
                                              desired_replication_level,
                                              hop_count,
                                              &handle->get->key,
                                              xquery,
                                              xquery_size,
                                              handle->bg,
                                              handle->peer_bf));
    GDS_CLIENTS_process_get (
      options
      | (forwarded
        ? 0
        : GNUNET_DHT_RO_LAST_HOP),
      type,
      hop_count,
      desired_replication_level,
      &handle->get->key);
  }
  /* clean up; note that 'bg' is owned by routing now! */
  GNUNET_CONTAINER_bloomfilter_free (handle->peer_bf);
 
  GNUNET_free (handle->get);
  GNUNET_free (handle);
}
 
 
static void
cb_handle_dht_p2p_get_local_hello (void *cls)
{
  struct HandleCallbackGet *handle = cls;
  enum GNUNET_BLOCK_Type type = ntohl (handle->get->type);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Handle getting local HELLO %s of type %u\n",
              GNUNET_h2s (&handle->get->key),
              type);
 
  if (GNUNET_BLOCK_TYPE_DHT_HELLO != type)
  {
    enum GNUNET_DHT_RouteOption options = ntohs (handle->get->options);
    const void *result_filter = (const void *) &handle->get[1];
    uint16_t msize = ntohs (handle->get->header.size);
    uint16_t result_filter_size = ntohs (handle->get->result_filter_size);
    const void *xquery = result_filter + result_filter_size;
    size_t xquery_size = msize - sizeof (*handle->get) - result_filter_size;
    /* The datacache callback runs once per *result*, never on completion,
       so it must not carry the continuation: no result would drop the
       request, several would run the continuation several times.  Both
       lookups below are synchronous. */
    if (0 != (options & GNUNET_DHT_RO_FIND_APPROXIMATE))
      handle->eval = GDS_DATACACHE_get_closest (&handle->get->key,
                                                type,
                                                xquery,
                                                xquery_size,
                                                handle->bg,
                                                &handle_local_result,
                                                handle->t->pi);
    else
      handle->eval = GDS_DATACACHE_handle_get (&handle->get->key,
                                               type,
                                               xquery,
                                               xquery_size,
                                               handle->bg,
                                               &handle_local_result,
                                               handle->t->pi);
  }
  cb_handle_dht_p2p_get_local_result (handle);
}
 
 
static void
cb_handle_dht_p2p_get_my_hello (void *cls)
{
  struct HandleCallbackGet *handle = cls;
  enum GNUNET_DHT_RouteOption options = ntohs (handle->get->options);
 
  if (0 != (options & GNUNET_DHT_RO_FIND_APPROXIMATE))
    handle_find_local_hello (handle->t->pi,
                             &handle->get->key,
                             handle->bg,
                             &cb_handle_dht_p2p_get_local_hello,
                             handle);
  else
    cb_handle_dht_p2p_get_local_hello (handle);
}
 
 
static void
handle_dht_p2p_get (void *cls,
                    const struct PeerGetMessage *get)
{
  struct Target *t = cls;
  struct PeerInfo *peer = t->pi;
  uint16_t msize = ntohs (get->header.size);
  uint16_t result_filter_size = ntohs (get->result_filter_size);
  uint16_t hop_count = ntohs (get->hop_count);
  enum GNUNET_BLOCK_Type type = ntohl (get->type);
  enum GNUNET_DHT_RouteOption options = ntohs (get->options);
  const void *result_filter = (const void *) &get[1];
  const void *xquery = result_filter + result_filter_size;
  size_t xquery_size = msize - sizeof (*get) - result_filter_size;
 
  /* parse and validate message */
  GNUNET_STATISTICS_update (GDS_stats,
                            "# P2P GET requests received",
                            1,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# P2P GET bytes received",
                            msize,
                            GNUNET_NO);
  if (GNUNET_NO ==
      GNUNET_BLOCK_check_query (GDS_block_context,
                                type,
                                &get->key,
                                xquery,
                                xquery_size))
  {
    /* request invalid */
    GNUNET_break_op (0);
    return;
  }
 
  {
    const struct GNUNET_PeerIdentity *my_identity;
    struct HandleCallbackGet *handle;
 
    handle = GNUNET_new (struct HandleCallbackGet);
    handle->t = t;
    handle->get = GNUNET_memdup (get, msize);
    handle->eval = GNUNET_BLOCK_REPLY_OK_MORE;
 
    my_identity = GNUNET_PILS_get_identity (GDS_pils);
    GNUNET_assert (NULL != my_identity);
 
    handle->peer_bf = GNUNET_CONTAINER_bloomfilter_init (get->bloomfilter,
                                                         DHT_BLOOM_SIZE,
                                                         GNUNET_CONSTANTS_BLOOMFILTER_K);
    GNUNET_break_op (GNUNET_YES ==
                     GNUNET_CONTAINER_bloomfilter_test (handle->peer_bf,
                                                        &peer->phash));
    handle->bg = GNUNET_BLOCK_group_create (GDS_block_context,
                                            type,
                                            result_filter,
                                            result_filter_size,
                                            "filter-size",
                                            result_filter_size,
                                            NULL);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "GET for %s at %s after %u hops\n",
                GNUNET_h2s (&get->key),
                GNUNET_i2s (my_identity),
                (unsigned int) hop_count);
    /* local lookup (this may update the bg) */
    if ( (0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
         (GDS_am_closest_peer (&get->key,
                               handle->peer_bf)) )
    {
      if ( (GNUNET_BLOCK_TYPE_DHT_HELLO == type) ||
           (GNUNET_BLOCK_TYPE_ANY == type) )
      {
        GNUNET_STATISTICS_update (GDS_stats,
                                  "# P2P HELLO lookup requests processed",
                                  1,
                                  GNUNET_NO);
        handle_find_my_hello (peer,
                              &get->key,
                              handle->bg,
                              &cb_handle_dht_p2p_get_my_hello,
                              handle);
      }
      else
        cb_handle_dht_p2p_get_local_hello (handle);
    }
    else
    {
      GNUNET_STATISTICS_update (GDS_stats,
                                "# P2P GET requests ONLY routed",
                                1,
                                GNUNET_NO);
      cb_handle_dht_p2p_get_local_result (handle);
    }
  }
}
 
 
static void
process_reply_with_path (const struct GNUNET_DATACACHE_Block *bd,
                         const struct GNUNET_HashCode *query_hash,
                         unsigned int get_path_length,
                         const struct GNUNET_DHT_PathElement *get_path)
{
  /* forward to local clients */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Forwarding reply to local clients\n");
  if (! GDS_CLIENTS_handle_reply (bd,
                                  query_hash,
                                  get_path_length,
                                  get_path))
  {
    GNUNET_break (0);
    return;
  }
  GDS_CLIENTS_process_get_resp (bd,
                                get_path,
                                get_path_length);
  if (GNUNET_YES == cache_results)
  {
    struct GNUNET_DHT_PathElement xput_path[GNUNET_NZL (get_path_length
                                                        + bd->put_path_length)];
    struct GNUNET_DATACACHE_Block bdx = *bd;
 
    if (NULL != bd->put_path)
      GNUNET_memcpy (xput_path,
                     bd->put_path,
                     bd->put_path_length * sizeof(struct
                                                  GNUNET_DHT_PathElement));
    GNUNET_memcpy (&xput_path[bd->put_path_length],
                   get_path,
                   get_path_length * sizeof(struct GNUNET_DHT_PathElement));
    bdx.put_path = xput_path;
    bdx.put_path_length += get_path_length;
    GDS_DATACACHE_handle_put (&bdx);
  }
  /* forward to other peers */
  GDS_ROUTING_process (bd,
                       query_hash,
                       get_path_length,
                       get_path);
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_p2p_result (void *cls,
                      const struct PeerResultMessage *prm)
{
  uint16_t msize = ntohs (prm->header.size) - sizeof (*prm);
  enum GNUNET_DHT_RouteOption ro = ntohs (prm->options);
  bool truncated = (0 != (ro & GNUNET_DHT_RO_TRUNCATED));
  bool tracked = (0 != (ro & GNUNET_DHT_RO_RECORD_ROUTE));
 
  uint16_t get_path_length = ntohs (prm->get_path_length);
  uint16_t put_path_length = ntohs (prm->put_path_length);
  size_t vsize = (truncated ? sizeof (struct GNUNET_PeerIdentity) : 0)
                 + (tracked ? sizeof (struct GNUNET_CRYPTO_EddsaSignature) : 0);
 
  (void) cls;
  if ( (msize < vsize) ||
       (msize - vsize <
        (get_path_length + put_path_length)
        * sizeof(struct GNUNET_DHT_PathElement)) ||
       (get_path_length >
        GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_DHT_PathElement)) ||
       (put_path_length >
        GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_DHT_PathElement)) )
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_dht_p2p_result (void *cls,
                       const struct PeerResultMessage *prm)
{
  struct Target *t = cls;
  struct PeerInfo *peer = t->pi;
  uint16_t msize = ntohs (prm->header.size) - sizeof (*prm);
  enum GNUNET_DHT_RouteOption ro = ntohs (prm->options);
  bool truncated = (0 != (ro & GNUNET_DHT_RO_TRUNCATED));
  bool tracked = (0 != (ro & GNUNET_DHT_RO_RECORD_ROUTE));
  uint16_t get_path_length = ntohs (prm->get_path_length);
  uint16_t put_path_length = ntohs (prm->put_path_length);
  const struct GNUNET_PeerIdentity *trunc_peer
    = truncated
    ? (const struct GNUNET_PeerIdentity *) &prm[1]
    : NULL;
  const struct GNUNET_DHT_PathElement *put_path
    = truncated
    ? (const struct GNUNET_DHT_PathElement *) &trunc_peer[1]
    : (const struct GNUNET_DHT_PathElement *) &prm[1];
  const struct GNUNET_DHT_PathElement *get_path
    = &put_path[put_path_length];
  const struct GNUNET_CRYPTO_EddsaSignature *last_sig
    = tracked
    ? (const struct GNUNET_CRYPTO_EddsaSignature *) &get_path[get_path_length]
    : NULL;
  const void *data
    = tracked
    ? (const void *) &last_sig[1]
    : (const void *) &get_path[get_path_length];
  size_t vsize = (truncated ? sizeof (struct GNUNET_PeerIdentity) : 0)
                 + (tracked ? sizeof (struct GNUNET_CRYPTO_EddsaSignature) : 0);
  struct GNUNET_DATACACHE_Block bd = {
    .expiration_time  = GNUNET_TIME_absolute_ntoh (prm->expiration_time),
    .put_path = put_path,
    .put_path_length = put_path_length,
    .key = prm->key,
    .type = ntohl (prm->type),
    .ro = ro,
    .data = data,
    .data_size = msize - vsize - (get_path_length + put_path_length)
                 * sizeof(struct GNUNET_DHT_PathElement)
  };
 
  /* parse and validate message */
  if (GNUNET_TIME_absolute_is_past (bd.expiration_time))
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Expired results discarded",
                              1,
                              GNUNET_NO);
    return;
  }
  if (GNUNET_OK !=
      GNUNET_BLOCK_check_block (GDS_block_context,
                                bd.type,
                                bd.data,
                                bd.data_size))
  {
    GNUNET_break_op (0);
    return;
  }
  GNUNET_STATISTICS_update (GDS_stats,
                            "# P2P RESULTS received",
                            1,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# P2P RESULT bytes received",
                            msize,
                            GNUNET_NO);
  {
    enum GNUNET_GenericReturnValue ret;
 
    ret = GNUNET_BLOCK_get_key (GDS_block_context,
                                bd.type,
                                bd.data,
                                bd.data_size,
                                &bd.key);
    if (GNUNET_NO == ret)
      bd.key = prm->key;
  }
 
  /* if we got a HELLO, consider it for our own routing table */
  hello_check (&bd);
 
  /* Need to append 'peer' to 'get_path' */
  if (tracked)
  {
    struct GNUNET_DHT_PathElement xget_path[get_path_length + 1];
    struct GNUNET_DHT_PathElement *gp = xget_path;
    unsigned int failure_offset;
 
    GNUNET_memcpy (xget_path,
                   get_path,
                   get_path_length * sizeof(struct GNUNET_DHT_PathElement));
    xget_path[get_path_length].pred = peer->id;
    /* use memcpy(), as last_sig may not be aligned */
    memcpy (&xget_path[get_path_length].sig,
            last_sig,
            sizeof (*last_sig));
#if SANITY_CHECKS
    {
      const struct GNUNET_PeerIdentity *my_identity;
      my_identity = GNUNET_PILS_get_identity (GDS_pils);
      GNUNET_assert (NULL != my_identity);
      /* TODO: might want to eventually implement probabilistic
        load-based path verification, but for now it is all or nothing */
      failure_offset
        = GNUNET_DHT_verify_path (bd.data,
                                  bd.data_size,
                                  bd.expiration_time,
                                  trunc_peer,
                                  put_path,
                                  put_path_length,
                                  gp,
                                  get_path_length + 1,
                                  my_identity);
    }
#else
    failure_offset = 0;
#endif
    if (0 != failure_offset)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Recorded path invalid at offset %u, truncating\n",
                  failure_offset);
      GNUNET_assert (failure_offset <= bd.put_path_length + get_path_length
                     + 1);
      if (failure_offset < bd.put_path_length)
      {
        /* failure on put path */
        trunc_peer = &bd.put_path[failure_offset - 1].pred;
        bd.ro |= GNUNET_DHT_RO_TRUNCATED;
        bd.put_path = &bd.put_path[failure_offset];
        bd.put_path_length -= failure_offset;
        truncated = true;
      }
      else
      {
        /* failure on get path */
        failure_offset -= bd.put_path_length;
        if (0 == failure_offset)
          trunc_peer = &bd.put_path[bd.put_path_length - 1].pred;
        else
          trunc_peer = &gp[failure_offset - 1].pred;
        get_path_length -= failure_offset;
        gp = &gp[failure_offset];
        bd.put_path_length = 0;
        bd.put_path = NULL;
        bd.ro |= GNUNET_DHT_RO_TRUNCATED;
        truncated = true;
      }
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Extending GET path of length %u with %s\n",
                get_path_length,
                GNUNET_i2s (&peer->id));
    if (truncated)
    {
      GNUNET_assert (NULL != trunc_peer);
      bd.trunc_peer = *trunc_peer;
    }
    process_reply_with_path (&bd,
                             &prm->key,
                             get_path_length + 1,
                             gp);
  }
  else
  {
    if (truncated)
    {
      GNUNET_assert (NULL != trunc_peer);
      bd.trunc_peer = *trunc_peer;
    }
    process_reply_with_path (&bd,
                             &prm->key,
                             0,
                             NULL);
  }
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_p2p_hello (void *cls,
                     const struct GNUNET_MessageHeader *hello)
{
  struct Target *t = cls;
  struct PeerInfo *peer = t->pi;
  enum GNUNET_GenericReturnValue ret;
  size_t hellob_size;
  void *hellob;
  struct GNUNET_TIME_Absolute expiration;
 
  ret = GNUNET_HELLO_dht_msg_to_block (hello,
                                       &peer->id,
                                       &hellob,
                                       &hellob_size,
                                       &expiration);
  GNUNET_free (hellob);
  return ret;
}
 
 
static void
handle_dht_p2p_hello (void *cls,
                      const struct GNUNET_MessageHeader *hello)
{
  struct Target *t = cls;
  struct PeerInfo *peer = t->pi;
 
  GNUNET_free (peer->hello);
  peer->hello_size = 0;
  GNUNET_break (GNUNET_OK ==
                GNUNET_HELLO_dht_msg_to_block (hello,
                                               &peer->id,
                                               &peer->hello,
                                               &peer->hello_size,
                                               &peer->hello_expiration));
}
 
 
void
GDS_u_receive (void *cls,
               void **tctx,
               void **sctx,
               const void *message,
               size_t message_size)
{
  struct Target *t = *tctx;
  struct GNUNET_MQ_MessageHandler core_handlers[] = {
    GNUNET_MQ_hd_var_size (dht_p2p_get,
                           GNUNET_MESSAGE_TYPE_DHT_P2P_GET,
                           struct PeerGetMessage,
                           t),
    GNUNET_MQ_hd_var_size (dht_p2p_put,
                           GNUNET_MESSAGE_TYPE_DHT_P2P_PUT,
                           struct PeerPutMessage,
                           t),
    GNUNET_MQ_hd_var_size (dht_p2p_result,
                           GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT,
                           struct PeerResultMessage,
                           t),
    GNUNET_MQ_hd_var_size (dht_p2p_hello,
                           GNUNET_MESSAGE_TYPE_DHT_P2P_HELLO,
                           struct GNUNET_MessageHeader,
                           t),
    GNUNET_MQ_handler_end ()
  };
  const struct GNUNET_MessageHeader *mh = message;
 
  (void) cls; /* the 'struct GDS_Underlay' */
  (void) sctx; /* our receiver address */
  if (NULL == t)
  {
    /* Received message claiming to originate from myself?
       Ignore! */
    GNUNET_break_op (0);
    return;
  }
  if (message_size < sizeof (*mh))
  {
    GNUNET_break_op (0);
    return;
  }
  if (message_size != ntohs (mh->size))
  {
    GNUNET_break_op (0);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Handling message of type %u from peer %s\n",
              ntohs (mh->type),
              GNUNET_i2s (&t->pi->id));
  if (GNUNET_OK !=
      GNUNET_MQ_handle_message (core_handlers,
                                mh))
  {
    GNUNET_break_op (0);
    return;
  }
}
 
 
void
GDS_try_connect (void *cls,
                 const struct GNUNET_PeerIdentity *pid,
                 const char *uri)
{
  const struct GNUNET_PeerIdentity *my_identity;
  struct GNUNET_HashCode phash;
  int peer_bucket;
  struct PeerBucket *bucket;
  (void) cls;
 
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
  GNUNET_assert (NULL != my_identity);
 
  if (0 == GNUNET_memcmp (my_identity, pid))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Got a HELLO for my own PID, ignoring it\n");
    return; /* that's us! */
  }
  GNUNET_CRYPTO_hash (pid,
                      sizeof(*pid),
                      &phash);
  peer_bucket = find_bucket (&phash);
  GNUNET_assert ( (peer_bucket >= 0) &&
                  ((unsigned int) peer_bucket < MAX_BUCKETS));
  bucket = &k_buckets[peer_bucket];
  for (struct PeerInfo *pi = bucket->head;
       NULL != pi;
       pi = pi->next)
    if (0 ==
        GNUNET_memcmp (&pi->id,
                       pid))
    {
      /* already connected */
      GDS_u_try_connect (pid,
                         uri);
      return;
    }
  if (bucket->peers_size >= bucket_size)
    return; /* do not care */
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Discovered peer %s at %s suitable for bucket %d (%u/%u), trying to connect\n",
              GNUNET_i2s (pid),
              uri,
              peer_bucket,
              bucket->peers_size,
              bucket_size);
  /* new peer that we like! */
  GDS_u_try_connect (pid,
                     uri);
}
 
 
void
GDS_NEIGHBOURS_broadcast (const struct GNUNET_MessageHeader *msg)
{
  for (unsigned int bc = 0; bc<closest_bucket; bc++)
  {
    struct PeerBucket *bucket = &k_buckets[bc];
    unsigned int count = 0;
 
    for (struct PeerInfo *pos = bucket->head;
         NULL != pos;
         pos = pos->next)
    {
      if (count >= bucket_size)
        break;   /* we only consider first #bucket_size entries per bucket */
      count++;
      do_send (pos,
               msg);
    }
  }
}
 
 
enum GNUNET_GenericReturnValue
GDS_NEIGHBOURS_init ()
{
 
  unsigned long long temp_config_num;
 
  disable_try_connect
    = GNUNET_CONFIGURATION_get_value_yesno (GDS_cfg,
                                            "DHT",
                                            "DISABLE_TRY_CONNECT");
  if (GNUNET_OK ==
      GNUNET_CONFIGURATION_get_value_number (GDS_cfg,
                                             "DHT",
                                             "bucket_size",
                                             &temp_config_num))
    bucket_size = (unsigned int) temp_config_num;
  cache_results
    = GNUNET_CONFIGURATION_get_value_yesno (GDS_cfg,
                                            "DHT",
                                            "CACHE_RESULTS");
  all_connected_peers = GNUNET_CONTAINER_multipeermap_create (256,
                                                              GNUNET_YES);
  return GNUNET_OK;
}
 
 
void
GDS_NEIGHBOURS_done ()
{
  if (NULL == all_connected_peers)
    return;
  GNUNET_assert (0 ==
                 GNUNET_CONTAINER_multipeermap_size (all_connected_peers));
  GNUNET_CONTAINER_multipeermap_destroy (all_connected_peers);
  all_connected_peers = NULL;
  GNUNET_assert (NULL == find_peer_task);
}
 
 
const struct GNUNET_PeerIdentity *
GDS_NEIGHBOURS_get_id ()
{
  return GNUNET_PILS_get_identity (GDS_pils);
}
 
 
/* end of gnunet-service-dht_neighbours.c */