-fixing #2578
[oweals/gnunet.git] / src / fs / gnunet-service-fs_pr.c
index aa18e2fc832a1ff5ad9dc59c64c612c343192f21..1df10abd296a3e7eb42bda6e39a3a62503efb5fc 100644 (file)
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
+     (C) 2009, 2010, 2011, 2012 Christian Grothoff (and other contributing authors)
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -74,7 +74,7 @@ struct GSF_PendingRequest
   /**
    * Array of hash codes of replies we've already seen.
    */
-  GNUNET_HashCode *replies_seen;
+  struct GNUNET_HashCode *replies_seen;
 
   /**
    * Bloomfilter masking replies we've already seen.
@@ -263,26 +263,24 @@ refresh_bloomfilter (struct GSF_PendingRequest *pr)
 struct GSF_PendingRequest *
 GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
                              enum GNUNET_BLOCK_Type type,
-                             const GNUNET_HashCode * query,
-                             const GNUNET_HashCode * namespace,
+                             const struct GNUNET_HashCode * query,
+                             const struct GNUNET_HashCode * namespace,
                              const struct GNUNET_PeerIdentity *target,
                              const char *bf_data, size_t bf_size,
                              uint32_t mingle, uint32_t anonymity_level,
                              uint32_t priority, int32_t ttl,
                              GNUNET_PEER_Id sender_pid,
                              GNUNET_PEER_Id origin_pid,
-                             const GNUNET_HashCode * replies_seen,
+                             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;
 
-#if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Creating request handle for `%s' of type %d\n",
               GNUNET_h2s (query), type);
-#endif
   GNUNET_STATISTICS_update (GSF_stats,
                             gettext_noop ("# Pending requests created"), 1,
                             GNUNET_NO);
@@ -326,9 +324,9 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
   {
     pr->replies_seen_size = replies_seen_count;
     pr->replies_seen =
-        GNUNET_malloc (sizeof (GNUNET_HashCode) * pr->replies_seen_size);
+        GNUNET_malloc (sizeof (struct GNUNET_HashCode) * pr->replies_seen_size);
     memcpy (pr->replies_seen, replies_seen,
-            replies_seen_count * sizeof (GNUNET_HashCode));
+            replies_seen_count * sizeof (struct GNUNET_HashCode));
     pr->replies_seen_count = replies_seen_count;
   }
   if (NULL != bf_data)
@@ -345,7 +343,7 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
   }
   GNUNET_CONTAINER_multihashmap_put (pr_map, query, pr,
                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
-  if (0 != (options & GSF_PRO_REQUEST_EXPIRES))
+  if (0 == (options & GSF_PRO_REQUEST_NEVER_EXPIRES))
   {
     pr->hnode =
         GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap, pr,
@@ -358,9 +356,10 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
       GNUNET_assert (dpr != NULL);
       if (pr == dpr)
         break;                  /* let the request live briefly... */
-      dpr->rh (dpr->rh_cls, GNUNET_BLOCK_EVALUATION_REQUEST_VALID, dpr,
-               UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_BLOCK_TYPE_ANY,
-               NULL, 0);
+      if (NULL != dpr->rh)
+       dpr->rh (dpr->rh_cls, GNUNET_BLOCK_EVALUATION_REQUEST_VALID, 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);
     }
   }
@@ -370,7 +369,6 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
   return pr;
 }
 
-
 /**
  * Obtain the public data associated with a pending request
  *
@@ -400,11 +398,11 @@ GSF_pending_request_is_compatible_ (struct GSF_PendingRequest *pra,
   if ((pra->public_data.type != prb->public_data.type) ||
       (0 !=
        memcmp (&pra->public_data.query, &prb->public_data.query,
-               sizeof (GNUNET_HashCode))) ||
+               sizeof (struct GNUNET_HashCode))) ||
       ((pra->public_data.type == GNUNET_BLOCK_TYPE_FS_SBLOCK) &&
        (0 !=
         memcmp (&pra->public_data.namespace, &prb->public_data.namespace,
-                sizeof (GNUNET_HashCode)))))
+                sizeof (struct GNUNET_HashCode)))))
     return GNUNET_NO;
   return GNUNET_OK;
 }
@@ -421,11 +419,11 @@ GSF_pending_request_is_compatible_ (struct GSF_PendingRequest *pra,
  */
 void
 GSF_pending_request_update_ (struct GSF_PendingRequest *pr,
-                             const GNUNET_HashCode * replies_seen,
+                             const struct GNUNET_HashCode * replies_seen,
                              unsigned int replies_seen_count)
 {
   unsigned int i;
-  GNUNET_HashCode mhash;
+  struct GNUNET_HashCode mhash;
 
   if (replies_seen_count + pr->replies_seen_count < pr->replies_seen_count)
     return;                     /* integer overflow */
@@ -436,7 +434,7 @@ GSF_pending_request_update_ (struct GSF_PendingRequest *pr,
       GNUNET_array_grow (pr->replies_seen, pr->replies_seen_size,
                          replies_seen_count + pr->replies_seen_count);
     memcpy (&pr->replies_seen[pr->replies_seen_count], replies_seen,
-            sizeof (GNUNET_HashCode) * replies_seen_count);
+            sizeof (struct GNUNET_HashCode) * replies_seen_count);
     pr->replies_seen_count += replies_seen_count;
     refresh_bloomfilter (pr);
   }
