gnunet-service-dht_neighbours.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2009-2017 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 "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_block_lib.h"
#include "gnunet_hello_lib.h"
#include "gnunet_constants.h"
#include "gnunet_protocols.h"
#include "gnunet_nse_service.h"
#include "gnunet_ats_service.h"
#include "gnunet_core_service.h"
#include "gnunet_datacache_lib.h"
#include "gnunet_transport_service.h"
#include "gnunet_hello_lib.h"
#include "gnunet_dht_service.h"
#include "gnunet_statistics_service.h"
#include "gnunet-service-dht.h"
#include "gnunet-service-dht_datacache.h"
#include "gnunet-service-dht_hello.h"
#include "gnunet-service-dht_neighbours.h"
#include "gnunet-service-dht_nse.h"
#include "gnunet-service-dht_routing.h"
#include "dht.h"

#define LOG_TRAFFIC(kind, ...) GNUNET_log_from(kind, "dht-traffic", __VA_ARGS__)

#define SANITY_CHECKS 1

#define MAX_BUCKETS sizeof(struct GNUNET_HashCode) * 8

#define DEFAULT_BUCKET_SIZE 8

#define FIND_PEER_REPLICATION_LEVEL 4

#define MAXIMUM_REPLICATION_LEVEL 16

#define MAXIMUM_PENDING_PER_PEER 64

#define DHT_MINIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 30)

#define DHT_MAXIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 10)

#define GET_TIMEOUT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)

extern struct GNUNET_TIME_Relative hello_expiration;


GNUNET_NETWORK_STRUCT_BEGIN

struct PeerPutMessage {
  struct GNUNET_MessageHeader header;

  uint32_t options GNUNET_PACKED;

  uint32_t type GNUNET_PACKED;

  uint32_t hop_count GNUNET_PACKED;

  uint32_t desired_replication_level GNUNET_PACKED;

  uint32_t put_path_length GNUNET_PACKED;

  struct GNUNET_TIME_AbsoluteNBO expiration_time;

  char bloomfilter[DHT_BLOOM_SIZE];

  struct GNUNET_HashCode key;

  /* put path (if tracked) */

  /* Payload */
};


struct PeerResultMessage {
  struct GNUNET_MessageHeader header;

  uint32_t type GNUNET_PACKED;

  uint32_t put_path_length GNUNET_PACKED;

  uint32_t get_path_length GNUNET_PACKED;

  struct GNUNET_TIME_AbsoluteNBO expiration_time;

  struct GNUNET_HashCode key;

  /* put path (if tracked) */

  /* get path (if tracked) */

  /* Payload */
};


struct PeerGetMessage {
  struct GNUNET_MessageHeader header;

  uint32_t options GNUNET_PACKED;

  uint32_t type GNUNET_PACKED;

  uint32_t hop_count GNUNET_PACKED;

  uint32_t desired_replication_level GNUNET_PACKED;

  uint32_t xquery_size;

  uint32_t bf_mutator;

  char bloomfilter[DHT_BLOOM_SIZE];

  struct GNUNET_HashCode key;

  /* xquery */

  /* result bloomfilter */
};
GNUNET_NETWORK_STRUCT_END


struct PeerInfo {
  struct PeerInfo *next;

  struct PeerInfo *prev;

  struct GNUNET_MQ_Handle *mq;

  const struct GNUNET_PeerIdentity *id;

  struct GNUNET_HashCode phash;

  int peer_bucket;
};


struct PeerBucket {
  struct PeerInfo *head;

  struct PeerInfo *tail;

  unsigned int peers_size;
};


struct ConnectInfo {
  struct GNUNET_TRANSPORT_OfferHelloHandle *oh;

  struct GNUNET_ATS_ConnectivitySuggestHandle *sh;

  uint32_t strength;
};


static int cache_results;

static int log_route_details_stderr;

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 struct GNUNET_CONTAINER_MultiPeerMap *all_desired_peers;

static unsigned int bucket_size = DEFAULT_BUCKET_SIZE;

static struct GNUNET_SCHEDULER_Task *find_peer_task;

static struct GNUNET_PeerIdentity my_identity;

struct GNUNET_HashCode my_identity_hash;

static struct GNUNET_CORE_Handle *core_api;

static struct GNUNET_ATS_ConnectivityHandle *ats_ch;


static int
find_bucket(const struct GNUNET_HashCode *hc)
{
  unsigned int bits;

  bits = GNUNET_CRYPTO_hash_matching_bits(&my_identity_hash, hc);
  if (bits == MAX_BUCKETS)
    {
      /* How can all bits match? Got my own ID? */
      GNUNET_break(0);
      return GNUNET_SYSERR;
    }
  return MAX_BUCKETS - bits - 1;
}


static void
offer_hello_done(void *cls)
{
  struct ConnectInfo *ci = cls;

  ci->oh = NULL;
}


static int
free_connect_info(void *cls,
                  const struct GNUNET_PeerIdentity *peer,
                  void *value)
{
  struct ConnectInfo *ci = value;

  (void)cls;
  GNUNET_assert(GNUNET_YES ==
                GNUNET_CONTAINER_multipeermap_remove(all_desired_peers,
                                                     peer,
                                                     ci));
  if (NULL != ci->sh)
    {
      GNUNET_ATS_connectivity_suggest_cancel(ci->sh);
      ci->sh = NULL;
    }
  if (NULL != ci->oh)
    {
      GNUNET_TRANSPORT_offer_hello_cancel(ci->oh);
      ci->oh = NULL;
    }
  GNUNET_free(ci);
  return GNUNET_YES;
}


