[datastore] Fix #3743

This change adds support for key == NULL to the datastore plugins
and replaces the offset argument with a next_uid and random arguments to
increase performance in the key == NULL case.

With the offset argument a datastore plugin would have to count all
matching keys before fetching the key at the right offset, which would
iterate over the entire database in the case of key == NULL.

The offset argument was used in two ways: to iterate over a set of
matching values and to start iteration at a random matching value. The new API
seperates these into two arguments: if random is true it will return a
random matching value, otherwise next_uid can be set to uid + 1 to return the
next matching value.

The random argument was not added to get_zero_anonymity. This function
is used to periodically insert zero anonymity values into the DHT. I
don't think it's necessary to randomize this.

5 files changed, 251 insertions, 277 deletions

diff --git a/src/fs/fs_api.h b/src/fs/fs_api.h
index e85de94a7..be22ea73e 100644
--- a/src/fs/fs_api.h
+++ b/src/fs/fs_api.h
@@ -1464,21 +1464,11 @@ struct GNUNET_FS_UnindexContext
   struct GNUNET_CRYPTO_FileHashContext *fhc;
 
   /**
-   * Which values have we seen already?
-   */
-  struct GNUNET_CONTAINER_MultiHashMap *seen_dh;
-
-  /**
    * Overall size of the file.
    */
   uint64_t file_size;
 
   /**
-   * Random offset given to #GNUNET_DATASTORE_get_key.
-   */
-  uint64_t roff;
-
-  /**
    * When did we start?
    */
   struct GNUNET_TIME_Absolute start_time;
diff --git a/src/fs/fs_unindex.c b/src/fs/fs_unindex.c
index ad1499f00..e1c7ea535 100644
--- a/src/fs/fs_unindex.c
+++ b/src/fs/fs_unindex.c
@@ -312,8 +312,6 @@ unindex_finish (struct GNUNET_FS_UnindexContext *uc)
   uc->fh = NULL;
   GNUNET_DATASTORE_disconnect (uc->dsh, GNUNET_NO);
   uc->dsh = NULL;
-  GNUNET_CONTAINER_multihashmap_destroy (uc->seen_dh);
-  uc->seen_dh = NULL;
   uc->state = UNINDEX_STATE_FS_NOTIFY;
   GNUNET_FS_unindex_sync_ (uc);
   uc->mq = GNUNET_CLIENT_connect (uc->h->cfg,
@@ -444,7 +442,6 @@ continue_after_remove (void *cls,
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                 _("Failed to remove UBlock: %s\n"),
                 msg);
-  GNUNET_CONTAINER_multihashmap_clear (uc->seen_dh);
   uc->ksk_offset++;
   GNUNET_FS_unindex_do_remove_kblocks_ (uc);
 }