@@ -479,7 +477,7 @@ GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
 {
   char lbuf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
   struct GetMessage *gm;
-  GNUNET_HashCode *ext;
+  struct GNUNET_HashCode *ext;
   size_t msize;
   unsigned int k;
   uint32_t bm;
@@ -489,12 +487,10 @@ GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
   int64_t ttl;
   int do_route;
 
-#if DEBUG_FS
   if (buf_size > 0)
     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);
-#endif
   k = 0;
   bm = 0;
   do_route = (0 == (pr->public_data.options & GSF_PRO_FORWARD_ONLY));
@@ -519,7 +515,7 @@ GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
     k++;
   }
   bf_size = GNUNET_CONTAINER_bloomfilter_get_size (pr->bf);
-  msize = sizeof (struct GetMessage) + bf_size + k * sizeof (GNUNET_HashCode);
+  msize = sizeof (struct GetMessage) + bf_size + k * sizeof (struct GNUNET_HashCode);
   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
   if (buf_size < msize)
     return msize;
@@ -534,6 +530,8 @@ GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
   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 - now.abs_value);
@@ -541,13 +539,13 @@ GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
   gm->filter_mutator = htonl (pr->mingle);
   gm->hash_bitmap = htonl (bm);
   gm->query = pr->public_data.query;
-  ext = (GNUNET_HashCode *) & gm[1];
+  ext = (struct GNUNET_HashCode *) & gm[1];
   k = 0;
   if (!do_route)
     GNUNET_PEER_resolve (pr->sender_pid,
                          (struct GNUNET_PeerIdentity *) &ext[k++]);
   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
-    memcpy (&ext[k++], &pr->public_data.namespace, sizeof (GNUNET_HashCode));
+    memcpy (&ext[k++], &pr->public_data.namespace, sizeof (struct GNUNET_HashCode));
   if (GNUNET_YES == pr->public_data.has_target)
     ext[k++] = pr->public_data.target.hashPubKey;
   if (pr->bf != NULL)
@@ -569,15 +567,13 @@ GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
  * @return GNUNET_YES (we should continue to iterate)
  */
 static int
