further datastore API improvements:

6 files changed, 496 insertions, 131 deletions

diff --git a/src/datastore/datastore_api.c b/src/datastore/datastore_api.c
index 960d7d34e..968e09b20 100644
--- a/src/datastore/datastore_api.c
+++ b/src/datastore/datastore_api.c
@@ -22,15 +22,6 @@
  * @file datastore/datastore_api.c
  * @brief Management for the datastore for files stored on a GNUnet node
  * @author Christian Grothoff
- *
- * TODO:
- * 1) clarify API (wrt. efficient UPDATE of priority/expiration after GET)
- * 2) implement INIT
- * 4) implement DROP
- * 5) implement PUT
- * 6) implement GET
- * 7) implement GET_RANDOM
- * 8) implement REMOVE
  */
 
 #include "platform.h"
diff --git a/src/datastore/gnunet-service-datastore.c b/src/datastore/gnunet-service-datastore.c
index db3524557..9a24865da 100644
--- a/src/datastore/gnunet-service-datastore.c
+++ b/src/datastore/gnunet-service-datastore.c
@@ -24,14 +24,8 @@
  * @author Christian Grothoff
  *
  * TODO:
- * 1) transmit and transmit flow-control (when do we signal client 'success'?
- *    ALSO: async transmit will need to address ref-counting issues on client!
- * 2) efficient "update" for client to raise priority / expiration
- *    (not possible with current datastore API, but plugin API has support!);
- *    [ maybe integrate desired priority/expiration updates directly
- *      with 'GET' request? ]
- * 3) semantics of "PUT" (plugin) if entry exists (should likely
- *   be similar to "UPDATE" (need to specify in PLUGIN API!)
+ * 1) semantics of "PUT" (plugin) if entry exists (should likely
+ *    be similar to "UPDATE" (need to specify in PLUGIN API!)
  * 4) quota management code!
  * 5) add bloomfilter for efficiency!
  */
@@ -42,6 +36,11 @@
 #include "plugin_datastore.h"
 #include "datastore.h"
 
+/**
+ * How many messages do we queue at most per client?
+ */
+#define MAX_PENDING 1024
+
 
 /**
  * Our datastore plugin.
@@ -75,19 +74,182 @@ struct DatastorePlugin
 
 
 /**
+ * Linked list of active reservations.
+ */
+struct ReservationList 
+{
+
+  /**
+   * This is a linked list.
+   */
+  struct ReservationList *next;
+
+  /**
+   * Client that made the reservation.
+   */
+  struct GNUNET_SERVER_Client *client;
+
+  /**
+   * Number of bytes (still) reserved.
+   */
+  uint64_t size;
+
+  /**
+   * Number of items (still) reserved.
+   */
+  uint64_t items;
+
+  /**
+   * Reservation identifier.
+   */
+  int32_t rid;
+
+};
+
+
+/**
  * Our datastore plugin (NULL if not available).
  */
 static struct DatastorePlugin *plugin;
 