static void
try_connect(const struct GNUNET_PeerIdentity *pid,
            const struct GNUNET_MessageHeader *h)
{
  int bucket;
  struct GNUNET_HashCode pid_hash;
  struct ConnectInfo *ci;
  uint32_t strength;

  GNUNET_CRYPTO_hash(pid,
                     sizeof(struct GNUNET_PeerIdentity),
                     &pid_hash);
  bucket = find_bucket(&pid_hash);
  if (bucket < 0)
    return; /* self? */
  ci = GNUNET_CONTAINER_multipeermap_get(all_desired_peers,
                                         pid);

  if (k_buckets[bucket].peers_size < bucket_size)
    strength = (bucket_size - k_buckets[bucket].peers_size) * bucket;
  else
    strength = bucket; /* minimum value of connectivity */
  if (GNUNET_YES ==
      GNUNET_CONTAINER_multipeermap_contains(all_connected_peers,
                                             pid))
    strength *= 2; /* double for connected peers */
  else if (k_buckets[bucket].peers_size > bucket_size)
    strength = 0; /* bucket full, we really do not care about more */

  if ((0 == strength) &&
      (NULL != ci))
    {
      /* release request */
      GNUNET_assert(GNUNET_YES ==
                    free_connect_info(NULL,
                                      pid,
                                      ci));
      return;
    }
  if (NULL == ci)
    {
      ci = GNUNET_new(struct ConnectInfo);
      GNUNET_assert(GNUNET_OK ==
                    GNUNET_CONTAINER_multipeermap_put(all_desired_peers,
                                                      pid,
                                                      ci,
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    }
  if ((NULL != ci->oh) &&
      (NULL != h))
    GNUNET_TRANSPORT_offer_hello_cancel(ci->oh);
  if (NULL != h)
    ci->oh = GNUNET_TRANSPORT_offer_hello(GDS_cfg,
                                          h,
                                          &offer_hello_done,
                                          ci);
  if ((NULL != ci->sh) &&
      (ci->strength != strength))
    GNUNET_ATS_connectivity_suggest_cancel(ci->sh);
  if (ci->strength != strength)
    ci->sh = GNUNET_ATS_connectivity_suggest(ats_ch,
                                             pid,
                                             strength);
  ci->strength = strength;
}


static int
update_desire_strength(void *cls,
                       const struct GNUNET_PeerIdentity *pid,
                       void *value)
{
  (void)cls;
  (void)value;
  try_connect(pid,
              NULL);
  return GNUNET_YES;
}


static void
update_connect_preferences()
{
  GNUNET_CONTAINER_multipeermap_iterate(all_desired_peers,
                                        &update_desire_strength,
                                        NULL);
}


static int
add_known_to_bloom(void *cls,
                   const struct GNUNET_PeerIdentity *key,
                   void *value)
{
  struct GNUNET_BLOCK_Group *bg = cls;
  struct GNUNET_HashCode key_hash;

  (void)cls;
  (void)value;
  GNUNET_CRYPTO_hash(key,
                     sizeof(struct GNUNET_PeerIdentity),
                     &key_hash);
  GNUNET_BLOCK_group_set_seen(bg,
                              &key_hash,
                              1);
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Adding known peer (%s) to bloomfilter for FIND PEER\n",
             GNUNET_i2s(key));
  return GNUNET_YES;
}


static void
send_find_peer_message(void *cls)
{
  struct GNUNET_TIME_Relative next_send_time;
  struct GNUNET_BLOCK_Group *bg;
  struct GNUNET_CONTAINER_BloomFilter *peer_bf;

  (void)cls;
  find_peer_task = NULL;
  if (newly_found_peers > bucket_size)
    {
      /* If we are finding many peers already, no need to send out our request right now! */
      find_peer_task =
        GNUNET_SCHEDULER_add_delayed(GNUNET_TIME_UNIT_MINUTES,
                                     &send_find_peer_message,
                                     NULL);
      newly_found_peers = 0;
      return;
    }
  bg = GNUNET_BLOCK_group_create(GDS_block_context,
                                 GNUNET_BLOCK_TYPE_DHT_HELLO,
                                 GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK,
                                                          UINT32_MAX),
                                 NULL,
                                 0,
                                 "filter-size",
                                 DHT_BLOOM_SIZE,
                                 NULL);
  GNUNET_CONTAINER_multipeermap_iterate(all_connected_peers,
                                        &add_known_to_bloom,
                                        bg);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# FIND PEER messages initiated"),
                           1,
                           GNUNET_NO);
  peer_bf
    = GNUNET_CONTAINER_bloomfilter_init(NULL,
                                        DHT_BLOOM_SIZE,
                                        GNUNET_CONSTANTS_BLOOMFILTER_K);
  // FIXME: pass priority!?
  GDS_NEIGHBOURS_handle_get(GNUNET_BLOCK_TYPE_DHT_HELLO,
                            GNUNET_DHT_RO_FIND_PEER | GNUNET_DHT_RO_RECORD_ROUTE,
                            FIND_PEER_REPLICATION_LEVEL,
                            0,
                            &my_identity_hash,
                            NULL,
                            0,
                            bg,
                            peer_bf);
  GNUNET_CONTAINER_bloomfilter_free(peer_bf);
  GNUNET_BLOCK_group_destroy(bg);
  /* schedule next round */
  next_send_time.rel_value_us =
    DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value_us +
    GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK,
                             DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value_us /
                             (newly_found_peers + 1));
  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);
}


static void *
handle_core_connect(void *cls,
                    const struct GNUNET_PeerIdentity *peer,
                    struct GNUNET_MQ_Handle *mq)
{
  struct PeerInfo *pi;

  (void)cls;
  /* Check for connect to self message */
  if (0 == GNUNET_memcmp(&my_identity,
                         peer))
    return NULL;
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Connected to %s\n",
             GNUNET_i2s(peer));
  GNUNET_assert(GNUNET_NO ==
                GNUNET_CONTAINER_multipeermap_get(all_connected_peers,
                                                  peer));
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# peers connected"),
                           1,
                           GNUNET_NO);
  pi = GNUNET_new(struct PeerInfo);
  pi->id = peer;
  pi->mq = mq;
  GNUNET_CRYPTO_hash(peer,
                     sizeof(struct GNUNET_PeerIdentity),
                     &pi->phash);
  pi->peer_bucket = find_bucket(&pi->phash);
  GNUNET_assert((pi->peer_bucket >= 0) &&
                ((unsigned int)pi->peer_bucket < MAX_BUCKETS));
  GNUNET_CONTAINER_DLL_insert_tail(k_buckets[pi->peer_bucket].head,
                                   k_buckets[pi->peer_bucket].tail,
                                   pi);
  k_buckets[pi->peer_bucket].peers_size++;
  closest_bucket = GNUNET_MAX(closest_bucket,
                              (unsigned int)pi->peer_bucket);
  GNUNET_assert(GNUNET_OK ==
                GNUNET_CONTAINER_multipeermap_put(all_connected_peers,
                                                  pi->id,
                                                  pi,
                                                  GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  if ((pi->peer_bucket > 0) &&
      (k_buckets[pi->peer_bucket].peers_size <= bucket_size))
    {
      update_connect_preferences();
      newly_found_peers++;
    }
  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);
    }
  return pi;
}


static void
handle_core_disconnect(void *cls,
                       const struct GNUNET_PeerIdentity *peer,
                       void *internal_cls)
{
  struct PeerInfo *to_remove = internal_cls;

  (void)cls;
  /* Check for disconnect from self message */
  if (NULL == to_remove)
    return;
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Disconnected %s\n",
             GNUNET_i2s(peer));
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# peers connected"),
                           -1,
                           GNUNET_NO);
  GNUNET_assert(GNUNET_YES ==
                GNUNET_CONTAINER_multipeermap_remove(all_connected_peers,
                                                     peer,
                                                     to_remove));
  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(to_remove->peer_bucket >= 0);
  GNUNET_CONTAINER_DLL_remove(k_buckets[to_remove->peer_bucket].head,
                              k_buckets[to_remove->peer_bucket].tail,
                              to_remove);
  GNUNET_assert(k_buckets[to_remove->peer_bucket].peers_size > 0);
  k_buckets[to_remove->peer_bucket].peers_size--;
  while ((closest_bucket > 0) &&
         (0 == k_buckets[to_remove->peer_bucket].peers_size))
    closest_bucket--;
  if (k_buckets[to_remove->peer_bucket].peers_size < bucket_size)
    update_connect_preferences();
  GNUNET_free(to_remove);
}


