gnunet-service-fs_pr.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2009-2013 GNUnet e.V.
 
     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     or (at your option) any later version.
 
     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     Affero General Public License for more details.
 
     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
     SPDX-License-Identifier: AGPL3.0-or-later
 */
 
#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_load_lib.h"
#include "gnunet-service-fs.h"
#include "gnunet-service-fs_cp.h"
#include "gnunet-service-fs_indexing.h"
#include "gnunet-service-fs_pe.h"
#include "gnunet-service-fs_pr.h"
#include "gnunet-service-fs_cadet.h"
 
 
#define DHT_GET_REPLICATION 5
 
#define MAX_DATASTORE_QUEUE (16 * MAX_QUEUE_PER_PEER)
 
#define CONTENT_BANDWIDTH_VALUE 800
 
#define MAX_RESULTS (100 * 1024)
 
#define INSANE_STATISTICS GNUNET_NO
 
#define CADET_RETRY_MAX 3
 
 
struct GSF_PendingRequest
{
  struct GSF_PendingRequestData public_data;
 
  GSF_PendingRequestReplyHandler rh;
 
  void *rh_cls;
 
  struct GNUNET_HashCode *replies_seen;
 
  struct GNUNET_BLOCK_Group *bg;
 
  struct GNUNET_CONTAINER_HeapNode *hnode;
 
  struct GNUNET_DATASTORE_QueueEntry *qe;
 
  struct GNUNET_DHT_GetHandle *gh;
 
  struct GSF_CadetRequest *cadet_request;
 
  GSF_LocalLookupContinuation llc_cont;
 
  void *llc_cont_cls;
 
  enum GNUNET_BLOCK_ReplyEvaluationResult local_result;
 
  GNUNET_PEER_Id sender_pid;
 
  GNUNET_PEER_Id origin_pid;
 
  struct GNUNET_TIME_Absolute qe_start;
 
  struct GNUNET_SCHEDULER_Task *warn_task;
 
  bool have_first_uid;
 
  bool seen_null;
 
  uint64_t first_uid;
 
  size_t result_count;
 
  unsigned int cadet_retry_count;
 
  unsigned int replies_seen_count;
 
