[GNUnet-SVN] r8739 - gnunet/src/datastore

[gnunet]

Author: grothoff
Date: 2009-07-17 15:09:54 -0600 (Fri, 17 Jul 2009)
New Revision: 8739

Modified:
   gnunet/src/datastore/perf_plugin_datastore.c
   gnunet/src/datastore/plugin_datastore_sqlite.c
Log:
implementing all-now iterator

Modified: gnunet/src/datastore/perf_plugin_datastore.c
===================================================================
--- gnunet/src/datastore/perf_plugin_datastore.c        2009-07-17 20:14:47 UTC 
(rev 8738)
+++ gnunet/src/datastore/perf_plugin_datastore.c        2009-07-17 21:09:54 UTC 
(rev 8739)
@@ -28,7 +28,7 @@
 #include "gnunet_protocols.h"
 #include "plugin_datastore.h"
 
-#define VERBOSE GNUNET_YES
+#define VERBOSE GNUNET_NO
 
 /**
  * Target datastore size (in bytes).  Realistic sizes are
@@ -36,10 +36,9 @@
  * those take too long to run them in the usual "make check"
  * sequence.  Hence the value used for shipping is tiny.
  */
-#define MAX_SIZE 1024LL * 1024
-/* * 128 */
+#define MAX_SIZE 1024LL * 1024 * 128
 
-#define ITERATIONS 2
+#define ITERATIONS 10
 
 /**
  * Number of put operations equivalent to 1/10th of MAX_SIZE
@@ -76,6 +75,7 @@
   struct GNUNET_DATASTORE_PluginFunctions * api;
   const char *msg;
   enum RunPhase phase;
+  unsigned int cnt;
 };
 
 
@@ -152,7 +152,8 @@
       crc->end = GNUNET_TIME_absolute_get();
       printf (crc->msg,
              crc->i,
-             (unsigned long long) (crc->end.value - crc->start.value));
+             (unsigned long long) (crc->end.value - crc->start.value),
+             crc->cnt);
       if (crc->phase != RP_AN_GET)
        {
          crc->phase++;
@@ -171,7 +172,7 @@
                                  &test, crc);
       return GNUNET_OK;
     }
-  fprintf (stderr, ".");
+  crc->cnt++;
   crc->api->next_request (next_cls,
                          GNUNET_NO);
   return GNUNET_OK;
@@ -238,8 +239,10 @@
       for (j=0;j<PUT_10;j++)
        putValue (crc->api, j, crc->i);
       crc->end = GNUNET_TIME_absolute_get ();
-      printf ("%3u insertion              took %20llums\n", crc->i,
-             (unsigned long long) (crc->end.value - crc->start.value));
+      printf ("%3u insertion took                      %20llums for %u\n",
+             crc->i,
+             (unsigned long long) (crc->end.value - crc->start.value),
+             (unsigned int) PUT_10);
       crc->i++;
       crc->phase = RP_LP_GET;
       GNUNET_SCHEDULER_add_after (crc->sched,
@@ -249,36 +252,41 @@
                                  &test, crc);
       break;
     case RP_LP_GET:
+      crc->cnt = 0;
       crc->start = GNUNET_TIME_absolute_get ();      
-      crc->msg = "%3u low priority iteration took %20llums\n";
+      crc->msg = "%3u low priority iteration took         %20llums for %u\n";
       crc->api->iter_low_priority (crc->api->cls, 0, 
                                   &iterateDummy,
                                   crc);
       break;
     case RP_AE_GET:
+      crc->cnt = 0;
       crc->start = GNUNET_TIME_absolute_get ();      
-      crc->msg = "%3u ascending expiration iteration took %20llums\n";
+      crc->msg = "%3u ascending expiration iteration took %20llums for %u\n";
       crc->api->iter_ascending_expiration (crc->api->cls, 0, 
                                      &iterateDummy,
                                      crc);
       break;
     case RP_ZA_GET:
+      crc->cnt = 0;
       crc->start = GNUNET_TIME_absolute_get ();      
-      crc->msg = "%3u zero anonymity iteration took %20llums\n";
+      crc->msg = "%3u zero anonymity iteration took       %20llums for %u\n";
       crc->api->iter_zero_anonymity (crc->api->cls, 0, 
                                     &iterateDummy,
                                     crc);
       break;
     case RP_MO_GET:
+      crc->cnt = 0;
       crc->start = GNUNET_TIME_absolute_get ();      
-      crc->msg = "%3u migration order iteration took %20llums\n";
+      crc->msg = "%3u migration order iteration took      %20llums for %u\n";
       crc->api->iter_migration_order (crc->api->cls, 0, 
                                      &iterateDummy,
                                      crc);
       break;
     case RP_AN_GET:
+      crc->cnt = 0;
       crc->start = GNUNET_TIME_absolute_get ();      
-      crc->msg = "%3u all now iteration took %20llums\n";
+      crc->msg = "%3u all now iteration took              %20llums for %u\n";
       crc->api->iter_all_now (crc->api->cls, 0,
                              &iterateDummy,
                              crc);

Modified: gnunet/src/datastore/plugin_datastore_sqlite.c
===================================================================
--- gnunet/src/datastore/plugin_datastore_sqlite.c      2009-07-17 20:14:47 UTC 
(rev 8738)
+++ gnunet/src/datastore/plugin_datastore_sqlite.c      2009-07-17 21:09:54 UTC 
(rev 8739)
@@ -29,7 +29,7 @@
 #include "plugin_datastore.h"
 #include <sqlite3.h>
 
-#define DEBUG_SQLITE GNUNET_YES
+#define DEBUG_SQLITE GNUNET_NO
 
 /**
  * After how many payload-changing operations
@@ -431,7 +431,7 @@
  *         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
+ *         no more values, GNUNET_SYSERR on error
  */
 typedef int (*PrepareFunction)(void *cls,
                               struct NextContext *nc);
