adding configure code for --enable-benchmarks, --enable-expensive-tests, some clean up
[oweals/gnunet.git] / src / datastore / datastore_api.c
index dde45f24ffeb4caee9918ce3aef8ba6bcc739b36..a3196530eab54a1f65c0dc9bf65e8370b7719f7f 100644 (file)
@@ -63,14 +63,14 @@ struct StatusContext
 struct ResultContext
 {
   /**
-   * Iterator to call with the result.
+   * Function to call with the result.
    */
-  GNUNET_DATASTORE_Iterator iter;
+  GNUNET_DATASTORE_DatumProcessor proc;
 
   /**
-   * Closure for iter.
+   * Closure for proc.
    */
-  void *iter_cls;
+  void *proc_cls;
 
 };
 
@@ -166,7 +166,7 @@ struct GNUNET_DATASTORE_QueueEntry
    * Note that the overall struct should end at a 
    * multiple of 64 bits.
    */
-  int32_t was_transmitted;
+  int was_transmitted;
   
 };
 
@@ -181,7 +181,6 @@ struct GNUNET_DATASTORE_Handle
    */
   const struct GNUNET_CONFIGURATION_Handle *cfg;
 
-
   /**
    * Current connection to the datastore service.
    */
@@ -235,6 +234,11 @@ struct GNUNET_DATASTORE_Handle
    */
   int in_receive;
 
+  /**
+   * We should ignore the next message(s) from the service.
+   */
+  unsigned int skip_next_messages;
+
 };
 
 
@@ -311,6 +315,11 @@ GNUNET_DATASTORE_disconnect (struct GNUNET_DATASTORE_Handle *h,
 {
   struct GNUNET_DATASTORE_QueueEntry *qe;
 
+  if (NULL != h->th)
+    {
+      GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
+      h->th = NULL;
+    }
   if (h->client != NULL)
     {
       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
@@ -324,7 +333,7 @@ GNUNET_DATASTORE_disconnect (struct GNUNET_DATASTORE_Handle *h,
   while (NULL != (qe = h->queue_head))
     {
       GNUNET_assert (NULL != qe->response_proc);
-      qe->response_proc (qe, NULL);
+      qe->response_proc (h, NULL);
     }
   if (GNUNET_YES == drop) 
     {
@@ -340,11 +349,13 @@ GNUNET_DATASTORE_disconnect (struct GNUNET_DATASTORE_Handle *h,
                                                   h))
            return;
          GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
+         h->client = NULL;
        }
       GNUNET_break (0);
     }
   GNUNET_STATISTICS_destroy (h->stats,
                             GNUNET_NO);
+  h->stats = NULL;
   GNUNET_free (h);
 }
 
@@ -366,8 +377,8 @@ timeout_queue_entry (void *cls,
                            1,
                            GNUNET_NO);
   qe->task = GNUNET_SCHEDULER_NO_TASK;
-  GNUNET_assert (qe->was_transmitted == GNUNET_NO);
-  qe->response_proc (qe, NULL);
+  GNUNET_assert (qe->was_transmitted == GNUNET_NO); 
+  qe->response_proc (qe->h, NULL);
 }
 
 
@@ -383,7 +394,7 @@ timeout_queue_entry (void *cls,
  * @param timeout timeout for the operation
  * @param response_proc function to call with replies (can be NULL)
  * @param qc client context (NOT a closure for response_proc)
- * @return NULL if the queue is full (and this entry was dropped)
+ * @return NULL if the queue is full 
  */
 static struct GNUNET_DATASTORE_QueueEntry *
 make_queue_entry (struct GNUNET_DATASTORE_Handle *h,
@@ -407,6 +418,14 @@ make_queue_entry (struct GNUNET_DATASTORE_Handle *h,
       c++;
       pos = pos->next;
     }
+  if (c >= max_queue_size)
+    {
+      GNUNET_STATISTICS_update (h->stats,
+                               gettext_noop ("# queue overflows"),
+                               1,
+                               GNUNET_NO);
+      return NULL;
+    }
   ret = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_QueueEntry) + msize);
   ret->h = h;
   ret->response_proc = response_proc;
@@ -440,25 +459,29 @@ make_queue_entry (struct GNUNET_DATASTORE_Handle *h,
                                     pos,
                                     ret);
   h->queue_size++;
