1 files changed, 252 insertions, 538 deletions

diff --git a/src/datastore/plugin_datastore_mysql.c b/src/datastore/plugin_datastore_mysql.c
index c3d9212d3..2eefd9b04 100644
--- a/src/datastore/plugin_datastore_mysql.c
+++ b/src/datastore/plugin_datastore_mysql.c
@@ -160,58 +160,6 @@ struct GNUNET_MysqlStatementHandle
 
 };
 
-/**
- * Context for the universal iterator.
- */
-struct NextRequestClosure;
-
-/**
- * Type of a function that will prepare
- * the next iteration.
- *
- * @param cls closure
- * @param nc the next context; NULL for the last
- *         call which gives the callback a chance to
- *         clean up the closure
- * @return GNUNET_OK on success, GNUNET_NO if there are
- *         no more values, GNUNET_SYSERR on error
- */
-typedef int (*PrepareFunction)(void *cls,
-                               struct NextRequestClosure *nc);
-
-
-struct NextRequestClosure
-{
-  struct Plugin *plugin;
-
-  struct GNUNET_TIME_Absolute now;
-
-  /**
-   * Function to call to prepare the next
-   * iteration.
-   */
-  PrepareFunction prep;
-
-  /**
-   * Closure for prep.
-   */
-  void *prep_cls;
-
-  MYSQL_BIND rbind[7];
-
-  enum GNUNET_BLOCK_Type type;
-
-  PluginIterator dviter;
-
-  void *dviter_cls;
-
-  unsigned int count;
-
-  int end_it;
-
-  int one_shot;
-};
-
 
 /**
  * Context for all functions in this plugin.
@@ -244,16 +192,6 @@ struct Plugin
   char *cnffile;
 
   /**
-   * Closure of the 'next_task' (must be freed if 'next_task' is cancelled).
-   */
-  struct NextRequestClosure *next_task_nc;
-
-  /**
-   * Pending task with scheduler for running the next request.
-   */
-  GNUNET_SCHEDULER_TaskIdentifier next_task;
-
-  /**
    * Prepared statements.
    */
 #define INSERT_ENTRY "INSERT INTO gn090 (repl,type,prio,anonLevel,expire,hash,vhash,value) VALUES (?,?,?,?,?,?,?,?)"
@@ -295,7 +233,7 @@ struct Plugin
 #define SELECT_SIZE "SELECT SUM(BIT_LENGTH(value) DIV 8) FROM gn090"
   struct GNUNET_MysqlStatementHandle *get_size;
 
-#define SELECT_IT_NON_ANONYMOUS "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 WHERE anonLevel=0 ORDER BY uid DESC LIMIT 1 OFFSET ?"
+#define SELECT_IT_NON_ANONYMOUS "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 WHERE anonLevel=0 AND type=? ORDER BY uid DESC LIMIT 1 OFFSET ?"
   struct GNUNET_MysqlStatementHandle *zero_iter;
 
 #define SELECT_IT_EXPIRATION "(SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 WHERE expire < ? ORDER BY prio ASC LIMIT 1) "\
@@ -372,7 +310,6 @@ get_my_cnf_path (const struct GNUNET_CONFIGURATION_Handle *cfg)
 }
 
 
