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__)
 
 
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;
 
  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;
 
 
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;
 
  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);
  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);
  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
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
cb_forwarded_dht_local_put (void *cls,
                            enum GNUNET_GenericReturnValue forwarded)
{
  if (GNUNET_OK != forwarded)
  {
    GNUNET_STATISTICS_update (GDS_stats,
                              "# Local PUT requests not routed",
                              1,
                              GNUNET_NO);
  }
}
 
 
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);
    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;
 
    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);
    /* route to other peers */
    GDS_NEIGHBOURS_handle_put (&bd,
                               replication_level,
                               0 /* hop count */,
                               peer_bf,
                               &cb_forwarded_dht_local_put,
                               NULL);
    GNUNET_CONTAINER_bloomfilter_free (peer_bf);
  }
  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->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);
}
 
 
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)
{
  if (NULL != retry_task)
  {
    GNUNET_SCHEDULER_cancel (retry_task);
    retry_task = NULL;
  }
}
 
 
#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 */