@@ -486,34 +483,15 @@ process_kblock_for_unindex (void *cls,
   const struct UBlock *ub;
   struct GNUNET_FS_Uri *chk_uri;
   struct GNUNET_HashCode query;
-  struct GNUNET_HashCode dh;
 
   uc->dqe = NULL;
   if (NULL == data)
   {
     /* no result */
-    GNUNET_CONTAINER_multihashmap_clear (uc->seen_dh);
     uc->ksk_offset++;
     GNUNET_FS_unindex_do_remove_kblocks_ (uc);
     return;
   }
-  GNUNET_CRYPTO_hash (data,
-                      size,
-                      &dh);
-  if (GNUNET_YES ==
-      GNUNET_CONTAINER_multihashmap_contains (uc->seen_dh,
-                                              &dh))
-  {
-    GNUNET_CONTAINER_multihashmap_clear (uc->seen_dh);
-    uc->ksk_offset++;
-    GNUNET_FS_unindex_do_remove_kblocks_ (uc);
-    return;
-  }
-  GNUNET_assert (GNUNET_OK ==
-                 GNUNET_CONTAINER_multihashmap_put (uc->seen_dh,
-                                                    &dh,
-                                                    uc,
-                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
   GNUNET_assert (GNUNET_BLOCK_TYPE_FS_UBLOCK == type);
   if (size < sizeof (struct UBlock))
   {
@@ -566,23 +544,24 @@ process_kblock_for_unindex (void *cls,
   GNUNET_FS_uri_destroy (chk_uri);
   /* matches! */
   uc->dqe = GNUNET_DATASTORE_remove (uc->dsh,
-                                     key,
+                                     key,
                                      size,
                                      data,
-                                     0 /* priority */,
+                                     0 /* priority */,
                                      1 /* queue size */,
-                                     &continue_after_remove,
-                                     uc);
+                                     &continue_after_remove,
+                                     uc);
   return;
  get_next:
   uc->dqe = GNUNET_DATASTORE_get_key (uc->dsh,
-                                      uc->roff++,
-                                      &uc->uquery,
-                                      GNUNET_BLOCK_TYPE_FS_UBLOCK,
-                                      0 /* priority */,
+                                      uid + 1 /* next_uid */,
+                                      false /* random */,
+                                      &uc->uquery,
+                                      GNUNET_BLOCK_TYPE_FS_UBLOCK,
+                                      0 /* priority */,
                                       1 /* queue size */,
-                                      &process_kblock_for_unindex,
-                                      uc);
+                                      &process_kblock_for_unindex,
+                                      uc);
 }
 
 
@@ -627,13 +606,14 @@ GNUNET_FS_unindex_do_remove_kblocks_ (struct GNUNET_FS_UnindexContext *uc)
                       sizeof (dpub),
                       &uc->uquery);
   uc->dqe = GNUNET_DATASTORE_get_key (uc->dsh,
-                                      uc->roff++,
-                                      &uc->uquery,
-                                      GNUNET_BLOCK_TYPE_FS_UBLOCK,
-                                      0 /* priority */,
+                                      0 /* next_uid */,
+                                      false /* random */,
+                                      &uc->uquery,
+                                      GNUNET_BLOCK_TYPE_FS_UBLOCK,
+                                      0 /* priority */,
                                       1 /* queue size */,
-                                      &process_kblock_for_unindex,
-                                      uc);
+                                      &process_kblock_for_unindex,
+                                      uc);
 }
 
 