static unsigned int
get_forward_count(uint32_t hop_count,
                  uint32_t target_replication)
{
  uint32_t random_value;
  uint32_t forward_count;
  float target_value;

  if (hop_count > GDS_NSE_get() * 4.0)
    {
      /* forcefully terminate */
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# requests TTL-dropped"),
                               1, GNUNET_NO);
      return 0;
    }
  if (hop_count > GDS_NSE_get() * 2.0)
    {
      /* Once we have reached our ideal number of hops, only forward to 1 peer */
      return 1;
    }
  /* bound by system-wide maximum */
  target_replication =
    GNUNET_MIN(MAXIMUM_REPLICATION_LEVEL, target_replication);
  target_value =
    1 + (target_replication - 1.0) / (GDS_NSE_get() +
                                      ((float)(target_replication - 1.0) *
                                       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(GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX);
  if (random_value < (target_value * UINT32_MAX))
    forward_count++;
  return forward_count;
}


static unsigned int
get_distance(const struct GNUNET_HashCode *target,
             const struct GNUNET_HashCode *have)
{
  unsigned int bucket;
  unsigned int msb;
  unsigned int lsb;
  unsigned int i;

  /* We have to represent the distance between two 2^9 (=512)-bit
   * numbers as a 2^5 (=32)-bit number with "0" being used for the
   * two numbers being identical; furthermore, we need to
   * guarantee that a difference in the number of matching
   * bits is always represented in the result.
   *
   * We use 2^32/2^9 numerical values to distinguish between
   * hash codes that have the same LSB bit distance and
   * use the highest 2^9 bits of the result to signify the
   * number of (mis)matching LSB bits; if we have 0 matching
   * and hence 512 mismatching LSB bits we return -1 (since
   * 512 itself cannot be represented with 9 bits) */

  /* first, calculate the most significant 9 bits of our
   * result, aka the number of LSBs */
  bucket = GNUNET_CRYPTO_hash_matching_bits(target,
                                            have);
  /* bucket is now a value between 0 and 512 */
  if (bucket == 512)
    return 0;                   /* perfect match */
  if (bucket == 0)
    return (unsigned int)-1;    /* LSB differs; use max (if we did the bit-shifting
                                 * below, we'd end up with max+1 (overflow)) */

  /* calculate the most significant bits of the final result */
  msb = (512 - bucket) << (32 - 9);
  /* calculate the 32-9 least significant bits of the final result by
   * looking at the differences in the 32-9 bits following the
   * mismatching bit at 'bucket' */
  lsb = 0;
  for (i = bucket + 1;
       (i < sizeof(struct GNUNET_HashCode) * 8) && (i < bucket + 1 + 32 - 9); i++)
    {
      if (GNUNET_CRYPTO_hash_get_bit(target, i) !=
          GNUNET_CRYPTO_hash_get_bit(have, i))
        lsb |= (1 << (bucket + 32 - 9 - i));    /* first bit set will be 10,
                                                 * last bit set will be 31 -- if
                                                 * i does not reach 512 first... */
    }
  return msb | lsb;
}


int
GDS_am_closest_peer(const struct GNUNET_HashCode *key,
                    const struct GNUNET_CONTAINER_BloomFilter *bloom)
{
  int bits;
  int other_bits;
  int bucket_num;
  struct PeerInfo *pos;

  if (0 == GNUNET_memcmp(&my_identity_hash,
                         key))
    return GNUNET_YES;
  bucket_num = find_bucket(key);
  GNUNET_assert(bucket_num >= 0);
  bits = GNUNET_CRYPTO_hash_matching_bits(&my_identity_hash,
                                          key);
  pos = k_buckets[bucket_num].head;
  while (NULL != pos)
    {
      if ((NULL != bloom) &&
          (GNUNET_YES ==
           GNUNET_CONTAINER_bloomfilter_test(bloom,
                                             &pos->phash)))
        {
          pos = pos->next;
          continue;             /* Skip already checked entries */
        }
      other_bits = GNUNET_CRYPTO_hash_matching_bits(&pos->phash,
                                                    key);
      if (other_bits > bits)
        return GNUNET_NO;
      if (other_bits == bits)   /* We match the same number of bits */
        return GNUNET_YES;
      pos = pos->next;
    }
  /* No peers closer, we are the closest! */
  return GNUNET_YES;
}


static struct PeerInfo *
select_peer(const struct GNUNET_HashCode *key,
            const struct GNUNET_CONTAINER_BloomFilter *bloom,
            uint32_t hops)
{
  unsigned int bc;
  unsigned int count;
  unsigned int selected;
  struct PeerInfo *pos;
  unsigned int dist;
  unsigned int smallest_distance;
  struct PeerInfo *chosen;

  if (hops >= GDS_NSE_get())
    {
      /* greedy selection (closest peer that is not in bloomfilter) */
      smallest_distance = UINT_MAX;
      chosen = NULL;
      for (bc = 0; bc <= closest_bucket; bc++)
        {
          pos = k_buckets[bc].head;
          count = 0;
          while ((pos != NULL) && (count < bucket_size))
            {
              if ((NULL == bloom) ||
                  (GNUNET_NO ==
                   GNUNET_CONTAINER_bloomfilter_test(bloom,
                                                     &pos->phash)))
                {
                  dist = get_distance(key,
                                      &pos->phash);
                  if (dist < smallest_distance)
                    {
                      chosen = pos;
                      smallest_distance = dist;
                    }
                }
              else
                {
                  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));
                  GNUNET_STATISTICS_update(GDS_stats,
                                           gettext_noop("# Peers excluded from routing due to Bloomfilter"),
                                           1,
                                           GNUNET_NO);
                  dist = get_distance(key,
                                      &pos->phash);
                  if (dist < smallest_distance)
                    {
                      chosen = NULL;
                      smallest_distance = dist;
                    }
                }
              count++;
              pos = pos->next;
            }
        }
      if (NULL == chosen)
        GNUNET_STATISTICS_update(GDS_stats,
                                 gettext_noop("# Peer selection failed"),
                                 1,
                                 GNUNET_NO);
      else
        GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                   "Selected peer `%s' in greedy routing for %s\n",
                   GNUNET_i2s(chosen->id),
                   GNUNET_h2s(key));
      return chosen;
    }

  /* select "random" peer */
  /* count number of peers that are available and not filtered */
  count = 0;
  for (bc = 0; bc <= closest_bucket; bc++)
    {
      pos = k_buckets[bc].head;
      while ((NULL != pos) && (count < bucket_size))
        {
          if ((NULL != bloom) &&
              (GNUNET_YES ==
               GNUNET_CONTAINER_bloomfilter_test(bloom,
                                                 &pos->phash)))
            {
              GNUNET_STATISTICS_update(GDS_stats,
                                       gettext_noop
                                         ("# Peers excluded from routing due to Bloomfilter"),
                                       1, GNUNET_NO);
              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));
              pos = pos->next;
              continue;         /* Ignore bloomfiltered peers */
            }
          count++;
          pos = pos->next;
        }
    }
  if (0 == count)               /* No peers to select from! */
    {
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# Peer selection failed"), 1,
                               GNUNET_NO);
      return NULL;
    }
  /* Now actually choose a peer */
  selected = GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK,
                                      count);
  count = 0;
  for (bc = 0; bc <= closest_bucket; bc++)
    {
      for (pos = k_buckets[bc].head; ((pos != NULL) && (count < bucket_size)); pos = pos->next)
        {
          if ((bloom != NULL) &&
              (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;
            }
        }
    }
  GNUNET_break(0);
  return NULL;
}


static unsigned int
get_target_peers(const struct GNUNET_HashCode *key,
                 struct GNUNET_CONTAINER_BloomFilter *bloom,
                 uint32_t hop_count,
                 uint32_t target_replication,
                 struct PeerInfo ***targets)
{
  unsigned int ret;
  unsigned int off;
  struct PeerInfo **rtargets;
  struct PeerInfo *nxt;