+/**
+ * Linked list of space reservations made by clients.
+ */
+static struct ReservationList *reservations;
+
+/**
+ * Static counter to produce reservation identifiers.
+ */
+static int reservation_gen;
+
+/**
+ * How much space are we allowed to use?
+ */
+static unsigned long long quota;
+
+
+/**
+ * Function called once the transmit operation has
+ * either failed or succeeded.
+ *
+ * @param cls closure
+ * @param status GNUNET_OK on success, GNUNET_SYSERR on error
+ */
+typedef void (*TransmitContinuation)(void *cls,
+                                     int status);
+
+struct TransmitCallbackContext 
+{
+  /**
+   * The message that we're asked to transmit.
+   */
+  struct GNUNET_MessageHeader *msg;
+
+  /**
+   * Client that we are transmitting to.
+   */
+  struct GNUNET_SERVER_Client *client;
+
+  /**
+   * Function to call once msg has been transmitted
+   * (or at least added to the buffer).
+   */
+  TransmitContinuation tc;
+
+  /**
+   * Closure for tc.
+   */
+  void *tc_cls;
+
+  /**
+   * GNUNET_YES if we are supposed to signal the server
+   * completion of the client's request.
+   */
+  int end;
+};
+
+
+/**
+ * Function called to notify a client about the socket
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the socket was closed for
+ * writing in the meantime.
+ *
+ * @param cls closure
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+transmit_callback (void *cls,
+                   size_t size, void *buf)
+{
+  struct TransmitCallbackContext *tcc = cls;
+  size_t msize;
+  
+  msize = ntohs(tcc->msg->size);
+  if (size == 0)
+    {
+      if (tcc->tc != NULL)
+        tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
+      if (GNUNET_YES == tcc->end)
+        GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
+      GNUNET_free (tcc->msg);
+      GNUNET_free (tcc);
+      return 0;
+    }
+  GNUNET_assert (size >= msize);
+  memcpy (buf, tcc->msg, msize);
+  if (tcc->tc != NULL)
+    tcc->tc (tcc->tc_cls, GNUNET_OK);
+  if (GNUNET_YES == tcc->end)
+    GNUNET_SERVER_receive_done (tcc->client, GNUNET_OK);     
+  GNUNET_free (tcc->msg);
+  GNUNET_free (tcc);
+  return msize;
+}
+
 
 /**
  * Transmit the given message to the client.
+ *
+ * @param client target of the message
+ * @param msg message to transmit, will be freed!
+ * @param end is this the last response (and we should
+ *        signal the server completion accodingly after
+ *        transmitting this message)?
  */
 static void
 transmit (struct GNUNET_SERVER_Client *client,
-          const struct GNUNET_MessageHeader *msg)
+          struct GNUNET_MessageHeader *msg,
+          TransmitContinuation tc,
+          void *tc_cls,
+          int end)
 {
-  /* FIXME! */
+  struct TransmitCallbackContext *tcc;
+
+  tcc = GNUNET_malloc (sizeof(struct TransmitCallbackContext));
+  tcc->msg = msg;
+  tcc->client = client;
+  tcc->tc = tc;
+  tcc->tc_cls = tc_cls;
+  tcc->end = end;
+
+  if (NULL ==
+      GNUNET_SERVER_notify_transmit_ready (client,
+                                           ntohs(msg->size),
+                                           GNUNET_TIME_UNIT_FOREVER_REL,
+                                           &transmit_callback,
+                                           tcc))
+    {
+      GNUNET_break (0);
+      if (GNUNET_YES == end)
+        GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+      if (NULL != tc)
+        tc (tc_cls, GNUNET_SYSERR);
+      GNUNET_free (msg);
+      GNUNET_free (tcc);
+    }
 }
 
 
@@ -112,8 +274,29 @@ transmit_status (struct GNUNET_SERVER_Client *client,
   sm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_STATUS);
   sm->status = htonl(code);
   memcpy (&sm[1], msg, slen);  
-  transmit (client, &sm->header);
-  GNUNET_free (sm);
+  transmit (client, &sm->header, NULL, NULL, GNUNET_YES);
+}
+
+
+/**
+ * Function called once the transmit operation has
+ * either failed or succeeded.
+ *
+ * @param cls closure
+ * @param status GNUNET_OK on success, GNUNET_SYSERR on error
+ */
+static void 
+get_next(void *next_cls,
+         int status)
+{
+  if (status != GNUNET_OK)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  _("Failed to transmit an item to the client; aborting iteration.\n"));    
+      plugin->api->next_request (next_cls, GNUNET_YES);
+      return;
+    }
+  plugin->api->next_request (next_cls, GNUNET_NO);
 }
 
 