@@ -826,8 +806,6 @@ GNUNET_FS_unindex_start (struct GNUNET_FS_Handle *h,
   uc->start_time = GNUNET_TIME_absolute_get ();
   uc->file_size = size;
   uc->client_info = cctx;
-  uc->seen_dh = GNUNET_CONTAINER_multihashmap_create (4,
-                                                       GNUNET_NO);
   GNUNET_FS_unindex_sync_ (uc);
   pi.status = GNUNET_FS_STATUS_UNINDEX_START;
   pi.value.unindex.eta = GNUNET_TIME_UNIT_FOREVER_REL;
diff --git a/src/fs/gnunet-service-fs_cadet_server.c b/src/fs/gnunet-service-fs_cadet_server.c
index b1a098175..f8619b812 100644
--- a/src/fs/gnunet-service-fs_cadet_server.c
+++ b/src/fs/gnunet-service-fs_cadet_server.c
@@ -345,12 +345,13 @@ handle_request (void *cls,
                             GNUNET_NO);
   refresh_timeout_task (sc);
   sc->qe = GNUNET_DATASTORE_get_key (GSF_dsh,
-                                     0,
-                                     &sqm->query,
-                                     ntohl (sqm->type),
-                                     0 /* priority */,
-                                     GSF_datastore_queue_size,
-                                     &handle_datastore_reply,
+                                     0 /* next_uid */,
+                                     false /* random */,
+                                     &sqm->query,
+                                     ntohl (sqm->type),
+                                     0 /* priority */,
+                                     GSF_datastore_queue_size,
+                                     &handle_datastore_reply,
                                      sc);
   if (NULL == sc->qe)
   {
diff --git a/src/fs/gnunet-service-fs_pr.c b/src/fs/gnunet-service-fs_pr.c
index b0fda24b5..b736b49c2 100644
--- a/src/fs/gnunet-service-fs_pr.c
+++ b/src/fs/gnunet-service-fs_pr.c
@@ -160,20 +160,27 @@ struct GSF_PendingRequest
   struct GNUNET_SCHEDULER_Task * warn_task;
 
   /**
-   * Current offset for querying our local datastore for results.
-   * Starts at a random value, incremented until we get the same
-   * UID again (detected using 'first_uid'), which is then used
-   * to termiante the iteration.
+   * Do we have a first UID yet?
+   */
+  bool have_first_uid;
+
+  /**
+   * Have we seen a NULL result yet?
    */
-  uint64_t local_result_offset;
+  bool seen_null;
 
   /**
    * Unique ID of the first result from the local datastore;
-   * used to detect wrap-around of the offset.
+   * used to terminate the loop.
    */
   uint64_t first_uid;
 
   /**
+   * Result count.
+   */
+  size_t result_count;
+
+  /**
    * How often have we retried this request via 'cadet'?
    * (used to bound overall retries).
    */
@@ -189,11 +196,6 @@ struct GSF_PendingRequest
    */
   unsigned int replies_seen_size;
 
-  /**
-   * Do we have a first UID yet?
-   */
-  unsigned int have_first_uid;
-
 };
 
 
@@ -332,8 +334,6 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
   if (NULL != target)
     extra += sizeof (struct GNUNET_PeerIdentity);
   pr = GNUNET_malloc (sizeof (struct GSF_PendingRequest) + extra);
-  pr->local_result_offset =
-      GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
   pr->public_data.query = *query;
   eptr = (struct GNUNET_HashCode *) &pr[1];
   if (NULL != target)
@@ -1340,6 +1340,123 @@ odc_warn_delay_task (void *cls)
 }
 
 
+/* 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_EVALUATION_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_EVALUATION_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_EVALUATION_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,
+                     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);
+}
+
+
 /**
  * We're processing (local) results for a search request
  * from another peer.  Pass applicable results to the
@@ -1369,69 +1486,71 @@ process_local_reply (void *cls,
                      uint64_t uid)
 {
   struct GSF_PendingRequest *pr = cls;
-  GSF_LocalLookupContinuation cont;
   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)
+  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 */
   {
-    pr->qe = NULL;
-    if (NULL == key)
-    {
+    /* No results */
 #if INSANE_STATISTICS
-      GNUNET_STATISTICS_update (GSF_stats,
-                                gettext_noop
-                                ("# Datastore lookups concluded (no results)"),
-                                1, GNUNET_NO);
+    GNUNET_STATISTICS_update (GSF_stats,
+                              gettext_noop
+                              ("# Datastore lookups concluded (no results)"),
+                              1, GNUNET_NO);
 #endif
-    }
-    if (GNUNET_NO == pr->have_first_uid)
-    {
-      pr->first_uid = uid;
-      pr->have_first_uid = 1;
-    }
-    else
-    {
-      if ((uid == pr->first_uid) && (key != NULL))
-      {
-        GNUNET_STATISTICS_update (GSF_stats,
-                                  gettext_noop
-                                  ("# Datastore lookups concluded (seen all)"),
-                                  1, GNUNET_NO);
-        key = NULL;             /* all replies seen! */
-      }
-      pr->have_first_uid++;
-      if ((pr->have_first_uid > MAX_RESULTS) && (key != NULL))
-      {
-        GNUNET_STATISTICS_update (GSF_stats,
-                                  gettext_noop
-                                  ("# Datastore lookups aborted (more than MAX_RESULTS)"),
-                                  1, GNUNET_NO);
-        key = NULL;             /* all replies seen! */
-      }
-    }
+    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_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
-                "No further local responses available.\n");
-#if INSANE_STATISTICS
-    if ((pr->public_data.type == GNUNET_BLOCK_TYPE_FS_DBLOCK) ||
-        (pr->public_data.type == GNUNET_BLOCK_TYPE_FS_IBLOCK))
-      GNUNET_STATISTICS_update (GSF_stats,
-                                gettext_noop
-                                ("# requested DBLOCK or IBLOCK not found"), 1,
-                                GNUNET_NO);
-#endif
-    goto check_error_and_continue;
+    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 (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
+  if (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Found ONDEMAND block, performing on-demand encoding\n");
@@ -1458,33 +1577,12 @@ process_local_reply (void *cls,
                               gettext_noop ("# on-demand lookups failed"), 1,
                               GNUNET_NO);
     GNUNET_SCHEDULER_cancel (pr->warn_task);
-    pr->warn_task =
-        GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
-                                      &warn_delay_task, pr);
-    pr->qe =
-        GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset - 1,
-                                  &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;                   /* we're done */
-    GNUNET_STATISTICS_update (GSF_stats,
-                              gettext_noop
-                              ("# Datastore lookups concluded (error queueing)"),
-                              1, GNUNET_NO);
-    goto check_error_and_continue;
+    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;
@@ -1496,34 +1594,9 @@ process_local_reply (void *cls,
     GNUNET_break (0);
     GNUNET_DATASTORE_remove (GSF_dsh, key, size, data, -1, -1,
                              NULL, NULL);
-    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, pr->local_result_offset - 1,
-                                  &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)
-    {
-      GNUNET_STATISTICS_update (GSF_stats,
-                                gettext_noop
-                                ("# Datastore lookups concluded (error queueing)"),
-                                1, GNUNET_NO);
-      goto check_error_and_continue;
-    }
+    start_local_query (pr,
+                       uid + 1 /* next_uid */,
+                       false /* random */);
     return;
   }
   prq.type = type;