  GNUNET_assert(NULL != bloom);
  ret = get_forward_count(hop_count,
                          target_replication);
  if (0 == ret)
    {
      *targets = NULL;
      return 0;
    }
  rtargets = GNUNET_new_array(ret,
                              struct PeerInfo *);
  for (off = 0; off < ret; off++)
    {
      nxt = select_peer(key,
                        bloom,
                        hop_count);
      if (NULL == nxt)
        break;
      rtargets[off] = nxt;
      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),
             ret);
  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,
             ret);
  return off;
}


int
GDS_NEIGHBOURS_handle_put(enum GNUNET_BLOCK_Type type,
                          enum GNUNET_DHT_RouteOption options,
                          uint32_t desired_replication_level,
                          struct GNUNET_TIME_Absolute expiration_time,
                          uint32_t hop_count,
                          struct GNUNET_CONTAINER_BloomFilter *bf,
                          const struct GNUNET_HashCode *key,
                          unsigned int put_path_length,
                          struct GNUNET_PeerIdentity *put_path,
                          const void *data,
                          size_t data_size)
{
  unsigned int target_count;
  unsigned int i;
  struct PeerInfo **targets;
  struct PeerInfo *target;
  size_t msize;
  struct GNUNET_MQ_Envelope *env;
  struct PeerPutMessage *ppm;
  struct GNUNET_PeerIdentity *pp;
  unsigned int skip_count;

  GNUNET_assert(NULL != bf);
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Adding myself (%s) to PUT bloomfilter for %s\n",
             GNUNET_i2s(&my_identity),
             GNUNET_h2s(key));
  GNUNET_CONTAINER_bloomfilter_add(bf,
                                   &my_identity_hash);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# PUT requests routed"),
                           1,
                           GNUNET_NO);
  target_count
    = get_target_peers(key,
                       bf,
                       hop_count,
                       desired_replication_level,
                       &targets);
  if (0 == target_count)
    {
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "Routing PUT for %s terminates after %u hops at %s\n",
                 GNUNET_h2s(key),
                 (unsigned int)hop_count,
                 GNUNET_i2s(&my_identity));
      return GNUNET_NO;
    }
  msize = put_path_length * sizeof(struct GNUNET_PeerIdentity) + data_size;
  if (msize + sizeof(struct PeerPutMessage)
      >= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
    {
      put_path_length = 0;
      msize = data_size;
    }
  if (msize + sizeof(struct PeerPutMessage)
      >= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
    {
      GNUNET_break(0);
      GNUNET_free(targets);
      return GNUNET_NO;
    }
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# PUT messages queued for transmission"),
                           target_count,
                           GNUNET_NO);
  skip_count = 0;
  for (i = 0; i < target_count; i++)
    {
      target = targets[i];
      if (GNUNET_MQ_get_length(target->mq) >= MAXIMUM_PENDING_PER_PEER)
        {
          /* skip */
          GNUNET_STATISTICS_update(GDS_stats,
                                   gettext_noop("# P2P messages dropped due to full queue"),
                                   1,
                                   GNUNET_NO);
          skip_count++;
          continue;
        }
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "Routing PUT for %s after %u hops to %s\n",
                 GNUNET_h2s(key),
                 (unsigned int)hop_count,
                 GNUNET_i2s(target->id));
      env = GNUNET_MQ_msg_extra(ppm,
                                msize,
                                GNUNET_MESSAGE_TYPE_DHT_P2P_PUT);
      ppm->options = htonl(options);
      ppm->type = htonl(type);
      ppm->hop_count = htonl(hop_count + 1);
      ppm->desired_replication_level = htonl(desired_replication_level);
      ppm->put_path_length = htonl(put_path_length);
      ppm->expiration_time = GNUNET_TIME_absolute_hton(expiration_time);
      GNUNET_break(GNUNET_YES ==
                   GNUNET_CONTAINER_bloomfilter_test(bf,
                                                     &target->phash));
      GNUNET_assert(GNUNET_OK ==
                    GNUNET_CONTAINER_bloomfilter_get_raw_data(bf,
                                                              ppm->bloomfilter,
                                                              DHT_BLOOM_SIZE));
      ppm->key = *key;
      pp = (struct GNUNET_PeerIdentity *)&ppm[1];
      GNUNET_memcpy(pp,
                    put_path,
                    sizeof(struct GNUNET_PeerIdentity) * put_path_length);
      GNUNET_memcpy(&pp[put_path_length],
                    data,
                    data_size);
      GNUNET_MQ_send(target->mq,
                     env);
    }
  GNUNET_free(targets);
  return (skip_count < target_count) ? GNUNET_OK : GNUNET_NO;
}


int
GDS_NEIGHBOURS_handle_get(enum GNUNET_BLOCK_Type type,
                          enum GNUNET_DHT_RouteOption options,
                          uint32_t desired_replication_level,
                          uint32_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)
{
  unsigned int target_count;
  struct PeerInfo **targets;
  struct PeerInfo *target;
  struct GNUNET_MQ_Envelope *env;
  size_t msize;
  struct PeerGetMessage *pgm;
  char *xq;
  size_t reply_bf_size;
  void *reply_bf;
  unsigned int skip_count;
  uint32_t bf_nonce;

  GNUNET_assert(NULL != peer_bf);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# 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\n",
             GNUNET_i2s(&my_identity),
             GNUNET_h2s(key));
  GNUNET_CONTAINER_bloomfilter_add(peer_bf,
                                   &my_identity_hash);
  if (0 == target_count)
    {
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "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,
                                   &bf_nonce,
                                   &reply_bf,
                                   &reply_bf_size))
    {
      reply_bf = NULL;
      reply_bf_size = 0;
      bf_nonce = GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK,
                                          UINT32_MAX);
    }
  msize = xquery_size + reply_bf_size;
  if (msize + sizeof(struct PeerGetMessage) >= GNUNET_MAX_MESSAGE_SIZE)
    {
      GNUNET_break(0);
      GNUNET_free_non_null(reply_bf);
      GNUNET_free(targets);
      return GNUNET_NO;
    }
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# GET messages queued for transmission"),
                           target_count,
                           GNUNET_NO);
  /* forward request */
  skip_count = 0;
  for (unsigned int i = 0; i < target_count; i++)
    {
      target = targets[i];
      if (GNUNET_MQ_get_length(target->mq) >= MAXIMUM_PENDING_PER_PEER)
        {
          /* skip */
          GNUNET_STATISTICS_update(GDS_stats,
                                   gettext_noop("# P2P messages dropped due to full queue"),
                                   1, GNUNET_NO);
          skip_count++;
          continue;
        }
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "Routing GET for %s after %u hops to %s\n",
                 GNUNET_h2s(key),
                 (unsigned int)hop_count,
                 GNUNET_i2s(target->id));
      env = GNUNET_MQ_msg_extra(pgm,
                                msize,
                                GNUNET_MESSAGE_TYPE_DHT_P2P_GET);
      pgm->options = htonl(options);
      pgm->type = htonl(type);
      pgm->hop_count = htonl(hop_count + 1);
      pgm->desired_replication_level = htonl(desired_replication_level);
      pgm->xquery_size = htonl(xquery_size);
      pgm->bf_mutator = bf_nonce;
      GNUNET_break(GNUNET_YES ==
                   GNUNET_CONTAINER_bloomfilter_test(peer_bf,
                                                     &target->phash));
      GNUNET_assert(GNUNET_OK ==
                    GNUNET_CONTAINER_bloomfilter_get_raw_data(peer_bf,
                                                              pgm->bloomfilter,
                                                              DHT_BLOOM_SIZE));
      pgm->key = *key;
      xq = (char *)&pgm[1];
      GNUNET_memcpy(xq,
                    xquery,
                    xquery_size);
      GNUNET_memcpy(&xq[xquery_size],
                    reply_bf,
                    reply_bf_size);
      GNUNET_MQ_send(target->mq,
                     env);
    }
  GNUNET_free(targets);
  GNUNET_free_non_null(reply_bf);
  return (skip_count < target_count) ? GNUNET_OK : GNUNET_NO;
}