-  if (c > max_queue_size)
-    {
-      GNUNET_STATISTICS_update (h->stats,
-                               gettext_noop ("# queue overflows"),
-                               1,
-                               GNUNET_NO);
-      response_proc (ret, NULL);
-      return NULL;
-    }
   ret->task = GNUNET_SCHEDULER_add_delayed (timeout,
                                            &timeout_queue_entry,
                                            ret);
   pos = ret->next;
   while (pos != NULL) 
     {
-      if (pos->max_queue < h->queue_size)
+      if ( (pos->max_queue < h->queue_size) &&
+          (pos->was_transmitted == GNUNET_NO) )
        {
          GNUNET_assert (pos->response_proc != NULL);
-         pos->response_proc (pos, NULL);
+         /* move 'pos' element to head so that it will be 
+            killed on 'NULL' call below */
+         GNUNET_CONTAINER_DLL_remove (h->queue_head,
+                                      h->queue_tail,
+                                      pos);
+         GNUNET_CONTAINER_DLL_insert (h->queue_head,
+                                      h->queue_tail,
+                                      pos);
+         GNUNET_STATISTICS_update (h->stats,
+                                   gettext_noop ("# Requests dropped from datastore queue"),
+                                   1,
+                                   GNUNET_NO);
+         pos->response_proc (h, NULL);
          break;
        }
       pos = pos->next;
@@ -539,6 +562,7 @@ do_disconnect (struct GNUNET_DATASTORE_Handle *h)
                            GNUNET_NO);
 #endif
   GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
+  h->skip_next_messages = 0;
   h->client = NULL;
   h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->retry_time,
                                                    &try_reconnect,
@@ -546,6 +570,37 @@ do_disconnect (struct GNUNET_DATASTORE_Handle *h)
 }
 
 
+/**
+ * Function called whenever we receive a message from
+ * the service.  Calls the appropriate handler.
+ *
+ * @param cls the 'struct GNUNET_DATASTORE_Handle'
+ * @param msg the received message
+ */
+static void 
+receive_cb (void *cls,
+           const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_DATASTORE_Handle *h = cls;
+  struct GNUNET_DATASTORE_QueueEntry *qe;
+
+  h->in_receive = GNUNET_NO;
+  if (h->skip_next_messages > 0)
+    {
+      h->skip_next_messages--;
+      process_queue (h);
+      return;
+   } 
+  if (NULL == (qe = h->queue_head))
+    {
+      GNUNET_break (0);
+      process_queue (h);
+      return; 
+    }
+  qe->response_proc (h, msg);
+}
+
+
 /**
  * Transmit request from queue to datastore service.
  *
@@ -591,10 +646,11 @@ transmit_request (void *cls,
   qe->was_transmitted = GNUNET_YES;
   GNUNET_SCHEDULER_cancel (qe->task);
   qe->task = GNUNET_SCHEDULER_NO_TASK;
+  GNUNET_assert (GNUNET_NO == h->in_receive);
   h->in_receive = GNUNET_YES;
   GNUNET_CLIENT_receive (h->client,
-                        qe->response_proc,
-                        qe,
+                        &receive_cb,
+                        h,
                         GNUNET_TIME_absolute_get_remaining (qe->timeout));
   GNUNET_STATISTICS_update (h->stats,
                            gettext_noop ("# bytes sent to datastore"),
@@ -647,6 +703,11 @@ process_queue (struct GNUNET_DATASTORE_Handle *h)
 #endif
       return; /* waiting for reconnect */
     }
+  if (GNUNET_YES == h->in_receive)
+    {
+      /* wait for response to previous query */
+      return; 
+    }
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Queueing %u byte request to DATASTORE\n",
@@ -658,6 +719,8 @@ process_queue (struct GNUNET_DATASTORE_Handle *h)
                                               GNUNET_YES,
                                               &transmit_request,
                                               h);
+  GNUNET_assert (GNUNET_NO == h->in_receive);
+  GNUNET_break (NULL != h->th);
 }
 
 
@@ -675,6 +738,13 @@ drop_status_cont (void *cls, int32_t result, const char *emsg)
 }
 
 