-clean_request (void *cls, const GNUNET_HashCode * key, void *value)
+clean_request (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   struct GSF_PendingRequest *pr = value;
   GSF_LocalLookupContinuation cont;
 
-#if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Cleaning up pending request for `%s'.\n", GNUNET_h2s (key));
-#endif
   if (NULL != (cont = pr->llc_cont))
   {
     pr->llc_cont = NULL;
@@ -769,19 +765,18 @@ update_request_performance_data (struct ProcessReplyClosure *prq,
  * @return GNUNET_YES (we should continue to iterate)
  */
 static int
-process_reply (void *cls, const GNUNET_HashCode * key, void *value)
+process_reply (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   struct ProcessReplyClosure *prq = cls;
   struct GSF_PendingRequest *pr = value;
-  GNUNET_HashCode chash;
+  struct GNUNET_HashCode chash;
+  struct GNUNET_TIME_Absolute last_transmission;
 
   if (NULL == pr->rh)
     return GNUNET_YES;
-#if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Matched result (type %u) for query `%s' with pending request\n",
               (unsigned int) prq->type, GNUNET_h2s (key));
-#endif
   GNUNET_STATISTICS_update (GSF_stats,
                             gettext_noop ("# replies received and matched"), 1,
                             GNUNET_NO);
@@ -790,7 +785,7 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
                              &pr->public_data.namespace,
                              (prq->type ==
                               GNUNET_BLOCK_TYPE_FS_SBLOCK) ?
-                             sizeof (GNUNET_HashCode) : 0, prq->data,
+                             sizeof (struct GNUNET_HashCode) : 0, prq->data,
                              prq->size);
   switch (prq->eval)
   {
@@ -803,19 +798,19 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
     GNUNET_LOAD_update (GSF_rt_entry_lifetime,
                         GNUNET_TIME_absolute_get_duration (pr->
                                                            public_data.start_time).rel_value);
+    if (!GSF_request_plan_reference_get_last_transmission_ (pr->public_data.rpr_head, prq->sender, &last_transmission))
+      last_transmission.abs_value = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value;
     /* pass on to other peers / local clients */
     pr->rh (pr->rh_cls, prq->eval, pr, prq->anonymity_level, prq->expiration,
-            prq->type, prq->data, prq->size);
+            last_transmission, prq->type, prq->data, prq->size);
     return GNUNET_YES;
   case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
     GNUNET_STATISTICS_update (GSF_stats,
                               gettext_noop
                               ("# duplicate replies discarded (bloomfilter)"),
                               1, GNUNET_NO);
-#if DEBUG_FS && 0
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Duplicate response `%s', discarding.\n", GNUNET_h2s (&mhash));
-#endif
+                "Duplicate response, discarding.\n");
     return GNUNET_YES;          /* duplicate */
   case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
     return GNUNET_YES;          /* wrong namespace */
@@ -835,11 +830,9 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
   GSF_pending_request_update_ (pr, &chash, 1);
   if (NULL == prq->sender)
   {
-#if DEBUG_FS
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Found result for query `%s' in local datastore\n",
                 GNUNET_h2s (key));
-#endif
     GNUNET_STATISTICS_update (GSF_stats,
                               gettext_noop ("# results found locally"), 1,
                               GNUNET_NO);
@@ -854,8 +847,10 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
   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.rpr_head, prq->sender, &last_transmission))
+    last_transmission.abs_value = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value;
   pr->rh (pr->rh_cls, prq->eval, pr, prq->anonymity_level, prq->expiration,
-          prq->type, prq->data, prq->size);
+          last_transmission, prq->type, prq->data, prq->size);
   return GNUNET_YES;
 }
 
@@ -890,49 +885,72 @@ struct PutMigrationContext
  *
  * @param cls closure
  * @param success GNUNET_SYSERR on failure
+ * @param min_expiration minimum expiration time required for content to be stored
  * @param msg NULL on success, otherwise an error message
  */
 static void
-put_migration_continuation (void *cls, int success, const char *msg)
+put_migration_continuation (void *cls, int success, 
+                           struct GNUNET_TIME_Absolute min_expiration,
+                           const char *msg)
 {
   struct PutMigrationContext *pmc = cls;
-  struct GNUNET_TIME_Relative delay;
-  struct GNUNET_TIME_Relative block_time;
   struct GSF_ConnectedPeer *cp;
+  struct GNUNET_TIME_Relative mig_pause;
   struct GSF_PeerPerformanceData *ppd;
 
-  delay = GNUNET_TIME_absolute_get_duration (pmc->start);
+  if (NULL != datastore_put_load)
+  {
+    if (GNUNET_SYSERR != success)
+    {
+      GNUNET_LOAD_update (datastore_put_load, 
+                         GNUNET_TIME_absolute_get_duration (pmc->start).rel_value);
+    }
+    else
+    {
+      /* on queue failure / timeout, increase the put load dramatically */
+      GNUNET_LOAD_update (datastore_put_load, 
+                         GNUNET_TIME_UNIT_MINUTES.rel_value);
+    }
+  }
   cp = GSF_peer_get_ (&pmc->origin);
-  if ((GNUNET_OK != success) && (GNUNET_NO == pmc->requested))
+  if (GNUNET_OK == success)
   {
-    /* block migration for a bit... */
     if (NULL != cp)
     {
       ppd = GSF_get_peer_performance_data_ (cp);
-      ppd->migration_duplication++;
-      block_time =
-          GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
-                                         5 * ppd->migration_duplication +
-                                         GNUNET_CRYPTO_random_u32
-                                         (GNUNET_CRYPTO_QUALITY_WEAK, 5));
-      GSF_block_peer_migration_ (cp, block_time);
+      ppd->migration_delay.rel_value /= 2;
     }
+    GNUNET_free (pmc);
+    return;
   }