void
GDS_NEIGHBOURS_handle_reply(const struct GNUNET_PeerIdentity *target,
                            enum GNUNET_BLOCK_Type type,
                            struct GNUNET_TIME_Absolute expiration_time,
                            const struct GNUNET_HashCode *key,
                            unsigned int put_path_length,
                            const struct GNUNET_PeerIdentity *put_path,
                            unsigned int get_path_length,
                            const struct GNUNET_PeerIdentity *get_path,
                            const void *data,
                            size_t data_size)
{
  struct PeerInfo *pi;
  struct GNUNET_MQ_Envelope *env;
  size_t msize;
  struct PeerResultMessage *prm;
  struct GNUNET_PeerIdentity *paths;

  msize = data_size + (get_path_length + put_path_length) *
          sizeof(struct GNUNET_PeerIdentity);
  if ((msize + sizeof(struct PeerResultMessage) >= GNUNET_MAX_MESSAGE_SIZE) ||
      (get_path_length >
       GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)) ||
      (put_path_length >
       GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)) ||
      (data_size > GNUNET_MAX_MESSAGE_SIZE))
    {
      GNUNET_break(0);
      return;
    }
  pi = GNUNET_CONTAINER_multipeermap_get(all_connected_peers,
                                         target);
  if (NULL == pi)
    {
      /* peer disconnected in the meantime, drop reply */
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "No matching peer for reply for key %s\n",
                 GNUNET_h2s(key));
      return;
    }
  if (GNUNET_MQ_get_length(pi->mq) >= MAXIMUM_PENDING_PER_PEER)
    {
      /* skip */
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# P2P messages dropped due to full queue"),
                               1,
                               GNUNET_NO);
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "Peer queue full, ignoring reply for key %s\n",
                 GNUNET_h2s(key));
      return;
    }

  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Forwarding reply for key %s to peer %s\n",
             GNUNET_h2s(key),
             GNUNET_i2s(target));
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop
                             ("# RESULT messages queued for transmission"), 1,
                           GNUNET_NO);
  env = GNUNET_MQ_msg_extra(prm,
                            msize,
                            GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT);
  prm->type = htonl(type);
  prm->put_path_length = htonl(put_path_length);
  prm->get_path_length = htonl(get_path_length);
  prm->expiration_time = GNUNET_TIME_absolute_hton(expiration_time);
  prm->key = *key;
  paths = (struct GNUNET_PeerIdentity *)&prm[1];
  GNUNET_memcpy(paths,
                put_path,
                put_path_length * sizeof(struct GNUNET_PeerIdentity));
  GNUNET_memcpy(&paths[put_path_length],
                get_path,
                get_path_length * sizeof(struct GNUNET_PeerIdentity));
  GNUNET_memcpy(&paths[put_path_length + get_path_length],
                data,
                data_size);
  GNUNET_MQ_send(pi->mq,
                 env);
}


static void
core_init(void *cls,
          const struct GNUNET_PeerIdentity *identity)
{
  (void)cls;
  GNUNET_log(GNUNET_ERROR_TYPE_INFO,
             "CORE called, I am %s\n",
             GNUNET_i2s(identity));
  my_identity = *identity;
  GNUNET_CRYPTO_hash(identity,
                     sizeof(struct GNUNET_PeerIdentity),
                     &my_identity_hash);
  GNUNET_SERVICE_resume(GDS_service);
}


static int
check_dht_p2p_put(void *cls,
                  const struct PeerPutMessage *put)
{
  uint32_t putlen;
  uint16_t msize;

  (void)cls;
  msize = ntohs(put->header.size);
  putlen = ntohl(put->put_path_length);
  if ((msize <
       sizeof(struct PeerPutMessage) +
       putlen * sizeof(struct GNUNET_PeerIdentity)) ||
      (putlen >
       GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)))
    {
      GNUNET_break_op(0);
      return GNUNET_SYSERR;
    }
  return GNUNET_OK;
}