+/**
+ * Free a queue entry.  Removes the given entry from the
+ * queue and releases associated resources.  Does NOT
+ * call the callback.
+ * 
+ * @param qe entry to free.
+ */
 static void
 free_queue_entry (struct GNUNET_DATASTORE_QueueEntry *qe)
 {
@@ -689,9 +759,11 @@ free_queue_entry (struct GNUNET_DATASTORE_QueueEntry *qe)
       qe->task = GNUNET_SCHEDULER_NO_TASK;
     }
   h->queue_size--;
+  qe->was_transmitted = GNUNET_SYSERR; /* use-after-free warning */
   GNUNET_free (qe);
 }
 
+
 /**
  * Type of a function to call when we receive a message
  * from the service.
@@ -704,18 +776,24 @@ process_status_message (void *cls,
                        const struct
                        GNUNET_MessageHeader * msg)
 {
-  struct GNUNET_DATASTORE_QueueEntry *qe = cls;
-  struct GNUNET_DATASTORE_Handle *h = qe->h;
-  struct StatusContext rc = qe->qc.sc;
+  struct GNUNET_DATASTORE_Handle *h = cls;
+  struct GNUNET_DATASTORE_QueueEntry *qe;
+  struct StatusContext rc;
   const struct StatusMessage *sm;
   const char *emsg;
   int32_t status;
   int was_transmitted;
 
-  h->in_receive = GNUNET_NO;
-  was_transmitted = qe->was_transmitted;
+  if (NULL == (qe = h->queue_head))
+    {
+      GNUNET_break (0);
+      do_disconnect (h);
+      return;
+    }
+  rc = qe->qc.sc;
   if (msg == NULL)
     {      
+      was_transmitted = qe->was_transmitted;
       free_queue_entry (qe);
       if (NULL == h->client)
        return; /* forced disconnect */
@@ -725,10 +803,11 @@ process_status_message (void *cls,
                 _("Failed to receive status response from database."));
       if (was_transmitted == GNUNET_YES)
        do_disconnect (h);
+      else
+       process_queue (h);
       return;
     }
   GNUNET_assert (GNUNET_YES == qe->was_transmitted);
-  GNUNET_assert (h->queue_head == qe);
   free_queue_entry (qe);
   if ( (ntohs(msg->size) < sizeof(struct StatusMessage)) ||
        (ntohs(msg->type) != GNUNET_MESSAGE_TYPE_DATASTORE_STATUS) ) 
@@ -829,9 +908,10 @@ GNUNET_DATASTORE_put (struct GNUNET_DATASTORE_Handle *h,
 
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-             "Asked to put %u bytes of data under key `%s'\n",
+             "Asked to put %u bytes of data under key `%s' for %llu ms\n",
              size,
-             GNUNET_h2s (key));
+             GNUNET_h2s (key),
+             GNUNET_TIME_absolute_get_remaining (expiration).rel_value);
 #endif
   msize = sizeof(struct DataMessage) + size;
   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