@@ -1535,14 +1608,15 @@ process_local_reply (void *cls,
   prq.eo = GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO;
   process_reply (&prq, key, pr);
   pr->local_result = prq.eval;
-  if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
+  if (GNUNET_BLOCK_EVALUATION_OK_LAST == prq.eval)
   {
     GNUNET_STATISTICS_update (GSF_stats,
                               gettext_noop
                               ("# Datastore lookups concluded (found last result)"),
                               1,
                               GNUNET_NO);
-    goto check_error_and_continue;
+    call_continuation (pr);
+    return;
   }
   if ((0 == (GSF_PRO_PRIORITY_UNLIMITED & pr->public_data.options)) &&
       ((GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
@@ -1554,66 +1628,12 @@ process_local_reply (void *cls,
                               gettext_noop ("# Datastore lookups concluded (load too high)"),
                               1,
                               GNUNET_NO);
-    goto check_error_and_continue;
-  }
-  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, pr->local_result_offset++,
-                                &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);
-  /* check if we successfully queued another datastore request;
-   * if so, return, otherwise call our continuation (if we have
-   * any) */
-check_error_and_continue:
-  if (NULL != pr->qe)
+    call_continuation (pr);
     return;
-  if (NULL != pr->warn_task)
-  {
-    GNUNET_SCHEDULER_cancel (pr->warn_task);
-    pr->warn_task = NULL;
   }
-  if (NULL == (cont = pr->llc_cont))
-    return;                     /* no continuation */
-  pr->llc_cont = NULL;
-  if (0 != (GSF_PRO_LOCAL_ONLY & pr->public_data.options))
-  {
-    if (GNUNET_BLOCK_EVALUATION_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_EVALUATION_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_EVALUATION_OK_LAST);
-    return;
-  }
-
-  cont (pr->llc_cont_cls, pr, pr->local_result);
+  start_local_query (pr,
+                     uid + 1 /* next_uid */,
+                     false /* random */);
 }
 
 
@@ -1657,43 +1677,14 @@ GSF_local_lookup_ (struct GSF_PendingRequest *pr,
   GNUNET_assert (NULL == pr->llc_cont);
   pr->llc_cont = cont;
   pr->llc_cont_cls = cont_cls;
-  pr->qe_start = GNUNET_TIME_absolute_get ();
-  pr->warn_task =
-      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
-                                    &warn_delay_task,
-                                    pr);
 #if INSANE_STATISTICS
   GNUNET_STATISTICS_update (GSF_stats,
                             gettext_noop ("# Datastore lookups initiated"), 1,
                             GNUNET_NO);
 #endif
-  pr->qe =
-      GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset++,
-                                &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_STATISTICS_update (GSF_stats,
-                            gettext_noop
-                            ("# Datastore lookups concluded (error queueing)"),
-                            1, GNUNET_NO);
-  GNUNET_SCHEDULER_cancel (pr->warn_task);
-  pr->warn_task = NULL;
-  pr->llc_cont = NULL;
-  if (NULL != cont)
-    cont (cont_cls, pr, pr->local_result);
+  start_local_query(pr,
+                    0 /* next_uid */,
+                    true /* random */);
 }
 
 
