datastore and fs fixes from Easter

14 files changed, 242 insertions, 132 deletions

diff --git a/src/fs/Makefile.am b/src/fs/Makefile.am
index f980f4206..20aa652ae 100644
--- a/src/fs/Makefile.am
+++ b/src/fs/Makefile.am
@@ -1,4 +1,3 @@
-
 INCLUDES = -I$(top_srcdir)/src/include
 
 if MINGW
@@ -173,8 +172,7 @@ check_SCRIPTS = \
  test_gnunet_fs_idx.py
 endif
 
-#if !DISABLE_TEST_RUN
-if 0 
+if !DISABLE_TEST_RUN
 TESTS = \
  test_fs_directory \
  test_fs_download \
diff --git a/src/fs/fs_download.c b/src/fs/fs_download.c
index 8192b8c1f..8eb2b4331 100644
--- a/src/fs/fs_download.c
+++ b/src/fs/fs_download.c
@@ -756,10 +756,12 @@ try_top_down_reconstruction (struct GNUNET_FS_DownloadContext *dc,
       child_block_size = GNUNET_FS_tree_compute_tree_size (drc->depth);
       GNUNET_assert (0 == (drc->offset - dr->offset) % child_block_size);
       chk_off = (drc->offset - dr->offset) / child_block_size;
-      GNUNET_assert (drc->state == BRS_INIT);
-      drc->state = BRS_CHK_SET;
-      drc->chk = chks[chk_off];
-      try_top_down_reconstruction (dc, drc);
+      if (drc->state == BRS_INIT)       
+        {
+          drc->state = BRS_CHK_SET;
+          drc->chk = chks[chk_off];
+          try_top_down_reconstruction (dc, drc);
+        }
       if (drc->state != BRS_DOWNLOAD_UP)
         up_done = GNUNET_NO; /* children not all done */
     } 
@@ -815,10 +817,11 @@ schedule_block_download (struct GNUNET_FS_DownloadContext *dc,
               dr->depth,
               GNUNET_h2s (&dr->chk.query));
 #endif