@@ -860,6 +940,8 @@ GNUNET_DATASTORE_put (struct GNUNET_DATASTORE_Handle *h,
   dm->type = htonl(type);
   dm->priority = htonl(priority);
   dm->anonymity = htonl(anonymity);
+  dm->replication = htonl (replication);
+  dm->reserved = htonl (0);
   dm->uid = GNUNET_htonll(0);
   dm->expiration = GNUNET_TIME_absolute_hton(expiration);
   dm->key = *key;
@@ -1097,7 +1179,7 @@ GNUNET_DATASTORE_update (struct GNUNET_DATASTORE_Handle *h,
 struct GNUNET_DATASTORE_QueueEntry *
 GNUNET_DATASTORE_remove (struct GNUNET_DATASTORE_Handle *h,
                          const GNUNET_HashCode *key,
-                         size_t size, 
+                        size_t size,
                         const void *data,
                         unsigned int queue_priority,
                         unsigned int max_queue_size,
@@ -1163,51 +1245,47 @@ GNUNET_DATASTORE_remove (struct GNUNET_DATASTORE_Handle *h,
  */
 static void 
 process_result_message (void *cls,
-                       const struct GNUNET_MessageHeader * msg)
+                       const struct GNUNET_MessageHeader *msg)
 {
-  struct GNUNET_DATASTORE_QueueEntry *qe = cls;
-  struct GNUNET_DATASTORE_Handle *h = qe->h;
-  struct ResultContext rc = qe->qc.rc;
+  struct GNUNET_DATASTORE_Handle *h = cls;
+  struct GNUNET_DATASTORE_QueueEntry *qe;
+  struct ResultContext rc;
   const struct DataMessage *dm;
-  int was_transmitted;
 
-  h->in_receive = GNUNET_NO;
   if (msg == NULL)
-   {
-      was_transmitted = qe->was_transmitted;
-      free_queue_entry (qe);
-      if (was_transmitted == GNUNET_YES)
+    {
+      qe = h->queue_head;
+      GNUNET_assert (NULL != qe);
+      if (qe->was_transmitted == GNUNET_YES)
        {
+         rc = qe->qc.rc;
          GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                      _("Failed to receive response from database.\n"));
          do_disconnect (h);
+         free_queue_entry (qe);
+         if (rc.proc != NULL)
+           rc.proc (rc.proc_cls,
+                    NULL, 0, NULL, 0, 0, 0, 
+                    GNUNET_TIME_UNIT_ZERO_ABS, 0);    
        }
       else
-       {
-#if DEBUG_DATASTORE
-         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                     "Request dropped due to finite datastore queue length.\n");
-#endif
-       }
-      if (rc.iter != NULL)
-       rc.iter (rc.iter_cls,
-                NULL, 0, NULL, 0, 0, 0, 
-                GNUNET_TIME_UNIT_ZERO_ABS, 0); 
+       process_queue (h);
       return;
     }
-  GNUNET_assert (GNUNET_YES == qe->was_transmitted);
-  GNUNET_assert (h->queue_head == qe);
   if (ntohs(msg->type) == GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END) 
     {
       GNUNET_break (ntohs(msg->size) == sizeof(struct GNUNET_MessageHeader));
+      qe = h->queue_head;
+      rc = qe->qc.rc;
+      GNUNET_assert (GNUNET_YES == qe->was_transmitted);
       free_queue_entry (qe);
 #if DEBUG_DATASTORE
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Received end of result set, new queue size is %u\n",
                  h->queue_size);
 #endif
-      if (rc.iter != NULL)
-       rc.iter (rc.iter_cls,
+      if (rc.proc != NULL)
+       rc.proc (rc.proc_cls,
                 NULL, 0, NULL, 0, 0, 0, 
                 GNUNET_TIME_UNIT_ZERO_ABS, 0); 
       h->retry_time.rel_value = 0;
@@ -1215,6 +1293,9 @@ process_result_message (void *cls,
       process_queue (h);
       return;
     }
+  qe = h->queue_head;
+  rc = qe->qc.rc;
+  GNUNET_assert (GNUNET_YES == qe->was_transmitted);
   if ( (ntohs(msg->size) < sizeof(struct DataMessage)) ||
        (ntohs(msg->type) != GNUNET_MESSAGE_TYPE_DATASTORE_DATA) ||
        (ntohs(msg->size) != sizeof(struct DataMessage) + ntohl (((const struct DataMessage*)msg)->size)) )
@@ -1223,37 +1304,16 @@ process_result_message (void *cls,
       free_queue_entry (qe);
       h->retry_time = GNUNET_TIME_UNIT_ZERO;
       do_disconnect (h);
-      if (rc.iter != NULL)
-       rc.iter (rc.iter_cls,
+      if (rc.proc != NULL)
+       rc.proc (rc.proc_cls,
                 NULL, 0, NULL, 0, 0, 0, 
-                GNUNET_TIME_UNIT_ZERO_ABS, 0); 
+                GNUNET_TIME_UNIT_ZERO_ABS, 0);
       return;
     }
   GNUNET_STATISTICS_update (h->stats,
                            gettext_noop ("# Results received"),
                            1,
                            GNUNET_NO);
-  if (rc.iter == NULL)
-    {
-      h->result_count++;
-      GNUNET_STATISTICS_update (h->stats,
-                               gettext_noop ("# Excess results received"),
-                               1,
-                               GNUNET_NO);
-      if (h->result_count > MAX_EXCESS_RESULTS)
-       {
-         free_queue_entry (qe);
-         GNUNET_STATISTICS_update (h->stats,
-                                   gettext_noop ("# Forced database connection resets"),
-                                   1,
-                                   GNUNET_NO);
-         h->retry_time = GNUNET_TIME_UNIT_ZERO;
-         do_disconnect (h);      
-         return;
-       }
-      GNUNET_DATASTORE_iterate_get_next (h);
-      return;
-    }
   dm = (const struct DataMessage*) msg;
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1263,16 +1323,19 @@ process_result_message (void *cls,
              ntohl(dm->size),
              GNUNET_h2s(&dm->key));
 #endif
+  free_queue_entry (qe);
   h->retry_time.rel_value = 0;
-  rc.iter (rc.iter_cls,
-          &dm->key,
-          ntohl(dm->size),
-          &dm[1],
-          ntohl(dm->type),
-          ntohl(dm->priority),
-          ntohl(dm->anonymity),
-          GNUNET_TIME_absolute_ntoh(dm->expiration),   
-          GNUNET_ntohll(dm->uid));
+  process_queue (h);
+  if (rc.proc != NULL)
+    rc.proc (rc.proc_cls,
+            &dm->key,
+            ntohl(dm->size),
+            &dm[1],
+            ntohl(dm->type),
+            ntohl(dm->priority),
+            ntohl(dm->anonymity),
+            GNUNET_TIME_absolute_ntoh(dm->expiration), 
+            GNUNET_ntohll(dm->uid));
 }
 
 
@@ -1288,33 +1351,33 @@ process_result_message (void *cls,
  * @param max_queue_size at what queue size should this request be dropped
  *        (if other requests of higher priority are in the queue)
  * @param timeout how long to wait at most for a response
- * @param iter function to call on a random value; it
+ * @param proc function to call on a random value; it
  *        will be called once with a value (if available)
  *        and always once with a value of NULL.
- * @param iter_cls closure for iter
+ * @param proc_cls closure for proc
  * @return NULL if the entry was not queued, otherwise a handle that can be used to
- *         cancel; note that even if NULL is returned, the callback will be invoked
- *         (or rather, will already have been invoked)
+ *         cancel
  */
 struct GNUNET_DATASTORE_QueueEntry *
 GNUNET_DATASTORE_get_for_replication (struct GNUNET_DATASTORE_Handle *h,
                                      unsigned int queue_priority,
                                      unsigned int max_queue_size,
                                      struct GNUNET_TIME_Relative timeout,
-                                     GNUNET_DATASTORE_Iterator iter
-                                     void *iter_cls)
+                                     GNUNET_DATASTORE_DatumProcessor proc
+                                     void *proc_cls)
 {
   struct GNUNET_DATASTORE_QueueEntry *qe;
   struct GNUNET_MessageHeader *m;
   union QueueContext qc;
 
+  GNUNET_assert (NULL != proc);
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-             "Asked to get random entry in %llu ms\n",
+             "Asked to get replication entry in %llu ms\n",
              (unsigned long long) timeout.rel_value);
 #endif
-  qc.rc.iter = iter;
-  qc.rc.iter_cls = iter_cls;
+  qc.rc.proc = proc;
+  qc.rc.proc_cls = proc_cls;
   qe = make_queue_entry (h, sizeof(struct GNUNET_MessageHeader),
                         queue_priority, max_queue_size, timeout,
                         &process_result_message, &qc);
@@ -1322,16 +1385,16 @@ GNUNET_DATASTORE_get_for_replication (struct GNUNET_DATASTORE_Handle *h,
     {
 #if DEBUG_DATASTORE
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                 "Could not create queue entry for GET RANDOM\n");
+                 "Could not create queue entry for GET REPLICATION\n");
 #endif
       return NULL;    
     }
   GNUNET_STATISTICS_update (h->stats,
-                           gettext_noop ("# GET RANDOM requests executed"),
+                           gettext_noop ("# GET REPLICATION requests executed"),
                            1,
                            GNUNET_NO);
   m = (struct GNUNET_MessageHeader*) &qe[1];
-  m->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_RANDOM);
+  m->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_REPLICATION);
   m->size = htons(sizeof (struct GNUNET_MessageHeader));
   process_queue (h);
   return qe;
