gnunet-service-dht_clients.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2009-2011, 2016-2017, 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-service-dht_clients.h"
#include "gnunet_pils_service.h"
#include "gnunet_util_lib.h"
 
#define SANITY_CHECKS 0
 
#define LOG_TRAFFIC(kind, ...) GNUNET_log_from (kind, "dht-traffic", \
                                                __VA_ARGS__)
 
#define LOG(kind, ...) GNUNET_log_from (kind, "dht-clients", __VA_ARGS__)
 
#define UNROUTED_RETRY_INTERVAL GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_SECONDS, 5)
 
#define MAXIMUM_GET_RETRY_INTERVAL GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_SECONDS, 5)
 
#define FAST_RETRY_WINDOW GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_MINUTES, 1)
 
#define UNROUTED_PUT_TIMEOUT GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_MINUTES, 5)
 
#define UNROUTED_PUT_RETRY_INTERVAL GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_SECONDS, 5)
 
#define MAX_PENDING_PUTS 1024
 
struct ClientHandle;
 
 
struct ClientQueryRecord
{
  struct GNUNET_HashCode key;
 
  struct ClientQueryRecord *next;
 
  struct ClientQueryRecord *prev;
 
  struct ClientHandle *ch;
 
  const void *xquery;
 
  struct GNUNET_HashCode *seen_replies;
 
  struct GNUNET_CONTAINER_HeapNode *hnode;
 
  struct GNUNET_TIME_Relative retry_frequency;
 
  struct GNUNET_TIME_Absolute retry_time;
 
  struct GNUNET_TIME_Absolute start_time;
 
  uint64_t unique_id;
 
  size_t xquery_size;
 
  unsigned int seen_replies_count;
 
  uint32_t replication;
 
  enum GNUNET_DHT_RouteOption msg_options;
 
  enum GNUNET_BLOCK_Type type;
};
 
 
struct ClientMonitorRecord
{
  struct ClientMonitorRecord *next;
 
  struct ClientMonitorRecord *prev;
 
  struct ClientHandle *ch;
 
  struct GNUNET_HashCode key;
 
  enum GNUNET_BLOCK_Type type;
 
  int16_t get;
 
  int16_t get_resp;
 
  uint16_t put;
 
};
 
 
struct ClientHandle
{
  struct ClientQueryRecord *cqr_head;
 
  struct ClientQueryRecord *cqr_tail;
 
  struct GNUNET_SERVICE_Client *client;
 
  struct GNUNET_MQ_Handle *mq;
};
 
 
struct GNUNET_BLOCK_Context *GDS_block_context;
 
struct GNUNET_STATISTICS_Handle *GDS_stats;
 
struct GNUNET_PILS_Handle *GDS_pils;
 
struct GNUNET_SERVICE_Handle *GDS_service;
 
const struct GNUNET_CONFIGURATION_Handle *GDS_cfg;
 
static struct ClientMonitorRecord *monitor_head;
 
static struct ClientMonitorRecord *monitor_tail;
 
static struct GNUNET_CONTAINER_MultiHashMap *forward_map;
 
static struct GNUNET_CONTAINER_Heap *retry_heap;
 
static struct GNUNET_SCHEDULER_Task *retry_task;
 
struct PendingPut
{
  struct PendingPut *next;
 
  struct PendingPut *prev;
 
  struct GNUNET_DATACACHE_Block bd;
 
  struct GNUNET_TIME_Absolute give_up_time;
 
  struct GNUNET_TIME_Absolute retry_time;
 
  struct GNUNET_TIME_Relative retry_frequency;
 
  uint32_t replication_level;
};
 
static struct PendingPut *pp_head;
 
static struct PendingPut *pp_tail;
 
static unsigned int pp_count;
 
static struct GNUNET_SCHEDULER_Task *put_retry_task;
 
 
static void
remove_client_query_record (struct ClientQueryRecord *record)
{
  struct ClientHandle *ch = record->ch;
 
  GNUNET_CONTAINER_DLL_remove (ch->cqr_head,
                               ch->cqr_tail,
                               record);
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multihashmap_remove (forward_map,
                                                       &record->key,
                                                       record));
  if (NULL != record->hnode)
    GNUNET_CONTAINER_heap_remove_node (record->hnode);
  GNUNET_array_grow (record->seen_replies,
                     record->seen_replies_count,
                     0);
  GNUNET_free (record);
}
 
 
static void *
client_connect_cb (void *cls,
                   struct GNUNET_SERVICE_Client *client,
                   struct GNUNET_MQ_Handle *mq)
{
  struct ClientHandle *ch;
 
