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