static void
handle_dht_p2p_put(void *cls,
                   const struct PeerPutMessage *put)
{
  struct PeerInfo *peer = cls;
  const struct GNUNET_PeerIdentity *put_path;
  const void *payload;
  uint32_t putlen;
  uint16_t msize;
  size_t payload_size;
  enum GNUNET_DHT_RouteOption options;
  struct GNUNET_CONTAINER_BloomFilter *bf;
  struct GNUNET_HashCode test_key;
  int forwarded;
  struct GNUNET_TIME_Absolute exp_time;

  exp_time = GNUNET_TIME_absolute_ntoh(put->expiration_time);
  if (0 == GNUNET_TIME_absolute_get_remaining(exp_time).rel_value_us)
    {
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# Expired PUTs discarded"),
                               1,
                               GNUNET_NO);
      return;
    }
  msize = ntohs(put->header.size);
  putlen = ntohl(put->put_path_length);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# P2P PUT requests received"),
                           1,
                           GNUNET_NO);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# P2P PUT bytes received"),
                           msize,
                           GNUNET_NO);
  put_path = (const struct GNUNET_PeerIdentity *)&put[1];
  payload = &put_path[putlen];
  options = ntohl(put->options);
  payload_size = msize - (sizeof(struct PeerPutMessage) +
                          putlen * sizeof(struct GNUNET_PeerIdentity));

  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "PUT for `%s' from %s\n",
             GNUNET_h2s(&put->key),
             GNUNET_i2s(peer->id));
  if (GNUNET_YES == log_route_details_stderr)
    {
      char *tmp;
      char *pp;

      pp = GNUNET_STRINGS_pp2s(put_path,
                               putlen);
      tmp = GNUNET_strdup(GNUNET_i2s(&my_identity));
      LOG_TRAFFIC(GNUNET_ERROR_TYPE_DEBUG,
                  "R5N PUT %s: %s->%s (%u, %u=>%u, PP: %s)\n",
                  GNUNET_h2s(&put->key),
                  GNUNET_i2s(peer->id),
                  tmp,
                  ntohl(put->hop_count),
                  GNUNET_CRYPTO_hash_matching_bits(&peer->phash,
                                                   &put->key),
                  GNUNET_CRYPTO_hash_matching_bits(&my_identity_hash,
                                                   &put->key),
                  pp);
      GNUNET_free(pp);
      GNUNET_free(tmp);
    }
  switch (GNUNET_BLOCK_get_key
            (GDS_block_context,
            ntohl(put->type),
            payload,
            payload_size,
            &test_key))
    {
    case GNUNET_YES:
      if (0 != memcmp(&test_key,
                      &put->key,
                      sizeof(struct GNUNET_HashCode)))
        {
          char *put_s = GNUNET_strdup(GNUNET_h2s_full(&put->key));

          GNUNET_break_op(0);
          GNUNET_log(GNUNET_ERROR_TYPE_WARNING,
                     "PUT with key `%s' for block with key %s\n",
                     put_s,
                     GNUNET_h2s_full(&test_key));
          GNUNET_free(put_s);
          return;
        }
      break;

    case GNUNET_NO:
      GNUNET_break_op(0);
      return;

    case GNUNET_SYSERR:
      /* cannot verify, good luck */
      break;
    }
  if (ntohl(put->type) == GNUNET_BLOCK_TYPE_REGEX)  /* FIXME: do for all tpyes */
    {
      switch (GNUNET_BLOCK_evaluate(GDS_block_context,
                                    ntohl(put->type),
                                    NULL, /* query group */
                                    GNUNET_BLOCK_EO_NONE,
                                    NULL,   /* query */
                                    NULL, 0, /* xquery */
                                    payload,
                                    payload_size))
        {
        case GNUNET_BLOCK_EVALUATION_OK_MORE:
        case GNUNET_BLOCK_EVALUATION_OK_LAST:
          break;

        case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
        case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
        case GNUNET_BLOCK_EVALUATION_RESULT_IRRELEVANT:
        case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
        case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
        case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
        default:
          GNUNET_break_op(0);
          return;
        }
    }

  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));
  {
    struct GNUNET_PeerIdentity pp[putlen + 1];

    /* extend 'put path' by sender */
    if (0 != (options & GNUNET_DHT_RO_RECORD_ROUTE))
      {
#if SANITY_CHECKS
        for (unsigned int i = 0; i <= putlen; i++)
          {
            for (unsigned int j = 0; j < i; j++)
              {
                GNUNET_break(0 != memcmp(&pp[i],
                                         &pp[j],
                                         sizeof(struct GNUNET_PeerIdentity)));
              }
            GNUNET_break(0 != memcmp(&pp[i],
                                     peer->id,
                                     sizeof(struct GNUNET_PeerIdentity)));
          }
#endif
        GNUNET_memcpy(pp,
                      put_path,
                      putlen * sizeof(struct GNUNET_PeerIdentity));
        pp[putlen] = *peer->id;
        putlen++;
      }
    else
      putlen = 0;

    /* give to local clients */
    GDS_CLIENTS_handle_reply(exp_time,
                             &put->key,
                             0,
                             NULL,
                             putlen,
                             pp,
                             ntohl(put->type),
                             payload_size,
                             payload);
    /* store locally */
    if ((0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
        (GDS_am_closest_peer(&put->key, bf)))
      GDS_DATACACHE_handle_put(exp_time,
                               &put->key,
                               putlen,
                               pp,
                               ntohl(put->type),
                               payload_size,
                               payload);
    /* route to other peers */
    forwarded = GDS_NEIGHBOURS_handle_put(ntohl(put->type),
                                          options,
                                          ntohl(put->desired_replication_level),
                                          exp_time,
                                          ntohl(put->hop_count),
                                          bf,
                                          &put->key,
                                          putlen,
                                          pp,
                                          payload,
                                          payload_size);
    /* notify monitoring clients */
    GDS_CLIENTS_process_put(options
                            | ((GNUNET_OK == forwarded)
                               ? GNUNET_DHT_RO_LAST_HOP
                               : 0),
                            ntohl(put->type),
                            ntohl(put->hop_count),
                            ntohl(put->desired_replication_level),
                            putlen, pp,
                            exp_time,
                            &put->key,
                            payload,
                            payload_size);
  }
  GNUNET_CONTAINER_bloomfilter_free(bf);
}


static void
handle_find_peer(const struct GNUNET_PeerIdentity *sender,
                 const struct GNUNET_HashCode *key,
                 struct GNUNET_BLOCK_Group *bg)
{
  int bucket_idx;
  struct PeerBucket *bucket;
  struct PeerInfo *peer;
  unsigned int choice;
  const struct GNUNET_HELLO_Message *hello;
  size_t hello_size;

  /* first, check about our own HELLO */
  if (NULL != GDS_my_hello)
    {
      hello_size = GNUNET_HELLO_size((const struct GNUNET_HELLO_Message *)GDS_my_hello);
      GNUNET_break(hello_size >= sizeof(struct GNUNET_MessageHeader));
      if (GNUNET_BLOCK_EVALUATION_OK_MORE ==
          GNUNET_BLOCK_evaluate(GDS_block_context,
                                GNUNET_BLOCK_TYPE_DHT_HELLO,
                                bg,
                                GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO,
                                &my_identity_hash,
                                NULL, 0,
                                GDS_my_hello,
                                hello_size))
        {
          GDS_NEIGHBOURS_handle_reply(sender,
                                      GNUNET_BLOCK_TYPE_DHT_HELLO,
                                      GNUNET_TIME_relative_to_absolute(hello_expiration),
                                      key,
                                      0,
                                      NULL,
                                      0,
                                      NULL,
                                      GDS_my_hello,
                                      hello_size);
        }
      else
        {
          GNUNET_STATISTICS_update(GDS_stats,
                                   gettext_noop("# FIND PEER requests ignored due to Bloomfilter"),
                                   1,
                                   GNUNET_NO);
        }
    }
  else
    {
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# FIND PEER requests ignored due to lack of HELLO"),
                               1,
                               GNUNET_NO);
    }

  /* then, also consider sending a random HELLO from the closest bucket */
  if (0 == memcmp(&my_identity_hash,
                  key,
                  sizeof(struct GNUNET_HashCode)))
    bucket_idx = closest_bucket;
  else
    bucket_idx = GNUNET_MIN((int)closest_bucket,
                            find_bucket(key));
  if (bucket_idx < 0)
    return;
  bucket = &k_buckets[bucket_idx];
  if (bucket->peers_size == 0)
    return;
  choice = GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK,
                                    bucket->peers_size);
  peer = bucket->head;
  while (choice > 0)
    {
      GNUNET_assert(NULL != peer);
      peer = peer->next;
      choice--;
    }
  choice = bucket->peers_size;
  do
    {
      peer = peer->next;
      if (0 == choice--)
        return;                 /* no non-masked peer available */
      if (NULL == peer)
        peer = bucket->head;
      hello = GDS_HELLO_get(peer->id);
    }
  while ((NULL == hello) ||
         (GNUNET_BLOCK_EVALUATION_OK_MORE !=
          GNUNET_BLOCK_evaluate(GDS_block_context,
                                GNUNET_BLOCK_TYPE_DHT_HELLO,
                                bg,
                                GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO,
                                &peer->phash,
                                NULL, 0,
                                hello,
                                (hello_size = GNUNET_HELLO_size(hello)))));
  GDS_NEIGHBOURS_handle_reply(sender,
                              GNUNET_BLOCK_TYPE_DHT_HELLO,
                              GNUNET_TIME_relative_to_absolute
                                (GNUNET_CONSTANTS_HELLO_ADDRESS_EXPIRATION),
                              key,
                              0,
                              NULL,
                              0,
                              NULL,
                              hello,
                              hello_size);
}


static void
handle_local_result(void *cls,
                    enum GNUNET_BLOCK_Type type,
                    struct GNUNET_TIME_Absolute expiration_time,
                    const struct GNUNET_HashCode *key,
                    unsigned int put_path_length,
                    const struct GNUNET_PeerIdentity *put_path,
                    unsigned int get_path_length,
                    const struct GNUNET_PeerIdentity *get_path,
                    const void *data,
                    size_t data_size)
{
  struct PeerInfo *peer = cls;
  char *pp;

  pp = GNUNET_STRINGS_pp2s(put_path,
                           put_path_length);
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Found local result for %s (PP: %s)\n",
             GNUNET_h2s(key),
             pp);
  GNUNET_free(pp);
  GDS_NEIGHBOURS_handle_reply(peer->id,
                              type,
                              expiration_time,
                              key,
                              put_path_length, put_path,
                              get_path_length, get_path,
                              data, data_size);
}