  (void) cls;
  ch = GNUNET_new (struct ClientHandle);
  ch->client = client;
  ch->mq = mq;
  return ch;
}
 
 
static void
client_disconnect_cb (void *cls,
                      struct GNUNET_SERVICE_Client *client,
                      void *app_ctx)
{
  struct ClientHandle *ch = app_ctx;
 
  (void) cls;
  (void) client;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Local client %p disconnects\n",
              ch);
  {
    struct ClientMonitorRecord *next;
 
    for (struct ClientMonitorRecord *monitor = monitor_head;
         NULL != monitor;
         monitor = next)
    {
      next = monitor->next;
      if (monitor->ch != ch)
        continue;
      GNUNET_CONTAINER_DLL_remove (monitor_head,
                                   monitor_tail,
                                   monitor);
      GNUNET_free (monitor);
    }
  }
 
  {
    struct ClientQueryRecord *cqr;
 
    while (NULL != (cqr = ch->cqr_head))
      remove_client_query_record (cqr);
  }
  GNUNET_free (ch);
}
 
 
static void
transmit_request (struct ClientQueryRecord *cqr)
{
  struct GNUNET_BLOCK_Group *bg;
  struct GNUNET_CONTAINER_BloomFilter *peer_bf;
  enum GNUNET_GenericReturnValue routed;
 
  GNUNET_STATISTICS_update (GDS_stats,
                            "# GET requests from clients injected",
                            1,
                            GNUNET_NO);
  bg = GNUNET_BLOCK_group_create (GDS_block_context,
                                  cqr->type,
                                  NULL, /* raw data */
                                  0, /* raw data size */
                                  "seen-set-size",
                                  cqr->seen_replies_count,
                                  NULL);
  GNUNET_BLOCK_group_set_seen (bg,
                               cqr->seen_replies,
                               cqr->seen_replies_count);
  peer_bf
    = GNUNET_CONTAINER_bloomfilter_init (NULL,
                                         DHT_BLOOM_SIZE,
                                         GNUNET_CONSTANTS_BLOOMFILTER_K);
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Initiating GET for %s, replication %u, already have %u replies\n",
       GNUNET_h2s (&cqr->key),
       cqr->replication,
       cqr->seen_replies_count);
  routed = GDS_NEIGHBOURS_handle_get (cqr->type,
                                      cqr->msg_options,
                                      cqr->replication,
                                      0 /* hop count */,
                                      &cqr->key,
                                      cqr->xquery,
                                      cqr->xquery_size,
                                      bg,
                                      peer_bf);
  GNUNET_BLOCK_group_destroy (bg);
  GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 
  /* Exponential back-off for retries.
   * max. is #GNUNET_TIME_STD_EXPONENTIAL_BACKOFF_THRESHOLD (15 min) */
  cqr->retry_frequency = GNUNET_TIME_STD_BACKOFF (cqr->retry_frequency);
  if (GNUNET_TIME_relative_cmp (
        GNUNET_TIME_absolute_get_duration (cqr->start_time),
        <,
        FAST_RETRY_WINDOW))
  {
    /* Somebody is still waiting for an answer; do not let the back-off
       turn one lost attempt into tens of seconds of latency. */
    cqr->retry_frequency
      = GNUNET_TIME_relative_min (MAXIMUM_GET_RETRY_INTERVAL,
                                  cqr->retry_frequency);
  }
  if (GNUNET_OK != routed)
  {
    /* We had no peer to route this to, so nothing was sent and nothing
       can answer.  Cap the interval so we come back as soon as the
       routing table recovers instead of idling for up to 15 minutes. */
    GNUNET_STATISTICS_update (GDS_stats,
                              "# GET requests from clients not routed",
                              1,
                              GNUNET_NO);
    cqr->retry_frequency
      = GNUNET_TIME_relative_min (UNROUTED_RETRY_INTERVAL,
                                  cqr->retry_frequency);
  }
  cqr->retry_time = GNUNET_TIME_relative_to_absolute (cqr->retry_frequency);
}
 
 
static void
transmit_next_request_task (void *cls)
{
  struct ClientQueryRecord *cqr;
 
  (void) cls;
  retry_task = NULL;
  while (NULL != (cqr = GNUNET_CONTAINER_heap_remove_root (retry_heap)))
  {
    cqr->hnode = NULL;
    if (! GNUNET_TIME_absolute_is_past (cqr->retry_time))
    {
      cqr->hnode
        = GNUNET_CONTAINER_heap_insert (retry_heap,
                                        cqr,
                                        cqr->retry_time.abs_value_us);
      retry_task
        = GNUNET_SCHEDULER_add_at (cqr->retry_time,
                                   &transmit_next_request_task,
                                   NULL);
      return;
    }
    transmit_request (cqr);
    cqr->hnode
      = GNUNET_CONTAINER_heap_insert (retry_heap,
                                      cqr,
                                      cqr->retry_time.abs_value_us);
  }
}
 
 
static enum GNUNET_GenericReturnValue
reset_retry_backoff (void *cls,
                     const struct GNUNET_HashCode *key,
                     void *value)
{
  struct ClientQueryRecord *cqr = value;
 
  (void) cls;
  (void) key;
  cqr->retry_frequency = GNUNET_TIME_UNIT_SECONDS;
  cqr->retry_time = GNUNET_TIME_absolute_get ();
  if (NULL != cqr->hnode)
    GNUNET_CONTAINER_heap_update_cost (cqr->hnode,
                                       cqr->retry_time.abs_value_us);
  return GNUNET_YES;
}
 
 
void
GDS_CLIENTS_retry_requests (void)
{
  if (0 == GNUNET_CONTAINER_multihashmap_size (forward_map))
    return;
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Routing table changed, re-transmitting %u pending GET request(s)\n",
       GNUNET_CONTAINER_multihashmap_size (forward_map));
  GNUNET_STATISTICS_update (GDS_stats,
                            "# GET requests retried after routing table change",
                            GNUNET_CONTAINER_multihashmap_size (forward_map),
                            GNUNET_NO);
  GNUNET_CONTAINER_multihashmap_iterate (forward_map,
                                         &reset_retry_backoff,
                                         NULL);
  if (NULL != retry_task)
    GNUNET_SCHEDULER_cancel (retry_task);
  retry_task = GNUNET_SCHEDULER_add_now (&transmit_next_request_task,
                                         NULL);
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_local_put (void *cls,
                     const struct GNUNET_DHT_ClientPutMessage *dht_msg)
{
  uint32_t replication_level = ntohl (dht_msg->desired_replication_level);
 
  (void) cls;
  if (replication_level > GNUNET_DHT_MAXIMUM_REPLICATION_LEVEL)
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
route_pending_put (struct PendingPut *pp);
 
 
static void
put_retry_task_cb (void *cls)
{
  struct PendingPut *ready_head = NULL;
  struct PendingPut *ready_tail = NULL;
  struct GNUNET_TIME_Absolute next = GNUNET_TIME_UNIT_FOREVER_ABS;
  struct PendingPut *nxt;
 
  (void) cls;
  put_retry_task = NULL;
  /* Splice the entries that are due out of the DLL first: routing one of
     them can re-queue it (and does, whenever it still finds no route). */
  for (struct PendingPut *pp = pp_head; NULL != pp; pp = nxt)
  {
    nxt = pp->next;
    if (! GNUNET_TIME_absolute_is_past (pp->retry_time))
      continue;
    GNUNET_CONTAINER_DLL_remove (pp_head,
                                 pp_tail,
                                 pp);
    pp_count--;
    GNUNET_CONTAINER_DLL_insert_tail (ready_head,
                                      ready_tail,
                                      pp);
  }
  while (NULL != (nxt = ready_head))
  {
    GNUNET_CONTAINER_DLL_remove (ready_head,
                                 ready_tail,
                                 nxt);
    route_pending_put (nxt);
  }
  /* Re-queueing above may have armed the task again; we know better. */
  if (NULL != put_retry_task)
  {
    GNUNET_SCHEDULER_cancel (put_retry_task);
    put_retry_task = NULL;
  }
  if (NULL == pp_head)
    return;
  for (struct PendingPut *pp = pp_head; NULL != pp; pp = pp->next)
    next = GNUNET_TIME_absolute_min (next,
                                     pp->retry_time);
  put_retry_task = GNUNET_SCHEDULER_add_at (next,
                                            &put_retry_task_cb,
                                            NULL);
}
 
 
static void
schedule_put_retry (void)
{
  if (NULL != put_retry_task)
    GNUNET_SCHEDULER_cancel (put_retry_task);
  put_retry_task = GNUNET_SCHEDULER_add_now (&put_retry_task_cb,
                                             NULL);
}
 
 
static void
cb_forwarded_dht_local_put (void *cls,
                            enum GNUNET_GenericReturnValue forwarded)
{
  struct PendingPut *pp = cls;
 
  if (GNUNET_OK == forwarded)
  {
    GNUNET_free (pp);
    return;
  }
  /* Nothing was sent, so nothing can store this block and no retry
     happens anywhere else: the DHT keeps no state about a PUT, and the
     client's continuation has already reported success (it fires when
     the request left the client, not when it was routed).  Hold on to
     the block and try again when we have somewhere to send it. */
  GNUNET_STATISTICS_update (GDS_stats,
                            "# Local PUT requests not routed",
                            1,
                            GNUNET_NO);
  if ( (GNUNET_TIME_absolute_is_past (pp->give_up_time)) ||
       (GNUNET_TIME_absolute_is_past (pp->bd.expiration_time)) ||
       (pp_count >= MAX_PENDING_PUTS) )
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Local PUT requests dropped (never routed)",
                              1,
                              GNUNET_NO);
    GNUNET_free (pp);
    return;
  }
  pp->retry_frequency
    = GNUNET_TIME_relative_min (UNROUTED_PUT_RETRY_INTERVAL,
                                GNUNET_TIME_STD_BACKOFF (pp->retry_frequency));
  pp->retry_time = GNUNET_TIME_relative_to_absolute (pp->retry_frequency);
  GNUNET_CONTAINER_DLL_insert (pp_head,
                               pp_tail,
                               pp);
  pp_count++;
  schedule_put_retry ();
}
 
 
static void
route_pending_put (struct PendingPut *pp)
{
  struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 
  peer_bf
    = GNUNET_CONTAINER_bloomfilter_init (NULL,
                                         DHT_BLOOM_SIZE,
                                         GNUNET_CONSTANTS_BLOOMFILTER_K);
  /* Takes over @a pp via the callback, which may run synchronously. */
  GDS_NEIGHBOURS_handle_put (&pp->bd,
                             pp->replication_level,
                             0 /* hop count */,
                             peer_bf,
                             &cb_forwarded_dht_local_put,
                             pp);
  GNUNET_CONTAINER_bloomfilter_free (peer_bf);
}
 
 
void
GDS_CLIENTS_retry_puts (void)
{
  if (NULL == pp_head)
    return;
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Routing table changed, re-trying %u unrouted PUT(s)\n",
       pp_count);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# PUT requests retried after routing table change",
                            pp_count,
                            GNUNET_NO);
  for (struct PendingPut *pp = pp_head; NULL != pp; pp = pp->next)
  {
    pp->retry_frequency = GNUNET_TIME_UNIT_SECONDS;
    pp->retry_time = GNUNET_TIME_absolute_get ();
  }
  schedule_put_retry ();
}
 
 
static void
handle_dht_local_put (void *cls,
                      const struct GNUNET_DHT_ClientPutMessage *dht_msg)
{
  struct ClientHandle *ch = cls;
  uint16_t size = ntohs (dht_msg->header.size);
  uint32_t replication_level
    = ntohl (dht_msg->desired_replication_level);
  struct GNUNET_DATACACHE_Block bd = {
    .key = dht_msg->key,
    .expiration_time = GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
    .data = &dht_msg[1],
    .data_size = size - sizeof (*dht_msg),
    .type = ntohl (dht_msg->type),
    .ro = ntohl (dht_msg->options)
  };
 
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Handling local PUT of %lu-bytes for query %s of type %u\n",
       (unsigned long) (size - sizeof(struct GNUNET_DHT_ClientPutMessage)),
       GNUNET_h2s (&dht_msg->key),
       (unsigned int) bd.type);
  if (NULL == GNUNET_PILS_get_identity (GDS_pils))
  {
    /* Without our own identity we can neither decide whether to store
       this block nor route it: GDS_am_closest_peer(),
       GDS_DATACACHE_handle_put() and GDS_NEIGHBOURS_handle_put() all
       require it.  Should not happen -- the service stays suspended
       until PILS answers -- but drop the request rather than abort. */
    GNUNET_break (0);
    GNUNET_STATISTICS_update (GDS_stats,
                              "# PUT requests dropped (no identity yet)",
                              1,
                              GNUNET_NO);
    GNUNET_SERVICE_client_continue (ch->client);
    return;
  }