@@ -1339,42 +1402,50 @@ GNUNET_DATASTORE_get_for_replication (struct GNUNET_DATASTORE_Handle *h,
 
 
 /**
- * Get a zero-anonymity value from the datastore.
+ * Get a single zero-anonymity value from the datastore.
  *
  * @param h handle to the datastore
+ * @param offset offset of the result (mod #num-results); set to
+ *               a random 64-bit value initially; then increment by
+ *               one each time; detect that all results have been found by uid
+ *               being again the first uid ever returned.
  * @param queue_priority ranking of this request in the priority queue
  * @param max_queue_size at what queue size should this request be dropped
  *        (if other requests of higher priority are in the queue)
  * @param timeout how long to wait at most for a response
- * @param type allowed type for the operation
- * @param iter function to call on a random value; it
+ * @param type allowed type for the operation (never zero)
+ * @param proc function to call on a random value; it
  *        will be called once with a value (if available)
- *        and always once with a value of NULL.
- * @param iter_cls closure for iter
+ *        or with NULL if none value exists.
+ * @param proc_cls closure for proc
  * @return NULL if the entry was not queued, otherwise a handle that can be used to
- *         cancel; note that even if NULL is returned, the callback will be invoked
- *         (or rather, will already have been invoked)
+ *         cancel
  */
 struct GNUNET_DATASTORE_QueueEntry *
-GNUNET_DATASTORE_iterate_zero_anonymity (struct GNUNET_DATASTORE_Handle *h,
-                                        unsigned int queue_priority,
-                                        unsigned int max_queue_size,
-                                        struct GNUNET_TIME_Relative timeout,
-                                        enum GNUNET_BLOCK_Type type,
-                                        GNUNET_DATASTORE_Iterator iter, 
-                                        void *iter_cls)
+GNUNET_DATASTORE_get_zero_anonymity (struct GNUNET_DATASTORE_Handle *h,
+                                    uint64_t offset,
+                                    unsigned int queue_priority,
+                                    unsigned int max_queue_size,
+                                    struct GNUNET_TIME_Relative timeout,
+                                    enum GNUNET_BLOCK_Type type,
+                                    GNUNET_DATASTORE_DatumProcessor proc, 
+                                    void *proc_cls)
 {
   struct GNUNET_DATASTORE_QueueEntry *qe;
   struct GetZeroAnonymityMessage *m;
   union QueueContext qc;
 
+  GNUNET_assert (NULL != proc);
+  GNUNET_assert (type != GNUNET_BLOCK_TYPE_ANY);
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-             "Asked to get zero-anonymity entry in %llu ms\n",
+             "Asked to get %llu-th zero-anonymity entry of type %d in %llu ms\n",
+             (unsigned long long) offset,
+             type,
              (unsigned long long) timeout.rel_value);
 #endif
-  qc.rc.iter = iter;
-  qc.rc.iter_cls = iter_cls;
+  qc.rc.proc = proc;
+  qc.rc.proc_cls = proc_cls;
   qe = make_queue_entry (h, sizeof(struct GetZeroAnonymityMessage),
                         queue_priority, max_queue_size, timeout,
                         &process_result_message, &qc);
@@ -1382,7 +1453,7 @@ GNUNET_DATASTORE_iterate_zero_anonymity (struct GNUNET_DATASTORE_Handle *h,
     {
 #if DEBUG_DATASTORE
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                 "Could not create queue entry for zero-anonymity iteration\n");
+                 "Could not create queue entry for zero-anonymity procation\n");
 #endif
       return NULL;    
     }