-  GNUNET_assert (GNUNET_NO ==
-                 GNUNET_CONTAINER_multihashmap_contains_value (dc->active,
-                                                               &dr->chk.query,
-                                                               dr));
+  if (GNUNET_NO !=
+      GNUNET_CONTAINER_multihashmap_contains_value (dc->active,
+                                                    &dr->chk.query,
+                                                    dr))
+    return; /* already active */
   GNUNET_CONTAINER_multihashmap_put (dc->active,
                                      &dr->chk.query,
                                      dr,
diff --git a/src/fs/fs_test_lib_data.conf b/src/fs/fs_test_lib_data.conf
index 68c5166b3..204bb90cf 100644
--- a/src/fs/fs_test_lib_data.conf
+++ b/src/fs/fs_test_lib_data.conf
@@ -43,7 +43,7 @@ HOSTNAME = localhost
 #TOTAL_QUOTA_OUT = 9321
 TOTAL_QUOTA_IN = 3932160
 TOTAL_QUOTA_OUT = 3932160
-DEBUG = YES
+#DEBUG = YES
 #PREFIX = valgrind --tool=memcheck --leak-check=yes
 #BINARY = /home/grothoff/bin/gnunet-service-core
 
@@ -53,8 +53,8 @@ HOSTNAME = localhost
 #OPTIONS = -L DEBUG
 CONTENT_CACHING = NO
 CONTENT_PUSHING = NO
-DEBUG = YES
-#PREFIX = valgrind --tool=memcheck --leak-check=yes 
+# DEBUG = YES
+# PREFIX = valgrind --tool=memcheck --leak-check=yes --trace-children=yes
 #BINARY = /home/grothoff/gn9/bin/gnunet-service-fs
 #PREFIX = xterm -e gdb -x cmd --args 
 
diff --git a/src/fs/gnunet-pseudonym.c b/src/fs/gnunet-pseudonym.c
index 769b4239d..68a760867 100644
--- a/src/fs/gnunet-pseudonym.c
+++ b/src/fs/gnunet-pseudonym.c
@@ -341,7 +341,7 @@ main (int argc, char *const *argv)
      0, &GNUNET_GETOPT_set_one, &no_remote_printing},
     {'r', "replication", "LEVEL",
      gettext_noop ("set the desired replication LEVEL"),
-     0, &GNUNET_GETOPT_set_uint, &bo.replication_level},
+     1, &GNUNET_GETOPT_set_uint, &bo.replication_level},
     {'R', "root", "ID",
      gettext_noop
      ("specify ID of the root of the namespace"),
diff --git a/src/fs/gnunet-service-fs_cp.c b/src/fs/gnunet-service-fs_cp.c
index 2522cbe7b..acad54501 100644
--- a/src/fs/gnunet-service-fs_cp.c
+++ b/src/fs/gnunet-service-fs_cp.c
@@ -704,9 +704,9 @@ copy_reply (void *cls,
 
 
 /**
- * Free the given client request.
+ * Free the given request.
  *
- * @param cls the client request to free
+ * @param cls the request to free
  * @param tc task context
  */ 
 static void
@@ -1182,6 +1182,7 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other,
                                     NULL, 0, /* replies_seen */
                                     &handle_p2p_reply,
                                     peerreq);
+  GNUNET_assert (NULL != pr);
   peerreq->pr = pr;
   GNUNET_break (GNUNET_OK ==
                 GNUNET_CONTAINER_multihashmap_put (cp->request_map,
@@ -1427,7 +1428,7 @@ cancel_pending_request (void *cls,
                         const GNUNET_HashCode *query,
                         void *value)
 {
-  struct PeerRequest *peerreq = cls;
+  struct PeerRequest *peerreq = value;
   struct GSF_PendingRequest *pr = peerreq->pr;
 
   GSF_pending_request_cancel_ (pr);
diff --git a/src/fs/gnunet-service-fs_indexing.c b/src/fs/gnunet-service-fs_indexing.c
index cc99d3962..dc6b82952 100644
--- a/src/fs/gnunet-service-fs_indexing.c
+++ b/src/fs/gnunet-service-fs_indexing.c
@@ -566,7 +566,7 @@ GNUNET_FS_handle_on_demand_block (const GNUNET_HashCode * key,
                                   uint32_t anonymity,
                                   struct GNUNET_TIME_Absolute
                                   expiration, uint64_t uid,
-                                  GNUNET_DATASTORE_Iterator cont,
+                                  GNUNET_DATASTORE_DatumProcessor cont,
                                   void *cont_cls)
 {
   const struct OnDemandBlock *odb;
diff --git a/src/fs/gnunet-service-fs_indexing.h b/src/fs/gnunet-service-fs_indexing.h
index 6a2c3d4a0..e1154830b 100644
--- a/src/fs/gnunet-service-fs_indexing.h
+++ b/src/fs/gnunet-service-fs_indexing.h
@@ -63,7 +63,7 @@ GNUNET_FS_handle_on_demand_block (const GNUNET_HashCode * key,
                                   uint32_t anonymity,
                                   struct GNUNET_TIME_Absolute
                                   expiration, uint64_t uid,
-                                  GNUNET_DATASTORE_Iterator cont,
+                                  GNUNET_DATASTORE_DatumProcessor cont,
                                   void *cont_cls);
 
 /**
diff --git a/src/fs/gnunet-service-fs_pe.c b/src/fs/gnunet-service-fs_pe.c
index 28036150f..4dc9de1b8 100644
--- a/src/fs/gnunet-service-fs_pe.c
+++ b/src/fs/gnunet-service-fs_pe.c
@@ -158,7 +158,7 @@ plan (struct PeerPlan *pp,
               rp->transmission_counter);
 #endif 
 
-
+  GNUNET_assert (rp->hn == NULL);
   if (GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission).rel_value == 0)
     rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap,
                                            rp,
diff --git a/src/fs/gnunet-service-fs_pr.c b/src/fs/gnunet-service-fs_pr.c
index 7406bed0f..c1074e8bf 100644
--- a/src/fs/gnunet-service-fs_pr.c
+++ b/src/fs/gnunet-service-fs_pr.c
@@ -100,6 +100,20 @@ struct GSF_PendingRequest
   GNUNET_PEER_Id sender_pid;
 
   /**
+   * 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.
+   */
+  uint64_t local_result_offset;
+
+  /**
+   * Unique ID of the first result from the local datastore;
+   * used to detect wrap-around of the offset.
+   */
+  uint64_t first_uid;
+
+  /**
    * Number of valid entries in the 'replies_seen' array.
    */
   unsigned int replies_seen_count;
@@ -113,7 +127,7 @@ struct GSF_PendingRequest
    * Mingle value we currently use for the bf.
    */
   uint32_t mingle;
-                            
+
 };
 
 
@@ -273,6 +287,8 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
               type);
 #endif 
   pr = GNUNET_malloc (sizeof (struct GSF_PendingRequest));
+  pr->local_result_offset = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                                      UINT64_MAX);                                                       
   pr->public_data.query = *query;
   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == type)
     {
@@ -535,7 +551,20 @@ clean_request (void *cls,
                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;
+      cont (pr->llc_cont_cls,
+            pr,
+            pr->local_result);
+    } 
   GSF_plan_notify_request_done_ (pr);
   GNUNET_free_non_null (pr->replies_seen);
   if (NULL != pr->bf)
@@ -560,6 +589,7 @@ clean_request (void *cls,
 void
 GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr)
 {
+  if (NULL == pr_map) return; /* already cleaned up! */
   GNUNET_assert (GNUNET_OK ==
                  GNUNET_CONTAINER_multihashmap_remove (pr_map,
                                                        &pr->public_data.query,
@@ -1023,13 +1053,22 @@ process_local_reply (void *cls,
   GNUNET_HashCode query;
   unsigned int old_rf;
   
+  pr->qe = NULL;
+  if (0 == pr->replies_seen_count)
+    {
+      pr->first_uid = uid;
+    }
+  else
+    {
+      if (uid == pr->first_uid)
+        key = NULL; /* all replies seen! */
+    }
   if (NULL == key)
     {
 #if DEBUG_FS
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                   "No further local responses available.\n");
 #endif
-      pr->qe = NULL;
       if (NULL != (cont = pr->llc_cont))
         {
           pr->llc_cont = NULL;
@@ -1041,9 +1080,10 @@ process_local_reply (void *cls,
     }
 #if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "New local response to `%s' of type %u.\n",
+              "Received reply for `%s' of type %d with UID %llu from datastore.\n",
               GNUNET_h2s (key),
-              type);
+              type,
+              (unsigned long long) uid);
 #endif
   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
     {
@@ -1061,8 +1101,22 @@ process_local_reply (void *cls,
                                             &process_local_reply,
                                             pr))
         {
-          if (pr->qe != NULL)
-            GNUNET_DATASTORE_iterate_get_next (GSF_dsh);
+          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
+                                             : 1 /* max queue size */,
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             &process_local_reply,
+                                             pr);
+          GNUNET_assert (NULL != pr->qe);
         }
       return;
     }
@@ -1085,7 +1139,22 @@ process_local_reply (void *cls,
                                -1, -1, 
                                GNUNET_TIME_UNIT_FOREVER_REL,
                                NULL, NULL);
-      GNUNET_DATASTORE_iterate_get_next (GSF_dsh);
+      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
+                                         : 1 /* max queue size */,
+                                         GNUNET_TIME_UNIT_FOREVER_REL,
+                                         &process_local_reply,
+                                         pr);
+      GNUNET_assert (NULL != pr->qe);
       return;
     }
   prq.type = type;
@@ -1097,12 +1166,16 @@ process_local_reply (void *cls,
     GSF_update_datastore_delay_ (pr->public_data.start_time);
   process_reply (&prq, key, pr);
   pr->local_result = prq.eval;
-  if (pr->qe == NULL)
+  if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
     {
-#if DEBUG_FS
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "Request cancelled, not asking datastore for more\n");
-#endif
+      if (NULL != (cont = pr->llc_cont))
+        {
+          pr->llc_cont = NULL;
+          cont (pr->llc_cont_cls,
+                pr,
+                pr->local_result);
+        }      
+      return;
     }
   if ( (0 == (GSF_PRO_PRIORITY_UNLIMITED & pr->public_data.options)) &&
        ( (GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
@@ -1116,8 +1189,6 @@ process_local_reply (void *cls,
                                 gettext_noop ("# processing result set cut short due to load"),
                                 1,
                                 GNUNET_NO);
-      GNUNET_DATASTORE_cancel (pr->qe);
-      pr->qe = NULL;
       if (NULL != (cont = pr->llc_cont))
         {
           pr->llc_cont = NULL;
@@ -1127,7 +1198,22 @@ process_local_reply (void *cls,
         }
       return;
     }
-  GNUNET_DATASTORE_iterate_get_next (GSF_dsh);
+  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
+                                     : 1 /* max queue size */,
+                                     GNUNET_TIME_UNIT_FOREVER_REL,
+                                     &process_local_reply,
+                                     pr);
+  GNUNET_assert (NULL != pr->qe);
 }
 
 
@@ -1147,20 +1233,21 @@ 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 = GNUNET_DATASTORE_iterate_key (GSF_dsh,
-                                         &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
-                                         : 1 /* max queue size */,
-                                         GNUNET_TIME_UNIT_FOREVER_REL,
-                                         &process_local_reply,
-                                         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
+                                     : 1 /* max queue size */,
+                                     GNUNET_TIME_UNIT_FOREVER_REL,
+                                     &process_local_reply,
+                                     pr);
 }
 
 
diff --git a/src/fs/gnunet-service-fs_put.c b/src/fs/gnunet-service-fs_put.c
index 121a90bcd..b15207ce8 100644
--- a/src/fs/gnunet-service-fs_put.c
+++ b/src/fs/gnunet-service-fs_put.c
@@ -35,25 +35,50 @@
 
 
 /**
- * Request to datastore for DHT PUTs (or NULL).
+ * Context for each zero-anonymity iterator.
  */
-static struct GNUNET_DATASTORE_QueueEntry *dht_qe;
+struct PutOperator
+{
 
-/**
- * Type we will request for the next DHT PUT round from the datastore.
- */
-static enum GNUNET_BLOCK_Type dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
+  /**
+   * Request to datastore for DHT PUTs (or NULL).
+   */
+  struct GNUNET_DATASTORE_QueueEntry *dht_qe;
+
+  /**
+   * Type we request from the datastore.
+   */
+  enum GNUNET_BLOCK_Type dht_put_type;
+
+  /**
+   * ID of task that collects blocks for DHT PUTs.
+   */
+  GNUNET_SCHEDULER_TaskIdentifier dht_task;
+  
+  /**
+   * How many entires with zero anonymity of our type do we currently
+   * estimate to have in the database?
+   */
+  uint64_t zero_anonymity_count_estimate;
+
+  /**
+   * Current offset when iterating the database.
+   */
+  uint64_t current_offset;
+};
 
-/**
- * ID of task that collects blocks for DHT PUTs.
- */
-static GNUNET_SCHEDULER_TaskIdentifier dht_task;
 
 /**
- * How many entires with zero anonymity do we currently estimate
- * to have in the database?
+ * ANY-terminated list of our operators (one per type
+ * of block that we're putting into the DHT).
  */
-static unsigned int zero_anonymity_count_estimate;
+static struct PutOperator operators[] = 
+  {
+    { NULL, GNUNET_BLOCK_TYPE_FS_KBLOCK, 0, 0, 0 },
+    { NULL, GNUNET_BLOCK_TYPE_FS_SBLOCK, 0, 0, 0 },
+    { NULL, GNUNET_BLOCK_TYPE_FS_NBLOCK, 0, 0, 0 },
+    { NULL, GNUNET_BLOCK_TYPE_ANY, 0, 0, 0 }
+  };
 
 
 /**
@@ -67,26 +92,26 @@ gather_dht_put_blocks (void *cls,
                        const struct GNUNET_SCHEDULER_TaskContext *tc);
 
 
-
 /**
- * If the DHT PUT gathering task is not currently running, consider
- * (re)scheduling it with the appropriate delay.
+ * Task that is run periodically to obtain blocks for DHT PUTs.
+ * 
+ * @param cls type of blocks to gather
+ * @param tc scheduler context (unused)
  */
 static void
-consider_dht_put_gathering (void *cls)
+delay_dht_put_blocks (void *cls,
+                      const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
+  struct PutOperator *po = cls;
   struct GNUNET_TIME_Relative delay;
 
-  if (GSF_dsh == NULL)
-    return;
-  if (dht_qe != NULL)
+  po->dht_task = GNUNET_SCHEDULER_NO_TASK;
+  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
     return;
-  if (dht_task != GNUNET_SCHEDULER_NO_TASK)
-    return;
-  if (zero_anonymity_count_estimate > 0)
+  if (po->zero_anonymity_count_estimate > 0)
     {
       delay = GNUNET_TIME_relative_divide (GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY,
-                                           zero_anonymity_count_estimate);
+                                           po->zero_anonymity_count_estimate);
       delay = GNUNET_TIME_relative_min (delay,
                                         MAX_DHT_PUT_FREQ);
     }
@@ -96,20 +121,9 @@ consider_dht_put_gathering (void *cls)
          (hopefully) appear */
       delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
     }