#if SANITY_CHECKS > 0
  if (GNUNET_OK !=
      GNUNET_BLOCK_check_block (GDS_block_context,
                                bd.type,
                                bd.data,
                                bd.data_size))
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_continue (ch->client);
    return;
  }
#endif
  GNUNET_STATISTICS_update (GDS_stats,
                            "# PUT requests received from clients",
                            1,
                            GNUNET_NO);
  LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
               "CLIENT-PUT %s\n",
               GNUNET_h2s_full (&dht_msg->key));
  /* give to local clients */
  GNUNET_break (GDS_CLIENTS_handle_reply (&bd,
                                          &bd.key,
                                          0, NULL /* get path */));
 
  {
    struct GNUNET_CONTAINER_BloomFilter *peer_bf;
    struct PendingPut *pp;
 
    peer_bf
      = GNUNET_CONTAINER_bloomfilter_init (NULL,
                                           DHT_BLOOM_SIZE,
                                           GNUNET_CONSTANTS_BLOOMFILTER_K);
    /* store locally */
    if ( (0 != (bd.ro & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
         (GDS_am_closest_peer (&dht_msg->key,
                               peer_bf)))
      GDS_DATACACHE_handle_put (&bd);
    GNUNET_CONTAINER_bloomfilter_free (peer_bf);
    /* route to other peers; the copy survives a failure to route so that
       we can try again, the client message does not */
    pp = GNUNET_malloc (sizeof (*pp) + bd.data_size);
    pp->bd = bd;
    pp->bd.data = &pp[1];
    GNUNET_memcpy (&pp[1],
                   bd.data,
                   bd.data_size);
    pp->replication_level = replication_level;
    pp->retry_frequency = GNUNET_TIME_UNIT_SECONDS;
    pp->give_up_time
      = GNUNET_TIME_relative_to_absolute (UNROUTED_PUT_TIMEOUT);
    route_pending_put (pp);
  }
  GDS_CLIENTS_process_put (
    &bd,
    0,  /* hop count */
    replication_level);
  GNUNET_SERVICE_client_continue (ch->client);
}
 
 
static void
handle_local_result (void *cls,
                     const struct GNUNET_DATACACHE_Block *bd)
{
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Datacache provided result for query key %s\n",
              GNUNET_h2s (&bd->key));
  GNUNET_break (GDS_CLIENTS_handle_reply (bd,
                                          &bd->key,
                                          0, NULL /* get_path */));
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_local_get (void *cls,
                     const struct GNUNET_DHT_ClientGetMessage *get)
{
  (void) cls;
  (void) get;
  /* always well-formed */
  return GNUNET_OK;
}
 
 
static void
handle_dht_local_get (void *cls,
                      const struct GNUNET_DHT_ClientGetMessage *get)
{
  struct ClientHandle *ch = cls;
  struct ClientQueryRecord *cqr;
  uint16_t size = ntohs (get->header.size);
  const char *xquery = (const char *) &get[1];
  size_t xquery_size = size - sizeof(struct GNUNET_DHT_ClientGetMessage);
 
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Received GET request for %s from local client %p, xq: %.*s\n",
       GNUNET_h2s (&get->key),
       ch->client,
       (int) xquery_size,
       xquery);
  GNUNET_STATISTICS_update (GDS_stats,
                            "# GET requests received from clients",
                            1,
                            GNUNET_NO);
  LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
               "CLIENT-GET %s\n",
               GNUNET_h2s_full (&get->key));
  if (NULL == GNUNET_PILS_get_identity (GDS_pils))
  {
    /* GDS_NEIGHBOURS_handle_get() requires our identity.  Should not
       happen -- the service stays suspended until PILS answers -- but
       drop the request rather than abort. */
    GNUNET_break (0);
    GNUNET_STATISTICS_update (GDS_stats,
                              "# GET requests dropped (no identity yet)",
                              1,
                              GNUNET_NO);
    GNUNET_SERVICE_client_continue (ch->client);
    return;
  }
 
  cqr = GNUNET_malloc (sizeof(struct ClientQueryRecord) + xquery_size);
  cqr->key = get->key;
  cqr->ch = ch;
  cqr->xquery = (const void *) &cqr[1];
  GNUNET_memcpy (&cqr[1],
                 xquery,
                 xquery_size);
  cqr->hnode = GNUNET_CONTAINER_heap_insert (retry_heap,
                                             cqr,
                                             0);
  cqr->retry_frequency = GNUNET_TIME_UNIT_SECONDS;
  cqr->retry_time = GNUNET_TIME_absolute_get ();
  cqr->start_time = cqr->retry_time;
  cqr->unique_id = get->unique_id;
  cqr->xquery_size = xquery_size;
  cqr->replication = ntohl (get->desired_replication_level);
  cqr->msg_options = ntohl (get->options);
  cqr->type = ntohl (get->type);
  GNUNET_CONTAINER_DLL_insert (ch->cqr_head,
                               ch->cqr_tail,
                               cqr);
  GNUNET_CONTAINER_multihashmap_put (forward_map,
                                     &cqr->key,
                                     cqr,
                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  GDS_CLIENTS_process_get (cqr->msg_options,
                           cqr->type,
                           0, /* hop count */
                           cqr->replication,
                           &get->key);
  /* start remote requests */
  if (NULL != retry_task)
    GNUNET_SCHEDULER_cancel (retry_task);
  retry_task = GNUNET_SCHEDULER_add_now (&transmit_next_request_task,
                                         NULL);
  /* perform local lookup */
  GDS_DATACACHE_handle_get (&get->key,
                            cqr->type,
                            cqr->xquery,
                            xquery_size,
                            NULL,
                            &handle_local_result,
                            ch);
  GNUNET_SERVICE_client_continue (ch->client);
}
 
 
struct FindByUniqueIdContext
{
  struct ClientQueryRecord *cqr;
 