-
 /**
  * Free a prepared statement.
  *
@@ -381,8 +318,7 @@ get_my_cnf_path (const struct GNUNET_CONFIGURATION_Handle *cfg)
  */
 static void
 prepared_statement_destroy (struct Plugin *plugin, 
-                            struct GNUNET_MysqlStatementHandle
-                            *s)
+                            struct GNUNET_MysqlStatementHandle *s)
 {
   GNUNET_CONTAINER_DLL_remove (plugin->shead,
                                plugin->stail,
@@ -397,6 +333,8 @@ prepared_statement_destroy (struct Plugin *plugin,
 /**
  * Close database connection and all prepared statements (we got a DB
  * disconnect error).
+ * 
+ * @param plugin plugin context
  */
 static int
 iclose (struct Plugin *plugin)
@@ -420,10 +358,11 @@ iclose (struct Plugin *plugin)
  * Open the connection with the database (and initialize
  * our default options).
  *
+ * @param plugin plugin context
  * @return GNUNET_OK on success
  */
 static int
-iopen (struct Plugin *ret)
+iopen (struct Plugin *plugin)
 {
   char *mysql_dbname;
   char *mysql_server;
@@ -433,67 +372,67 @@ iopen (struct Plugin *ret)
   my_bool reconnect;
   unsigned int timeout;
 
-  ret->dbf = mysql_init (NULL);
-  if (ret->dbf == NULL)
+  plugin->dbf = mysql_init (NULL);
+  if (plugin->dbf == NULL)
     return GNUNET_SYSERR;
-  if (ret->cnffile != NULL)
-    mysql_options (ret->dbf, MYSQL_READ_DEFAULT_FILE, ret->cnffile);
-  mysql_options (ret->dbf, MYSQL_READ_DEFAULT_GROUP, "client");
+  if (plugin->cnffile != NULL)
+    mysql_options (plugin->dbf, MYSQL_READ_DEFAULT_FILE, plugin->cnffile);
+  mysql_options (plugin->dbf, MYSQL_READ_DEFAULT_GROUP, "client");
   reconnect = 0;
-  mysql_options (ret->dbf, MYSQL_OPT_RECONNECT, &reconnect);
-  mysql_options (ret->dbf,
+  mysql_options (plugin->dbf, MYSQL_OPT_RECONNECT, &reconnect);
+  mysql_options (plugin->dbf,
                  MYSQL_OPT_CONNECT_TIMEOUT, (const void *) &timeout);
-  mysql_options(ret->dbf, MYSQL_SET_CHARSET_NAME, "UTF8");
+  mysql_options(plugin->dbf, MYSQL_SET_CHARSET_NAME, "UTF8");
   timeout = 60; /* in seconds */
-  mysql_options (ret->dbf, MYSQL_OPT_READ_TIMEOUT, (const void *) &timeout);
-  mysql_options (ret->dbf, MYSQL_OPT_WRITE_TIMEOUT, (const void *) &timeout);
+  mysql_options (plugin->dbf, MYSQL_OPT_READ_TIMEOUT, (const void *) &timeout);
+  mysql_options (plugin->dbf, MYSQL_OPT_WRITE_TIMEOUT, (const void *) &timeout);
   mysql_dbname = NULL;
-  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (ret->env->cfg,
+  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
                                                      "datastore-mysql", "DATABASE"))
     GNUNET_assert (GNUNET_OK == 
-                   GNUNET_CONFIGURATION_get_value_string (ret->env->cfg,
+                   GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
                                                           "datastore-mysql", "DATABASE", 
                                                           &mysql_dbname));
   else
     mysql_dbname = GNUNET_strdup ("gnunet");
   mysql_user = NULL;
-  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (ret->env->cfg,
+  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
                                                      "datastore-mysql", "USER"))
     {
       GNUNET_assert (GNUNET_OK == 
-                    GNUNET_CONFIGURATION_get_value_string (ret->env->cfg,
+                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
                                                            "datastore-mysql", "USER", 
                                                            &mysql_user));
     }
   mysql_password = NULL;
-  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (ret->env->cfg,
+  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
                                                      "datastore-mysql", "PASSWORD"))
     {
       GNUNET_assert (GNUNET_OK ==
-                    GNUNET_CONFIGURATION_get_value_string (ret->env->cfg,
+                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
                                                            "datastore-mysql", "PASSWORD",
                                                            &mysql_password));
     }
   mysql_server = NULL;
-  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (ret->env->cfg,
+  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
                                                      "datastore-mysql", "HOST"))
     {
       GNUNET_assert (GNUNET_OK == 
-                    GNUNET_CONFIGURATION_get_value_string (ret->env->cfg,
+                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
                                                            "datastore-mysql", "HOST", 
                                                            &mysql_server));
     }
   mysql_port = 0;
-  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (ret->env->cfg,
+  if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
                                                      "datastore-mysql", "PORT"))
     {
       GNUNET_assert (GNUNET_OK ==
-                    GNUNET_CONFIGURATION_get_value_number (ret->env->cfg, "datastore-mysql",
+                    GNUNET_CONFIGURATION_get_value_number (plugin->env->cfg, "datastore-mysql",
                                                            "PORT", &mysql_port));
     }
 
   GNUNET_assert (mysql_dbname != NULL);
-  mysql_real_connect (ret->dbf, 
+  mysql_real_connect (plugin->dbf, 
                       mysql_server, 
                       mysql_user, mysql_password,
                       mysql_dbname, 
@@ -503,10 +442,10 @@ iopen (struct Plugin *ret)
   GNUNET_free_non_null (mysql_user);
   GNUNET_free_non_null (mysql_password);
   GNUNET_free (mysql_dbname);
-  if (mysql_error (ret->dbf)[0])
+  if (mysql_error (plugin->dbf)[0])
     {
       LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR,
-                 "mysql_real_connect", ret);
+                 "mysql_real_connect", plugin);
       return GNUNET_SYSERR;
     }
   return GNUNET_OK;
@@ -686,19 +625,6 @@ init_params (struct Plugin *plugin,
   return GNUNET_OK;
 }
 
-/**
- * Type of a callback that will be called for each
- * data set returned from MySQL.
- *
- * @param cls user-defined argument
- * @param num_values number of elements in values
- * @param values values returned by MySQL
- * @return GNUNET_OK to continue iterating, GNUNET_SYSERR to abort
- */
-typedef int (*GNUNET_MysqlDataProcessor) (void *cls,
-                                          unsigned int num_values,
-                                          MYSQL_BIND *values);
-
 
 /**
  * Run a prepared SELECT statement.
@@ -708,40 +634,31 @@ typedef int (*GNUNET_MysqlDataProcessor) (void *cls,
  * @param result_size number of elements in results array
  * @param results pointer to already initialized MYSQL_BIND
  *        array (of sufficient size) for passing results
- * @param processor function to call on each result
- * @param processor_cls extra argument to processor
- * @param ... pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
+ * @param ap pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
  *        values (size + buffer-reference for pointers); terminated
  *        with "-1"
- * @return GNUNET_SYSERR on error, otherwise
- *         the number of successfully affected (or queried) rows
+ * @return GNUNET_SYSERR on error, otherwise GNUNET_OK or GNUNET_NO (no result)
  */
 static int
-prepared_statement_run_select (struct Plugin *plugin,
-                               struct GNUNET_MysqlStatementHandle *s,
-                               unsigned int result_size,
-                               MYSQL_BIND *results,
-                               GNUNET_MysqlDataProcessor processor, void *processor_cls,
-                               ...)
+prepared_statement_run_select_va (struct Plugin *plugin,
+                                  struct GNUNET_MysqlStatementHandle *s,
+                                  unsigned int result_size,
+                                  MYSQL_BIND *results,
+                                  va_list ap)
 {
-  va_list ap;
   int ret;
   unsigned int rsize;
-  int total;
 
   if (GNUNET_OK != prepare_statement (plugin, s))
     {
       GNUNET_break (0);
       return GNUNET_SYSERR;
     }
-  va_start (ap, processor_cls);
   if (GNUNET_OK != init_params (plugin, s, ap))
     {
       GNUNET_break (0);
-      va_end (ap);
       return GNUNET_SYSERR;
     }
-  va_end (ap);
   rsize = mysql_stmt_field_count (s->statement);
   if (rsize > result_size)
     {
@@ -757,29 +674,53 @@ prepared_statement_run_select (struct Plugin *plugin,
       iclose (plugin);
       return GNUNET_SYSERR;
     }
-
-  total = 0;
-  while (1)
+  ret = mysql_stmt_fetch (s->statement);
+  if (ret == MYSQL_NO_DATA)
+    return GNUNET_NO;
+  if (ret != 0)
     {
-      ret = mysql_stmt_fetch (s->statement);
-      if (ret == MYSQL_NO_DATA)
-        break;
-      if (ret != 0)
-        {
-          GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                      _("`%s' failed at %s:%d with error: %s\n"),
-                      "mysql_stmt_fetch",
-                      __FILE__, __LINE__, mysql_stmt_error (s->statement));
-          iclose (plugin);
-          return GNUNET_SYSERR;
-        }
-      if (processor != NULL)
-        if (GNUNET_OK != processor (processor_cls, rsize, results))
-          break;
-      total++;
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  _("`%s' failed at %s:%d with error: %s\n"),
+                  "mysql_stmt_fetch",
+                  __FILE__, __LINE__, mysql_stmt_error (s->statement));
+      iclose (plugin);
+      return GNUNET_SYSERR;
     }
   mysql_stmt_reset (s->statement);
-  return total;
+  return GNUNET_OK;
+}
+
+
+/**
+ * Run a prepared SELECT statement.
+ *
+ * @param plugin plugin context
+ * @param s statement to run
+ * @param result_size number of elements in results array
+ * @param results pointer to already initialized MYSQL_BIND
+ *        array (of sufficient size) for passing results
+ * @param ... pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
+ *        values (size + buffer-reference for pointers); terminated
+ *        with "-1"
+ * @return GNUNET_SYSERR on error, otherwise
+ *         the number of successfully affected (or queried) rows
+ */
+static int
+prepared_statement_run_select (struct Plugin *plugin,
+                               struct GNUNET_MysqlStatementHandle *s,
+                               unsigned int result_size,
+                               MYSQL_BIND *results,
+                               ...)
+{
+  va_list ap;
+  int ret;
+
+  va_start (ap, results);
+  ret = prepared_statement_run_select_va (plugin, s, 
+                                          result_size, results,
+                                          ap);
+  va_end (ap);
+  return ret;
 }
 
 