@@ -480,6 +480,11 @@
   unsigned long long last_rowid;
 
   /**
+   * Key of the last result.
+   */
+  GNUNET_HashCode lastKey;  
+
+  /**
    * Expiration time of the last value visited.
    */
   struct GNUNET_TIME_Absolute lastExpiration;
@@ -502,23 +507,16 @@
 
 
 /**
- * Function invoked on behalf of a "PluginIterator"
- * asking the database plugin to call the iterator
- * with the next item.
+ * Continuation of "sqlite_next_request".
  *
- * @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).
+ * @param cls the next context
  */
 static void 
-sqlite_next_request (void *next_cls,
-                    int end_it)
+sqlite_next_request_cont (void *cls,
+                         const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   static struct GNUNET_TIME_Absolute zero;
-  struct NextContext * nc= next_cls;
+  struct NextContext * nc= cls;
   struct Plugin *plugin;
   unsigned long long rowid;
   sqlite3_stmt *stmtd;
@@ -531,13 +529,11 @@
   const GNUNET_HashCode *key;
   const void *data;
 
+ 
   plugin = nc->plugin;
-  sqlite3_reset (nc->stmt);
-  if ( (GNUNET_YES == end_it) ||
-       (GNUNET_YES == nc->end_it) ||
+  if ( (GNUNET_YES == nc->end_it) ||
        (GNUNET_OK != (nc->prep(nc->prep_cls,
-                              nc))) ||
-       (SQLITE_ROW != sqlite3_step (nc->stmt)) )
+                              nc))) )
     {
     END:
       nc->iter (nc->iter_cls, 
@@ -589,11 +585,12 @@
     }
 
   priority = sqlite3_column_int (nc->stmt, 2);
-  nc->lastPriority = priority;
   anonymity = sqlite3_column_int (nc->stmt, 3);
   expiration.value = sqlite3_column_int64 (nc->stmt, 4);
+  key = sqlite3_column_blob (nc->stmt, 5);
+  nc->lastPriority = priority;
   nc->lastExpiration = expiration;
-  key = sqlite3_column_blob (nc->stmt, 5);
+  memcpy (&nc->lastKey, key, sizeof(GNUNET_HashCode));
   data = sqlite3_column_blob (nc->stmt, 6);
   nc->count++;
   ret = nc->iter (nc->iter_cls,
@@ -623,6 +620,35 @@
 
 
 /**
+ * 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)
+{
+  struct NextContext * nc= next_cls;
+
+  if (GNUNET_YES == end_it)
+    nc->end_it = GNUNET_YES;
+  GNUNET_SCHEDULER_add_continuation (nc->plugin->env->sched,
+                                    GNUNET_NO,                              
+                                    &sqlite_next_request_cont,
+                                    nc,
+                                    GNUNET_SCHEDULER_REASON_PREREQ_DONE);
+}
+
+
+
+/**
  * Store an item in the datastore.
  *
  * @param cls closure
@@ -776,7 +802,6 @@
   int is_migr;
   int limit_nonanonymous;
   uint32_t type;
-  GNUNET_HashCode key;  
 };
 
 
@@ -790,27 +815,52 @@
 
   if (nc == NULL)
     {
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Asked to clean up iterator state.\n");
+#endif
       sqlite3_finalize (ic->stmt_1);
       sqlite3_finalize (ic->stmt_2);
       return GNUNET_SYSERR;
     }
+  sqlite3_reset (ic->stmt_1);
+  sqlite3_reset (ic->stmt_2);
   plugin = nc->plugin;
   if (ic->is_prio)
     {
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Restricting to results larger than the last priority %u\n",
+                 nc->lastPriority);
+#endif
       sqlite3_bind_int (ic->stmt_1, 1, nc->lastPriority);
       sqlite3_bind_int (ic->stmt_2, 1, nc->lastPriority);
     }
   else
     {
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Restricting to results larger than the last expiration 
%llu\n",
+                 (unsigned long long) nc->lastExpiration.value);
+#endif
       sqlite3_bind_int64 (ic->stmt_1, 1, nc->lastExpiration.value);
       sqlite3_bind_int64 (ic->stmt_2, 1, nc->lastExpiration.value);
     }
+#if DEBUG_SQLITE
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+             "Restricting to results larger than the last key `%s'\n",
+             GNUNET_h2s(&nc->lastKey));
+#endif
   sqlite3_bind_blob (ic->stmt_1, 2, 
-                    &ic->key, 
+                    &nc->lastKey, 
                     sizeof (GNUNET_HashCode),
                     SQLITE_TRANSIENT);
   if (SQLITE_ROW == (ret = sqlite3_step (ic->stmt_1)))
-    {
+    {      
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Result found using iterator 1\n");
+#endif
       nc->stmt = ic->stmt_1;
       return GNUNET_OK;
     }
@@ -829,6 +879,10 @@
                "sqlite3_reset");
   if (SQLITE_ROW == (ret = sqlite3_step (ic->stmt_2))) 
     {
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Result found using iterator 2\n");
+#endif
       nc->stmt = ic->stmt_2;
       return GNUNET_OK;
     }
@@ -915,13 +969,13 @@
     {
       nc->lastPriority = 0;
       nc->lastExpiration.value = 0;
-      memset (&ic->key, 0, sizeof (GNUNET_HashCode));
+      memset (&nc->lastKey, 0, sizeof (GNUNET_HashCode));
     }
   else
     {
       nc->lastPriority = 0x7FFFFFFF;
       nc->lastExpiration.value = 0x7FFFFFFFFFFFFFFFLL;
-      memset (&ic->key, 255, sizeof (GNUNET_HashCode));
+      memset (&nc->lastKey, 255, sizeof (GNUNET_HashCode));
     }
   sqlite_next_request (nc, GNUNET_NO);
 }
@@ -974,9 +1028,9 @@
   char *q2;
 
   now = GNUNET_TIME_absolute_get ();
-  GNUNET_asprintf (&q1, SELECT_IT_EXPIRATION_TIME_1,
+  GNUNET_asprintf (&q1, SELECT_IT_NON_ANONYMOUS_1,
                   now.value);
-  GNUNET_asprintf (&q2, SELECT_IT_EXPIRATION_TIME_2,
+  GNUNET_asprintf (&q2, SELECT_IT_NON_ANONYMOUS_2,
                   now.value);
   basic_iter (cls,
              type, 
@@ -1060,7 +1114,42 @@
 }
 
 
+static int
+all_next_prepare (void *cls,
+                 struct NextContext *nc)
+{
+  struct Plugin *plugin;
+  int ret;
 
+  if (nc == NULL)
+    {
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Asked to clean up iterator state.\n");
+#endif
+      return GNUNET_SYSERR;
+    }
+  plugin = nc->plugin;
+  if (SQLITE_ROW == (ret = sqlite3_step (nc->stmt)))
+    {      
+#if DEBUG_SQLITE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Result found\n");
+#endif
+      return GNUNET_OK;
+    }
+  if (ret != SQLITE_DONE)
+    {
+      LOG_SQLITE (plugin, NULL,
+                 GNUNET_ERROR_TYPE_ERROR |
+                 GNUNET_ERROR_TYPE_BULK,
+                 "sqlite3_step");
+      return GNUNET_SYSERR;
+    }
+  return GNUNET_NO;
+}
+
+
 /**
  * Select a subset of the items in the datastore and call
  * the given iterator for each of them.
@@ -1078,7 +1167,28 @@
                            void *iter_cls)
 {
   static struct GNUNET_TIME_Absolute zero;
-  iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
+  struct Plugin *plugin = cls;
+  struct NextContext *nc;
+  sqlite3_stmt *stmt;
+
+  if (sq_prepare (plugin->dbh, 
+                 "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ 
FROM gn080",
+                 &stmt) != SQLITE_OK)
+    {
+      LOG_SQLITE (plugin, NULL,
+                  GNUNET_ERROR_TYPE_ERROR |
+                  GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare");
+      iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
+      return;
+    }
+  nc = GNUNET_malloc (sizeof(struct NextContext));
+  nc->plugin = plugin;
+  nc->iter = iter;
+  nc->iter_cls = iter_cls;
+  nc->stmt = stmt;
+  nc->prep = &all_next_prepare;
+  nc->prep_cls = NULL;
+  sqlite_next_request (nc, GNUNET_NO);
 }
 
 
@@ -1117,6 +1227,7 @@
   else
     limit_off = 0;
   sqoff = 1;
+  sqlite3_reset (nc->stmt);
   ret = sqlite3_bind_blob (nc->stmt,
                           sqoff++,
                           &gnc->key,