  uint64_t unique_id;
};
 
 
static enum GNUNET_GenericReturnValue
find_by_unique_id (void *cls,
                   const struct GNUNET_HashCode *key,
                   void *value)
{
  struct FindByUniqueIdContext *fui_ctx = cls;
  struct ClientQueryRecord *cqr = value;
 
  if (cqr->unique_id != fui_ctx->unique_id)
    return GNUNET_YES;
  fui_ctx->cqr = cqr;
  return GNUNET_NO;
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_local_get_result_seen (
  void *cls,
  const struct GNUNET_DHT_ClientGetResultSeenMessage *seen)
{
  uint16_t size = ntohs (seen->header.size);
  unsigned int hash_count =
    (size - sizeof(*seen))
    / sizeof(struct GNUNET_HashCode);
 
  if (size != sizeof(*seen) + hash_count * sizeof(struct GNUNET_HashCode))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_dht_local_get_result_seen (
  void *cls,
  const struct GNUNET_DHT_ClientGetResultSeenMessage *seen)
{
  struct ClientHandle *ch = cls;
  uint16_t size = ntohs (seen->header.size);
  unsigned int hash_count = (size - sizeof(*seen))
                            / sizeof(struct GNUNET_HashCode);
  const struct GNUNET_HashCode *hc = (const struct GNUNET_HashCode*) &seen[1];
  struct FindByUniqueIdContext fui_ctx = {
    .unique_id = seen->unique_id
  };
  unsigned int old_count;
  struct ClientQueryRecord *cqr;
 
  GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
                                              &seen->key,
                                              &find_by_unique_id,
                                              &fui_ctx);
  if (NULL == (cqr = fui_ctx.cqr))
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (ch->client);
    return;
  }
  /* finally, update 'seen' list */
  old_count = cqr->seen_replies_count;
  GNUNET_array_grow (cqr->seen_replies,
                     cqr->seen_replies_count,
                     cqr->seen_replies_count + hash_count);
  GNUNET_memcpy (&cqr->seen_replies[old_count],
                 hc,
                 sizeof(struct GNUNET_HashCode) * hash_count);
  GNUNET_SERVICE_client_continue (ch->client);
}
 
 
struct RemoveByUniqueIdContext
{
  struct ClientHandle *ch;
 