diff --git a/src/fs/gnunet-service-fs_put.c b/src/fs/gnunet-service-fs_put.c
index bb4cb4ecb..cd062bf2b 100644
--- a/src/fs/gnunet-service-fs_put.c
+++ b/src/fs/gnunet-service-fs_put.c
@@ -72,9 +72,14 @@ struct PutOperator
   uint64_t zero_anonymity_count_estimate;
 
   /**
-   * Current offset when iterating the database.
+   * Count of results received from the database.
    */
-  uint64_t current_offset;
+  uint64_t result_count;
+
+  /**
+   * Next UID to request when iterating the database.
+   */
+  uint64_t next_uid;
 };
 
 
@@ -177,37 +182,43 @@ delay_dht_put_task (void *cls)
  */
 static void
 process_dht_put_content (void *cls,
-                         const struct GNUNET_HashCode * key,
-                         size_t size,
+                         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)
+                         enum GNUNET_BLOCK_Type type,
+                         uint32_t priority,
+                         uint32_t anonymity,
+                         struct GNUNET_TIME_Absolute expiration,
+                         uint64_t uid)
 {
   struct PutOperator *po = cls;
 
   po->dht_qe = NULL;
   if (key == NULL)
   {
-    po->zero_anonymity_count_estimate = po->current_offset - 1;
-    po->current_offset = 0;
+    po->zero_anonymity_count_estimate = po->result_count;
+    po->result_count = 0;
+    po->next_uid = 0;
     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
     return;
   }
+  po->result_count++;
+  po->next_uid = uid + 1;
   po->zero_anonymity_count_estimate =
-      GNUNET_MAX (po->current_offset, po->zero_anonymity_count_estimate);
+    GNUNET_MAX (po->result_count, po->zero_anonymity_count_estimate);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Retrieved block `%s' of type %u for DHT PUT\n", GNUNET_h2s (key),
               type);
   po->dht_put = GNUNET_DHT_put (GSF_dht,
                                 key,
                                 DEFAULT_PUT_REPLICATION,
-                                GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
+                                GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                 type,
                                 size,
                                 data,
-                                expiration,
-                                &delay_dht_put_blocks, po);
+                                expiration,
+                                &delay_dht_put_blocks,
+                                po);
 }
 
 
@@ -223,10 +234,13 @@ gather_dht_put_blocks (void *cls)
 
   po->dht_task = NULL;
   po->dht_qe =
-      GNUNET_DATASTORE_get_zero_anonymity (GSF_dsh, po->current_offset++, 0,
+      GNUNET_DATASTORE_get_zero_anonymity (GSF_dsh,
+                                           po->next_uid,
+                                           0,
                                            UINT_MAX,
                                            po->dht_put_type,
-                                           &process_dht_put_content, po);
+                                           &process_dht_put_content,
+                                           po);
   if (NULL == po->dht_qe)
     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 }