-  dht_task = GNUNET_SCHEDULER_add_delayed (delay,
-                                           &gather_dht_put_blocks,
-                                           cls);
-}
-
-
-/**
- * Function called upon completion of the DHT PUT operation.
- */
-static void
-dht_put_continuation (void *cls,
-                      const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-  GNUNET_DATASTORE_iterate_get_next (GSF_dsh);
+  po->dht_task = GNUNET_SCHEDULER_add_delayed (delay,
+                                               &gather_dht_put_blocks,
+                                               po);
 }
 
 
@@ -138,31 +152,19 @@ process_dht_put_content (void *cls,
                          struct GNUNET_TIME_Absolute
                          expiration, uint64_t uid)
 { 
-  static unsigned int counter;
-  static GNUNET_HashCode last_vhash;
-  static GNUNET_HashCode vhash;
+  struct PutOperator *po = cls;
 
+  po->dht_qe = NULL;
   if (key == NULL)
     {
-      dht_qe = NULL;
-      consider_dht_put_gathering (cls);
+      po->zero_anonymity_count_estimate = po->current_offset - 1;
+      po->current_offset = 0;
+      po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_blocks,
+                                               po);
       return;
     }
-  /* slightly funky code to estimate the total number of values with zero
-     anonymity from the maximum observed length of a monotonically increasing 
-     sequence of hashes over the contents */
-  GNUNET_CRYPTO_hash (data, size, &vhash);
-  if (GNUNET_CRYPTO_hash_cmp (&vhash, &last_vhash) <= 0)
-    {
-      if (zero_anonymity_count_estimate > 0)
-        zero_anonymity_count_estimate /= 2;
-      counter = 0;
-    }
-  last_vhash = vhash;
-  if (counter < 31)
-    counter++;
-  if (zero_anonymity_count_estimate < (1 << counter))
-    zero_anonymity_count_estimate = (1 << counter);
+  po->zero_anonymity_count_estimate = GNUNET_MAX (po->current_offset,
+                                                  po->zero_anonymity_count_estimate);
 #if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Retrieved block `%s' of type %u for DHT PUT\n",