  uint64_t unique_id;
};
 
 
static enum GNUNET_GenericReturnValue
remove_by_unique_id (void *cls,
                     const struct GNUNET_HashCode *key,
                     void *value)
{
  const struct RemoveByUniqueIdContext *ctx = cls;
  struct ClientQueryRecord *cqr = value;
 
  if (cqr->unique_id != ctx->unique_id)
    return GNUNET_YES;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Removing client %p's record for key %s (by unique id)\n",
              ctx->ch->client,
              GNUNET_h2s (key));
  remove_client_query_record (cqr);
  return GNUNET_YES;
}
 
 
static void
handle_dht_local_get_stop (
  void *cls,
  const struct GNUNET_DHT_ClientGetStopMessage *dht_stop_msg)
{
  struct ClientHandle *ch = cls;
  struct RemoveByUniqueIdContext ctx;
 
  GNUNET_STATISTICS_update (GDS_stats,
                            "# GET STOP requests received from clients",
                            1,
                            GNUNET_NO);
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Received GET STOP request for %s from local client %p\n",
       GNUNET_h2s (&dht_stop_msg->key),
       ch->client);
  ctx.ch = ch;
  ctx.unique_id = dht_stop_msg->unique_id;
  GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
                                              &dht_stop_msg->key,
                                              &remove_by_unique_id,
                                              &ctx);
  GNUNET_SERVICE_client_continue (ch->client);
}
 
 
struct ForwardReplyContext
{
  const struct GNUNET_DATACACHE_Block *bd;
 
  const struct GNUNET_DHT_PathElement *get_path;
 
  unsigned int get_path_length;
 
};
 
 
static enum GNUNET_GenericReturnValue
forward_reply (void *cls,
               const struct GNUNET_HashCode *query_hash,
               void *value)
{
  struct ForwardReplyContext *frc = cls;
  struct ClientQueryRecord *record = value;
  const struct GNUNET_DATACACHE_Block *bd = frc->bd;
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_DHT_ClientResultMessage *reply;
  enum GNUNET_BLOCK_ReplyEvaluationResult eval;
  bool do_free;
  struct GNUNET_HashCode ch;
  struct GNUNET_DHT_PathElement *paths;
  bool truncated = (0 != (bd->ro & GNUNET_DHT_RO_TRUNCATED));
  size_t xsize = bd->data_size;
 
  LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
               "CLIENT-RESULT %s\n",
               GNUNET_h2s_full (&bd->key));
  if ( (record->type != GNUNET_BLOCK_TYPE_ANY) &&
       (record->type != bd->type) )
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Record type mismatch, not passing request for key %s to local client\n",
         GNUNET_h2s (&bd->key));
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Key match, type mismatches in REPLY to CLIENT",
                              1,
                              GNUNET_NO);
    return GNUNET_YES;          /* type mismatch */
  }
  if ( (0 == (record->msg_options & GNUNET_DHT_RO_FIND_APPROXIMATE)) &&
       (0 != GNUNET_memcmp (&bd->key,
                            query_hash)) )
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Inexact key match, but exact match required",
                              1,
                              GNUNET_NO);
    return GNUNET_YES;          /* type mismatch */
  }
  GNUNET_CRYPTO_hash (bd->data,
                      bd->data_size,
                      &ch);
  for (unsigned int i = 0; i < record->seen_replies_count; i++)
    if (0 ==
        GNUNET_memcmp (&record->seen_replies[i],
                       &ch))
    {
      LOG (GNUNET_ERROR_TYPE_DEBUG,
           "Duplicate reply, not passing request for key %s to local client\n",
           GNUNET_h2s (&bd->key));
      GNUNET_STATISTICS_update (GDS_stats,
                                "# Duplicate REPLIES to CLIENT request dropped",
                                1,
                                GNUNET_NO);
      return GNUNET_YES;        /* duplicate */
    }
  eval
    = GNUNET_BLOCK_check_reply (GDS_block_context,
                                record->type,
                                NULL,
                                &bd->key,
                                record->xquery,
                                record->xquery_size,
                                bd->data,
                                bd->data_size);
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Evaluation result is %d for key %s for local client's query\n",
       (int) eval,
       GNUNET_h2s (&bd->key));
  switch (eval)
  {
  case GNUNET_BLOCK_REPLY_OK_LAST:
    do_free = true;
    break;
  case GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED:
  case GNUNET_BLOCK_REPLY_OK_MORE:
    GNUNET_array_append (record->seen_replies,
                         record->seen_replies_count,
                         ch);
    do_free = false;
    break;
  case GNUNET_BLOCK_REPLY_OK_DUPLICATE:
    /* should be impossible to encounter here */
    GNUNET_break (0);
    return GNUNET_YES;
  case GNUNET_BLOCK_REPLY_IRRELEVANT:
    return GNUNET_YES;
  default:
    GNUNET_break (0);
    return GNUNET_NO;
  }
  GNUNET_STATISTICS_update (GDS_stats,
                            "# RESULTS queued for clients",
                            1,
                            GNUNET_NO);
  xsize += (frc->get_path_length + bd->put_path_length)
           * sizeof(struct GNUNET_DHT_PathElement);
  if (truncated)
    xsize += sizeof (struct GNUNET_PeerIdentity);
 
#if SUPER_REDUNDANT_CHECK
  {
    const struct GNUNET_PeerIdentity *my_identity;
    my_identity = GNUNET_PILS_get_identity (GDS_pils);
    GNUNET_assert (NULL != my_identity);
    GNUNET_break (0 ==
                  GNUNET_DHT_verify_path (bd->data,
                                          bd->data_size,
                                          bd->expiration_time,
                                          truncated
                                          ? &bd->trunc_peer
                                          : NULL,
                                          bd->put_path,
                                          bd->put_path_length,
                                          frc->get_path,
                                          frc->get_path_length,
                                          my_identity));
  }