static int
check_dht_p2p_get(void *cls,
                  const struct PeerGetMessage *get)
{
  uint32_t xquery_size;
  uint16_t msize;

  (void)cls;
  msize = ntohs(get->header.size);
  xquery_size = ntohl(get->xquery_size);
  if (msize < sizeof(struct PeerGetMessage) + xquery_size)
    {
      GNUNET_break_op(0);
      return GNUNET_SYSERR;
    }
  return GNUNET_OK;
}


static void
handle_dht_p2p_get(void *cls,
                   const struct PeerGetMessage *get)
{
  struct PeerInfo *peer = cls;
  uint32_t xquery_size;
  size_t reply_bf_size;
  uint16_t msize;
  enum GNUNET_BLOCK_Type type;
  enum GNUNET_DHT_RouteOption options;
  enum GNUNET_BLOCK_EvaluationResult eval;
  struct GNUNET_BLOCK_Group *bg;
  struct GNUNET_CONTAINER_BloomFilter *peer_bf;
  const char *xquery;
  int forwarded;

  /* parse and validate message */
  msize = ntohs(get->header.size);
  xquery_size = ntohl(get->xquery_size);
  reply_bf_size = msize - (sizeof(struct PeerGetMessage) + xquery_size);
  type = ntohl(get->type);
  options = ntohl(get->options);
  xquery = (const char *)&get[1];
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# P2P GET requests received"),
                           1,
                           GNUNET_NO);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# P2P GET bytes received"),
                           msize,
                           GNUNET_NO);
  if (GNUNET_YES == log_route_details_stderr)
    {
      char *tmp;

      tmp = GNUNET_strdup(GNUNET_i2s(&my_identity));
      LOG_TRAFFIC(GNUNET_ERROR_TYPE_DEBUG,
                  "R5N GET %s: %s->%s (%u, %u=>%u) xq: %.*s\n",
                  GNUNET_h2s(&get->key),
                  GNUNET_i2s(peer->id),
                  tmp,
                  ntohl(get->hop_count),
                  GNUNET_CRYPTO_hash_matching_bits(&peer->phash,
                                                   &get->key),
                  GNUNET_CRYPTO_hash_matching_bits(&my_identity_hash,
                                                   &get->key),
                  ntohl(get->xquery_size),
                  xquery);
      GNUNET_free(tmp);
    }
  eval
    = GNUNET_BLOCK_evaluate(GDS_block_context,
                            type,
                            NULL,
                            GNUNET_BLOCK_EO_NONE,
                            &get->key,
                            xquery,
                            xquery_size,
                            NULL,
                            0);
  if (eval != GNUNET_BLOCK_EVALUATION_REQUEST_VALID)
    {
      /* request invalid or block type not supported */
      GNUNET_break_op(eval == GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED);
      return;
    }
  peer_bf = GNUNET_CONTAINER_bloomfilter_init(get->bloomfilter,
                                              DHT_BLOOM_SIZE,
                                              GNUNET_CONSTANTS_BLOOMFILTER_K);
  GNUNET_break_op(GNUNET_YES ==
                  GNUNET_CONTAINER_bloomfilter_test(peer_bf,
                                                    &peer->phash));
  bg = GNUNET_BLOCK_group_create(GDS_block_context,
                                 type,
                                 get->bf_mutator,
                                 &xquery[xquery_size],
                                 reply_bf_size,
                                 "filter-size",
                                 reply_bf_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)ntohl(get->hop_count));
  /* local lookup (this may update the reply_bf) */
  if ((0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
      (GDS_am_closest_peer(&get->key,
                           peer_bf)))
    {
      if ((0 != (options & GNUNET_DHT_RO_FIND_PEER)))
        {
          GNUNET_STATISTICS_update(GDS_stats,
                                   gettext_noop("# P2P FIND PEER requests processed"),
                                   1,
                                   GNUNET_NO);
          handle_find_peer(peer->id,
                           &get->key,
                           bg);
        }
      else
        {
          eval = GDS_DATACACHE_handle_get(&get->key,
                                          type,
                                          xquery,
                                          xquery_size,
                                          bg,
                                          &handle_local_result,
                                          peer);
        }
    }
  else
    {
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# P2P GET requests ONLY routed"),
                               1,
                               GNUNET_NO);
    }

  /* remember request for routing replies */
  GDS_ROUTING_add(peer->id,
                  type,
                  bg,       /* bg now owned by routing, but valid at least until end of this function! */
                  options,
                  &get->key,
                  xquery,
                  xquery_size);

  /* P2P forwarding */
  forwarded = GNUNET_NO;
  if (eval != GNUNET_BLOCK_EVALUATION_OK_LAST)
    forwarded = GDS_NEIGHBOURS_handle_get(type,
                                          options,
                                          ntohl(get->desired_replication_level),
                                          ntohl(get->hop_count),
                                          &get->key,
                                          xquery,
                                          xquery_size,
                                          bg,
                                          peer_bf);
  GDS_CLIENTS_process_get(options
                          | (GNUNET_OK == forwarded)
                          ? GNUNET_DHT_RO_LAST_HOP : 0,
                          type,
                          ntohl(get->hop_count),
                          ntohl(get->desired_replication_level),
                          0,
                          NULL,
                          &get->key);

  /* clean up; note that 'bg' is owned by routing now! */
  GNUNET_CONTAINER_bloomfilter_free(peer_bf);
}


static int
check_dht_p2p_result(void *cls,
                     const struct PeerResultMessage *prm)
{
  uint32_t get_path_length;
  uint32_t put_path_length;
  uint16_t msize;

  (void)cls;
  msize = ntohs(prm->header.size);
  put_path_length = ntohl(prm->put_path_length);
  get_path_length = ntohl(prm->get_path_length);
  if ((msize <
       sizeof(struct PeerResultMessage) + (get_path_length +
                                           put_path_length) *
       sizeof(struct GNUNET_PeerIdentity)) ||
      (get_path_length >
       GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)) ||
      (put_path_length >
       GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)))
    {
      GNUNET_break_op(0);
      return GNUNET_SYSERR;
    }
  return GNUNET_OK;
}


static void
process_reply_with_path(struct GNUNET_TIME_Absolute expiration_time,
                        const struct GNUNET_HashCode *key,
                        unsigned int get_path_length,
                        const struct GNUNET_PeerIdentity *get_path,
                        unsigned int put_path_length,
                        const struct GNUNET_PeerIdentity *put_path,
                        enum GNUNET_BLOCK_Type type,
                        size_t data_size,
                        const void *data)
{
  /* forward to local clients */
  GDS_CLIENTS_handle_reply(expiration_time,
                           key,
                           get_path_length,
                           get_path,
                           put_path_length,
                           put_path,
                           type,
                           data_size,
                           data);
  GDS_CLIENTS_process_get_resp(type,
                               get_path,
                               get_path_length,
                               put_path,
                               put_path_length,
                               expiration_time,
                               key,
                               data,
                               data_size);
  if (GNUNET_YES == cache_results)
    {
      struct GNUNET_PeerIdentity xput_path[get_path_length + 1 + put_path_length];

      GNUNET_memcpy(xput_path,
                    put_path,
                    put_path_length * sizeof(struct GNUNET_PeerIdentity));
      GNUNET_memcpy(&xput_path[put_path_length],
                    get_path,
                    get_path_length * sizeof(struct GNUNET_PeerIdentity));

      GDS_DATACACHE_handle_put(expiration_time,
                               key,
                               get_path_length + put_path_length,
                               xput_path,
                               type,
                               data_size,
                               data);
    }
  /* forward to other peers */
  GDS_ROUTING_process(type,
                      expiration_time,
                      key,
                      put_path_length,
                      put_path,
                      get_path_length,
                      get_path,
                      data,
                      data_size);
}