-  else
+  if ( (GNUNET_NO == success) && 
+       (GNUNET_NO == pmc->requested) && 
+       (NULL != cp) )
   {
-    if (NULL != cp)
+    ppd = GSF_get_peer_performance_data_ (cp);
+    if (min_expiration.abs_value > 0)
     {
-      ppd = GSF_get_peer_performance_data_ (cp);
-      ppd->migration_duplication = 0;   /* reset counter */
+      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 = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                                     ppd->migration_delay.rel_value);
+      ppd->migration_delay = GNUNET_TIME_relative_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);
-  /* on failure, increase the put load dramatically */
-  if (NULL != datastore_put_load)
-    GNUNET_LOAD_update (datastore_put_load, 
-                       GNUNET_TIME_UNIT_HOURS.rel_value);
-  if (GNUNET_OK == success)
-    return;
   GNUNET_STATISTICS_update (GSF_stats,
                             gettext_noop ("# Datastore `PUT' failures"), 1,
                             GNUNET_NO);
@@ -984,7 +1002,7 @@ test_put_load_too_high (uint32_t priority)
  */
 static void
 handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp,
-                  const GNUNET_HashCode * key,
+                  const struct GNUNET_HashCode * key,
                   const struct GNUNET_PeerIdentity *get_path,
                   unsigned int get_path_length,
                   const struct GNUNET_PeerIdentity *put_path,
@@ -1001,17 +1019,18 @@ handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp,
   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)))
   {
-#if DEBUG_FS
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Replicating result for query `%s' with priority %u\n",
                 GNUNET_h2s (key), prq.priority);
-#endif
     pmc = GNUNET_malloc (sizeof (struct PutMigrationContext));
     pmc->start = GNUNET_TIME_absolute_get ();
     pmc->requested = GNUNET_YES;
@@ -1023,7 +1042,7 @@ handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp,
                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
                               &put_migration_continuation, pmc))
     {
-      put_migration_continuation (pmc, GNUNET_NO, NULL);
+      put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
     }
   }
 }
@@ -1040,7 +1059,7 @@ GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
   const void *xquery;
   size_t xquery_size;
   struct GNUNET_PeerIdentity pi;
-  char buf[sizeof (GNUNET_HashCode) * 2];
+  char buf[sizeof (struct GNUNET_HashCode) * 2] GNUNET_ALIGN;
 
   if (0 != pr->public_data.anonymity_level)
     return;
@@ -1054,8 +1073,8 @@ GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
   {
     xquery = buf;
-    memcpy (buf, &pr->public_data.namespace, sizeof (GNUNET_HashCode));
-    xquery_size = sizeof (GNUNET_HashCode);
+    memcpy (buf, &pr->public_data.namespace, sizeof (struct GNUNET_HashCode));
+    xquery_size = sizeof (struct GNUNET_HashCode);
   }
   if (0 != (pr->public_data.options & GSF_PRO_FORWARD_ONLY))
   {
@@ -1065,7 +1084,7 @@ GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
     xquery_size += sizeof (struct GNUNET_PeerIdentity);
   }
   pr->gh =
-      GNUNET_DHT_get_start (GSF_dht, GNUNET_TIME_UNIT_FOREVER_REL,
+      GNUNET_DHT_get_start (GSF_dht, 
                             pr->public_data.type, &pr->public_data.query,
                             5 /* DEFAULT_GET_REPLICATION */ ,
                             GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
@@ -1086,9 +1105,8 @@ warn_delay_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   struct GSF_PendingRequest *pr = cls;
 
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              _("Datastore lookup already took %llu ms!\n"),
-              (unsigned long long)
-              GNUNET_TIME_absolute_get_duration (pr->qe_start).rel_value);
+              _("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);
@@ -1107,9 +1125,8 @@ odc_warn_delay_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   struct GSF_PendingRequest *pr = cls;
 
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              _("On-demand lookup already took %llu ms!\n"),
-              (unsigned long long)
-              GNUNET_TIME_absolute_get_duration (pr->qe_start).rel_value);
+              _("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);
@@ -1134,7 +1151,7 @@ odc_warn_delay_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
  *        maybe 0 if no unique identifier is available
  */
 static void
-process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
+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,
                      struct GNUNET_TIME_Absolute expiration, uint64_t uid)
@@ -1142,7 +1159,7 @@ process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
   struct GSF_PendingRequest *pr = cls;
   GSF_LocalLookupContinuation cont;
   struct ProcessReplyClosure prq;
-  GNUNET_HashCode query;
+  struct GNUNET_HashCode query;
   unsigned int old_rf;
 
   GNUNET_SCHEDULER_cancel (pr->warn_task);
@@ -1185,10 +1202,8 @@ process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
   }
   if (NULL == key)
   {
-#if DEBUG_FS
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
                 "No further local responses available.\n");
-#endif
     if ((pr->public_data.type == GNUNET_BLOCK_TYPE_FS_DBLOCK) ||
         (pr->public_data.type == GNUNET_BLOCK_TYPE_FS_IBLOCK))
       GNUNET_STATISTICS_update (GSF_stats,
@@ -1197,17 +1212,13 @@ process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
                                 GNUNET_NO);
     goto check_error_and_continue;
   }
-#if DEBUG_FS
   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);
-#endif
   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
   {
-#if DEBUG_FS
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Found ONDEMAND block, performing on-demand encoding\n");
-#endif
     GNUNET_STATISTICS_update (GSF_stats,
                               gettext_noop
                               ("# on-demand blocks matched requests"), 1,
@@ -1321,9 +1332,7 @@ process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
       ((GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
        (pr->public_data.results_found > 5 + 2 * pr->public_data.priority)))
   {
-#if DEBUG_FS > 2
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Load too high, done with request\n");
-#endif
     GNUNET_STATISTICS_update (GSF_stats,
                               gettext_noop
                               ("# Datastore lookups concluded (load too high)"),
@@ -1468,7 +1477,7 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
   size_t dsize;
   enum GNUNET_BLOCK_Type type;
   struct GNUNET_TIME_Absolute expiration;
-  GNUNET_HashCode query;
+  struct GNUNET_HashCode query;
   struct ProcessReplyClosure prq;
   struct GNUNET_TIME_Relative block_time;
   double putl;
@@ -1484,6 +1493,9 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
   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 (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
     return GNUNET_SYSERR;
   if (GNUNET_OK !=
@@ -1497,10 +1509,7 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
                             GNUNET_NO);
   /* now, lookup 'query' */
   prq.data = (const void *) &put[1];
-  if (NULL != cp)
-    prq.sender = cp;
-  else
-    prq.sender = NULL;
+  prq.sender = cp;
   prq.size = dsize;
   prq.type = type;
   prq.expiration = expiration;
@@ -1514,16 +1523,15 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
     GSF_connected_peer_change_preference_ (cp,
                                            CONTENT_BANDWIDTH_VALUE +
                                            1000 * prq.priority);
-    GSF_get_peer_performance_data_ (cp)->trust += 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)))
   {
-#if DEBUG_FS
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Replicating result for query `%s' with priority %u\n",
                 GNUNET_h2s (&query), prq.priority);