#endif
 
  env = GNUNET_MQ_msg_extra (reply,
                             xsize,
                             GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT);
  reply->type = htonl (bd->type);
  reply->options = htonl (bd->ro);
  reply->get_path_length = htonl (frc->get_path_length);
  reply->put_path_length = htonl (bd->put_path_length);
  reply->unique_id = record->unique_id;
  reply->expiration = GNUNET_TIME_absolute_hton (bd->expiration_time);
  reply->key = *query_hash;
  if (truncated)
  {
    void *tgt = &reply[1];
 
    GNUNET_memcpy (tgt,
                   &bd->trunc_peer,
                   sizeof (struct GNUNET_PeerIdentity));
    paths = (struct GNUNET_DHT_PathElement *)
            (tgt + sizeof (struct GNUNET_PeerIdentity));
  }
  else
  {
    paths = (struct GNUNET_DHT_PathElement *) &reply[1];
  }
  GNUNET_memcpy (paths,
                 bd->put_path,
                 sizeof(struct GNUNET_DHT_PathElement)
                 * bd->put_path_length);
  GNUNET_memcpy (&paths[bd->put_path_length],
                 frc->get_path,
                 sizeof(struct GNUNET_DHT_PathElement)
                 * frc->get_path_length);
  GNUNET_memcpy (&paths[frc->get_path_length + bd->put_path_length],
                 bd->data,
                 bd->data_size);
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Sending reply to query %s for client %p\n",
       GNUNET_h2s (query_hash),
       record->ch->client);
  GNUNET_MQ_send (record->ch->mq,
                  env);
  if (GNUNET_YES == do_free)
    remove_client_query_record (record);
  return GNUNET_YES;
}
 
 
bool
GDS_CLIENTS_handle_reply (const struct GNUNET_DATACACHE_Block *bd,
                          const struct GNUNET_HashCode *query_hash,
                          unsigned int get_path_length,
                          const struct GNUNET_DHT_PathElement *get_path)
{
  struct ForwardReplyContext frc;
  size_t msize = sizeof (struct GNUNET_DHT_ClientResultMessage)
                 + bd->data_size
                 + (get_path_length + bd->put_path_length)
                 * sizeof(struct GNUNET_DHT_PathElement);
#if SANITY_CHECKS > 1
  bool truncated = (0 != (bd->ro & GNUNET_DHT_RO_TRUNCATED));
#endif
 
  if (msize >= GNUNET_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    return false;
  }
#if SANITY_CHECKS > 1
  {
    const struct GNUNET_PeerIdentity *my_identity;
    my_identity = GNUNET_PILS_get_identity (GDS_pils);
    GNUNET_assert (NULL != my_identity);
    if (0 !=
        GNUNET_DHT_verify_path (bd->data,
                                bd->data_size,
                                bd->expiration_time,
                                truncated
                                ? &bd->trunc_peer
                                : NULL,
                                bd->put_path,
                                bd->put_path_length,
                                get_path,
                                get_path_length,
                                my_identity))
    {
      GNUNET_break (0);
      return false;
    }
  }
#endif
  frc.bd = bd;
  frc.get_path = get_path;
  frc.get_path_length = get_path_length;
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Forwarding reply for query hash %s with GPL %u and PPL %u to client\n",
       GNUNET_h2s (query_hash),
       get_path_length,
       bd->put_path_length);
  if (0 ==
      GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
                                                  query_hash,
                                                  &forward_reply,
                                                  &frc))
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "No matching client for reply for query %s\n",
         GNUNET_h2s (query_hash));
    GNUNET_STATISTICS_update (GDS_stats,
                              "# REPLIES ignored for CLIENTS (no match)",
                              1,
                              GNUNET_NO);
  }
  return true;
}
 
 
/* **************** HELLO logic ***************** */
 
 
static void
handle_dht_local_hello_get (void *cls,
                            const struct GNUNET_MessageHeader *msg)
{
  struct ClientHandle *ch = cls;
  const struct GNUNET_PeerIdentity *my_identity;
  struct GNUNET_HELLO_Parser *p;
  struct GNUNET_MessageHeader *hdr;
  struct GNUNET_MQ_Envelope *env;
 
  my_identity = GNUNET_PILS_get_identity (GDS_pils);
 
  if (NULL != GDS_my_hello)
    p = GNUNET_HELLO_parser_from_msg (GDS_my_hello, my_identity);
  else
    p = NULL;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Handling request from local client for my HELLO\n");
 
  if (NULL != p)
  {
    char *url;
    size_t slen;
 
    url = GNUNET_HELLO_parser_to_url (p);
    slen = strlen (url) + 1;
 
    env = GNUNET_MQ_msg_extra (hdr,
                               slen,
                               GNUNET_MESSAGE_TYPE_DHT_CLIENT_HELLO_URL);
    memcpy (&hdr[1],
            url,
            slen);
    GNUNET_free (url);
    GNUNET_HELLO_parser_free (p);
  }
  else
  {
    env = GNUNET_MQ_msg_extra (hdr, 0, GNUNET_MESSAGE_TYPE_DHT_CLIENT_HELLO_URL)
    ;
  }
 
  GNUNET_MQ_send (ch->mq,
                  env);
  GNUNET_SERVICE_client_continue (ch->client);
}
 
 
static enum GNUNET_GenericReturnValue
check_dht_local_hello_offer (void *cls,
                             const struct GNUNET_MessageHeader *hdr)
{
  uint16_t len = ntohs (hdr->size);
  const char *buf = (const char *) &hdr[1];
 
  (void) cls;
  if ('\0' != buf[len - sizeof (*hdr) - 1])
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_dht_local_hello_offer (void *cls,
                              const struct GNUNET_MessageHeader *msg)
{
  struct ClientHandle *ch = cls;
  const char *url = (const char *) &msg[1];
  struct GNUNET_HELLO_Parser *b;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Local client provided HELLO URL %s\n",
              url);
  b = GNUNET_HELLO_parser_from_url (url);
  if (NULL == b)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (ch->client);
    return;
  }
  GNUNET_SERVICE_client_continue (ch->client);
  GNUNET_HELLO_parser_iterate (b,
                               &GDS_try_connect,
                               NULL);
  GNUNET_HELLO_parser_free (b);
}
 
 
/* ************* logic for monitors ************** */
 
 
static void
handle_dht_local_monitor (void *cls,
                          const struct GNUNET_DHT_MonitorStartStopMessage *msg)
{
  struct ClientHandle *ch = cls;
  struct ClientMonitorRecord *r;
 
  r = GNUNET_new (struct ClientMonitorRecord);
  r->ch = ch;
  r->type = ntohl (msg->type);
  r->get = ntohs (msg->get);
  r->get_resp = ntohs (msg->get_resp);
  r->put = ntohs (msg->put);
  if (0 != ntohs (msg->filter_key))
    r->key = msg->key;
  GNUNET_CONTAINER_DLL_insert (monitor_head,
                               monitor_tail,
                               r);
  GNUNET_SERVICE_client_continue (ch->client);
}
 
 
static void
handle_dht_local_monitor_stop (
  void *cls,
  const struct GNUNET_DHT_MonitorStartStopMessage *msg)
{
  struct ClientHandle *ch = cls;
 
  GNUNET_SERVICE_client_continue (ch->client);
  for (struct ClientMonitorRecord *r = monitor_head;
       NULL != r;
       r = r->next)
  {
    bool keys_match;
 
    keys_match =
      (GNUNET_is_zero (&r->key))
      ? (0 == ntohs (msg->filter_key))
      : ( (0 != ntohs (msg->filter_key)) &&
          (! GNUNET_memcmp (&r->key,
                            &msg->key)) );
    if ( (ch == r->ch) &&
         (ntohl (msg->type) == r->type) &&
         (r->get == msg->get) &&
         (r->get_resp == msg->get_resp) &&
         (r->put == msg->put) &&
         keys_match)
    {
      GNUNET_CONTAINER_DLL_remove (monitor_head,
                                   monitor_tail,
                                   r);
      GNUNET_free (r);
      return;     /* Delete only ONE entry */
    }
  }
}
 
 
typedef void
(*MonitorAction)(void *cls,
                 struct ClientMonitorRecord *m);
 
 