@@ -1394,55 +1465,57 @@ GNUNET_DATASTORE_iterate_zero_anonymity (struct GNUNET_DATASTORE_Handle *h,
   m->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_ZERO_ANONYMITY);
   m->header.size = htons(sizeof (struct GetZeroAnonymityMessage));
   m->type = htonl ((uint32_t) type);
+  m->offset = GNUNET_htonll (offset);
   process_queue (h);
   return qe;
 }
 
 
-
 /**
- * Iterate over the results for a particular key
- * in the datastore.  The iterator will only be called
- * once initially; if the first call did contain a
- * result, further results can be obtained by calling
- * "GNUNET_DATASTORE_iterate_get_next" with the given argument.
+ * Get a result for a particular key from the datastore.  The processor
+ * will only be called once.
  *
  * @param h handle to the datastore
+ * @param offset offset of the result (mod #num-results); set to
+ *               a random 64-bit value initially; then increment by
+ *               one each time; detect that all results have been found by uid
+ *               being again the first uid ever returned.
  * @param key maybe NULL (to match all entries)
  * @param type desired type, 0 for any
  * @param queue_priority ranking of this request in the priority queue
  * @param max_queue_size at what queue size should this request be dropped
  *        (if other requests of higher priority are in the queue)
  * @param timeout how long to wait at most for a response
- * @param iter function to call on each matching value;
+ * @param proc function to call on each matching value;
  *        will be called once with a NULL value at the end
- * @param iter_cls closure for iter
+ * @param proc_cls closure for proc
  * @return NULL if the entry was not queued, otherwise a handle that can be used to
- *         cancel; note that even if NULL is returned, the callback will be invoked
- *         (or rather, will already have been invoked)
+ *         cancel
  */
 struct GNUNET_DATASTORE_QueueEntry *