  unsigned int replies_seen_size;
};
 
 
static struct GNUNET_CONTAINER_MultiHashMap *pr_map;
 
 
static struct GNUNET_LOAD_Value *datastore_put_load;
 
 
static int active_to_migration;
 
 
static struct GNUNET_CONTAINER_Heap *requests_by_expiration_heap;
 
 
static unsigned long long max_pending_requests = (32 * 1024);
 
 
static void
refresh_bloomfilter (enum GNUNET_BLOCK_Type type,
                     struct GSF_PendingRequest *pr)
{
  if (NULL != pr->bg)
  {
    GNUNET_BLOCK_group_destroy (pr->bg);
    pr->bg = NULL;
  }
  if (GNUNET_BLOCK_TYPE_FS_UBLOCK != type)
    return; /* no need */
  pr->bg =
    GNUNET_BLOCK_group_create (GSF_block_ctx,
                               type,
                               NULL,
                               0,
                               "seen-set-size",
                               pr->replies_seen_count,
                               NULL);
  if (NULL == pr->bg)
    return;
  GNUNET_break (GNUNET_OK ==
                GNUNET_BLOCK_group_set_seen (pr->bg,
                                             pr->replies_seen,
                                             pr->replies_seen_count));
}
 
 
struct GSF_PendingRequest *
GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
                             enum GNUNET_BLOCK_Type type,
                             const struct GNUNET_HashCode *query,
                             const struct GNUNET_PeerIdentity *target,
                             const char *bf_data,
                             size_t bf_size,
                             uint32_t anonymity_level,
                             uint32_t priority,
                             int32_t ttl,
                             GNUNET_PEER_Id sender_pid,
                             GNUNET_PEER_Id origin_pid,
                             const struct GNUNET_HashCode *replies_seen,
                             unsigned int replies_seen_count,
                             GSF_PendingRequestReplyHandler rh,
                             void *rh_cls)
{
  struct GSF_PendingRequest *pr;
  struct GSF_PendingRequest *dpr;
  size_t extra;
  struct GNUNET_HashCode *eptr;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Creating request handle for `%s' of type %d\n",
              GNUNET_h2s (query),
              type);
#if INSANE_STATISTICS
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Pending requests created"),
                            1,
                            GNUNET_NO);
#endif
  extra = 0;
  if (NULL != target)
    extra += sizeof(struct GNUNET_PeerIdentity);
  pr = GNUNET_malloc (sizeof(struct GSF_PendingRequest) + extra);
  pr->public_data.query = *query;
  eptr = (struct GNUNET_HashCode *) &pr[1];
  if (NULL != target)
  {
    pr->public_data.target = (struct GNUNET_PeerIdentity *) eptr;
    GNUNET_memcpy (eptr, target, sizeof(struct GNUNET_PeerIdentity));
  }
  pr->public_data.anonymity_level = anonymity_level;
  pr->public_data.priority = priority;
  pr->public_data.original_priority = priority;
  pr->public_data.options = options;
  pr->public_data.type = type;
  pr->public_data.start_time = GNUNET_TIME_absolute_get ();
  pr->sender_pid = sender_pid;
  pr->origin_pid = origin_pid;
  pr->rh = rh;
  pr->rh_cls = rh_cls;
  GNUNET_assert ((sender_pid != 0) || (0 == (options & GSF_PRO_FORWARD_ONLY)));
  if (ttl >= 0)
    pr->public_data.ttl = GNUNET_TIME_relative_to_absolute (
      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, (uint32_t) ttl));
  else
    pr->public_data.ttl = GNUNET_TIME_absolute_subtract (
      pr->public_data.start_time,
      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
                                     (uint32_t) (-ttl)));
  if (replies_seen_count > 0)
  {
    pr->replies_seen_size = replies_seen_count;
    pr->replies_seen =
      GNUNET_new_array (pr->replies_seen_size, struct GNUNET_HashCode);
    GNUNET_memcpy (pr->replies_seen,
                   replies_seen,
                   replies_seen_count * sizeof(struct GNUNET_HashCode));
    pr->replies_seen_count = replies_seen_count;
  }
  if ((NULL != bf_data) &&
      (GNUNET_BLOCK_TYPE_FS_UBLOCK == pr->public_data.type))
  {
    pr->bg = GNUNET_BLOCK_group_create (GSF_block_ctx,
                                        pr->public_data.type,
                                        bf_data,
                                        bf_size,
                                        "seen-set-size",
                                        0,
                                        NULL);
  }
  else if ((replies_seen_count > 0) &&
           (0 != (options & GSF_PRO_BLOOMFILTER_FULL_REFRESH)))
  {
    refresh_bloomfilter (pr->public_data.type, pr);
  }
  GNUNET_CONTAINER_multihashmap_put (pr_map,
                                     &pr->public_data.query,
                                     pr,
                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  if (0 == (options & GSF_PRO_REQUEST_NEVER_EXPIRES))
  {
    pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
                                              pr,
                                              pr->public_data.ttl.abs_value_us);
    /* make sure we don't track too many requests */
    while (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) >
           max_pending_requests)
    {
      dpr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
      GNUNET_assert (NULL != dpr);
      if (pr == dpr)
        break;     /* let the request live briefly... */
      if (NULL != dpr->rh)
        dpr->rh (dpr->rh_cls,
                 GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED,
                 dpr,
                 UINT32_MAX,
                 GNUNET_TIME_UNIT_FOREVER_ABS,
                 GNUNET_TIME_UNIT_FOREVER_ABS,
                 GNUNET_BLOCK_TYPE_ANY,
                 NULL,
                 0);
      GSF_pending_request_cancel_ (dpr, GNUNET_YES);
    }
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Pending requests active"),
                            1,
                            GNUNET_NO);
  return pr;
}
 
 
struct GSF_PendingRequestData *
GSF_pending_request_get_data_ (struct GSF_PendingRequest *pr)
{
  return &pr->public_data;
}
 
 
int
GSF_pending_request_is_compatible_ (struct GSF_PendingRequest *pra,
                                    struct GSF_PendingRequest *prb)
{
  if ((pra->public_data.type != prb->public_data.type) ||
      (0 != memcmp (&pra->public_data.query,
                    &prb->public_data.query,
                    sizeof(struct GNUNET_HashCode))))
    return GNUNET_NO;
  return GNUNET_OK;
}
 
 
void
GSF_pending_request_update_ (struct GSF_PendingRequest *pr,
                             const struct GNUNET_HashCode *replies_seen,
                             unsigned int replies_seen_count)
{
  if (replies_seen_count + pr->replies_seen_count < pr->replies_seen_count)
    return; /* integer overflow */
  if (0 != (pr->public_data.options & GSF_PRO_BLOOMFILTER_FULL_REFRESH))
  {
    /* we're responsible for the BF, full refresh */
    if (replies_seen_count + pr->replies_seen_count > pr->replies_seen_size)
      GNUNET_array_grow (pr->replies_seen,
                         pr->replies_seen_size,
                         replies_seen_count + pr->replies_seen_count);
    GNUNET_memcpy (&pr->replies_seen[pr->replies_seen_count],
                   replies_seen,
                   sizeof(struct GNUNET_HashCode) * replies_seen_count);
    pr->replies_seen_count += replies_seen_count;
    refresh_bloomfilter (pr->public_data.type, pr);
  }
  else
  {
    if (NULL == pr->bg)
    {
      /* we're not the initiator, but the initiator did not give us
       * any bloom-filter, so we need to create one on-the-fly */
      refresh_bloomfilter (pr->public_data.type, pr);
    }
    else
    {
      GNUNET_break (GNUNET_OK ==
                    GNUNET_BLOCK_group_set_seen (pr->bg,
                                                 replies_seen,
                                                 pr->replies_seen_count));
    }
  }
  if (NULL != pr->gh)
    GNUNET_DHT_get_filter_known_results (pr->gh,
                                         replies_seen_count,
                                         replies_seen);
}
 
 
struct GNUNET_MQ_Envelope *
GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr)
{
  struct GNUNET_MQ_Envelope *env;
  struct GetMessage *gm;
  struct GNUNET_PeerIdentity *ext;
  unsigned int k;
  uint32_t bm;
  uint32_t prio;
  size_t bf_size;
  struct GNUNET_TIME_Absolute now;
  int64_t ttl;
  int do_route;
  void *bf_data;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Building request message for `%s' of type %d\n",
              GNUNET_h2s (&pr->public_data.query),
              pr->public_data.type);
  k = 0;
  bm = 0;
  do_route = (0 == (pr->public_data.options & GSF_PRO_FORWARD_ONLY));
  if ((! do_route) && (pr->sender_pid == 0))
  {
    GNUNET_break (0);
    do_route = GNUNET_YES;
  }
  if (! do_route)
  {
    bm |= GET_MESSAGE_BIT_RETURN_TO;
    k++;
  }
  if (NULL != pr->public_data.target)
  {
    bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
    k++;
  }
  if (GNUNET_OK !=
      GNUNET_BLOCK_group_serialize (pr->bg,
                                    &bf_data,
                                    &bf_size))
  {
    bf_size = 0;
    bf_data = NULL;
  }
  env = GNUNET_MQ_msg_extra (gm,
                             bf_size + k * sizeof(struct GNUNET_PeerIdentity),
                             GNUNET_MESSAGE_TYPE_FS_GET);
  gm->type = htonl (pr->public_data.type);
  if (do_route)
    prio = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                     pr->public_data.priority + 1);
  else
    prio = 0;
  pr->public_data.priority -= prio;
  pr->public_data.num_transmissions++;
  pr->public_data.respect_offered += prio;
  gm->priority = htonl (prio);
  now = GNUNET_TIME_absolute_get ();
  ttl = (int64_t) (pr->public_data.ttl.abs_value_us - now.abs_value_us);
  gm->ttl = htonl (ttl / 1000LL / 1000LL);
  gm->reserved = htonl (0);
  gm->hash_bitmap = htonl (bm);
  gm->query = pr->public_data.query;
  ext = (struct GNUNET_PeerIdentity *) &gm[1];
  k = 0;
  if (! do_route)
    GNUNET_PEER_resolve (pr->sender_pid, &ext[k++]);
  if (NULL != pr->public_data.target)
    ext[k++] = *pr->public_data.target;
  GNUNET_memcpy (&ext[k], bf_data, bf_size);
  GNUNET_free (bf_data);
  return env;
}
 
 
static int
clean_request (void *cls, const struct GNUNET_HashCode *key, void *value)
{
  struct GSF_PendingRequest *pr = value;
  GSF_LocalLookupContinuation cont;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Cleaning up pending request for `%s'.\n",
              GNUNET_h2s (key));
  if (NULL != pr->cadet_request)
  {
    pr->cadet_retry_count = CADET_RETRY_MAX;
    GSF_cadet_query_cancel (pr->cadet_request);
    pr->cadet_request = NULL;
  }
  if (NULL != (cont = pr->llc_cont))
  {
    pr->llc_cont = NULL;
    cont (pr->llc_cont_cls,
          pr,
          pr->local_result);
  }
  GSF_plan_notify_request_done_ (pr);
  GNUNET_free (pr->replies_seen);
  GNUNET_BLOCK_group_destroy (pr->bg);
  pr->bg = NULL;
  GNUNET_PEER_change_rc (pr->sender_pid, -1);
  pr->sender_pid = 0;
  GNUNET_PEER_change_rc (pr->origin_pid, -1);
  pr->origin_pid = 0;
  if (NULL != pr->hnode)
  {
    GNUNET_CONTAINER_heap_remove_node (pr->hnode);
    pr->hnode = NULL;
  }
  if (NULL != pr->qe)
  {
    GNUNET_DATASTORE_cancel (pr->qe);
    pr->qe = NULL;
  }
  if (NULL != pr->gh)
  {
    GNUNET_DHT_get_stop (pr->gh);
    pr->gh = NULL;
  }
  if (NULL != pr->warn_task)
  {
    GNUNET_SCHEDULER_cancel (pr->warn_task);
    pr->warn_task = NULL;
  }
  GNUNET_assert (
    GNUNET_OK ==
    GNUNET_CONTAINER_multihashmap_remove (pr_map, &pr->public_data.query, pr));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Pending requests active"),
                            -1,
                            GNUNET_NO);
  GNUNET_free (pr);
  return GNUNET_YES;
}
 
 
void
GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr, int full_cleanup)
{
  GSF_LocalLookupContinuation cont;
 
  if (NULL == pr_map)
    return; /* already cleaned up! */
  if (GNUNET_NO == full_cleanup)
  {
    /* make request inactive (we're no longer interested in more results),
     * but do NOT remove from our data-structures, we still need it there
     * to prevent the request from looping */
    pr->rh = NULL;
    if (NULL != pr->cadet_request)
    {
      pr->cadet_retry_count = CADET_RETRY_MAX;
      GSF_cadet_query_cancel (pr->cadet_request);
      pr->cadet_request = NULL;
    }
    if (NULL != (cont = pr->llc_cont))
    {
      pr->llc_cont = NULL;
      cont (pr->llc_cont_cls,
            pr,
            pr->local_result);
    }
    GSF_plan_notify_request_done_ (pr);
    if (NULL != pr->qe)
    {
      GNUNET_DATASTORE_cancel (pr->qe);
      pr->qe = NULL;
    }
    if (NULL != pr->gh)
    {
      GNUNET_DHT_get_stop (pr->gh);
      pr->gh = NULL;
    }
    if (NULL != pr->warn_task)
    {
      GNUNET_SCHEDULER_cancel (pr->warn_task);
      pr->warn_task = NULL;
    }
    return;
  }
  GNUNET_assert (GNUNET_YES ==
                 clean_request (NULL, &pr->public_data.query, pr));
}
 
 
void
GSF_iterate_pending_requests_ (GSF_PendingRequestIterator it, void *cls)
{
  GNUNET_CONTAINER_multihashmap_iterate (
    pr_map,
    (GNUNET_CONTAINER_MultiHashMapIteratorCallback) it,
    cls);
}
 
 
struct ProcessReplyClosure
{
  const void *data;
 
  struct GSF_ConnectedPeer *sender;
 
  struct GNUNET_TIME_Absolute expiration;
 
  size_t size;
 
  enum GNUNET_BLOCK_Type type;
 
  uint32_t priority;
 
  uint32_t anonymity_level;
 
  enum GNUNET_BLOCK_ReplyEvaluationResult eval;
 
  int request_found;
};
 
 
static void
update_request_performance_data (struct ProcessReplyClosure *prq,
                                 struct GSF_PendingRequest *pr)
{
  if (prq->sender == NULL)
    return;
  GSF_peer_update_performance_ (prq->sender,
                                pr->public_data.start_time,
                                prq->priority);
}
 
 
static enum GNUNET_GenericReturnValue
process_reply (void *cls,
               const struct GNUNET_HashCode *key,
               void *value)
{
  struct ProcessReplyClosure *prq = cls;
  struct GSF_PendingRequest *pr = value;
  struct GNUNET_HashCode chash;
  struct GNUNET_TIME_Absolute last_transmission;
 
  if (NULL == pr->rh)
    return GNUNET_YES;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Matched result (type %u) for query `%s' with pending request\n",
              (unsigned int) prq->type,
              GNUNET_h2s (key));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# replies received and matched"),
                            1,
                            GNUNET_NO);
  prq->eval = GNUNET_BLOCK_check_reply (GSF_block_ctx,
                                        prq->type,
                                        pr->bg,
                                        key,
                                        NULL, 0,
                                        prq->data,
                                        prq->size);
  switch (prq->eval)
  {
  case GNUNET_BLOCK_REPLY_OK_MORE:
    update_request_performance_data (prq, pr);
    break;
  case GNUNET_BLOCK_REPLY_OK_LAST:
    /* short cut: stop processing early, no BF-update, etc. */
    update_request_performance_data (prq, pr);
    GNUNET_LOAD_update (GSF_rt_entry_lifetime,
                        GNUNET_TIME_absolute_get_duration (
                          pr->public_data.start_time)
                        .rel_value_us);
    if (GNUNET_YES !=
        GSF_request_plan_reference_get_last_transmission_ (pr->public_data
                                                           .pr_head,
                                                           prq->sender,
                                                           &last_transmission))
      last_transmission = GNUNET_TIME_UNIT_FOREVER_ABS;
    /* pass on to other peers / local clients */
    pr->rh (pr->rh_cls,
            prq->eval,
            pr,
            prq->anonymity_level,
            prq->expiration,
            last_transmission,
            prq->type,
            prq->data,
            prq->size);
    return GNUNET_YES;
  case GNUNET_BLOCK_REPLY_OK_DUPLICATE:
#if INSANE_STATISTICS
    GNUNET_STATISTICS_update (GSF_stats,
                              "# duplicate replies discarded (bloomfilter)",
                              1,
                              GNUNET_NO);
#endif
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Duplicate response, discarding.\n");
    return GNUNET_YES;   /* duplicate */
  case GNUNET_BLOCK_REPLY_IRRELEVANT:
    GNUNET_STATISTICS_update (GSF_stats,
                              "# irrelevant replies discarded",
                              1,
                              GNUNET_NO);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Irrelevant response, ignoring.\n");
    return GNUNET_YES;
  case GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED:
    GNUNET_break (0); /* bad installation? */
    return GNUNET_NO;
  }
  /* update bloomfilter */
  GNUNET_CRYPTO_hash (prq->data,
                      prq->size,
                      &chash);
  GSF_pending_request_update_ (pr,
                               &chash,
                               1);
  if (NULL == prq->sender)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Found result for query `%s' in local datastore\n",
                GNUNET_h2s (key));
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop ("# results found locally"),
                              1,
                              GNUNET_NO);
  }
  else
  {
    GSF_dht_lookup_ (pr);
  }
  prq->priority += pr->public_data.original_priority;
  pr->public_data.priority = 0;
  pr->public_data.original_priority = 0;
  pr->public_data.results_found++;
  prq->request_found = GNUNET_YES;
  /* finally, pass on to other peer / local client */
  if (! GSF_request_plan_reference_get_last_transmission_ (pr->public_data
                                                           .pr_head,
                                                           prq->sender,
                                                           &last_transmission))
    last_transmission = GNUNET_TIME_UNIT_FOREVER_ABS;
  pr->rh (pr->rh_cls,
          prq->eval,
          pr,
          prq->anonymity_level,
          prq->expiration,
          last_transmission,
          prq->type,
          prq->data,
          prq->size);
  return GNUNET_YES;
}
 
 
struct PutMigrationContext
{
  struct GNUNET_TIME_Absolute start;
 
  struct GNUNET_PeerIdentity origin;
 
  int requested;
};
 
 
static void
put_migration_continuation (void *cls,
                            int success,
                            struct GNUNET_TIME_Absolute min_expiration,
                            const char *msg)
{
  struct PutMigrationContext *pmc = cls;
  struct GSF_ConnectedPeer *cp;
  struct GNUNET_TIME_Relative mig_pause;
  struct GSF_PeerPerformanceData *ppd;
 
  if (NULL != datastore_put_load)
  {
    if (GNUNET_SYSERR != success)
    {
      GNUNET_LOAD_update (datastore_put_load,
                          GNUNET_TIME_absolute_get_duration (pmc->start)
                          .rel_value_us);
    }
    else
    {
      /* on queue failure / timeout, increase the put load dramatically */
      GNUNET_LOAD_update (datastore_put_load,
                          GNUNET_TIME_UNIT_MINUTES.rel_value_us);
    }
  }
  cp = GSF_peer_get_ (&pmc->origin);
  if (GNUNET_OK == success)
  {
    if (NULL != cp)
    {
      ppd = GSF_get_peer_performance_data_ (cp);
      ppd->migration_delay.rel_value_us /= 2;
    }
    GNUNET_free (pmc);
    return;
  }
  if ((GNUNET_NO == success) && (GNUNET_NO == pmc->requested) && (NULL != cp))
  {
    ppd = GSF_get_peer_performance_data_ (cp);
    if (min_expiration.abs_value_us > 0)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Asking to stop migration for %s because datastore is full\n",
                  GNUNET_STRINGS_relative_time_to_string (
                    GNUNET_TIME_absolute_get_remaining (min_expiration),
                    GNUNET_YES));
      GSF_block_peer_migration_ (cp, min_expiration);
    }
    else
    {
      ppd->migration_delay = GNUNET_TIME_relative_max (GNUNET_TIME_UNIT_SECONDS,
                                                       ppd->migration_delay);
      ppd->migration_delay =
        GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_HOURS, ppd->migration_delay);
      mig_pause.rel_value_us =
        GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
                                  ppd->migration_delay.rel_value_us);
      ppd->migration_delay =
        GNUNET_TIME_relative_saturating_multiply (ppd->migration_delay, 2);
      GNUNET_log (
        GNUNET_ERROR_TYPE_DEBUG,
        "Replicated content already exists locally, asking to stop migration for %s\n",
        GNUNET_STRINGS_relative_time_to_string (mig_pause, GNUNET_YES));
      GSF_block_peer_migration_ (cp,
                                 GNUNET_TIME_relative_to_absolute (mig_pause));
    }
  }
  GNUNET_free (pmc);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Datastore `PUT' failures"),
                            1,
                            GNUNET_NO);
}
 
 
static int
test_put_load_too_high (uint32_t priority)
{
  double ld;
 
  if (NULL == datastore_put_load)
    return GNUNET_NO;
  if (GNUNET_LOAD_get_average (datastore_put_load) < 50)
    return GNUNET_NO; /* very fast */
  ld = GNUNET_LOAD_get_load (datastore_put_load);
  if (ld < 2.0 * (1 + priority))
    return GNUNET_NO;
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop (
                              "# storage requests dropped due to high load"),
                            1,
                            GNUNET_NO);
  return GNUNET_YES;
}
 
 
static void
handle_dht_reply (void *cls,
                  struct GNUNET_TIME_Absolute exp,
                  const struct GNUNET_HashCode *key,
                  const struct GNUNET_PeerIdentity *trunc_peer,
                  const struct GNUNET_DHT_PathElement *get_path,
                  unsigned int get_path_length,
                  const struct GNUNET_DHT_PathElement *put_path,
                  unsigned int put_path_length,
                  enum GNUNET_BLOCK_Type type,
                  size_t size,
                  const void *data)
{
  struct GSF_PendingRequest *pr = cls;
  struct ProcessReplyClosure prq;
  struct PutMigrationContext *pmc;
 
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Replies received from DHT"),
                            1,
                            GNUNET_NO);
  memset (&prq, 0, sizeof(prq));
  prq.data = data;
  prq.expiration = exp;
  /* do not allow migrated content to live longer than 1 year */
  prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (
                                               GNUNET_TIME_UNIT_YEARS),
                                             prq.expiration);
  prq.size = size;
  prq.type = type;
  process_reply (&prq,
                 key,
                 pr);
  if ((GNUNET_YES == active_to_migration) &&
      (GNUNET_NO == test_put_load_too_high (prq.priority)))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Replicating result for query `%s' with priority %u\n",
                GNUNET_h2s (key),
                prq.priority);
    pmc = GNUNET_new (struct PutMigrationContext);
    pmc->start = GNUNET_TIME_absolute_get ();
    pmc->requested = GNUNET_YES;
    if (NULL == GNUNET_DATASTORE_put (GSF_dsh,
                                      0,
                                      key,
                                      size,
                                      data,
                                      type,
                                      prq.priority,
                                      1 /* anonymity */,
                                      0 /* replication */,
                                      exp,
                                      1 + prq.priority,
                                      MAX_DATASTORE_QUEUE,
                                      &put_migration_continuation,
                                      pmc))
    {
      put_migration_continuation (pmc,
                                  GNUNET_SYSERR,
                                  GNUNET_TIME_UNIT_ZERO_ABS,
                                  NULL);
    }
  }
}
 
 
void
GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
{
  const void *xquery;
  size_t xquery_size;
  struct GNUNET_PeerIdentity pi;
  char buf[sizeof(struct GNUNET_HashCode) * 2] GNUNET_ALIGN;
 
  if (0 != pr->public_data.anonymity_level)
    return;
  if (NULL != pr->gh)
  {
    GNUNET_DHT_get_stop (pr->gh);
    pr->gh = NULL;
  }
  xquery = NULL;
  xquery_size = 0;
  if (0 != (pr->public_data.options & GSF_PRO_FORWARD_ONLY))
  {
    GNUNET_assert (0 != pr->sender_pid);
    GNUNET_PEER_resolve (pr->sender_pid, &pi);
    GNUNET_memcpy (&buf[xquery_size], &pi, sizeof(struct GNUNET_PeerIdentity));
    xquery_size += sizeof(struct GNUNET_PeerIdentity);
  }
  pr->gh = GNUNET_DHT_get_start (GSF_dht,
                                 pr->public_data.type,
                                 &pr->public_data.query,
                                 DHT_GET_REPLICATION,
                                 GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                 xquery,
                                 xquery_size,
                                 &handle_dht_reply,
                                 pr);
  if ((NULL != pr->gh) && (0 != pr->replies_seen_count))
    GNUNET_DHT_get_filter_known_results (pr->gh,
                                         pr->replies_seen_count,
                                         pr->replies_seen);
}
 
 
static void
cadet_reply_proc (void *cls,
                  enum GNUNET_BLOCK_Type type,
                  struct GNUNET_TIME_Absolute expiration,
                  size_t data_size,
                  const void *data)
{
  struct GSF_PendingRequest *pr = cls;
  struct ProcessReplyClosure prq;
  struct GNUNET_HashCode query;
 
  pr->cadet_request = NULL;
  if (GNUNET_OK !=
      GNUNET_BLOCK_check_block (GSF_block_ctx,
                                type,
                                data,
                                data_size))
  {
    GNUNET_break_op (0);
    return;
  }
  if (GNUNET_BLOCK_TYPE_ANY == type)
  {
    GNUNET_break (NULL == data);
    GNUNET_break (0 == data_size);
    pr->cadet_retry_count++;
    if (pr->cadet_retry_count >= CADET_RETRY_MAX)
      return;   /* give up on cadet */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Error retrieiving block via cadet\n");
    /* retry -- without delay, as this is non-anonymous
       and cadet/cadet connect will take some time anyway */
    pr->cadet_request = GSF_cadet_query (pr->public_data.target,
                                         &pr->public_data.query,
                                         pr->public_data.type,
                                         &cadet_reply_proc,
                                         pr);
    return;
  }
  if (GNUNET_YES !=
      GNUNET_BLOCK_get_key (GSF_block_ctx,
                            type,
                            data,
                            data_size,
                            &query))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Failed to derive key for block of type %d\n",
                (int) type);
    GNUNET_break_op (0);
    return;
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Replies received from CADET"),
                            1,
                            GNUNET_NO);
  memset (&prq, 0, sizeof(prq));
  prq.data = data;
  prq.expiration = expiration;
  /* do not allow migrated content to live longer than 1 year */
  prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (
                                               GNUNET_TIME_UNIT_YEARS),
                                             prq.expiration);
  prq.size = data_size;
  prq.type = type;
  process_reply (&prq,
                 &query,
                 pr);
}
 
 
void
GSF_cadet_lookup_ (struct GSF_PendingRequest *pr)
{
  if (0 != pr->public_data.anonymity_level)
    return;
  if (0 == pr->public_data.target)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Cannot do cadet-based download, target peer not known\n");
    return;
  }
  if (NULL != pr->cadet_request)
    return;
  pr->cadet_request = GSF_cadet_query (pr->public_data.target,
                                       &pr->public_data.query,
                                       pr->public_data.type,
                                       &cadet_reply_proc,
                                       pr);
}
 
 
static void
warn_delay_task (void *cls)
{
  struct GSF_PendingRequest *pr = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
              _ ("Datastore lookup already took %s!\n"),
              GNUNET_STRINGS_relative_time_to_string (
                GNUNET_TIME_absolute_get_duration (pr->qe_start),
                GNUNET_YES));
  pr->warn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
                                                &warn_delay_task,
                                                pr);
}
 
 
static void
odc_warn_delay_task (void *cls)
{
  struct GSF_PendingRequest *pr = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
              _ ("On-demand lookup already took %s!\n"),
              GNUNET_STRINGS_relative_time_to_string (
                GNUNET_TIME_absolute_get_duration (pr->qe_start),
                GNUNET_YES));
  pr->warn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
                                                &odc_warn_delay_task,
                                                pr);
}
 
 
/* Call our continuation (if we have any) */
static void
call_continuation (struct GSF_PendingRequest *pr)
{
  GSF_LocalLookupContinuation cont = pr->llc_cont;
 
  GNUNET_assert (NULL == pr->qe);
  if (NULL != pr->warn_task)
  {
    GNUNET_SCHEDULER_cancel (pr->warn_task);
    pr->warn_task = NULL;
  }
  if (NULL == cont)
    return; /* no continuation */
  pr->llc_cont = NULL;
  if (0 != (GSF_PRO_LOCAL_ONLY & pr->public_data.options))
  {
    if (GNUNET_BLOCK_REPLY_OK_LAST != pr->local_result)
    {
      /* Signal that we are done and that there won't be any
         additional results to allow client to clean up state. */
      pr->rh (pr->rh_cls,
              GNUNET_BLOCK_REPLY_OK_LAST,
              pr,
              UINT32_MAX,
              GNUNET_TIME_UNIT_ZERO_ABS,
              GNUNET_TIME_UNIT_FOREVER_ABS,
              GNUNET_BLOCK_TYPE_ANY,
              NULL,
              0);
    }
    /* Finally, call our continuation to signal that we are
       done with local processing of this request; i.e. to
       start reading again from the client. */
    cont (pr->llc_cont_cls,
          NULL,
          GNUNET_BLOCK_REPLY_OK_LAST);
    return;
  }
 
  cont (pr->llc_cont_cls,
        pr,
        pr->local_result);
}
 
 
/* Update stats and call continuation */
static void
no_more_local_results (struct GSF_PendingRequest *pr)
{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
              "No further local responses available.\n");
#if INSANE_STATISTICS
  if ((GNUNET_BLOCK_TYPE_FS_DBLOCK == pr->public_data.type) ||
      (GNUNET_BLOCK_TYPE_FS_IBLOCK == pr->public_data.type))
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# requested DBLOCK or IBLOCK not found"),
                              1,
                              GNUNET_NO);
#endif
  call_continuation (pr);
}
 
 
/* forward declaration */
static void
process_local_reply (void *cls,
                     const struct GNUNET_HashCode *key,
                     size_t size,
                     const void *data,
                     enum GNUNET_BLOCK_Type type,
                     uint32_t priority,
                     uint32_t anonymity,
                     uint32_t replication,
                     struct GNUNET_TIME_Absolute expiration,
                     uint64_t uid);
 
 
/* Start a local query */
static void
start_local_query (struct GSF_PendingRequest *pr,
                   uint64_t next_uid,
                   bool random)
{
  pr->qe_start = GNUNET_TIME_absolute_get ();
  pr->warn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
                                                &warn_delay_task,
                                                pr);
  pr->qe = GNUNET_DATASTORE_get_key (GSF_dsh,
                                     next_uid,
                                     random,
                                     &pr->public_data.query,
                                     pr->public_data.type ==
                                     GNUNET_BLOCK_TYPE_FS_DBLOCK
                                     ? GNUNET_BLOCK_TYPE_ANY
                                     : pr->public_data.type,
                                     (0 != (GSF_PRO_PRIORITY_UNLIMITED
                                            & pr->public_data.options))
                                     ? UINT_MAX
                                     : 1
                                     /* queue priority */,
                                     (0 != (GSF_PRO_PRIORITY_UNLIMITED
                                            & pr->public_data.options))
                                     ? UINT_MAX
                                     : GSF_datastore_queue_size
                                     /* max queue size */,
                                     &process_local_reply,
                                     pr);
  if (NULL != pr->qe)
    return;
  GNUNET_log (
    GNUNET_ERROR_TYPE_DEBUG,
    "ERROR Requesting `%s' of type %d with next_uid %llu from datastore.\n",
    GNUNET_h2s (&pr->public_data.query),
    pr->public_data.type,
    (unsigned long long) next_uid);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop (
                              "# Datastore lookups concluded (error queueing)"),
                            1,
                            GNUNET_NO);
  call_continuation (pr);
}
 
 
static void
process_local_reply (void *cls,
                     const struct GNUNET_HashCode *key,
                     size_t size,
                     const void *data,
                     enum GNUNET_BLOCK_Type type,
                     uint32_t priority,
                     uint32_t anonymity,
                     uint32_t replication,
                     struct GNUNET_TIME_Absolute expiration,
                     uint64_t uid)
{
  struct GSF_PendingRequest *pr = cls;
  struct ProcessReplyClosure prq;
  struct GNUNET_HashCode query;
  unsigned int old_rf;
 
  GNUNET_SCHEDULER_cancel (pr->warn_task);
  pr->warn_task = NULL;
  if (NULL == pr->qe)
    goto called_from_on_demand;
  pr->qe = NULL;
  if (
    (NULL == key) && pr->seen_null &&
    ! pr->have_first_uid)  /* We have hit the end for the 2nd time with no results */
  {
    /* No results */
#if INSANE_STATISTICS
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# Datastore lookups concluded (no results)"),
                              1,
                              GNUNET_NO);
#endif
    no_more_local_results (pr);
    return;
  }
  if (((NULL == key) &&
       pr->seen_null) || /* We have hit the end for the 2nd time OR */
      (pr->seen_null && pr->have_first_uid &&
       (uid >= pr->first_uid))) /* We have hit the end and past first UID */
  {
    /* Seen all results */
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# Datastore lookups concluded (seen all)"),
                              1,
                              GNUNET_NO);
    no_more_local_results (pr);
    return;
  }
  if (NULL == key)
  {
    GNUNET_assert (! pr->seen_null);
    pr->seen_null = true;
    start_local_query (pr, 0 /* next_uid */, false /* random */);
    return;
  }
  if (! pr->have_first_uid)
  {
    pr->first_uid = uid;
    pr->have_first_uid = true;
  }
  pr->result_count++;
  if (pr->result_count > MAX_RESULTS)
  {
    GNUNET_STATISTICS_update (
      GSF_stats,
      gettext_noop ("# Datastore lookups aborted (more than MAX_RESULTS)"),
      1,
      GNUNET_NO);
    no_more_local_results (pr);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received reply for `%s' of type %d with UID %llu from datastore.\n",
              GNUNET_h2s (key),
              type,
              (unsigned long long) uid);
  if (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Found ONDEMAND block, performing on-demand encoding\n");
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# on-demand blocks matched requests"),
                              1,
                              GNUNET_NO);
    pr->qe_start = GNUNET_TIME_absolute_get ();
    pr->warn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
                                                  &odc_warn_delay_task,
                                                  pr);
    if (GNUNET_OK == GNUNET_FS_handle_on_demand_block (key,
                                                       size,
                                                       data,
                                                       type,
                                                       priority,
                                                       anonymity,
                                                       replication,
                                                       expiration,
                                                       uid,
                                                       &process_local_reply,
                                                       pr))
    {
      GNUNET_STATISTICS_update (GSF_stats,
                                gettext_noop (
                                  "# on-demand lookups performed successfully"),
                                1,
                                GNUNET_NO);
      return;     /* we're done */
    }
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop ("# on-demand lookups failed"),
                              1,
                              GNUNET_NO);
    GNUNET_SCHEDULER_cancel (pr->warn_task);
    start_local_query (pr, uid + 1 /* next_uid */, false /* random */);
    return;
  }
called_from_on_demand:
  old_rf = pr->public_data.results_found;
  memset (&prq, 0, sizeof(prq));
  prq.data = data;
  prq.expiration = expiration;
  prq.size = size;
  if (GNUNET_OK !=
      GNUNET_BLOCK_get_key (GSF_block_ctx,
                            type,
                            data,
                            size,
                            &query))
  {
    GNUNET_break (0);
    GNUNET_DATASTORE_remove (GSF_dsh,
                             key,
                             size,
                             data,
                             UINT_MAX,
                             UINT_MAX,
                             NULL,
                             NULL);
    start_local_query (pr, uid + 1 /* next_uid */, false /* random */);
    return;
  }
  prq.type = type;
  prq.priority = priority;
  prq.request_found = GNUNET_NO;
  prq.anonymity_level = anonymity;
  if ((0 == old_rf) && (0 == pr->public_data.results_found))
    GSF_update_datastore_delay_ (pr->public_data.start_time);
  process_reply (&prq,
                 key,
                 pr);
  pr->local_result = prq.eval;
  if (GNUNET_BLOCK_REPLY_OK_LAST == prq.eval)
  {
    GNUNET_STATISTICS_update (
      GSF_stats,
      gettext_noop ("# Datastore lookups concluded (found last result)"),
      1,
      GNUNET_NO);
    call_continuation (pr);
    return;
  }
  if ((0 == (GSF_PRO_PRIORITY_UNLIMITED & pr->public_data.options)) &&
      ((GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
       (pr->public_data.results_found > 5 + 2 * pr->public_data.priority)))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Load too high, done with request\n");
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# Datastore lookups concluded (load too high)"),
                              1,
                              GNUNET_NO);
    call_continuation (pr);
    return;
  }
  start_local_query (pr, uid + 1 /* next_uid */, false /* random */);
}
 
 
int
GSF_pending_request_test_target_ (struct GSF_PendingRequest *pr,
                                  const struct GNUNET_PeerIdentity *target)
{
  struct GNUNET_PeerIdentity pi;
 
  if (0 == pr->origin_pid)
    return GNUNET_YES;
  GNUNET_PEER_resolve (pr->origin_pid, &pi);
  return (0 == memcmp (&pi, target, sizeof(struct GNUNET_PeerIdentity)))
         ? GNUNET_NO
         : GNUNET_YES;
}
 
 
void
GSF_local_lookup_ (struct GSF_PendingRequest *pr,
                   GSF_LocalLookupContinuation cont,
                   void *cont_cls)
{
  GNUNET_assert (NULL == pr->gh);
  GNUNET_assert (NULL == pr->cadet_request);
  GNUNET_assert (NULL == pr->llc_cont);
  pr->llc_cont = cont;
  pr->llc_cont_cls = cont_cls;
#if INSANE_STATISTICS
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Datastore lookups initiated"),
                            1,
                            GNUNET_NO);
#endif
  start_local_query (pr, 0 /* next_uid */, true /* random */);
}
 
 
void
handle_p2p_put (void *cls,
                const struct PutMessage *put)
{
  struct GSF_ConnectedPeer *cp = cls;
  uint16_t msize;
  size_t dsize;
  enum GNUNET_BLOCK_Type type;
  struct GNUNET_TIME_Absolute expiration;
  struct GNUNET_HashCode query;
  struct ProcessReplyClosure prq;
  struct GNUNET_TIME_Relative block_time;
  double putl;
  struct PutMigrationContext *pmc;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received P2P PUT from %s\n",
              GNUNET_i2s (GSF_get_peer_performance_data_ (cp)->peer));
  GSF_cover_content_count++;
  msize = ntohs (put->header.size);
  dsize = msize - sizeof(struct PutMessage);
  type = ntohl (put->type);
  expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
  /* do not allow migrated content to live longer than 1 year */
  expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (
                                           GNUNET_TIME_UNIT_YEARS),
                                         expiration);
  if (GNUNET_OK !=
      GNUNET_BLOCK_check_block (GSF_block_ctx,
                                type,
                                &put[1],
                                dsize))
  {
    GNUNET_break_op (0);
    return;
  }
  if (GNUNET_OK !=
      GNUNET_BLOCK_get_key (GSF_block_ctx,
                            type,
                            &put[1],
                            dsize,
                            &query))
  {
    GNUNET_break_op (0);
    return;
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# GAP PUT messages received"),
                            1,
                            GNUNET_NO);
  /* now, lookup 'query' */
  prq.data = (const void *) &put[1];
  prq.sender = cp;
  prq.size = dsize;
  prq.type = type;
  prq.expiration = expiration;
  prq.priority = 0;
  prq.anonymity_level = UINT32_MAX;
  prq.request_found = GNUNET_NO;
  GNUNET_CONTAINER_multihashmap_get_multiple (pr_map,
                                              &query,
                                              &process_reply,
                                              &prq);
  if (NULL != cp)
  {
    GSF_connected_peer_change_preference_ (cp,
                                           CONTENT_BANDWIDTH_VALUE
                                           + 1000 * prq.priority);
    GSF_get_peer_performance_data_ (cp)->respect += prq.priority;
  }
  if ((GNUNET_YES == active_to_migration) && (NULL != cp) &&
      (GNUNET_NO == test_put_load_too_high (prq.priority)))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Replicating result for query `%s' with priority %u\n",
                GNUNET_h2s (&query),
                prq.priority);
    pmc = GNUNET_new (struct PutMigrationContext);
    pmc->start = GNUNET_TIME_absolute_get ();
    pmc->requested = prq.request_found;
    GNUNET_assert (0 != GSF_get_peer_performance_data_ (cp)->pid);
    GNUNET_PEER_resolve (GSF_get_peer_performance_data_ (cp)->pid,
                         &pmc->origin);
    if (NULL == GNUNET_DATASTORE_put (GSF_dsh,
                                      0,
                                      &query,
                                      dsize,
                                      &put[1],
                                      type,
                                      prq.priority,
                                      1 /* anonymity */,
                                      0 /* replication */,
                                      expiration,
                                      1 + prq.priority,
                                      MAX_DATASTORE_QUEUE,
                                      &put_migration_continuation,
                                      pmc))
    {
      put_migration_continuation (pmc,
                                  GNUNET_SYSERR,
                                  GNUNET_TIME_UNIT_ZERO_ABS,
                                  NULL);
    }
  }
  else if (NULL != cp)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Choosing not to keep content `%s' (%d/%d)\n",
                GNUNET_h2s (&query),
                active_to_migration,
                test_put_load_too_high (prq.priority));
  }
  putl = GNUNET_LOAD_get_load (datastore_put_load);
  if ((NULL != cp) && (GNUNET_NO == prq.request_found) &&
      ((GNUNET_YES != active_to_migration) ||
       (putl > 2.5 * (1 + prq.priority))))
  {
    if (GNUNET_YES != active_to_migration)
      putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
    block_time = GNUNET_TIME_relative_multiply (
      GNUNET_TIME_UNIT_MILLISECONDS,
      5000 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                       (unsigned int) (60000 * putl * putl)));
    GNUNET_log (
      GNUNET_ERROR_TYPE_DEBUG,
      "Asking to stop migration for %s because of load %f and events %d/%d\n",
      GNUNET_STRINGS_relative_time_to_string (block_time, GNUNET_YES),
      putl,
      active_to_migration,
      (GNUNET_NO == prq.request_found));
    GSF_block_peer_migration_ (cp,
                               GNUNET_TIME_relative_to_absolute (block_time));
  }
}
 
 
int
GSF_pending_request_test_active_ (struct GSF_PendingRequest *pr)
{
  return (NULL != pr->rh) ? GNUNET_YES : GNUNET_NO;
}
 
 
void
GSF_pending_request_init_ ()
{
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number (GSF_cfg,
                                             "fs",
                                             "MAX_PENDING_REQUESTS",
                                             &max_pending_requests))
  {
    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
                               "fs",
                               "MAX_PENDING_REQUESTS");
  }
  active_to_migration =
    GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg, "FS", "CONTENT_CACHING");
  datastore_put_load = GNUNET_LOAD_value_init (DATASTORE_LOAD_AUTODECLINE);
  pr_map = GNUNET_CONTAINER_multihashmap_create (32 * 1024, GNUNET_YES);
  requests_by_expiration_heap =
    GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
}
 
 
void
GSF_pending_request_done_ ()
{
  GNUNET_CONTAINER_multihashmap_iterate (pr_map, &clean_request, NULL);
  GNUNET_CONTAINER_multihashmap_destroy (pr_map);
  pr_map = NULL;
  GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
  requests_by_expiration_heap = NULL;
  GNUNET_LOAD_value_free (datastore_put_load);
  datastore_put_load = NULL;
}
 
 
/* end of gnunet-service-fs_pr.c */