static void
for_matching_monitors (enum GNUNET_BLOCK_Type type,
                       const struct GNUNET_HashCode *key,
                       MonitorAction cb,
                       void *cb_cls)
{
  struct ClientHandle **cl = NULL;
  unsigned int cl_size = 0;
 
  for (struct ClientMonitorRecord *m = monitor_head;
       NULL != m;
       m = m->next)
  {
    bool found = false;
 
    if ( (GNUNET_BLOCK_TYPE_ANY != m->type) &&
         (m->type != type) )
      continue;
    if ( (! GNUNET_is_zero (&m->key)) &&
         (0 ==
          GNUNET_memcmp (key,
                         &m->key)) )
      continue;
    /* Don't send duplicates */
    for (unsigned i = 0; i < cl_size; i++)
      if (cl[i] == m->ch)
      {
        found = true;
        break;
      }
    if (found)
      continue;
    GNUNET_array_append (cl,
                         cl_size,
                         m->ch);
    cb (cb_cls,
        m);
  }
  GNUNET_free (cl);
}
 
 
struct GetActionContext
{
  enum GNUNET_DHT_RouteOption options;
  enum GNUNET_BLOCK_Type type;
  uint32_t hop_count;
  uint32_t desired_replication_level;
  struct GNUNET_PeerIdentity trunc_peer;
  const struct GNUNET_HashCode *key;
};
 
 
static void
get_action (void *cls,
            struct ClientMonitorRecord *m)
{
  struct GetActionContext *gac = cls;
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_DHT_MonitorGetMessage *mmsg;
 
  env = GNUNET_MQ_msg (mmsg,
                       GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET);
  mmsg->options = htonl (gac->options);
  mmsg->type = htonl (gac->type);
  mmsg->hop_count = htonl (gac->hop_count);
  mmsg->desired_replication_level = htonl (gac->desired_replication_level);
  mmsg->key = *gac->key;
  GNUNET_MQ_send (m->ch->mq,
                  env);
}
 
 
void
GDS_CLIENTS_process_get (enum GNUNET_DHT_RouteOption options,
                         enum GNUNET_BLOCK_Type type,
                         uint32_t hop_count,
                         uint32_t desired_replication_level,
                         const struct GNUNET_HashCode *key)
{
  struct GetActionContext gac = {
    .options = options,
    .type = type,
    .hop_count = hop_count,
    .desired_replication_level = desired_replication_level,
    .key = key
  };
 
  for_matching_monitors (type,
                         key,
                         &get_action,
                         &gac);
}
 
 
struct ResponseActionContext
{
  const struct GNUNET_DATACACHE_Block *bd;
  const struct GNUNET_DHT_PathElement *get_path;
  unsigned int get_path_length;
};
 
 
static void
response_action (void *cls,
                 struct ClientMonitorRecord *m)
{
  const struct ResponseActionContext *resp_ctx = cls;
  const struct GNUNET_DATACACHE_Block *bd = resp_ctx->bd;
  bool truncated = (0 != (bd->ro & GNUNET_DHT_RO_TRUNCATED));
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_DHT_MonitorGetRespMessage *mmsg;
  struct GNUNET_DHT_PathElement *path;
  size_t msize;
 
  msize = bd->data_size;
  msize += (resp_ctx->get_path_length + bd->put_path_length)
           * sizeof(struct GNUNET_DHT_PathElement);
  if (truncated)
    msize += sizeof (struct GNUNET_PeerIdentity);
  env = GNUNET_MQ_msg_extra (mmsg,
                             msize,
                             GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET_RESP);
  mmsg->type = htonl (bd->type);
  mmsg->put_path_length = htonl (bd->put_path_length);
  mmsg->get_path_length = htonl (resp_ctx->get_path_length);
  mmsg->expiration_time = GNUNET_TIME_absolute_hton (bd->expiration_time);
  mmsg->key = bd->key;
  if (truncated)
  {
    void *tgt = &mmsg[1];
 
    GNUNET_memcpy (tgt,
                   &bd->trunc_peer,
                   sizeof (struct GNUNET_PeerIdentity));
    path = (struct GNUNET_DHT_PathElement *)
           (tgt + sizeof (struct GNUNET_PeerIdentity));
  }
  else
  {
    path = (struct GNUNET_DHT_PathElement *) &mmsg[1];
  }
  GNUNET_memcpy (path,
                 bd->put_path,
                 bd->put_path_length * sizeof(struct GNUNET_DHT_PathElement));
  GNUNET_memcpy (path,
                 resp_ctx->get_path,
                 resp_ctx->get_path_length
                 * sizeof(struct GNUNET_DHT_PathElement));
  GNUNET_memcpy (&path[resp_ctx->get_path_length],
                 bd->data,
                 bd->data_size);
  GNUNET_MQ_send (m->ch->mq,
                  env);
}
 
 
void
GDS_CLIENTS_process_get_resp (const struct GNUNET_DATACACHE_Block *bd,
                              const struct GNUNET_DHT_PathElement *get_path,
                              unsigned int get_path_length)
{
  struct ResponseActionContext rac = {
    .bd = bd,
    .get_path = get_path,
    .get_path_length = get_path_length
  };
 
  for_matching_monitors (bd->type,
                         &bd->key,
                         &response_action,
                         &rac);
}
 
 
struct PutActionContext
{
  const struct GNUNET_DATACACHE_Block *bd;
  uint32_t hop_count;
  uint32_t desired_replication_level;
};
 
 
static void
put_action (void *cls,
            struct ClientMonitorRecord *m)
{
  const struct PutActionContext *put_ctx = cls;
  const struct GNUNET_DATACACHE_Block *bd = put_ctx->bd;
  bool truncated = (0 != (bd->ro & GNUNET_DHT_RO_TRUNCATED));
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_DHT_MonitorPutMessage *mmsg;
  struct GNUNET_DHT_PathElement *msg_path;
  size_t msize;
 