-GNUNET_DATASTORE_iterate_key (struct GNUNET_DATASTORE_Handle *h,
-                             const GNUNET_HashCode * key,
-                             enum GNUNET_BLOCK_Type type,
-                             unsigned int queue_priority,
-                             unsigned int max_queue_size,
-                             struct GNUNET_TIME_Relative timeout,
-                             GNUNET_DATASTORE_Iterator iter, 
-                             void *iter_cls)
+GNUNET_DATASTORE_get_key (struct GNUNET_DATASTORE_Handle *h,
+                         uint64_t offset,
+                         const GNUNET_HashCode * key,
+                         enum GNUNET_BLOCK_Type type,
+                         unsigned int queue_priority,
+                         unsigned int max_queue_size,
+                         struct GNUNET_TIME_Relative timeout,
+                         GNUNET_DATASTORE_DatumProcessor proc, 
+                         void *proc_cls)
 {
   struct GNUNET_DATASTORE_QueueEntry *qe;
   struct GetMessage *gm;
   union QueueContext qc;
 
+  GNUNET_assert (NULL != proc);
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Asked to look for data of type %u under key `%s'\n",
              (unsigned int) type,
              GNUNET_h2s (key));
 #endif
-  qc.rc.iter = iter;
-  qc.rc.iter_cls = iter_cls;
+  qc.rc.proc = proc;
+  qc.rc.proc_cls = proc_cls;
   qe = make_queue_entry (h, sizeof(struct GetMessage),
                         queue_priority, max_queue_size, timeout,
                         &process_result_message, &qc);
@@ -1462,6 +1535,7 @@ GNUNET_DATASTORE_iterate_key (struct GNUNET_DATASTORE_Handle *h,
   gm = (struct GetMessage*) &qe[1];
   gm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET);
   gm->type = htonl(type);
+  gm->offset = GNUNET_htonll (offset);
   if (key != NULL)
     {
       gm->header.size = htons(sizeof (struct GetMessage));
@@ -1476,26 +1550,6 @@ GNUNET_DATASTORE_iterate_key (struct GNUNET_DATASTORE_Handle *h,
 }
 
 
-/**
- * Function called to trigger obtaining the next result
- * from the datastore.
- * 
- * @param h handle to the datastore
- */
-void 
-GNUNET_DATASTORE_iterate_get_next (struct GNUNET_DATASTORE_Handle *h)
-{
-  struct GNUNET_DATASTORE_QueueEntry *qe = h->queue_head;
-
-  GNUNET_assert (&process_result_message == qe->response_proc);
-  h->in_receive = GNUNET_YES;
-  GNUNET_CLIENT_receive (h->client,
-                        qe->response_proc,
-                        qe,
-                        GNUNET_TIME_absolute_get_remaining (qe->timeout));
-}
-
-
 /**
  * Cancel a datastore operation.  The final callback from the
  * operation must not have been done yet.
@@ -1507,6 +1561,7 @@ GNUNET_DATASTORE_cancel (struct GNUNET_DATASTORE_QueueEntry *qe)
 {
   struct GNUNET_DATASTORE_Handle *h;
 
+  GNUNET_assert (GNUNET_SYSERR != qe->was_transmitted);
   h = qe->h;
 #if DEBUG_DATASTORE
   GNUNET_log  (GNUNET_ERROR_TYPE_DEBUG,
@@ -1518,8 +1573,7 @@ GNUNET_DATASTORE_cancel (struct GNUNET_DATASTORE_QueueEntry *qe)
   if (GNUNET_YES == qe->was_transmitted) 
     {
       free_queue_entry (qe);
-      h->retry_time = GNUNET_TIME_UNIT_ZERO;
-      do_disconnect (h);
+      h->skip_next_messages++;
       return;
     }
   free_queue_entry (qe);