-#endif
     pmc = GNUNET_malloc (sizeof (struct PutMigrationContext));
     pmc->start = GNUNET_TIME_absolute_get ();
     pmc->requested = prq.request_found;
@@ -1538,22 +1546,21 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
                               &put_migration_continuation, pmc))
     {
-      put_migration_continuation (pmc, GNUNET_NO, NULL);
+      put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
     }
   }
-  else
+  else if (NULL != cp)
   {
-#if DEBUG_FS
     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));
-#endif
   }
   putl = GNUNET_LOAD_get_load (datastore_put_load);
-  if ((NULL != (cp = prq.sender)) && (GNUNET_NO == prq.request_found) &&
-      ((GNUNET_YES != active_to_migration) ||
-       (putl > 2.5 * (1 + prq.priority))))
+  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);
@@ -1563,7 +1570,13 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
                                        GNUNET_CRYPTO_random_u32
                                        (GNUNET_CRYPTO_QUALITY_WEAK,
                                         (unsigned int) (60000 * putl * putl)));
-    GSF_block_peer_migration_ (cp, block_time);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
+               "Asking to stop migration for %llu ms because of load %f and events %d/%d\n",
+               (unsigned long long) block_time.rel_value,
+               putl,
+               active_to_migration,
+               (GNUNET_NO == prq.request_found));
+    GSF_block_peer_migration_ (cp, GNUNET_TIME_relative_to_absolute (block_time));
   }
   return GNUNET_OK;
 }
@@ -1582,19 +1595,15 @@ GSF_pending_request_init_ ()
                                              "MAX_PENDING_REQUESTS",
                                              &max_pending_requests))
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                _
-                ("Configuration fails to specify `%s', assuming default value."),
-                "MAX_PENDING_REQUESTS");
+    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
+                              "fs", "MAX_PENDING_REQUESTS");
   }
   if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_size (GSF_cfg, "ats", "WAN_QUOTA_OUT",
                                            &bps))
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                _
-                ("Configuration fails to specify `%s', assuming default value."),
-                "WAN_QUOTA_OUT");
+    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
+                              "ats", "WAN_QUOTA_OUT");
     bps = 65536;
   }
   /* queue size should be #queries we can have pending and satisfy within