  msize = bd->data_size
          + bd->put_path_length
          * sizeof(struct GNUNET_DHT_PathElement);
  if (truncated)
    msize += sizeof (struct GNUNET_PeerIdentity);
  env = GNUNET_MQ_msg_extra (mmsg,
                             msize,
                             GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT);
  mmsg->options = htonl (bd->ro);
  mmsg->type = htonl (bd->type);
  mmsg->hop_count = htonl (put_ctx->hop_count);
  mmsg->desired_replication_level = htonl (put_ctx->desired_replication_level);
  mmsg->put_path_length = htonl (bd->put_path_length);
  mmsg->key = bd->key;
  mmsg->expiration_time = GNUNET_TIME_absolute_hton (bd->expiration_time);
  if (truncated)
  {
    void *tgt = &mmsg[1];
 
    GNUNET_memcpy (tgt,
                   &bd->trunc_peer,
                   sizeof (struct GNUNET_PeerIdentity));
    msg_path = (struct GNUNET_DHT_PathElement *)
               (tgt + sizeof (struct GNUNET_PeerIdentity));
  }
  else
  {
    msg_path = (struct GNUNET_DHT_PathElement *) &mmsg[1];
  }
  GNUNET_memcpy (msg_path,
                 bd->put_path,
                 bd->put_path_length * sizeof(struct GNUNET_DHT_PathElement));
  GNUNET_memcpy (&msg_path[bd->put_path_length],
                 bd->data,
                 bd->data_size);
  GNUNET_MQ_send (m->ch->mq,
                  env);
}
 
 
void
GDS_CLIENTS_process_put (const struct GNUNET_DATACACHE_Block *bd,
                         uint32_t hop_count,
                         uint32_t desired_replication_level)
{
  struct PutActionContext put_ctx = {
    .bd = bd,
    .hop_count = hop_count,
    .desired_replication_level = desired_replication_level
  };
 
  for_matching_monitors (bd->type,
                         &bd->key,
                         &put_action,
                         &put_ctx);
}
 
 
/* ********************** Initialization logic ***************** */
 
 
void
GDS_CLIENTS_init (void)
{
  forward_map
    = GNUNET_CONTAINER_multihashmap_create (1024,
                                            GNUNET_YES);
  retry_heap
    = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
}
 
 
void
GDS_CLIENTS_stop (void)
{
  struct PendingPut *pp;
 
  if (NULL != retry_task)
  {
    GNUNET_SCHEDULER_cancel (retry_task);
    retry_task = NULL;
  }
  if (NULL != put_retry_task)
  {
    GNUNET_SCHEDULER_cancel (put_retry_task);
    put_retry_task = NULL;
  }
  while (NULL != (pp = pp_head))
  {
    GNUNET_CONTAINER_DLL_remove (pp_head,
                                 pp_tail,
                                 pp);
    pp_count--;
    GNUNET_free (pp);
  }
}
 
 
#define GDS_DHT_SERVICE_INIT(name, run)          \
        GNUNET_SERVICE_MAIN \
          (GNUNET_OS_project_data_gnunet (), \
          name,                            \
          GNUNET_SERVICE_OPTION_NONE, \
          run, \
          &client_connect_cb, \
          &client_disconnect_cb, \
          NULL, \
          GNUNET_MQ_hd_var_size (dht_local_put, \
                                 GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT, \
                                 struct GNUNET_DHT_ClientPutMessage, \
                                 NULL), \
          GNUNET_MQ_hd_var_size (dht_local_get, \
                                 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET, \
                                 struct GNUNET_DHT_ClientGetMessage, \
                                 NULL), \
          GNUNET_MQ_hd_fixed_size (dht_local_get_stop, \
                                   GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP, \
                                   struct GNUNET_DHT_ClientGetStopMessage, \
                                   NULL), \
          GNUNET_MQ_hd_fixed_size (dht_local_monitor, \
                                   GNUNET_MESSAGE_TYPE_DHT_MONITOR_START, \
                                   struct GNUNET_DHT_MonitorStartStopMessage, \
                                   NULL), \
          GNUNET_MQ_hd_fixed_size (dht_local_monitor_stop, \
                                   GNUNET_MESSAGE_TYPE_DHT_MONITOR_STOP, \
                                   struct GNUNET_DHT_MonitorStartStopMessage, \
                                   NULL), \
          GNUNET_MQ_hd_var_size (dht_local_get_result_seen, \
                                 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_RESULTS_KNOWN, \
                                 struct GNUNET_DHT_ClientGetResultSeenMessage, \
                                 NULL), \
          GNUNET_MQ_hd_fixed_size (dht_local_hello_get,              \
                                   GNUNET_MESSAGE_TYPE_DHT_CLIENT_HELLO_GET, \
                                   struct GNUNET_MessageHeader, \
                                   NULL), \
          GNUNET_MQ_hd_var_size (dht_local_hello_offer, \
                                 GNUNET_MESSAGE_TYPE_DHT_CLIENT_HELLO_URL, \
                                 struct GNUNET_MessageHeader, \
                                 NULL), \
          GNUNET_MQ_handler_end ())
 
void GDS_CLIENTS_done (void);
 
void __attribute__ ((destructor))
GDS_CLIENTS_done (void)
{
  if (NULL != retry_heap)
  {
    GNUNET_assert (0 == GNUNET_CONTAINER_heap_get_size (retry_heap));
    GNUNET_CONTAINER_heap_destroy (retry_heap);
    retry_heap = NULL;
  }
  if (NULL != forward_map)
  {
    GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (forward_map));
    GNUNET_CONTAINER_multihashmap_destroy (forward_map);
    forward_map = NULL;
  }
}
 
 
/* end of gnunet-service-dht_clients.c */