@@ -122,6 +305,7 @@ transmit_status (struct GNUNET_SERVER_Client *client,
  * to the client.
  *
  * @param cls closure, pointer to the client (of type GNUNET_SERVER_Client).
+ * @param next_cls closure to use to ask for the next item
  * @param key key for the content
  * @param size number of bytes in data
  * @param data content stored
@@ -137,6 +321,7 @@ transmit_status (struct GNUNET_SERVER_Client *client,
  */
 static int
 transmit_item (void *cls,
+               void *next_cls,
                const GNUNET_HashCode * key,
                uint32_t size,
                const void *data,
@@ -147,15 +332,17 @@ transmit_item (void *cls,
                expiration, unsigned long long uid)
 {
   struct GNUNET_SERVER_Client *client = cls;
-  struct GNUNET_MessageHeader end;
+  struct GNUNET_MessageHeader *end;
   struct DataMessage *dm;
 
   if (key == NULL)
     {
       /* transmit 'DATA_END' */
-      end.size = htons(sizeof(struct GNUNET_MessageHeader));
-      end.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END);
-      transmit (client, &end);
+      end = GNUNET_malloc (sizeof(struct GNUNET_MessageHeader));
+      end->size = htons(sizeof(struct GNUNET_MessageHeader));
+      end->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END);
+      transmit (client, end, NULL, NULL, GNUNET_YES);
+      GNUNET_SERVER_client_drop (client);
       return GNUNET_OK;
     }
   dm = GNUNET_malloc (sizeof(struct DataMessage) + size);
@@ -170,8 +357,7 @@ transmit_item (void *cls,
   dm->uid = GNUNET_htonll(uid);
   dm->key = *key;
   memcpy (&dm[1], data, size);
-  transmit (client, &dm->header);
-  GNUNET_free (dm);
+  transmit (client, &dm->header, &get_next, next_cls, GNUNET_NO);
   return GNUNET_OK;
 }
 
@@ -188,8 +374,20 @@ handle_reserve (void *cls,
              struct GNUNET_SERVER_Client *client,
              const struct GNUNET_MessageHeader *message)
 {
-  transmit_status (client, GNUNET_SYSERR, "not implemented");
-  GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+  const struct ReserveMessage *msg = (const struct ReserveMessage*) message;
+  struct ReservationList *e;
+
+  /* FIXME: check if we have that much space... */
+  e = GNUNET_malloc (sizeof(struct ReservationList));
+  e->next = reservations;
+  reservations = e;
+  e->client = client;
+  e->size = GNUNET_ntohll(msg->size);
+  e->items = GNUNET_ntohll(msg->items);
+  e->rid = ++reservation_gen;
+  if (reservation_gen < 0)
+    reservation_gen = 0; /* wrap around */
+  transmit_status (client, e->rid, NULL);
 }
 
 
@@ -205,8 +403,32 @@ handle_release_reserve (void *cls,
                         struct GNUNET_SERVER_Client *client,
                         const struct GNUNET_MessageHeader *message)
 {
-  transmit_status (client, GNUNET_SYSERR, "not implemented");
-  GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+  const struct ReleaseReserveMessage *msg = (const struct ReleaseReserveMessage*) message;
+  struct ReservationList *pos;
+  struct ReservationList *prev;
+  struct ReservationList *next;
+  
+  int rid = ntohl(msg->rid);
+  next = reservations;
+  prev = NULL;
+  while (NULL != (pos = next))
+    {
+      next = pos->next;
+      if (rid == pos->rid)
+        {
+          if (prev == NULL)
+            reservations = next;
+          else
+            prev->next = next;
+          /* FIXME: released remaining reserved space! */
+          GNUNET_free (pos);
+          transmit_status (client, GNUNET_OK, NULL);
+          return;
+        }       
+      prev = pos;
+      pos = next;
+    }
+  transmit_status (client, GNUNET_SYSERR, "Could not find matching reservation");
 }
 
 
@@ -284,7 +506,6 @@ handle_put (void *cls,
                           &msg);
   transmit_status (client, ret, msg);
   GNUNET_free_non_null (msg);
-  GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 }
 
 
@@ -312,13 +533,13 @@ handle_get (void *cls,
       return;
     }
   msg = (const struct GetMessage*) message;
+  GNUNET_SERVER_client_drop (client);
   plugin->api->get (plugin->api->cls,
                     ((size == sizeof(struct GetMessage)) ? &msg->key : NULL),
                     NULL,
                     ntohl(msg->type),
                     &transmit_item,
                     client);    
-  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -347,7 +568,6 @@ handle_update (void *cls,
                              &emsg);
   transmit_status (client, ret, emsg);
   GNUNET_free_non_null (emsg);
-  GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 }
 
 
@@ -363,20 +583,38 @@ handle_get_random (void *cls,
                    struct GNUNET_SERVER_Client *client,
                    const struct GNUNET_MessageHeader *message)
 {
+  GNUNET_SERVER_client_drop (client);
   plugin->api->iter_migration_order (plugin->api->cls,
                                      0,
                                      &transmit_item,
                                      client);  
-  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
 /**
+ * Context for the 'remove_callback'.
+ */
+struct RemoveContext 
+{
+  /**
+   * Client for whom we're doing the remvoing.
+   */
+  struct GNUNET_SERVER_Client *client;
+
+  /**
+   * GNUNET_YES if we managed to remove something.
+   */
+  int found;
+};
+
+
+/**
  * Callback function that will cause the item that is passed
  * in to be deleted (by returning GNUNET_NO).
  */
 static int
 remove_callback (void *cls,
+                 void *next_cls,
                  const GNUNET_HashCode * key,
                  uint32_t size,
                  const void *data,
@@ -386,8 +624,19 @@ remove_callback (void *cls,
                  struct GNUNET_TIME_Absolute
                  expiration, unsigned long long uid)
 {
-  int *found = cls;
-  *found = GNUNET_YES;
+  struct RemoveContext *rc = cls;
+  if (key == NULL)
+    {
+      if (GNUNET_YES == rc->found)
+        transmit_status (rc->client, GNUNET_OK, NULL);
+      else
+        transmit_status (rc->client, GNUNET_SYSERR, _("Content not found"));       
+      GNUNET_SERVER_client_drop (rc->client);
+      GNUNET_free (rc);
+      return GNUNET_OK; /* last item */
+    }
+  rc->found = GNUNET_YES;
+  plugin->api->next_request (next_cls, GNUNET_YES);
   return GNUNET_NO;
 }
 
@@ -406,7 +655,7 @@ handle_remove (void *cls,
 {
   const struct DataMessage *dm = check_data (message);
   GNUNET_HashCode vhash;
-  int found;
+  struct RemoveContext *rc;
 
   if (dm == NULL)
     {
@@ -414,7 +663,9 @@ handle_remove (void *cls,
       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
       return;
     }
-  found = GNUNET_NO;
+  rc = GNUNET_malloc (sizeof(struct RemoveContext));
+  GNUNET_SERVER_client_keep (client);
+  rc->client = client;
   GNUNET_CRYPTO_hash (&dm[1],
                       ntohl(dm->size),
                       &vhash);
@@ -423,12 +674,7 @@ handle_remove (void *cls,
                     &vhash,
                     ntohl(dm->type),
                     &remove_callback,
-                    &found);
-  if (GNUNET_YES == found)
-    transmit_status (client, GNUNET_OK, NULL);
-  else
-    transmit_status (client, GNUNET_SYSERR, _("Content not found"));
-  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+                    rc);
 }
 
 
@@ -549,6 +795,23 @@ cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 
 
 /**
+ * Function that removes all active reservations made
+ * by the given client and releases the space for other
+ * requests.
+ *
+ * @param cls closure
+ * @param client identification of the client
+ */
+static void
+cleanup_reservations (void *cls,
+                      struct GNUNET_SERVER_Client
+                      * client)
+{
+  /* FIXME */
+}
+
+
+/**
  * Process datastore requests.
  *
  * @param cls closure
@@ -562,9 +825,20 @@ run (void *cls,
      struct GNUNET_SERVER_Handle *server,
      struct GNUNET_CONFIGURATION_Handle *cfg)
 {
+  if (GNUNET_OK !=
+      GNUNET_CONFIGURATION_get_value_number (cfg,
+                                             "DATASTORE", "QUOTA", &quota))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  _("No `%s' specified for `%s' in configuration!\n"),
+                  "QUOTA",
+                  "DATASTORE");
+      return;
+    }
   plugin = load_plugin (cfg, sched);
   if (NULL == plugin)
     return;
+  GNUNET_SERVER_disconnect_notify (server, &cleanup_reservations, NULL);
   GNUNET_SERVER_add_handlers (server, handlers);
   GNUNET_SCHEDULER_add_delayed (sched,
                                 GNUNET_YES,
diff --git a/src/datastore/plugin_datastore.h b/src/datastore/plugin_datastore.h
index cc68f3557..1214ae1be 100644
--- a/src/datastore/plugin_datastore.h
+++ b/src/datastore/plugin_datastore.h
@@ -55,15 +55,64 @@ struct GNUNET_DATASTORE_PluginEnvironment
 
 
 /**
+ * Function invoked on behalf of a "PluginIterator"
+ * asking the database plugin to call the iterator
+ * with the next item.
+ *
+ * @param next_cls whatever argument was given
+ *        to the PluginIterator as "next_cls".
+ * @param end_it set to GNUNET_YES if we
+ *        should terminate the iteration early
+ *        (iterator should be still called once more
+ *         to signal the end of the iteration).
+ */
+typedef void (*PluginNextRequest)(void *next_cls,
+                                  int end_it);
+
+
+/**
+ * An iterator over a set of items stored in the datastore.
+ *
+ * @param cls closure
+ * @param next_cls closure to pass to the "next" function.
+ * @param key key for the content
+ * @param size number of bytes in data
+ * @param data content stored
+ * @param type type of the content
+ * @param priority priority of the content
+ * @param anonymity anonymity-level for the content
+ * @param expiration expiration time for the content
+ * @param uid unique identifier for the datum;
+ *        maybe 0 if no unique identifier is available
+ *
+ * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
+ *         (continue on call to "next", of course),
+ *         GNUNET_NO to delete the item and continue (if supported)
+ */
+typedef int (*PluginIterator) (void *cls,
+                               void *next_cls,
+                               const GNUNET_HashCode * key,
+                               uint32_t size,
+                               const void *data,
+                               uint32_t type,
+                               uint32_t priority,
+                               uint32_t anonymity,
+                               struct GNUNET_TIME_Absolute
+                               expiration, 
+                               uint64_t uid);
+
+/**
  * Get an estimate of how much space the database is
  * currently using.
  * @return number of bytes used on disk
  */
-typedef unsigned long long (*GNUNET_DATASTORE_GetSize) (void *cls);
+typedef unsigned long long (*PluginGetSize) (void *cls);
 
 
 /**
- * Store an item in the datastore.
+ * Store an item in the datastore.  If the item is already present,
+ * the priorities are summed up and the higher expiration time and
+ * lower anonymity level is used.
  *
  * @param cls closure
  * @param key key for the item
@@ -76,15 +125,14 @@ typedef unsigned long long (*GNUNET_DATASTORE_GetSize) (void *cls);
  * @param msg set to an error message (on failure)
  * @return GNUNET_OK on success, GNUNET_SYSERR on failure
  */
-typedef int
-  (*GNUNET_DATASTORE_Put) (void *cls,
-                           const GNUNET_HashCode * key,
-                           uint32_t size,
-                           const void *data,
-                           uint32_t type,
-                           uint32_t priority,
-                           uint32_t anonymity,
-                           struct GNUNET_TIME_Absolute expiration,
+typedef int (*PluginPut) (void *cls,
+                          const GNUNET_HashCode * key,
+                          uint32_t size,
+                          const void *data,
+                          uint32_t type,
+                          uint32_t priority,
+                          uint32_t anonymity,
+                          struct GNUNET_TIME_Absolute expiration,
                            char **msg);
 
 
@@ -101,16 +149,19 @@ typedef int
  *        there may be!
  * @param type entries of which type are relevant?
  *     Use 0 for any type.
- * @param iter function to call on each matching value;
- *        will be called once with a NULL value at the end
+ * @param iter function to call on each matching value; however,
+ *        after the first call to "iter", the plugin must wait
+ *        until "NextRequest" was called before giving the iterator
+ *        the next item; finally, the "iter" should be called once
+ *        once with a NULL value at the end ("next_cls" should be NULL
+ *        for that last call)
  * @param iter_cls closure for iter
  */
-typedef void
-  (*GNUNET_DATASTORE_Get) (void *cls,
-                           const GNUNET_HashCode * key,
-                           const GNUNET_HashCode * vhash,
-                           uint32_t type,
-                           GNUNET_DATASTORE_Iterator iter, void *iter_cls);
+typedef void (*PluginGet) (void *cls,
+                           const GNUNET_HashCode * key,
+                           const GNUNET_HashCode * vhash,
+                           uint32_t type,
+                           PluginIterator iter, void *iter_cls);
 
 
 /**
@@ -135,11 +186,10 @@ typedef void
  * @param msg set to an error message (on error)
  * @return GNUNET_OK on success
  */
-typedef int
-  (*GNUNET_DATASTORE_Update) (void *cls,
-                              unsigned long long uid,
-                              int delta, struct GNUNET_TIME_Absolute expire,
-                              char **msg);
+typedef int (*PluginUpdate) (void *cls,
+                             uint64_t uid,
+                             int delta, struct GNUNET_TIME_Absolute expire,
+                             char **msg);
 
 
 /**
@@ -148,20 +198,23 @@ typedef int
  *
  * @param type entries of which type should be considered?
  *        Use 0 for any type.
- * @param iter function to call on each matching value;
- *        will be called once with a NULL value at the end
+ * @param iter function to call on each matching value; however,
+ *        after the first call to "iter", the plugin must wait
+ *        until "NextRequest" was called before giving the iterator
+ *        the next item; finally, the "iter" should be called once
+ *        once with a NULL value at the end ("next_cls" should be NULL
+ *        for that last call)
  * @param iter_cls closure for iter
  */
-typedef void
-  (*GNUNET_DATASTORE_Selector) (void *cls,
+typedef void (*PluginSelector) (void *cls,
                                 uint32_t type,
-                                GNUNET_DATASTORE_Iterator iter,
+                                PluginIterator iter,
                                 void *iter_cls);
 
 /**
  * Drop database.
  */
-typedef void (*GNUNET_DATASTORE_Drop) (void *cls);
+typedef void (*PluginDrop) (void *cls);
 
 
 
@@ -173,7 +226,8 @@ struct GNUNET_DATASTORE_PluginFunctions
 {
 
   /**
-   * Closure to use for all of the following callbacks.
+   * Closure to use for all of the following callbacks
+   * (except "next_request").
    */
   void *cls;
 
@@ -181,18 +235,26 @@ struct GNUNET_DATASTORE_PluginFunctions
    * Get the current on-disk size of the SQ store.  Estimates are
    * fine, if that's the only thing available.
    */
-  GNUNET_DATASTORE_GetSize get_size;
+  PluginGetSize get_size;
 
   /**
    * Function to store an item in the datastore.
    */
-  GNUNET_DATASTORE_Put put;
+  PluginPut put;
+
+  /**
+   * Function called by iterators whenever they want the next value;
+   * note that unlike all of the other callbacks, this one does get a
+   * the "next_cls" closure which is usually different from the "cls"
+   * member of this struct!
+   */
+  PluginNextRequest next_request;
 
   /**
    * Function to iterate over the results for a particular key
    * in the datastore.
    */
-  GNUNET_DATASTORE_Get get;
+  PluginGet get;
 
   /**
    * Update the priority for a particular key in the datastore.  If
@@ -202,24 +264,24 @@ struct GNUNET_DATASTORE_PluginFunctions
    * priority should be added to the existing priority, ignoring the
    * priority in value.
    */
-  GNUNET_DATASTORE_Update update;
+  PluginUpdate update;
 
   /**
    * Iterate over the items in the datastore in ascending
    * order of priority.
    */
-  GNUNET_DATASTORE_Selector iter_low_priority;
+  PluginSelector iter_low_priority;
 
   /**
    * Iterate over content with anonymity zero.
    */
-  GNUNET_DATASTORE_Selector iter_zero_anonymity;
+  PluginSelector iter_zero_anonymity;
 
   /**
    * Iterate over the items in the datastore in ascending order of
    * expiration time. 
    */
-  GNUNET_DATASTORE_Selector iter_ascending_expiration;
+  PluginSelector iter_ascending_expiration;
 
   /**
    * Iterate over the items in the datastore in migration
@@ -227,7 +289,7 @@ struct GNUNET_DATASTORE_PluginFunctions
    * (and then signal 'end' with a second call).  This is
    * a significant difference from all the other iterators!
    */
-  GNUNET_DATASTORE_Selector iter_migration_order;
+  PluginSelector iter_migration_order;
 
   /**
    * Iterate over all the items in the datastore
@@ -235,13 +297,13 @@ struct GNUNET_DATASTORE_PluginFunctions
    * (can lock datastore while this happens, focus
    * is on doing it fast).
    */
-  GNUNET_DATASTORE_Selector iter_all_now;
+  PluginSelector iter_all_now;
 
   /**
    * Delete the database.  The next operation is
    * guaranteed to be unloading of the module.
    */
-  GNUNET_DATASTORE_Drop drop;
+  PluginDrop drop;
 
 };
 
diff --git a/src/datastore/plugin_datastore_sqlite.c b/src/datastore/plugin_datastore_sqlite.c
index 234519079..6bee1a0c9 100644
--- a/src/datastore/plugin_datastore_sqlite.c
+++ b/src/datastore/plugin_datastore_sqlite.c
@@ -51,6 +51,26 @@ static unsigned long long sqlite_plugin_get_size (void *cls)
 }
 
 
+
+/**
+ * Function invoked on behalf of a "PluginIterator"
+ * asking the database plugin to call the iterator
+ * with the next item.
+ *
+ * @param next_cls whatever argument was given
+ *        to the PluginIterator as "next_cls".
+ * @param end_it set to GNUNET_YES if we
+ *        should terminate the iteration early
+ *        (iterator should be still called once more
+ *         to signal the end of the iteration).
+ */
+static void 
+sqlite_next_request (void *next_cls,
+                     int end_it)
+{
+}
+
+
 /**
  * Store an item in the datastore.
  *
@@ -103,10 +123,10 @@ sqlite_plugin_get (void *cls,
                    const GNUNET_HashCode * key,
                    const GNUNET_HashCode * vhash,
                    uint32_t type,
-                   GNUNET_DATASTORE_Iterator iter, void *iter_cls)
+                   PluginIterator iter, void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
 }
 
 
@@ -155,12 +175,12 @@ sqlite_plugin_update (void *cls,
  */
 static void
 sqlite_plugin_iter_low_priority (void *cls,
-                        uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
-                        void *iter_cls)
+                                 uint32_t type,
+                                 PluginIterator iter,
+                                 void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
 }
 
 
@@ -176,12 +196,12 @@ sqlite_plugin_iter_low_priority (void *cls,
  */
 static void
 sqlite_plugin_iter_zero_anonymity (void *cls,
-                        uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
-                        void *iter_cls)
+                                   uint32_t type,
+                                   PluginIterator iter,
+                                   void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
 }
 
 
@@ -198,12 +218,12 @@ sqlite_plugin_iter_zero_anonymity (void *cls,
  */
 static void
 sqlite_plugin_iter_ascending_expiration (void *cls,
-                        uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
-                        void *iter_cls)
+                                         uint32_t type,
+                                         PluginIterator iter,
+                                         void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
 }
 
 
@@ -220,12 +240,12 @@ sqlite_plugin_iter_ascending_expiration (void *cls,
  */
 static void
 sqlite_plugin_iter_migration_order (void *cls,
-                        uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
-                        void *iter_cls)
+                                    uint32_t type,
+                                    PluginIterator iter,
+                                    void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
 }
 
 
@@ -242,12 +262,12 @@ sqlite_plugin_iter_migration_order (void *cls,
  */
 static void
 sqlite_plugin_iter_all_now (void *cls,
-                        uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
-                        void *iter_cls)
+                            uint32_t type,
+                            PluginIterator iter,
+                            void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
 }
 
 
@@ -276,6 +296,7 @@ libgnunet_plugin_datastore_sqlite_init (void *cls)
   api->cls = plugin;
   api->get_size = &sqlite_plugin_get_size;
   api->put = &sqlite_plugin_put;
+  api->next_request = &sqlite_next_request;
   api->get = &sqlite_plugin_get;
   api->update = &sqlite_plugin_update;
   api->iter_low_priority = &sqlite_plugin_iter_low_priority;
diff --git a/src/datastore/plugin_datastore_template.c b/src/datastore/plugin_datastore_template.c
index 1e3d4343b..4c8f9e1f9 100644
--- a/src/datastore/plugin_datastore_template.c
+++ b/src/datastore/plugin_datastore_template.c
@@ -82,6 +82,25 @@ template_plugin_put (void *cls,
 
 
 /**
+ * Function invoked on behalf of a "PluginIterator"
+ * asking the database plugin to call the iterator
+ * with the next item.
+ *
+ * @param next_cls whatever argument was given
+ *        to the PluginIterator as "next_cls".
+ * @param end_it set to GNUNET_YES if we
+ *        should terminate the iteration early
+ *        (iterator should be still called once more
+ *         to signal the end of the iteration).
+ */
+static void 
+template_next_request (void *next_cls,
+                       int end_it)
+{
+}
+
+
+/**
  * Iterate over the results for a particular key
  * in the datastore.
  *
@@ -100,10 +119,10 @@ template_plugin_put (void *cls,
  */
 static void
 template_plugin_get (void *cls,
-                   const GNUNET_HashCode * key,
-                   const GNUNET_HashCode * vhash,
-                   uint32_t type,
-                   GNUNET_DATASTORE_Iterator iter, void *iter_cls)
+                     const GNUNET_HashCode * key,
+                     const GNUNET_HashCode * vhash,
+                     uint32_t type,
+                     PluginIterator iter, void *iter_cls)
 {
 }
 
@@ -153,9 +172,9 @@ template_plugin_update (void *cls,
  */
 static void
 template_plugin_iter_low_priority (void *cls,
-                        uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
-                        void *iter_cls)
+                                   uint32_t type,
+                                   PluginIterator iter,
+                                   void *iter_cls)
 {
 }
 
@@ -174,7 +193,7 @@ template_plugin_iter_low_priority (void *cls,
 static void
 template_plugin_iter_zero_anonymity (void *cls,
                         uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
+                        PluginIterator iter,
                         void *iter_cls)
 {
 }
@@ -194,7 +213,7 @@ template_plugin_iter_zero_anonymity (void *cls,
 static void
 template_plugin_iter_ascending_expiration (void *cls,
                         uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
+                        PluginIterator iter,
                         void *iter_cls)
 {
 }
@@ -214,7 +233,7 @@ template_plugin_iter_ascending_expiration (void *cls,
 static void
 template_plugin_iter_migration_order (void *cls,
                         uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
+                        PluginIterator iter,
                         void *iter_cls)
 {
 }
@@ -234,7 +253,7 @@ template_plugin_iter_migration_order (void *cls,
 static void
 template_plugin_iter_all_now (void *cls,
                         uint32_t type,
-                        GNUNET_DATASTORE_Iterator iter,
+                        PluginIterator iter,
                         void *iter_cls)
 {
 }
@@ -265,6 +284,7 @@ libgnunet_plugin_datastore_template_init (void *cls)
   api->cls = plugin;
   api->get_size = &template_plugin_get_size;
   api->put = &template_plugin_put;
+  api->next_request = &template_next_request;
   api->get = &template_plugin_get;
   api->update = &template_plugin_update;
   api->iter_low_priority = &template_plugin_iter_low_priority;
diff --git a/src/include/gnunet_datastore_service.h b/src/include/gnunet_datastore_service.h
index 626a5afcc..10b2157d3 100644
--- a/src/include/gnunet_datastore_service.h
+++ b/src/include/gnunet_datastore_service.h
@@ -189,19 +189,16 @@ GNUNET_DATASTORE_update (struct GNUNET_DATASTORE_Handle *h,
  * @param expiration expiration time for the content
  * @param uid unique identifier for the datum;
  *        maybe 0 if no unique identifier is available
- *
- * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue,
- *         GNUNET_NO to delete the item and continue (if supported)
  */
-typedef int (*GNUNET_DATASTORE_Iterator) (void *cls,
-                                          const GNUNET_HashCode * key,
-                                          uint32_t size,
-                                          const void *data,
-                                          uint32_t type,
-                                          uint32_t priority,
-                                          uint32_t anonymity,
-                                          struct GNUNET_TIME_Absolute
-                                          expiration, unsigned long long uid);
+typedef void (*GNUNET_DATASTORE_Iterator) (void *cls,
+                                           const GNUNET_HashCode * key,
+                                           uint32_t size,
+                                           const void *data,
+                                           uint32_t type,
+                                           uint32_t priority,
+                                           uint32_t anonymity,
+                                           struct GNUNET_TIME_Absolute
+                                           expiration, uint64_t uid);
 
 
 /**