@@ -854,23 +795,6 @@ do_delete_entry (struct Plugin *plugin,
 
 
 /**
- * Function that simply returns GNUNET_OK
- *
- * @param cls closure, not used
- * @param num_values not used
- * @param values not used
- * @return GNUNET_OK
- */
-static int
-return_ok (void *cls, 
-           unsigned int num_values, 
-           MYSQL_BIND *values)
-{
-  return GNUNET_OK;
-}
-
-
-/**
  * Get an estimate of how much space the database is
  * currently using.
  *
@@ -878,7 +802,7 @@ return_ok (void *cls,
  * @return number of bytes used on disk
  */
 static unsigned long long
-mysql_plugin_get_size (void *cls)
+mysql_plugin_estimate_size (void *cls)
 {
   struct Plugin *plugin = cls;
   MYSQL_BIND cbind[1];
@@ -893,7 +817,6 @@ mysql_plugin_get_size (void *cls)
       prepared_statement_run_select (plugin,
                                      plugin->get_size,
                                      1, cbind, 
-                                     &return_ok, NULL,
                                      -1))
     return 0;
   return total;
@@ -929,7 +852,6 @@ mysql_plugin_put (void *cls,
 {
   struct Plugin *plugin = cls;
   unsigned int irepl = replication;
-  unsigned int itype = type;
   unsigned int ipriority = priority;
   unsigned int ianonymity = anonymity;
   unsigned long long lexpiration = expiration.abs_value;
@@ -952,7 +874,7 @@ mysql_plugin_put (void *cls,
                               plugin->insert_entry,
                               NULL,
                               MYSQL_TYPE_LONG, &irepl, GNUNET_YES,
-                              MYSQL_TYPE_LONG, &itype, GNUNET_YES,
+                              MYSQL_TYPE_LONG, &type, GNUNET_YES,
                               MYSQL_TYPE_LONG, &ipriority, GNUNET_YES,
                               MYSQL_TYPE_LONG, &ianonymity, GNUNET_YES,
                               MYSQL_TYPE_LONGLONG, &lexpiration, GNUNET_YES,
@@ -1034,20 +956,23 @@ mysql_plugin_update (void *cls,
 }
 
 
-
-
 /**
- * Continuation of "mysql_next_request".
+ * Run the given select statement and call 'proc' on the resulting
+ * values (which must be in particular positions).
  *
- * @param next_cls the next context
- * @param tc the task context (unused)
+ * @param plugin the plugin handle
+ * @param stmt select statement to run
+ * @param proc function to call on result
+ * @param proc_cls closure for proc
+ * @param ... arguments to initialize stmt
  */
 static void 
-mysql_next_request_cont (void *next_cls,
-                         const struct GNUNET_SCHEDULER_TaskContext *tc)
+execute_select (struct Plugin *plugin,
+                struct GNUNET_MysqlStatementHandle *stmt,
+                PluginDatumProcessor proc, void *proc_cls,
+                ...)
 {
-  struct NextRequestClosure *nrc = next_cls;
-  struct Plugin *plugin;
+  va_list ap;
   int ret;
   unsigned int type;
   unsigned int priority;
@@ -1059,19 +984,10 @@ mysql_next_request_cont (void *next_cls,
   char value[GNUNET_DATASTORE_MAX_VALUE_SIZE];
   GNUNET_HashCode key;
   struct GNUNET_TIME_Absolute expiration;
-  MYSQL_BIND *rbind = nrc->rbind;
-
-  plugin = nrc->plugin;
-  plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
-  plugin->next_task_nc = NULL;
+  MYSQL_BIND rbind[7];
 
-  if (GNUNET_YES == nrc->end_it) 
-    goto END_SET;
-  GNUNET_assert (nrc->plugin->next_task == GNUNET_SCHEDULER_NO_TASK);
-  nrc->now = GNUNET_TIME_absolute_get ();
   hashSize = sizeof (GNUNET_HashCode);
-  memset (nrc->rbind, 0, sizeof (nrc->rbind));
-  rbind = nrc->rbind;
+  memset (rbind, 0, sizeof (rbind));
   rbind[0].buffer_type = MYSQL_TYPE_LONG;
   rbind[0].buffer = &type;
   rbind[0].is_unsigned = 1;
@@ -1096,16 +1012,28 @@ mysql_next_request_cont (void *next_cls,
   rbind[6].buffer = &uid;
   rbind[6].is_unsigned = 1;
 
-  if (GNUNET_OK != nrc->prep (nrc->prep_cls,
-                              nrc))
-    goto END_SET;
-  GNUNET_assert (nrc->plugin->next_task == GNUNET_SCHEDULER_NO_TASK);
+  va_start (ap, proc_cls);
+  ret = prepared_statement_run_select_va (plugin,
+                                          stmt,
+                                          7, rbind,
+                                          ap);
+  va_end (ap);
+  if (ret <= 0)
+    {
+      proc (proc_cls, 
+            NULL, 0, NULL, 0, 0, 0, 
+            GNUNET_TIME_UNIT_ZERO_ABS, 0);
+      return;
+    }
   GNUNET_assert (size <= sizeof(value));
   if ( (rbind[4].buffer_length != sizeof (GNUNET_HashCode)) ||
        (hashSize != sizeof (GNUNET_HashCode)) )
     {
       GNUNET_break (0);
-      goto END_SET;
+      proc (proc_cls, 
+            NULL, 0, NULL, 0, 0, 0, 
+            GNUNET_TIME_UNIT_ZERO_ABS, 0);
+      return;
     }     
 #if DEBUG_MYSQL
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1116,18 +1044,13 @@ mysql_next_request_cont (void *next_cls,
               anonymity,
               exp);
 #endif
+  GNUNET_assert (size < MAX_DATUM_SIZE);
   expiration.abs_value = exp;
-  ret = nrc->dviter (nrc->dviter_cls, 
-                     (nrc->one_shot == GNUNET_YES) ? NULL : nrc,
-                     &key,
-                     size, value,
-                     type, priority, anonymity, expiration,
-                     uid);
-  if (ret == GNUNET_SYSERR)
-    {
-      nrc->end_it = GNUNET_YES;
-      return;
-    }
+  ret = proc (proc_cls, 
+              &key,
+              size, value,
+              type, priority, anonymity, expiration,
+              uid);
   if (ret == GNUNET_NO)
     {
       do_delete_entry (plugin, uid);
@@ -1135,189 +1058,50 @@ mysql_next_request_cont (void *next_cls,
         plugin->env->duc (plugin->env->cls,
                           - size);
     }
-  if (nrc->one_shot == GNUNET_YES)
-    GNUNET_free (nrc);
-  return;
- END_SET:
-  /* call dviter with "end of set" */
-  GNUNET_assert (nrc->plugin->next_task == GNUNET_SCHEDULER_NO_TASK);
-  nrc->dviter (nrc->dviter_cls, 
-               NULL, NULL, 0, NULL, 0, 0, 0, 
-               GNUNET_TIME_UNIT_ZERO_ABS, 0);
-  GNUNET_assert (nrc->plugin->next_task == GNUNET_SCHEDULER_NO_TASK);
-  nrc->prep (nrc->prep_cls, NULL);
-  GNUNET_assert (nrc->plugin->next_task == GNUNET_SCHEDULER_NO_TASK);
-  GNUNET_free (nrc);
 }
 
 
-/**
- * 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 
-mysql_plugin_next_request (void *next_cls,
-                           int end_it)
-{
-  struct NextRequestClosure *nrc = next_cls;
-
-  if (GNUNET_YES == end_it)
-    nrc->end_it = GNUNET_YES;
-  nrc->plugin->next_task_nc = nrc;
-  nrc->plugin->next_task = GNUNET_SCHEDULER_add_now (&mysql_next_request_cont,
-                                                     nrc);
-}  
-
 
 /**
- * Context for 'get_statement_prepare'.
- */
-struct GetContext
-{
-  GNUNET_HashCode key;
-  GNUNET_HashCode vhash;
-
-  unsigned int prio;
-  unsigned int anonymity;
-  unsigned long long expiration;
-  unsigned long long vkey;
-  unsigned long long total;
-  unsigned int off;
-  unsigned int count;
-  int have_vhash;
-};
-
-
-static int
-get_statement_prepare (void *cls,
-                       struct NextRequestClosure *nrc)
-{
-  struct GetContext *gc = cls;
-  struct Plugin *plugin;
-  int ret;
-  unsigned long hashSize;
-  
-  if (NULL == nrc)
-    {
-      GNUNET_free (gc);
-      return GNUNET_NO;
-    }
-  if (gc->count == gc->total)
-    return GNUNET_NO;
-  plugin = nrc->plugin;
-  hashSize = sizeof (GNUNET_HashCode);
-  if (++gc->off >= gc->total)
-    gc->off = 0;
-#if DEBUG_MYSQL
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Obtaining result number %d/%lld at offset %u for GET `%s'\n",
-              gc->count+1,
-              gc->total,
-              gc->off,
-              GNUNET_h2s (&gc->key));  
-#endif
-  if (nrc->type != 0)
-    {
-      if (gc->have_vhash)
-        {
-          ret = prepared_statement_run_select (plugin,
-                                               plugin->select_entry_by_hash_vhash_and_type, 
-                                               7, nrc->rbind, 
-                                               &return_ok, NULL, 
-                                               MYSQL_TYPE_BLOB, &gc->key, hashSize, &hashSize,
-                                               MYSQL_TYPE_BLOB, &gc->vhash, hashSize, &hashSize,
-                                               MYSQL_TYPE_LONG, &nrc->type, GNUNET_YES, 
-                                               MYSQL_TYPE_LONG, &gc->off, GNUNET_YES,
-                                               -1);
-        }
-      else
-        {
-          ret =
-            prepared_statement_run_select (plugin,
-                                           plugin->select_entry_by_hash_and_type, 
-                                           7, nrc->rbind, 
-                                           &return_ok, NULL,
-                                           MYSQL_TYPE_BLOB, &gc->key, hashSize, &hashSize,
-                                           MYSQL_TYPE_LONG, &nrc->type, GNUNET_YES, 
-                                           MYSQL_TYPE_LONG, &gc->off, GNUNET_YES,
-                                           -1);
-        }
-    }
-  else
-    {
-      if (gc->have_vhash)
-        {
-          ret =
-            prepared_statement_run_select (plugin,
-                                           plugin->select_entry_by_hash_and_vhash, 
-                                           7, nrc->rbind, 
-                                           &return_ok, NULL,
-                                           MYSQL_TYPE_BLOB, &gc->key, hashSize, &hashSize, 
-                                           MYSQL_TYPE_BLOB, &gc->vhash, hashSize, &hashSize, 
-                                           MYSQL_TYPE_LONG, &gc->off, GNUNET_YES, 
-                                           -1);
-        }
-      else
-        {
-          ret =
-            prepared_statement_run_select (plugin,
-                                           plugin->select_entry_by_hash, 
-                                           7, nrc->rbind, 
-                                           &return_ok, NULL,
-                                           MYSQL_TYPE_BLOB, &gc->key, hashSize, &hashSize,
-                                           MYSQL_TYPE_LONG, &gc->off, GNUNET_YES, 
-                                           -1);
-        }
-    }
-  gc->count++;
-  return ret;
-}
-
-
-/**
- * Iterate over the results for a particular key in the datastore.
+ * Get one of the results for a particular key in the datastore.
  *
  * @param cls closure
- * @param key maybe NULL (to match all entries)
+ * @param offset offset of the result (mod #num-results); 
+ *               specific ordering does not matter for the offset
+ * @param key key to match, never NULL
  * @param vhash hash of the value, maybe NULL (to
  *        match all values that have the right key).
  *        Note that for DBlocks there is no difference
  *        betwen key and vhash, but for other blocks
  *        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_cls closure for iter
+ *     Use 0 for any type.
+ * @param proc function to call on each matching value; however,
+ *        after the first call to "proc", the plugin must wait
+ *        until "NextRequest" was called before giving the processor
+ *        the next item; finally, the "proc" should be called once
+ *        once with a NULL value at the end ("next_cls" should be NULL
+ *        for that last call)
+ * @param proc_cls closure for proc
  */
 static void
-mysql_plugin_get (void *cls,
-                  const GNUNET_HashCode *key,
-                  const GNUNET_HashCode *vhash,
-                  enum GNUNET_BLOCK_Type type,
-                  PluginIterator iter, void *iter_cls)
+mysql_plugin_get_key (void *cls,
+                      uint64_t offset,
+                      const GNUNET_HashCode *key,
+                      const GNUNET_HashCode *vhash,
+                      enum GNUNET_BLOCK_Type type,                    
+                      PluginDatumProcessor proc, void *proc_cls)
 {
   struct Plugin *plugin = cls;
-  unsigned int itype = type;
   int ret;
   MYSQL_BIND cbind[1];
-  struct GetContext *gc;
-  struct NextRequestClosure *nrc;
   long long total;
   unsigned long hashSize;
   unsigned long hashSize2;
+  unsigned long long off;
 
   GNUNET_assert (key != NULL);
-  if (iter == NULL) 
-    return;
+  GNUNET_assert (NULL != proc);
   hashSize = sizeof (GNUNET_HashCode);
   hashSize2 = sizeof (GNUNET_HashCode);
   memset (cbind, 0, sizeof (cbind));
@@ -1333,10 +1117,9 @@ mysql_plugin_get (void *cls,
             prepared_statement_run_select (plugin,
                                            plugin->count_entry_by_hash_vhash_and_type, 
                                            1, cbind, 
-                                           &return_ok, NULL,
                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize, 
                                            MYSQL_TYPE_BLOB, vhash, hashSize2, &hashSize2, 
-                                           MYSQL_TYPE_LONG, &itype, GNUNET_YES,
+                                           MYSQL_TYPE_LONG, &type, GNUNET_YES,
                                            -1);
         }
       else
@@ -1345,9 +1128,8 @@ mysql_plugin_get (void *cls,
             prepared_statement_run_select (plugin,
                                            plugin->count_entry_by_hash_and_type, 
                                            1, cbind, 
-                                           &return_ok, NULL,
                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize, 
-                                           MYSQL_TYPE_LONG, &itype, GNUNET_YES,
+                                           MYSQL_TYPE_LONG, &type, GNUNET_YES,
                                            -1);
         }
     }
@@ -1359,7 +1141,6 @@ mysql_plugin_get (void *cls,
             prepared_statement_run_select (plugin,
                                            plugin->count_entry_by_hash_and_vhash, 
                                            1, cbind,
-                                           &return_ok, NULL,
                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize, 
                                            MYSQL_TYPE_BLOB, vhash, hashSize2, &hashSize2, 
                                            -1);
@@ -1371,79 +1152,81 @@ mysql_plugin_get (void *cls,
             prepared_statement_run_select (plugin,
                                            plugin->count_entry_by_hash,
                                            1, cbind, 
-                                           &return_ok, NULL, 
                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize, 
                                            -1);
         }
     }
   if ((ret != GNUNET_OK) || (0 >= total))
     {
-      iter (iter_cls, 
-            NULL, NULL, 0, NULL, 0, 0, 0, 
+      proc (proc_cls, 
+            NULL, 0, NULL, 0, 0, 0, 
             GNUNET_TIME_UNIT_ZERO_ABS, 0);
       return;
     }
+  offset = offset % total;
+  off = (unsigned long long) offset;
 #if DEBUG_MYSQL
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Iterating over %lld results for GET `%s'\n",
+              "Obtaining %llu/%lld result for GET `%s'\n",
+              off,
               total,
               GNUNET_h2s (key));
 #endif
-  gc = GNUNET_malloc (sizeof (struct GetContext));
-  gc->key = *key;
-  if (vhash != NULL)
+
+  if (type != GNUNET_BLOCK_TYPE_ANY)
     {
-      gc->have_vhash = GNUNET_YES;
-      gc->vhash = *vhash;
+      if (NULL != vhash)
+        {
+          execute_select (plugin,
+                          plugin->select_entry_by_hash_vhash_and_type, 
+                          proc, proc_cls,
+                          MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
+                          MYSQL_TYPE_BLOB, vhash, hashSize, &hashSize,
+                          MYSQL_TYPE_LONG, &type, GNUNET_YES, 
+                          MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
+                          -1);
+        }
+      else
+        {
+          execute_select (plugin,
+                          plugin->select_entry_by_hash_and_type, 
+                          proc, proc_cls,
+                          MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
+                          MYSQL_TYPE_LONG, &type, GNUNET_YES, 
+                          MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
+                          -1);
+        }
+    }
+  else
+    {
+      if (NULL != vhash)
+        {
+          execute_select (plugin,
+                          plugin->select_entry_by_hash_and_vhash, 
+                          proc, proc_cls,
+                          MYSQL_TYPE_BLOB, key, hashSize, &hashSize, 
+                          MYSQL_TYPE_BLOB, vhash, hashSize, &hashSize, 
+                          MYSQL_TYPE_LONGLONG, &off, GNUNET_YES, 
+                          -1);
+        }
+      else
+        {
+          execute_select (plugin,
+                          plugin->select_entry_by_hash, 
+                          proc, proc_cls,
+                          MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
+                          MYSQL_TYPE_LONGLONG, &off, GNUNET_YES, 
+                          -1);
+        }
     }
-  gc->total = total;
-  gc->off = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, total);
-  
-
-  nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
-  nrc->plugin = plugin;
-  nrc->type = type;  
-  nrc->dviter = iter;
-  nrc->dviter_cls = iter_cls;
-  nrc->prep = &get_statement_prepare;
-  nrc->prep_cls = gc;
-  mysql_plugin_next_request (nrc, GNUNET_NO);
-}
-
-
-/**
- * Run the prepared statement to get the next data item ready.
- * 
- * @param cls not used
- * @param nrc closure for the next request iterator
- * @return GNUNET_OK on success, GNUNET_NO if there is no additional item
- */
-static int
-iterator_zero_prepare (void *cls,
-                       struct NextRequestClosure *nrc)
-{
-  struct Plugin *plugin;
-  int ret;
-
-  if (nrc == NULL)
-    return GNUNET_NO;
-  plugin = nrc->plugin;
-  ret = prepared_statement_run_select (plugin,
-                                       plugin->zero_iter,
-                                       7, nrc->rbind,
-                                       &return_ok, NULL,
-                                       MYSQL_TYPE_LONG, &nrc->count, GNUNET_YES,
-                                       -1);
-  nrc->count++;
-  return ret;
 }
 
 
 /**
- * Select a subset of the items in the datastore and call
- * the given iterator for each of them.
+ * Get a zero-anonymity datum from the datastore.
  *
  * @param cls our "struct Plugin*"
+ * @param offset offset of the result
  * @param type entries of which type should be considered?
  *        Use 0 for any type.
  * @param iter function to call on each matching value;
@@ -1451,47 +1234,27 @@ iterator_zero_prepare (void *cls,
  * @param iter_cls closure for iter
  */
 static void
-mysql_plugin_iter_zero_anonymity (void *cls,
-                                  enum GNUNET_BLOCK_Type type,
-                                  PluginIterator iter,
-                                  void *iter_cls)
+mysql_plugin_get_zero_anonymity (void *cls,
+                                 uint64_t offset,
+                                 enum GNUNET_BLOCK_Type type,
+                                 PluginDatumProcessor proc, void *proc_cls)
 {
   struct Plugin *plugin = cls;
-  struct NextRequestClosure *nrc;
-
-  nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
-  nrc->plugin = plugin;
-  nrc->type = type;  
-  nrc->dviter = iter;
-  nrc->dviter_cls = iter_cls;
-  nrc->prep = &iterator_zero_prepare;
-  mysql_plugin_next_request (nrc, GNUNET_NO);
-}
+  unsigned long long off;
 
+  off = (unsigned long long) offset;
+  execute_select (plugin,
+                  plugin->zero_iter,
+                  proc, proc_cls,
+                  MYSQL_TYPE_LONG, &type, GNUNET_YES,
+                  MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
+                  -1);
 
-/**
- * Run the SELECT statement for the replication function.
- * 
- * @param cls the 'struct Plugin'
- * @param nrc the context (not used)
- */
-static int
-replication_prepare (void *cls,
-                     struct NextRequestClosure *nrc)
-{
-  struct Plugin *plugin = cls;
-
-  return prepared_statement_run_select (plugin,
-                                        plugin->select_replication, 
-                                        7, nrc->rbind, 
-                                        &return_ok, NULL,
-                                        -1);
 }
 
 
-
 /**
- * Context for 'repl_iter' function.
+ * Context for 'repl_proc' function.
  */
 struct ReplCtx
 {
@@ -1504,22 +1267,21 @@ struct ReplCtx
   /**
    * Function to call for the result (or the NULL).
    */
-  PluginIterator iter;
+  PluginDatumProcessor proc;
   
   /**
-   * Closure for iter.
+   * Closure for proc.
    */
-  void *iter_cls;
+  void *proc_cls;
 };
 
 
 /**
- * Wrapper for the iterator for 'sqlite_plugin_replication_get'.
+ * Wrapper for the processor for 'mysql_plugin_get_replication'.
  * Decrements the replication counter and calls the original
  * iterator.
  *
  * @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
@@ -1535,8 +1297,7 @@ struct ReplCtx
  *         GNUNET_NO to delete the item and continue (if supported)
  */
 static int
-repl_iter (void *cls,
-           void *next_cls,
+repl_proc (void *cls,
            const GNUNET_HashCode *key,
            uint32_t size,
            const void *data,
@@ -1552,8 +1313,8 @@ repl_iter (void *cls,
   int ret;
   int iret;
 
-  ret = rc->iter (rc->iter_cls,
-                  next_cls, key,
+  ret = rc->proc (rc->proc_cls,
+                  key,
                   size, data, 
                   type, priority, anonymity, expiration,
                   uid);
@@ -1561,10 +1322,10 @@ repl_iter (void *cls,
     {
       oid = (unsigned long long) uid;
       iret = prepared_statement_run (plugin,
-                                    plugin->dec_repl,
-                                    NULL,
-                                    MYSQL_TYPE_LONGLONG, &oid, GNUNET_YES, 
-                                    -1);
+                                     plugin->dec_repl,
+                                     NULL,
+                                     MYSQL_TYPE_LONGLONG, &oid, GNUNET_YES, 
+                                     -1);
       if (iret == GNUNET_SYSERR)
         {
           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
@@ -1577,94 +1338,56 @@ repl_iter (void *cls,
 
 
 /**
- * Get a random item for replication.  Returns a single, not expired, random item
- * from those with the highest replication counters.  The item's 
- * replication counter is decremented by one IF it was positive before.
- * Call 'iter' with all values ZERO or NULL if the datastore is empty.
+ * Get a random item for replication.  Returns a single, not expired,
+ * random item from those with the highest replication counters.  The
+ * item's replication counter is decremented by one IF it was positive
+ * before.  Call 'proc' with all values ZERO or NULL if the datastore
+ * is empty.
  *
  * @param cls closure
- * @param iter function to call the value (once only).
- * @param iter_cls closure for iter
+ * @param proc function to call the value (once only).
+ * @param iter_cls closure for proc
  */
 static void
-mysql_plugin_replication_get (void *cls,
-                              PluginIterator iter, void *iter_cls)
+mysql_plugin_get_replication (void *cls,
+                              PluginDatumProcessor proc, void *proc_cls)
 {
   struct Plugin *plugin = cls;
-  struct NextRequestClosure *nrc;
   struct ReplCtx rc;
   
   rc.plugin = plugin;
-  rc.iter = iter;
-  rc.iter_cls = iter_cls;
-  nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
-  nrc->plugin = plugin;
-  nrc->now = GNUNET_TIME_absolute_get ();
-  nrc->prep = &replication_prepare;
-  nrc->prep_cls = plugin;
-  nrc->type = 0;
-  nrc->dviter = &repl_iter;
-  nrc->dviter_cls = &rc;
-  nrc->end_it = GNUNET_NO;
-  nrc->one_shot = GNUNET_YES;
-  mysql_next_request_cont (nrc, NULL);
-}
-
+  rc.proc = proc;
+  rc.proc_cls = proc_cls;
+  execute_select (plugin,
+                  plugin->select_replication, 
+                  &repl_proc, &rc,
+                  -1);
 
-/**
- * Run the SELECT statement for the expiration function.
- * 
- * @param cls the 'struct Plugin'
- * @param nrc the context (not used)
- * @return GNUNET_OK on success, GNUNET_NO if there are
- *         no more values, GNUNET_SYSERR on error
- */
-static int
-expiration_prepare (void *cls,
-                    struct NextRequestClosure *nrc)
-{
-  struct Plugin *plugin = cls;
-  long long nt;
-
-  if (NULL == nrc)
-    return GNUNET_NO;
-  nt = (long long) nrc->now.abs_value;
-  return prepared_statement_run_select
-    (plugin,
-     plugin->select_expiration, 
-     7, nrc->rbind, 
-     &return_ok, NULL,
-     MYSQL_TYPE_LONGLONG, &nt, GNUNET_YES, 
-     -1);
 }
 
 
 /**
  * Get a random item for expiration.
- * Call 'iter' with all values ZERO or NULL if the datastore is empty.
+ * Call 'proc' with all values ZERO or NULL if the datastore is empty.
  *
  * @param cls closure
- * @param iter function to call the value (once only).
- * @param iter_cls closure for iter
+ * @param proc function to call the value (once only).
+ * @param proc_cls closure for proc
  */
 static void
-mysql_plugin_expiration_get (void *cls,
-                             PluginIterator iter, void *iter_cls)
+mysql_plugin_get_expiration (void *cls,
+                             PluginDatumProcessor proc, void *proc_cls)
 {
   struct Plugin *plugin = cls;
-  struct NextRequestClosure *nrc;
-
-  nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
-  nrc->plugin = plugin;
-  nrc->now = GNUNET_TIME_absolute_get ();
-  nrc->prep = &expiration_prepare;
-  nrc->prep_cls = plugin;
-  nrc->type = 0;
-  nrc->dviter = iter;
-  nrc->dviter_cls = iter_cls;
-  nrc->end_it = GNUNET_NO;
-  nrc->one_shot = GNUNET_YES;
-  mysql_next_request_cont (nrc, NULL);
+  long long nt;
+
+  nt = (long long) GNUNET_TIME_absolute_get().abs_value;
+  execute_select (plugin,
+                  plugin->select_expiration, 
+                  proc, proc_cls,
+                  MYSQL_TYPE_LONGLONG, &nt, GNUNET_YES, 
+                  -1);
+
 }
 
 
@@ -1760,14 +1483,13 @@ libgnunet_plugin_datastore_mysql_init (void *cls)
 
   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
   api->cls = plugin;
-  api->get_size = &mysql_plugin_get_size;
+  api->estimate_size = &mysql_plugin_estimate_size;
   api->put = &mysql_plugin_put;
-  api->next_request = &mysql_plugin_next_request;
-  api->get = &mysql_plugin_get;
-  api->replication_get = &mysql_plugin_replication_get;
-  api->expiration_get = &mysql_plugin_expiration_get;
   api->update = &mysql_plugin_update;
-  api->iter_zero_anonymity = &mysql_plugin_iter_zero_anonymity;
+  api->get_key = &mysql_plugin_get_key;
+  api->get_replication = &mysql_plugin_get_replication;
+  api->get_expiration = &mysql_plugin_get_expiration;
+  api->get_zero_anonymity = &mysql_plugin_get_zero_anonymity;
   api->drop = &mysql_plugin_drop;
   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
                    "mysql", _("Mysql database running\n"));
@@ -1787,14 +1509,6 @@ libgnunet_plugin_datastore_mysql_done (void *cls)
   struct Plugin *plugin = api->cls;
 
   iclose (plugin);
-  if (plugin->next_task != GNUNET_SCHEDULER_NO_TASK)
-    {
-      GNUNET_SCHEDULER_cancel (plugin->next_task);
-      plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
-      plugin->next_task_nc->prep (plugin->next_task_nc->prep_cls, NULL);
-      GNUNET_free (plugin->next_task_nc);
-      plugin->next_task_nc = NULL;
-    }
   GNUNET_free_non_null (plugin->cnffile);
   GNUNET_free (plugin);
   GNUNET_free (api);