@@ -178,8 +180,8 @@ process_dht_put_content (void *cls,
                   data,
                   expiration,
                   GNUNET_TIME_UNIT_FOREVER_REL,
-                  &dht_put_continuation,
-                  cls);
+                  &delay_dht_put_blocks,
+                  po);
 }
 
 
@@ -193,17 +195,20 @@ static void
 gather_dht_put_blocks (void *cls,
                        const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
-  dht_task = GNUNET_SCHEDULER_NO_TASK;
-  if (GSF_dsh == NULL)
+  struct PutOperator *po = cls;
+
+  po->dht_task = GNUNET_SCHEDULER_NO_TASK;
+  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
     return;
-  if (dht_put_type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
-    dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
-  dht_qe = GNUNET_DATASTORE_iterate_zero_anonymity (GSF_dsh, 
+  po->dht_qe = GNUNET_DATASTORE_get_zero_anonymity (GSF_dsh, 
+                                                    po->current_offset++,
                                                     0, UINT_MAX,
                                                     GNUNET_TIME_UNIT_FOREVER_REL,
-                                                    dht_put_type++,
-                                                    &process_dht_put_content, NULL);
-  GNUNET_assert (dht_qe != NULL);
+                                                    po->dht_put_type,
+                                                    &process_dht_put_content, po);
+  if (NULL == po->dht_qe)
+    po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_blocks,
+                                             po);
 }
 
 