static void
handle_dht_p2p_result(void *cls,
                      const struct PeerResultMessage *prm)
{
  struct PeerInfo *peer = cls;
  const struct GNUNET_PeerIdentity *put_path;
  const struct GNUNET_PeerIdentity *get_path;
  const void *data;
  uint32_t get_path_length;
  uint32_t put_path_length;
  uint16_t msize;
  size_t data_size;
  enum GNUNET_BLOCK_Type type;
  struct GNUNET_TIME_Absolute exp_time;

  /* parse and validate message */
  exp_time = GNUNET_TIME_absolute_ntoh(prm->expiration_time);
  if (0 == GNUNET_TIME_absolute_get_remaining(exp_time).rel_value_us)
    {
      GNUNET_STATISTICS_update(GDS_stats,
                               gettext_noop("# Expired results discarded"),
                               1,
                               GNUNET_NO);
      return;
    }
  msize = ntohs(prm->header.size);
  put_path_length = ntohl(prm->put_path_length);
  get_path_length = ntohl(prm->get_path_length);
  put_path = (const struct GNUNET_PeerIdentity *)&prm[1];
  get_path = &put_path[put_path_length];
  type = ntohl(prm->type);
  data = (const void *)&get_path[get_path_length];
  data_size = msize - (sizeof(struct PeerResultMessage) +
                       (get_path_length +
                        put_path_length) * sizeof(struct GNUNET_PeerIdentity));
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# P2P RESULTS received"),
                           1,
                           GNUNET_NO);
  GNUNET_STATISTICS_update(GDS_stats,
                           gettext_noop("# P2P RESULT bytes received"),
                           msize,
                           GNUNET_NO);
  if (GNUNET_YES == log_route_details_stderr)
    {
      char *tmp;
      char *pp;
      char *gp;

      gp = GNUNET_STRINGS_pp2s(get_path,
                               get_path_length);
      pp = GNUNET_STRINGS_pp2s(put_path,
                               put_path_length);
      tmp = GNUNET_strdup(GNUNET_i2s(&my_identity));
      LOG_TRAFFIC(GNUNET_ERROR_TYPE_DEBUG,
                  "R5N RESULT %s: %s->%s (GP: %s, PP: %s)\n",
                  GNUNET_h2s(&prm->key),
                  GNUNET_i2s(peer->id),
                  tmp,
                  gp,
                  pp);
      GNUNET_free(gp);
      GNUNET_free(pp);
      GNUNET_free(tmp);
    }
  /* if we got a HELLO, consider it for our own routing table */
  if (GNUNET_BLOCK_TYPE_DHT_HELLO == type)
    {
      const struct GNUNET_MessageHeader *h;
      struct GNUNET_PeerIdentity pid;

      /* Should be a HELLO, validate and consider using it! */
      if (data_size < sizeof(struct GNUNET_HELLO_Message))
        {
          GNUNET_break_op(0);
          return;
        }
      h = data;
      if (data_size != ntohs(h->size))
        {
          GNUNET_break_op(0);
          return;
        }
      if (GNUNET_OK !=
          GNUNET_HELLO_get_id((const struct GNUNET_HELLO_Message *)h,
                              &pid))
        {
          GNUNET_break_op(0);
          return;
        }
      if ((GNUNET_YES != disable_try_connect) &&
          (0 != memcmp(&my_identity,
                       &pid,
                       sizeof(struct GNUNET_PeerIdentity))))
        try_connect(&pid,
                    h);
    }

  /* First, check if 'peer' is already on the path, and if
     so, truncate it instead of expanding. */
  for (unsigned int i = 0; i <= get_path_length; i++)
    if (0 == memcmp(&get_path[i],
                    peer->id,
                    sizeof(struct GNUNET_PeerIdentity)))
      {
        process_reply_with_path(exp_time,
                                &prm->key,
                                i,
                                get_path,
                                put_path_length,
                                put_path,
                                type,
                                data_size,
                                data);
        return;
      }

  /* Need to append 'peer' to 'get_path' (normal case) */
  {
    struct GNUNET_PeerIdentity xget_path[get_path_length + 1];

    GNUNET_memcpy(xget_path,
                  get_path,
                  get_path_length * sizeof(struct GNUNET_PeerIdentity));
    xget_path[get_path_length] = *peer->id;

    process_reply_with_path(exp_time,
                            &prm->key,
                            get_path_length + 1,
                            xget_path,
                            put_path_length,
                            put_path,
                            type,
                            data_size,
                            data);
  }
}


int
GDS_NEIGHBOURS_init()
{
  struct GNUNET_MQ_MessageHandler core_handlers[] = {
    GNUNET_MQ_hd_var_size(dht_p2p_get,
                          GNUNET_MESSAGE_TYPE_DHT_P2P_GET,
                          struct PeerGetMessage,
                          NULL),
    GNUNET_MQ_hd_var_size(dht_p2p_put,
                          GNUNET_MESSAGE_TYPE_DHT_P2P_PUT,
                          struct PeerPutMessage,
                          NULL),
    GNUNET_MQ_hd_var_size(dht_p2p_result,
                          GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT,
                          struct PeerResultMessage,
                          NULL),
    GNUNET_MQ_handler_end()
  };
  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");

  log_route_details_stderr =
    (NULL != getenv("GNUNET_DHT_ROUTE_DEBUG")) ? GNUNET_YES : GNUNET_NO;
  ats_ch = GNUNET_ATS_connectivity_init(GDS_cfg);
  core_api = GNUNET_CORE_connect(GDS_cfg,
                                 NULL,
                                 &core_init,
                                 &handle_core_connect,
                                 &handle_core_disconnect,
                                 core_handlers);
  if (NULL == core_api)
    return GNUNET_SYSERR;
  all_connected_peers = GNUNET_CONTAINER_multipeermap_create(256,
                                                             GNUNET_YES);
  all_desired_peers = GNUNET_CONTAINER_multipeermap_create(256,
                                                           GNUNET_NO);
  return GNUNET_OK;
}


void
GDS_NEIGHBOURS_done()
{
  if (NULL == core_api)
    return;
  GNUNET_CORE_disconnect(core_api);
  core_api = NULL;
  GNUNET_assert(0 ==
                GNUNET_CONTAINER_multipeermap_size(all_connected_peers));
  GNUNET_CONTAINER_multipeermap_destroy(all_connected_peers);
  all_connected_peers = NULL;
  GNUNET_CONTAINER_multipeermap_iterate(all_desired_peers,
                                        &free_connect_info,
                                        NULL);
  GNUNET_CONTAINER_multipeermap_destroy(all_desired_peers);
  all_desired_peers = NULL;
  GNUNET_ATS_connectivity_done(ats_ch);
  ats_ch = NULL;
  GNUNET_assert(NULL == find_peer_task);
}


struct GNUNET_PeerIdentity *
GDS_NEIGHBOURS_get_id()
{
  return &my_identity;
}


/* end of gnunet-service-dht_neighbours.c */