@@ -213,7 +218,14 @@ gather_dht_put_blocks (void *cls,
 void
 GSF_put_init_ ()
 {
-  dht_task = GNUNET_SCHEDULER_add_now (&gather_dht_put_blocks, NULL);
+  unsigned int i;
+
+  i = 0;
+  while (operators[i].dht_put_type != GNUNET_BLOCK_TYPE_ANY)
+    {
+      operators[i].dht_task = GNUNET_SCHEDULER_add_now (&gather_dht_put_blocks, &operators[i]);
+      i++;
+    }
 }
 
 
@@ -223,15 +235,23 @@ GSF_put_init_ ()
 void
 GSF_put_done_ ()
 {
-  if (GNUNET_SCHEDULER_NO_TASK != dht_task)
-    {
-      GNUNET_SCHEDULER_cancel (dht_task);
-      dht_task = GNUNET_SCHEDULER_NO_TASK;
-    }
-  if (NULL != dht_qe)
+  struct PutOperator *po;
+  unsigned int i;
+
+  i = 0;
+  while ((po = &operators[i])->dht_put_type != GNUNET_BLOCK_TYPE_ANY)
     {
-      GNUNET_DATASTORE_cancel (dht_qe);
-      dht_qe = NULL;
+      if (GNUNET_SCHEDULER_NO_TASK != po->dht_task)
+        {
+          GNUNET_SCHEDULER_cancel (po->dht_task);
+          po->dht_task = GNUNET_SCHEDULER_NO_TASK;
+        }
+      if (NULL != po->dht_qe)
+        {
+          GNUNET_DATASTORE_cancel (po->dht_qe);
+          po->dht_qe = NULL;
+        }
+      i++;
     }
 }
 
diff --git a/src/fs/test_fs_download_data.conf b/src/fs/test_fs_download_data.conf
index 0a7eb311a..6bbae9dc9 100644
--- a/src/fs/test_fs_download_data.conf
+++ b/src/fs/test_fs_download_data.conf
@@ -36,7 +36,8 @@ HOSTNAME = localhost
 [fs]
 PORT = 42471
 HOSTNAME = localhost
-ACTIVEMIGRATION = NO
+CONTENT_CACHING = NO
+CONTENT_PUSHING = NO
 # DEBUG = YES
 #PREFIX = valgrind --tool=memcheck --leak-check=yes
 #BINARY = /home/grothoff/bin/gnunet-service-fs
diff --git a/src/fs/test_gnunet_fs_idx.py.in b/src/fs/test_gnunet_fs_idx.py.in
index 3bb3681c6..c97ffd883 100755
--- a/src/fs/test_gnunet_fs_idx.py.in
+++ b/src/fs/test_gnunet_fs_idx.py.in
@@ -31,7 +31,7 @@ try:
   pub.expect ("URI is `gnunet://fs/chk/PC0M19QMQC0BPSHR6BGA228PP6INER1D610MGEMOMEM87222FN8HVUO7PQGO0O9HD2GVLHF2N5IDHEQUNK6LKE428FPO96SKQEA486O.PG7K85JGQ6N599MD5HEP3CHEVFPKQD9JB6NPSLVA3T1SKDS66CFI499VS6MGQ88B0QUAVT1282TCRD4GGFVUKDLGI8F0SPIANA3J2LG.35147'.\r")
   pub.expect (pexpect.EOF)
 
-  down = pexpect.spawn ('gnunet-download -c test_gnunet_fs_idx_data.conf -o \"COPYING\" gnunet://fs/chk/PC0M19QMQC0BPSHR6BGA228PP6INER1D610MGEMOMEM87222FN8HVUO7PQGO0O9HD2GVLHF2N5IDHEQUNK6LKE428FPO96SKQEA486O.PG7K85JGQ6N599MD5HEP3CHEVFPKQD9JB6NPSLVA3T1SKDS66CFI499VS6MGQ88B0QUAVT1282TCRD4GGFVUKDLGI8F0SPIANA3J2LG.35147')
+  down = pexpect.spawn ('gnunet-download -c test_gnunet_fs_idx_data.conf -o COPYING gnunet://fs/chk/PC0M19QMQC0BPSHR6BGA228PP6INER1D610MGEMOMEM87222FN8HVUO7PQGO0O9HD2GVLHF2N5IDHEQUNK6LKE428FPO96SKQEA486O.PG7K85JGQ6N599MD5HEP3CHEVFPKQD9JB6NPSLVA3T1SKDS66CFI499VS6MGQ88B0QUAVT1282TCRD4GGFVUKDLGI8F0SPIANA3J2LG.35147')
   down.expect (re.compile ("Downloading `COPYING\' done \(.*\).\r"));
   down.expect (pexpect.EOF);
   os.system ('rm COPYING');
diff --git a/src/fs/test_gnunet_fs_ns_data.conf b/src/fs/test_gnunet_fs_ns_data.conf
index 65bac0a15..2086cd0fd 100644
--- a/src/fs/test_gnunet_fs_ns_data.conf
+++ b/src/fs/test_gnunet_fs_ns_data.conf
@@ -36,7 +36,7 @@ HOSTNAME = localhost
 [fs]
 PORT = 47471
 HOSTNAME = localhost
-#DEBUG = YES
+DEBUG = YES
 #PREFIX = valgrind --tool=memcheck --leak-check=yes
 #BINARY = /home/grothoff/bin/gnunet-service-fs
 
diff --git a/src/fs/test_gnunet_service_fs_migration_data.conf b/src/fs/test_gnunet_service_fs_migration_data.conf
index a72a98e97..3ab61d76c 100644
--- a/src/fs/test_gnunet_service_fs_migration_data.conf
+++ b/src/fs/test_gnunet_service_fs_migration_data.conf
@@ -53,7 +53,7 @@ HOSTNAME = localhost
 ACTIVEMIGRATION = YES
 CONTENT_CACHING = YES
 CONTENT_PUSHING = YES
-DEBUG = YES
+#DEBUG = YES
 #PREFIX = valgrind --tool=memcheck --leak-check=yes 
 #PREFIX = xterm -e gdb -x cmd --args