src/datastore/plugin_datastore_sqlite.c

   1  /*
   2      This file is part of GNUnet
   3      (C) 2009 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 2, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file datastore/plugin_datastore_sqlite.c
  23  * @brief sqlite-based datastore backend
  24  * @author Christian Grothoff
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_arm_service.h"
  29 #include "gnunet_statistics_service.h"
  30 #include "plugin_datastore.h"
  31 #include <sqlite3.h>
  32
  33 #define DEBUG_SQLITE GNUNET_YES
  34
  35 /**
  36  * After how many payload-changing operations
  37  * do we sync our statistics?
  38  */
  39 #define MAX_STAT_SYNC_LAG 50
  40
  41 #define QUOTA_STAT_NAME gettext_noop ("file-sharing datastore utilization (in bytes)")
  42
  43 /**
  44  * Log an error message at log-level 'level' that indicates
  45  * a failure of the command 'cmd' on file 'filename'
  46  * with the message given by strerror(errno).
  47  */
  48 #define LOG_SQLITE(db, msg, level, cmd) do { GNUNET_log_from (level, "sqlite", _("`%s' failed at %s:%d with error: %s\n"), cmd, __FILE__, __LINE__, sqlite3_errmsg(db->dbh)); if (msg != NULL) GNUNET_asprintf(msg, _("`%s' failed with error: %s"), cmd, sqlite3_errmsg(db->dbh)); } while(0)
  49
  50 #define SELECT_IT_LOW_PRIORITY_1 \
  51   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio = ? AND hash > ?) "\
  52   "ORDER BY hash ASC LIMIT 1"
  53
  54 #define SELECT_IT_LOW_PRIORITY_2 \
  55   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio > ?) "\
  56   "ORDER BY prio ASC, hash ASC LIMIT 1"
  57
  58 #define SELECT_IT_NON_ANONYMOUS_1 \
  59   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio = ? AND hash < ? AND anonLevel = 0 AND expire > %llu) "\
  60   " ORDER BY hash DESC LIMIT 1"
  61
  62 #define SELECT_IT_NON_ANONYMOUS_2 \
  63   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio < ? AND anonLevel = 0 AND expire > %llu)"\
  64   " ORDER BY prio DESC, hash DESC LIMIT 1"
  65
  66 #define SELECT_IT_EXPIRATION_TIME_1 \
  67   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire = ? AND hash > ?) "\
  68   " ORDER BY hash ASC LIMIT 1"
  69
  70 #define SELECT_IT_EXPIRATION_TIME_2 \
  71   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire > ?) "\
  72   " ORDER BY expire ASC, hash ASC LIMIT 1"
  73
  74 #define SELECT_IT_MIGRATION_ORDER_1 \
  75   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire = ? AND hash < ?) "\
  76   " ORDER BY hash DESC LIMIT 1"
  77
  78 #define SELECT_IT_MIGRATION_ORDER_2 \
  79   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire < ? AND expire > %llu) "\
  80   " ORDER BY expire DESC, hash DESC LIMIT 1"
  81
  82 /**
  83  * After how many ms "busy" should a DB operation fail for good?
  84  * A low value makes sure that we are more responsive to requests
  85  * (especially PUTs).  A high value guarantees a higher success
  86  * rate (SELECTs in iterate can take several seconds despite LIMIT=1).
  87  *
  88  * The default value of 250ms should ensure that users do not experience
  89  * huge latencies while at the same time allowing operations to succeed
  90  * with reasonable probability.
  91  */
  92 #define BUSY_TIMEOUT_MS 250
  93
  94
  95 /**
  96  * Context for all functions in this plugin.
  97  */
  98 struct Plugin
  99 {
 100   /**
 101    * Our execution environment.
 102    */
 103   struct GNUNET_DATASTORE_PluginEnvironment *env;
 104
 105   /**
 106    * Database filename.
 107    */
 108   char *fn;
 109
 110   /**
 111    * Native SQLite database handle.
 112    */
 113   sqlite3 *dbh;
 114
 115   /**
 116    * Precompiled SQL for update.
 117    */
 118   sqlite3_stmt *updPrio;
 119
 120   /**
 121    * Precompiled SQL for insertion.
 122    */
 123   sqlite3_stmt *insertContent;
 124
 125   /**
 126    * Handle to the statistics service.
 127    */
 128   struct GNUNET_STATISTICS_Handle *statistics;
 129
 130   /**
 131    * How much data are we currently storing
 132    * in the database?
 133    */
 134   unsigned long long payload;
 135
 136   /**
 137    * Number of updates that were made to the
 138    * payload value since we last synchronized
 139    * it with the statistics service.
 140    */
 141   unsigned int lastSync;
 142
 143   /**
 144    * Should the database be dropped on shutdown?
 145    */
 146   int drop_on_shutdown;
 147 };
 148
 149
 150 /**
 151  * @brief Prepare a SQL statement
 152  *
 153  * @param dbh handle to the database
 154  * @param zSql SQL statement, UTF-8 encoded
 155  * @param ppStmt set to the prepared statement
 156  * @return 0 on success
 157  */
 158 static int
 159 sq_prepare (sqlite3 * dbh, const char *zSql,
 160             sqlite3_stmt ** ppStmt)
 161 {
 162   char *dummy;
 163   return sqlite3_prepare (dbh,
 164                           zSql,
 165                           strlen (zSql), ppStmt, (const char **) &dummy);
 166 }
 167
 168
 169 /**
 170  * Create our database indices.
 171  *
 172  * @param dbh handle to the database
 173  */
 174 static void
 175 create_indices (sqlite3 * dbh)
 176 {
 177   /* create indices */
 178   sqlite3_exec (dbh,
 179                 "CREATE INDEX idx_hash ON gn080 (hash)", NULL, NULL, NULL);
 180   sqlite3_exec (dbh,
 181                 "CREATE INDEX idx_hash_vhash ON gn080 (hash,vhash)", NULL,
 182                 NULL, NULL);
 183   sqlite3_exec (dbh, "CREATE INDEX idx_prio ON gn080 (prio)", NULL, NULL,
 184                 NULL);
 185   sqlite3_exec (dbh, "CREATE INDEX idx_expire ON gn080 (expire)", NULL, NULL,
 186                 NULL);
 187   sqlite3_exec (dbh, "CREATE INDEX idx_comb3 ON gn080 (prio,anonLevel)", NULL,
 188                 NULL, NULL);
 189   sqlite3_exec (dbh, "CREATE INDEX idx_comb4 ON gn080 (prio,hash,anonLevel)",
 190                 NULL, NULL, NULL);
 191   sqlite3_exec (dbh, "CREATE INDEX idx_comb7 ON gn080 (expire,hash)", NULL,
 192                 NULL, NULL);
 193 }
 194
 195
 196
 197 #if 1
 198 #define CHECK(a) GNUNET_break(a)
 199 #define ENULL NULL
 200 #else
 201 #define ENULL &e
 202 #define ENULL_DEFINED 1
 203 #define CHECK(a) if (! a) { GNUNET_log(GNUNET_ERROR_TYPE_ERRROR, "%s\n", e); sqlite3_free(e); }
 204 #endif
 205
 206
 207
 208
 209 /**
 210  * Initialize the database connections and associated
 211  * data structures (create tables and indices
 212  * as needed as well).
 213  *
 214  * @param cfg our configuration
 215  * @param plugin the plugin context (state for this module)
 216  * @return GNUNET_OK on success
 217  */
 218 static int
 219 database_setup (const struct GNUNET_CONFIGURATION_Handle *cfg,
 220                 struct Plugin *plugin)
 221 {
 222   sqlite3_stmt *stmt;
 223   char *afsdir;
 224 #if ENULL_DEFINED
 225   char *e;
 226 #endif
 227
 228   if (GNUNET_OK !=
 229       GNUNET_CONFIGURATION_get_value_filename (cfg,
 230                                                "datastore-sqlite",
 231                                                "FILENAME",
 232                                                &afsdir))
 233     {
 234       GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
 235                        "sqlite",
 236                        _("Option `%s' in section `%s' missing in configuration!\n"),
 237                        "FILENAME",
 238                        "datastore-sqlite");
 239       return GNUNET_SYSERR;
 240     }
 241   if (GNUNET_OK != GNUNET_DISK_directory_create_for_file (afsdir))
 242     {
 243       GNUNET_break (0);
 244       GNUNET_free (afsdir);
 245       return GNUNET_SYSERR;
 246     }
 247   plugin->fn = GNUNET_STRINGS_to_utf8 (afsdir, strlen (afsdir),
 248 #ifdef ENABLE_NLS
 249                                               nl_langinfo (CODESET)
 250 #else
 251                                               "UTF-8"   /* good luck */
 252 #endif
 253                                               );
 254   GNUNET_free (afsdir);
 255
 256   /* Open database and precompile statements */
 257   if (sqlite3_open (plugin->fn, &plugin->dbh) != SQLITE_OK)
 258     {
 259       GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
 260                        "sqlite",
 261                        _("Unable to initialize SQLite: %s.\n"),
 262                        sqlite3_errmsg (plugin->dbh));
 263       return GNUNET_SYSERR;
 264     }
 265   CHECK (SQLITE_OK ==
 266          sqlite3_exec (plugin->dbh,
 267                        "PRAGMA temp_store=MEMORY", NULL, NULL, ENULL));
 268   CHECK (SQLITE_OK ==
 269          sqlite3_exec (plugin->dbh,
 270                        "PRAGMA synchronous=OFF", NULL, NULL, ENULL));
 271   CHECK (SQLITE_OK ==
 272          sqlite3_exec (plugin->dbh,
 273                        "PRAGMA count_changes=OFF", NULL, NULL, ENULL));
 274   CHECK (SQLITE_OK ==
 275          sqlite3_exec (plugin->dbh, "PRAGMA page_size=4092", NULL, NULL, ENULL));
 276
 277   CHECK (SQLITE_OK == sqlite3_busy_timeout (plugin->dbh, BUSY_TIMEOUT_MS));
 278
 279
 280   /* We have to do it here, because otherwise precompiling SQL might fail */
 281   CHECK (SQLITE_OK ==
 282          sq_prepare (plugin->dbh,
 283                      "SELECT 1 FROM sqlite_master WHERE tbl_name = 'gn080'",
 284                      &stmt));
 285   if ( (sqlite3_step (stmt) == SQLITE_DONE) &&
 286        (sqlite3_exec (plugin->dbh,
 287                       "CREATE TABLE gn080 ("
 288                       "  size INT4 NOT NULL DEFAULT 0,"
 289                       "  type INT4 NOT NULL DEFAULT 0,"
 290                       "  prio INT4 NOT NULL DEFAULT 0,"
 291                       "  anonLevel INT4 NOT NULL DEFAULT 0,"
 292                       "  expire INT8 NOT NULL DEFAULT 0,"
 293                       "  hash TEXT NOT NULL DEFAULT '',"
 294                       "  vhash TEXT NOT NULL DEFAULT '',"
 295                       "  value BLOB NOT NULL DEFAULT '')", NULL, NULL,
 296                       NULL) != SQLITE_OK) )
 297     {
 298       LOG_SQLITE (plugin, NULL,
 299                   GNUNET_ERROR_TYPE_ERROR,
 300                   "sqlite3_exec");
 301       sqlite3_finalize (stmt);
 302       return GNUNET_SYSERR;
 303     }
 304   sqlite3_finalize (stmt);
 305   create_indices (plugin->dbh);
 306
 307   CHECK (SQLITE_OK ==
 308          sq_prepare (plugin->dbh,
 309                      "SELECT 1 FROM sqlite_master WHERE tbl_name = 'gn071'",
 310                      &stmt));
 311   if ( (sqlite3_step (stmt) == SQLITE_DONE) &&
 312        (sqlite3_exec (plugin->dbh,
 313                       "CREATE TABLE gn071 ("
 314                       "  key TEXT NOT NULL DEFAULT '',"
 315                       "  value INTEGER NOT NULL DEFAULT 0)", NULL, NULL,
 316                       NULL) != SQLITE_OK) )
 317     {
 318       LOG_SQLITE (plugin, NULL,
 319                   GNUNET_ERROR_TYPE_ERROR, "sqlite3_exec");
 320       sqlite3_finalize (stmt);
 321       return GNUNET_SYSERR;
 322     }
 323   sqlite3_finalize (stmt);
 324
 325   if ((sq_prepare (plugin->dbh,
 326                    "UPDATE gn080 SET prio = prio + ?, expire = MAX(expire,?) WHERE "
 327                    "_ROWID_ = ?",
 328                    &plugin->updPrio) != SQLITE_OK) ||
 329       (sq_prepare (plugin->dbh,
 330                    "INSERT INTO gn080 (size, type, prio, "
 331                    "anonLevel, expire, hash, vhash, value) VALUES "
 332                    "(?, ?, ?, ?, ?, ?, ?, ?)",
 333                    &plugin->insertContent) != SQLITE_OK))
 334     {
 335       LOG_SQLITE (plugin, NULL,
 336                   GNUNET_ERROR_TYPE_ERROR, "precompiling");
 337       return GNUNET_SYSERR;
 338     }
 339   return GNUNET_OK;
 340 }
 341
 342
 343 /**
 344  * Synchronize our utilization statistics with the
 345  * statistics service.
 346  * @param plugin the plugin context (state for this module)
 347  */
 348 static void
 349 sync_stats (struct Plugin *plugin)
 350 {
 351   GNUNET_STATISTICS_set (plugin->statistics,
 352                          QUOTA_STAT_NAME,
 353                          plugin->payload,
 354                          GNUNET_YES);
 355   plugin->lastSync = 0;
 356 }
 357
 358
 359 /**
 360  * Shutdown database connection and associate data
 361  * structures.
 362  * @param plugin the plugin context (state for this module)
 363  */
 364 static void
 365 database_shutdown (struct Plugin *plugin)
 366 {
 367   if (plugin->lastSync > 0)
 368     sync_stats (plugin);
 369   if (plugin->updPrio != NULL)
 370     sqlite3_finalize (plugin->updPrio);
 371   if (plugin->insertContent != NULL)
 372     sqlite3_finalize (plugin->insertContent);
 373   sqlite3_close (plugin->dbh);
 374   GNUNET_free_non_null (plugin->fn);
 375 }
 376
 377
 378 /**
 379  * Get an estimate of how much space the database is
 380  * currently using.
 381  *
 382  * @param cls our plugin context
 383  * @return number of bytes used on disk
 384  */
 385 static unsigned long long sqlite_plugin_get_size (void *cls)
 386 {
 387   struct Plugin *plugin = cls;
 388   return plugin->payload;
 389 }
 390
 391
 392 /**
 393  * Delete the database entry with the given
 394  * row identifier.
 395  *
 396  * @param plugin the plugin context (state for this module)
 397  * @param rid the ID of the row to delete
 398  */
 399 static int
 400 delete_by_rowid (struct Plugin* plugin,
 401                  unsigned long long rid)
 402 {
 403   sqlite3_stmt *stmt;
 404
 405   if (sq_prepare (plugin->dbh,
 406                   "DELETE FROM gn080 WHERE _ROWID_ = ?", &stmt) != SQLITE_OK)
 407     {
 408       LOG_SQLITE (plugin, NULL,
 409                   GNUNET_ERROR_TYPE_ERROR |
 410                   GNUNET_ERROR_TYPE_BULK, "sq_prepare");
 411       return GNUNET_SYSERR;
 412     }
 413   sqlite3_bind_int64 (stmt, 1, rid);
 414   if (SQLITE_DONE != sqlite3_step (stmt))
 415     {
 416       LOG_SQLITE (plugin, NULL,
 417                   GNUNET_ERROR_TYPE_ERROR |
 418                   GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 419       sqlite3_finalize (stmt);
 420       return GNUNET_SYSERR;
 421     }
 422   sqlite3_finalize (stmt);
 423   return GNUNET_OK;
 424 }
 425
 426
 427 /**
 428  * Context for the universal iterator.
 429  */
 430 struct NextContext;
 431
 432 /**
 433  * Type of a function that will prepare
 434  * the next iteration.
 435  *
 436  * @param cls closure
 437  * @param nc the next context; NULL for the last
 438  *         call which gives the callback a chance to
 439  *         clean up the closure
 440  * @return GNUNET_OK on success, GNUNET_NO if there are
 441  *         no more values, GNUNET_SYSERR on error
 442  */
 443 typedef int (*PrepareFunction)(void *cls,
 444                                struct NextContext *nc);
 445
 446
 447 /**
 448  * Context we keep for the "next request" callback.
 449  */
 450 struct NextContext
 451 {
 452   /**
 453    * Internal state.
 454    */
 455   struct Plugin *plugin;
 456
 457   /**
 458    * Function to call on the next value.
 459    */
 460   PluginIterator iter;
 461
 462   /**
 463    * Closure for iter.
 464    */
 465   void *iter_cls;
 466
 467   /**
 468    * Function to call to prepare the next
 469    * iteration.
 470    */
 471   PrepareFunction prep;
 472
 473   /**
 474    * Closure for prep.
 475    */
 476   void *prep_cls;
 477
 478   /**
 479    * Statement that the iterator will get the data
 480    * from (updated or set by prep).
 481    */
 482   sqlite3_stmt *stmt;
 483
 484   /**
 485    * Row ID of the last result.
 486    */
 487   unsigned long long last_rowid;
 488
 489   /**
 490    * Key of the last result.
 491    */
 492   GNUNET_HashCode lastKey;
 493
 494   /**
 495    * Expiration time of the last value visited.
 496    */
 497   struct GNUNET_TIME_Absolute lastExpiration;
 498
 499   /**
 500    * Priority of the last value visited.
 501    */
 502   unsigned int lastPriority;
 503
 504   /**
 505    * Number of results processed so far.
 506    */
 507   unsigned int count;
 508
 509   /**
 510    * Set to GNUNET_YES if we must stop now.
 511    */
 512   int end_it;
 513 };
 514
 515
 516 /**
 517  * Continuation of "sqlite_next_request".
 518  *
 519  * @param cls the next context
 520  * @param tc the task context (unused)
 521  */
 522 static void
 523 sqlite_next_request_cont (void *cls,
 524                           const struct GNUNET_SCHEDULER_TaskContext *tc)
 525 {
 526   struct NextContext * nc= cls;
 527   struct Plugin *plugin;
 528   unsigned long long rowid;
 529   sqlite3_stmt *stmtd;
 530   int ret;
 531   unsigned int type;
 532   unsigned int size;
 533   unsigned int priority;
 534   unsigned int anonymity;
 535   struct GNUNET_TIME_Absolute expiration;
 536   const GNUNET_HashCode *key;
 537   const void *data;
 538
 539
 540   plugin = nc->plugin;
 541   if ( (GNUNET_YES == nc->end_it) ||
 542        (GNUNET_OK != (nc->prep(nc->prep_cls,
 543                                nc))) )
 544     {
 545     END:
 546       nc->iter (nc->iter_cls,
 547                 NULL, NULL, 0, NULL, 0, 0, 0,
 548                 GNUNET_TIME_UNIT_ZERO_ABS, 0);
 549       nc->prep (nc->prep_cls, NULL);
 550       GNUNET_free (nc);
 551       return;
 552     }
 553
 554   rowid = sqlite3_column_int64 (nc->stmt, 7);
 555   nc->last_rowid = rowid;
 556   type = sqlite3_column_int (nc->stmt, 1);
 557   size = sqlite3_column_bytes (nc->stmt, 6);
 558   if (sqlite3_column_bytes (nc->stmt, 5) != sizeof (GNUNET_HashCode))
 559     {
 560       GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING,
 561                        "sqlite",
 562                        _("Invalid data in database.  Trying to fix (by deletion).\n"));
 563       if (SQLITE_OK != sqlite3_reset (nc->stmt))
 564         LOG_SQLITE (nc->plugin, NULL,
 565                     GNUNET_ERROR_TYPE_ERROR |
 566                     GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 567       if (sq_prepare
 568           (nc->plugin->dbh,
 569            "DELETE FROM gn080 WHERE NOT LENGTH(hash) = ?",
 570            &stmtd) != SQLITE_OK)
 571         {
 572           LOG_SQLITE (nc->plugin, NULL,
 573                       GNUNET_ERROR_TYPE_ERROR |
 574                       GNUNET_ERROR_TYPE_BULK,
 575                       "sq_prepare");
 576           goto END;
 577         }
 578
 579       if (SQLITE_OK != sqlite3_bind_int (stmtd, 1, sizeof (GNUNET_HashCode)))
 580         LOG_SQLITE (nc->plugin, NULL,
 581                     GNUNET_ERROR_TYPE_ERROR |
 582                     GNUNET_ERROR_TYPE_BULK, "sqlite3_bind_int");
 583       if (SQLITE_DONE != sqlite3_step (stmtd))
 584         LOG_SQLITE (nc->plugin, NULL,
 585                     GNUNET_ERROR_TYPE_ERROR |
 586                     GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 587       if (SQLITE_OK != sqlite3_finalize (stmtd))
 588         LOG_SQLITE (nc->plugin, NULL,
 589                     GNUNET_ERROR_TYPE_ERROR |
 590                     GNUNET_ERROR_TYPE_BULK, "sqlite3_finalize");
 591       goto END;
 592     }
 593
 594   priority = sqlite3_column_int (nc->stmt, 2);
 595   anonymity = sqlite3_column_int (nc->stmt, 3);
 596   expiration.value = sqlite3_column_int64 (nc->stmt, 4);
 597   key = sqlite3_column_blob (nc->stmt, 5);
 598   nc->lastPriority = priority;
 599   nc->lastExpiration = expiration;
 600   memcpy (&nc->lastKey, key, sizeof(GNUNET_HashCode));
 601   data = sqlite3_column_blob (nc->stmt, 6);
 602   nc->count++;
 603   ret = nc->iter (nc->iter_cls,
 604                   nc,
 605                   key,
 606                   size,
 607                   data,
 608                   type,
 609                   priority,
 610                   anonymity,
 611                   expiration,
 612                   rowid);
 613   if (ret == GNUNET_SYSERR)
 614     {
 615       nc->end_it = GNUNET_YES;
 616       return;
 617     }
 618 #if DEBUG_SQLITE
 619   if (ret == GNUNET_NO)
 620     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 621                      "sqlite",
 622                      "Asked to remove entry %llu (%u bytes)\n",
 623                      (unsigned long long) rowid,
 624                      size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 625 #endif
 626   if ( (ret == GNUNET_NO) &&
 627        (GNUNET_OK == delete_by_rowid (plugin, rowid)) )
 628     {
 629       plugin->payload -= (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 630       plugin->lastSync++;
 631 #if DEBUG_SQLITE
 632       if (ret == GNUNET_NO)
 633         GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 634                          "sqlite",
 635                          "Removed entry %llu (%u bytes), new payload is %llu\n",
 636                          (unsigned long long) rowid,
 637                          size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
 638                          (unsigned long long) plugin->payload);
 639 #endif
 640       if (plugin->lastSync >= MAX_STAT_SYNC_LAG)
 641         sync_stats (plugin);
 642     }
 643 }
 644
 645
 646 /**
 647  * Function invoked on behalf of a "PluginIterator"
 648  * asking the database plugin to call the iterator
 649  * with the next item.
 650  *
 651  * @param next_cls whatever argument was given
 652  *        to the PluginIterator as "next_cls".
 653  * @param end_it set to GNUNET_YES if we
 654  *        should terminate the iteration early
 655  *        (iterator should be still called once more
 656  *         to signal the end of the iteration).
 657  */
 658 static void
 659 sqlite_next_request (void *next_cls,
 660                      int end_it)
 661 {
 662   struct NextContext * nc= next_cls;
 663
 664   if (GNUNET_YES == end_it)
 665     nc->end_it = GNUNET_YES;
 666   GNUNET_SCHEDULER_add_continuation (nc->plugin->env->sched,
 667                                      GNUNET_NO,
 668                                      &sqlite_next_request_cont,
 669                                      nc,
 670                                      GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 671 }
 672
 673
 674
 675 /**
 676  * Store an item in the datastore.
 677  *
 678  * @param cls closure
 679  * @param key key for the item
 680  * @param size number of bytes in data
 681  * @param data content stored
 682  * @param type type of the content
 683  * @param priority priority of the content
 684  * @param anonymity anonymity-level for the content
 685  * @param expiration expiration time for the content
 686  * @param msg set to an error message
 687  * @return GNUNET_OK on success
 688  */
 689 static int
 690 sqlite_plugin_put (void *cls,
 691                    const GNUNET_HashCode * key,
 692                    uint32_t size,
 693                    const void *data,
 694                    uint32_t type,
 695                    uint32_t priority,
 696                    uint32_t anonymity,
 697                    struct GNUNET_TIME_Absolute expiration,
 698                    char ** msg)
 699 {
 700   struct Plugin *plugin = cls;
 701   int n;
 702   sqlite3_stmt *stmt;
 703   GNUNET_HashCode vhash;
 704
 705 #if DEBUG_SQLITE
 706   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 707                    "sqlite",
 708                    "Storing in database block with type %u/key `%s'/priority %u/expiration %llu (%lld).\n",
 709                    type,
 710                    GNUNET_h2s(key),
 711                    priority,
 712                    (unsigned long long) GNUNET_TIME_absolute_get_remaining (expiration).value,
 713                    (long long) expiration.value);
 714 #endif
 715   GNUNET_CRYPTO_hash (data, size, &vhash);
 716   stmt = plugin->insertContent;
 717   if ((SQLITE_OK != sqlite3_bind_int (stmt, 1, size)) ||
 718       (SQLITE_OK != sqlite3_bind_int (stmt, 2, type)) ||
 719       (SQLITE_OK != sqlite3_bind_int (stmt, 3, priority)) ||
 720       (SQLITE_OK != sqlite3_bind_int (stmt, 4, anonymity)) ||
 721       (SQLITE_OK != sqlite3_bind_int64 (stmt, 5, (sqlite3_int64) expiration.value)) ||
 722       (SQLITE_OK !=
 723        sqlite3_bind_blob (stmt, 6, key, sizeof (GNUNET_HashCode),
 724                           SQLITE_TRANSIENT)) ||
 725       (SQLITE_OK !=
 726        sqlite3_bind_blob (stmt, 7, &vhash, sizeof (GNUNET_HashCode),
 727                           SQLITE_TRANSIENT))
 728       || (SQLITE_OK !=
 729           sqlite3_bind_blob (stmt, 8, data, size,
 730                              SQLITE_TRANSIENT)))
 731     {
 732       LOG_SQLITE (plugin,
 733                   msg,
 734                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_bind_XXXX");
 735       if (SQLITE_OK != sqlite3_reset (stmt))
 736         LOG_SQLITE (plugin, NULL,
 737                     GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 738       return GNUNET_SYSERR;
 739     }
 740   n = sqlite3_step (stmt);
 741   if (n != SQLITE_DONE)
 742     {
 743       if (n == SQLITE_BUSY)
 744         {
 745           LOG_SQLITE (plugin, msg,
 746                       GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 747           sqlite3_reset (stmt);
 748           GNUNET_break (0);
 749           return GNUNET_NO;
 750         }
 751       LOG_SQLITE (plugin, msg,
 752                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 753       sqlite3_reset (stmt);
 754       return GNUNET_SYSERR;
 755     }
 756   if (SQLITE_OK != sqlite3_reset (stmt))
 757     LOG_SQLITE (plugin, NULL,
 758                 GNUNET_ERROR_TYPE_ERROR |
 759                 GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 760   plugin->lastSync++;
 761   plugin->payload += size + GNUNET_DATASTORE_ENTRY_OVERHEAD;
 762 #if DEBUG_SQLITE
 763   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 764                    "sqlite",
 765                    "Stored new entry (%u bytes), new payload is %llu\n",
 766                    size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
 767                    (unsigned long long) plugin->payload);
 768 #endif
 769   if (plugin->lastSync >= MAX_STAT_SYNC_LAG)
 770     sync_stats (plugin);
 771   return GNUNET_OK;
 772 }
 773
 774
 775 /**
 776  * Update the priority for a particular key in the datastore.  If
 777  * the expiration time in value is different than the time found in
 778  * the datastore, the higher value should be kept.  For the
 779  * anonymity level, the lower value is to be used.  The specified
 780  * priority should be added to the existing priority, ignoring the
 781  * priority in value.
 782  *
 783  * Note that it is possible for multiple values to match this put.
 784  * In that case, all of the respective values are updated.
 785  *
 786  * @param cls the plugin context (state for this module)
 787  * @param uid unique identifier of the datum
 788  * @param delta by how much should the priority
 789  *     change?  If priority + delta < 0 the
 790  *     priority should be set to 0 (never go
 791  *     negative).
 792  * @param expire new expiration time should be the
 793  *     MAX of any existing expiration time and
 794  *     this value
 795  * @param msg set to an error message
 796  * @return GNUNET_OK on success
 797  */
 798 static int
 799 sqlite_plugin_update (void *cls,
 800                       uint64_t uid,
 801                       int delta, struct GNUNET_TIME_Absolute expire,
 802                       char **msg)
 803 {
 804   struct Plugin *plugin = cls;
 805   int n;
 806
 807   sqlite3_bind_int (plugin->updPrio, 1, delta);
 808   sqlite3_bind_int64 (plugin->updPrio, 2, expire.value);
 809   sqlite3_bind_int64 (plugin->updPrio, 3, uid);
 810   n = sqlite3_step (plugin->updPrio);
 811   if (n != SQLITE_DONE)
 812     LOG_SQLITE (plugin, msg,
 813                 GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
 814                 "sqlite3_step");
 815 #if DEBUG_SQLITE
 816   else
 817     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 818                      "sqlite",
 819                      "Block updated\n");
 820 #endif
 821   sqlite3_reset (plugin->updPrio);
 822
 823   if (n == SQLITE_BUSY)
 824     return GNUNET_NO;
 825   return n == SQLITE_DONE ? GNUNET_OK : GNUNET_SYSERR;
 826 }
 827
 828
 829 /**
 830  * Internal context for an iteration.
 831  */
 832 struct IterContext
 833 {
 834   /**
 835    * FIXME.
 836    */
 837   sqlite3_stmt *stmt_1;
 838
 839   /**
 840    * FIXME.
 841    */
 842   sqlite3_stmt *stmt_2;
 843
 844   /**
 845    * FIXME.
 846    */
 847   int is_asc;
 848
 849   /**
 850    * FIXME.
 851    */
 852   int is_prio;
 853
 854   /**
 855    * FIXME.
 856    */
 857   int is_migr;
 858
 859   /**
 860    * FIXME.
 861    */
 862   int limit_nonanonymous;
 863
 864   /**
 865    * Desired type for blocks returned by this iterator.
 866    */
 867   uint32_t type;
 868 };
 869
 870
 871 /**
 872  * Prepare our SQL query to obtain the next record from the database.
 873  *
 874  * @param cls our "struct IterContext"
 875  * @param nc NULL to terminate the iteration, otherwise our context for
 876  *           getting the next result.
 877  * @return GNUNET_OK on success, GNUNET_NO if there are no more results,
 878  *         GNUNET_SYSERR on error (or end of iteration)
 879  */
 880 static int
 881 iter_next_prepare (void *cls,
 882                    struct NextContext *nc)
 883 {
 884   struct IterContext *ic = cls;
 885   struct Plugin *plugin;
 886   int ret;
 887
 888   if (nc == NULL)
 889     {
 890 #if DEBUG_SQLITE
 891       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 892                   "Asked to clean up iterator state.\n");
 893 #endif
 894       sqlite3_finalize (ic->stmt_1);
 895       sqlite3_finalize (ic->stmt_2);
 896       return GNUNET_SYSERR;
 897     }
 898   sqlite3_reset (ic->stmt_1);
 899   sqlite3_reset (ic->stmt_2);
 900   plugin = nc->plugin;
 901   if (ic->is_prio)
 902     {
 903 #if DEBUG_SQLITE
 904       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 905                   "Restricting to results larger than the last priority %u\n",
 906                   nc->lastPriority);
 907 #endif
 908       sqlite3_bind_int (ic->stmt_1, 1, nc->lastPriority);
 909       sqlite3_bind_int (ic->stmt_2, 1, nc->lastPriority);
 910     }
 911   else
 912     {
 913 #if DEBUG_SQLITE
 914       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 915                   "Restricting to results larger than the last expiration %llu\n",
 916                   (unsigned long long) nc->lastExpiration.value);
 917 #endif
 918       sqlite3_bind_int64 (ic->stmt_1, 1, nc->lastExpiration.value);
 919       sqlite3_bind_int64 (ic->stmt_2, 1, nc->lastExpiration.value);
 920     }
 921 #if DEBUG_SQLITE
 922   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 923               "Restricting to results larger than the last key `%s'\n",
 924               GNUNET_h2s(&nc->lastKey));
 925 #endif
 926   sqlite3_bind_blob (ic->stmt_1, 2,
 927                      &nc->lastKey,
 928                      sizeof (GNUNET_HashCode),
 929                      SQLITE_TRANSIENT);
 930   if (SQLITE_ROW == (ret = sqlite3_step (ic->stmt_1)))
 931     {
 932 #if DEBUG_SQLITE
 933       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 934                   "Result found using iterator 1\n");
 935 #endif
 936       nc->stmt = ic->stmt_1;
 937       return GNUNET_OK;
 938     }
 939   if (ret != SQLITE_DONE)
 940     {
 941       LOG_SQLITE (plugin, NULL,
 942                   GNUNET_ERROR_TYPE_ERROR |
 943                   GNUNET_ERROR_TYPE_BULK,
 944                   "sqlite3_step");
 945       return GNUNET_SYSERR;
 946     }
 947   if (SQLITE_OK != sqlite3_reset (ic->stmt_1))
 948     LOG_SQLITE (plugin, NULL,
 949                 GNUNET_ERROR_TYPE_ERROR |
 950                 GNUNET_ERROR_TYPE_BULK,
 951                 "sqlite3_reset");
 952   if (SQLITE_ROW == (ret = sqlite3_step (ic->stmt_2)))
 953     {
 954 #if DEBUG_SQLITE
 955       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 956                   "Result found using iterator 2\n");
 957 #endif
 958       nc->stmt = ic->stmt_2;
 959       return GNUNET_OK;
 960     }
 961   if (ret != SQLITE_DONE)
 962     {
 963       LOG_SQLITE (plugin, NULL,
 964                   GNUNET_ERROR_TYPE_ERROR |
 965                   GNUNET_ERROR_TYPE_BULK,
 966                   "sqlite3_step");
 967       return GNUNET_SYSERR;
 968     }
 969   if (SQLITE_OK != sqlite3_reset (ic->stmt_2))
 970     LOG_SQLITE (plugin, NULL,
 971                 GNUNET_ERROR_TYPE_ERROR |
 972                 GNUNET_ERROR_TYPE_BULK,
 973                 "sqlite3_reset");
 974 #if DEBUG_SQLITE
 975   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 976               "No result found using either iterator\n");
 977 #endif
 978   return GNUNET_NO;
 979 }
 980
 981
 982 /**
 983  * Call a method for each key in the database and
 984  * call the callback method on it.
 985  *
 986  * @param plugin our plugin context
 987  * @param type entries of which type should be considered?
 988  * @param is_asc are we iterating in ascending order?
 989  * @param is_prio are we iterating by priority (otherwise by expiration)
 990  * @param is_migr are we iterating in migration order?
 991  * @param limit_nonanonymous are we restricting results to those with anonymity
 992  *              level zero?
 993  * @param stmt_str_1 first SQL statement to execute
 994  * @param stmt_str_2 SQL statement to execute to get "more" results (inner iteration)
 995  * @param iter function to call on each matching value;
 996  *        will be called once with a NULL value at the end
 997  * @param iter_cls closure for iter
 998  */
 999 static void
1000 basic_iter (struct Plugin *plugin,
1001             uint32_t type,
1002             int is_asc,
1003             int is_prio,
1004             int is_migr,
1005             int limit_nonanonymous,
1006             const char *stmt_str_1,
1007             const char *stmt_str_2,
1008             PluginIterator iter,
1009             void *iter_cls)
1010 {
1011   struct NextContext *nc;
1012   struct IterContext *ic;
1013   sqlite3_stmt *stmt_1;
1014   sqlite3_stmt *stmt_2;
1015
1016 #if DEBUG_SQLITE
1017   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1018               "At %llu, using queries `%s' and `%s'\n",
1019               (unsigned long long) GNUNET_TIME_absolute_get ().value,
1020               stmt_str_1,
1021               stmt_str_2);
1022 #endif
1023   if (sq_prepare (plugin->dbh, stmt_str_1, &stmt_1) != SQLITE_OK)
1024     {
1025       LOG_SQLITE (plugin, NULL,
1026                   GNUNET_ERROR_TYPE_ERROR |
1027                   GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare");
1028       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1029       return;
1030     }
1031   if (sq_prepare (plugin->dbh, stmt_str_2, &stmt_2) != SQLITE_OK)
1032     {
1033       LOG_SQLITE (plugin, NULL,
1034                   GNUNET_ERROR_TYPE_ERROR |
1035                   GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare");
1036       sqlite3_finalize (stmt_1);
1037       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1038       return;
1039     }
1040   nc = GNUNET_malloc (sizeof(struct NextContext) +
1041                       sizeof(struct IterContext));
1042   nc->plugin = plugin;
1043   nc->iter = iter;
1044   nc->iter_cls = iter_cls;
1045   nc->stmt = NULL;
1046   ic = (struct IterContext*) &nc[1];
1047   ic->stmt_1 = stmt_1;
1048   ic->stmt_2 = stmt_2;
1049   ic->type = type;
1050   ic->is_asc = is_asc;
1051   ic->is_prio = is_prio;
1052   ic->is_migr = is_migr;
1053   ic->limit_nonanonymous = limit_nonanonymous;
1054   nc->prep = &iter_next_prepare;
1055   nc->prep_cls = ic;
1056   if (is_asc)
1057     {
1058       nc->lastPriority = 0;
1059       nc->lastExpiration.value = 0;
1060       memset (&nc->lastKey, 0, sizeof (GNUNET_HashCode));
1061     }
1062   else
1063     {
1064       nc->lastPriority = 0x7FFFFFFF;
1065       nc->lastExpiration.value = 0x7FFFFFFFFFFFFFFFLL;
1066       memset (&nc->lastKey, 255, sizeof (GNUNET_HashCode));
1067     }
1068   sqlite_next_request (nc, GNUNET_NO);
1069 }
1070
1071
1072 /**
1073  * Select a subset of the items in the datastore and call
1074  * the given iterator for each of them.
1075  *
1076  * @param cls our plugin context
1077  * @param type entries of which type should be considered?
1078  *        Use 0 for any type.
1079  * @param iter function to call on each matching value;
1080  *        will be called once with a NULL value at the end
1081  * @param iter_cls closure for iter
1082  */
1083 static void
1084 sqlite_plugin_iter_low_priority (void *cls,
1085                                  uint32_t type,
1086                                  PluginIterator iter,
1087                                  void *iter_cls)
1088 {
1089   basic_iter (cls,
1090               type,
1091               GNUNET_YES, GNUNET_YES,
1092               GNUNET_NO, GNUNET_NO,
1093               SELECT_IT_LOW_PRIORITY_1,
1094               SELECT_IT_LOW_PRIORITY_2,
1095               iter, iter_cls);
1096 }
1097
1098
1099 /**
1100  * Select a subset of the items in the datastore and call
1101  * the given iterator for each of them.
1102  *
1103  * @param cls our plugin context
1104  * @param type entries of which type should be considered?
1105  *        Use 0 for any type.
1106  * @param iter function to call on each matching value;
1107  *        will be called once with a NULL value at the end
1108  * @param iter_cls closure for iter
1109  */
1110 static void
1111 sqlite_plugin_iter_zero_anonymity (void *cls,
1112                                    uint32_t type,
1113                                    PluginIterator iter,
1114                                    void *iter_cls)
1115 {
1116   struct GNUNET_TIME_Absolute now;
1117   char *q1;
1118   char *q2;
1119
1120   now = GNUNET_TIME_absolute_get ();
1121   GNUNET_asprintf (&q1, SELECT_IT_NON_ANONYMOUS_1,
1122                    (unsigned long long) now.value);
1123   GNUNET_asprintf (&q2, SELECT_IT_NON_ANONYMOUS_2,
1124                    (unsigned long long) now.value);
1125   basic_iter (cls,
1126               type,
1127               GNUNET_NO, GNUNET_YES,
1128               GNUNET_NO, GNUNET_YES,
1129               q1,
1130               q2,
1131               iter, iter_cls);
1132   GNUNET_free (q1);
1133   GNUNET_free (q2);
1134 }
1135
1136
1137
1138 /**
1139  * Select a subset of the items in the datastore and call
1140  * the given iterator for each of them.
1141  *
1142  * @param cls our plugin context
1143  * @param type entries of which type should be considered?
1144  *        Use 0 for any type.
1145  * @param iter function to call on each matching value;
1146  *        will be called once with a NULL value at the end
1147  * @param iter_cls closure for iter
1148  */
1149 static void
1150 sqlite_plugin_iter_ascending_expiration (void *cls,
1151                                          uint32_t type,
1152                                          PluginIterator iter,
1153                                          void *iter_cls)
1154 {
1155   struct GNUNET_TIME_Absolute now;
1156   char *q1;
1157   char *q2;
1158
1159   now = GNUNET_TIME_absolute_get ();
1160   GNUNET_asprintf (&q1, SELECT_IT_EXPIRATION_TIME_1,
1161                    (unsigned long long) 0*now.value);
1162   GNUNET_asprintf (&q2, SELECT_IT_EXPIRATION_TIME_2,
1163                    (unsigned long long) 0*now.value);
1164   basic_iter (cls,
1165               type,
1166               GNUNET_YES, GNUNET_NO,
1167               GNUNET_NO, GNUNET_NO,
1168               q1, q2,
1169               iter, iter_cls);
1170   GNUNET_free (q1);
1171   GNUNET_free (q2);
1172 }
1173
1174
1175 /**
1176  * Select a subset of the items in the datastore and call
1177  * the given iterator for each of them.
1178  *
1179  * @param cls our plugin context
1180  * @param type entries of which type should be considered?
1181  *        Use 0 for any type.
1182  * @param iter function to call on each matching value;
1183  *        will be called once with a NULL value at the end
1184  * @param iter_cls closure for iter
1185  */
1186 static void
1187 sqlite_plugin_iter_migration_order (void *cls,
1188                                     uint32_t type,
1189                                     PluginIterator iter,
1190                                     void *iter_cls)
1191 {
1192   struct GNUNET_TIME_Absolute now;
1193   char *q;
1194
1195   now = GNUNET_TIME_absolute_get ();
1196   GNUNET_asprintf (&q, SELECT_IT_MIGRATION_ORDER_2,
1197                    (unsigned long long) now.value);
1198   basic_iter (cls,
1199               type,
1200               GNUNET_NO, GNUNET_NO,
1201               GNUNET_YES, GNUNET_NO,
1202               SELECT_IT_MIGRATION_ORDER_1,
1203               q,
1204               iter, iter_cls);
1205   GNUNET_free (q);
1206 }
1207
1208
1209 /**
1210  * Call sqlite using the already prepared query to get
1211  * the next result.
1212  *
1213  * @param cls not used
1214  * @param nc context with the prepared query
1215  * @return GNUNET_OK on success, GNUNET_SYSERR on error, GNUNET_NO if
1216  *        there are no more results
1217  */
1218 static int
1219 all_next_prepare (void *cls,
1220                   struct NextContext *nc)
1221 {
1222   struct Plugin *plugin;
1223   int ret;
1224
1225   if (nc == NULL)
1226     {
1227 #if DEBUG_SQLITE
1228       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1229                   "Asked to clean up iterator state.\n");
1230 #endif
1231       return GNUNET_SYSERR;
1232     }
1233   plugin = nc->plugin;
1234   if (SQLITE_ROW == (ret = sqlite3_step (nc->stmt)))
1235     {
1236       return GNUNET_OK;
1237     }
1238   if (ret != SQLITE_DONE)
1239     {
1240       LOG_SQLITE (plugin, NULL,
1241                   GNUNET_ERROR_TYPE_ERROR |
1242                   GNUNET_ERROR_TYPE_BULK,
1243                   "sqlite3_step");
1244       return GNUNET_SYSERR;
1245     }
1246   return GNUNET_NO;
1247 }
1248
1249
1250 /**
1251  * Select a subset of the items in the datastore and call
1252  * the given iterator for each of them.
1253  *
1254  * @param cls our plugin context
1255  * @param type entries of which type should be considered?
1256  *        Use 0 for any type.
1257  * @param iter function to call on each matching value;
1258  *        will be called once with a NULL value at the end
1259  * @param iter_cls closure for iter
1260  */
1261 static void
1262 sqlite_plugin_iter_all_now (void *cls,
1263                             uint32_t type,
1264                             PluginIterator iter,
1265                             void *iter_cls)
1266 {
1267   struct Plugin *plugin = cls;
1268   struct NextContext *nc;
1269   sqlite3_stmt *stmt;
1270
1271   if (sq_prepare (plugin->dbh,
1272                   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080",
1273                   &stmt) != SQLITE_OK)
1274     {
1275       LOG_SQLITE (plugin, NULL,
1276                   GNUNET_ERROR_TYPE_ERROR |
1277                   GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare");
1278       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1279       return;
1280     }
1281   nc = GNUNET_malloc (sizeof(struct NextContext));
1282   nc->plugin = plugin;
1283   nc->iter = iter;
1284   nc->iter_cls = iter_cls;
1285   nc->stmt = stmt;
1286   nc->prep = &all_next_prepare;
1287   nc->prep_cls = NULL;
1288   sqlite_next_request (nc, GNUNET_NO);
1289 }
1290
1291
1292 /**
1293  * FIXME.
1294  */
1295 struct GetNextContext
1296 {
1297
1298   /**
1299    * FIXME.
1300    */
1301   int total;
1302
1303   /**
1304    * FIXME.
1305    */
1306   int off;
1307
1308   /**
1309    * FIXME.
1310    */
1311   int have_vhash;
1312
1313   /**
1314    * FIXME.
1315    */
1316   unsigned int type;
1317
1318   /**
1319    * FIXME.
1320    */
1321   sqlite3_stmt *stmt;
1322
1323   /**
1324    * FIXME.
1325    */
1326   GNUNET_HashCode key;
1327
1328   /**
1329    * FIXME.
1330    */
1331   GNUNET_HashCode vhash;
1332 };
1333
1334
1335
1336 /**
1337  * FIXME.
1338  *
1339  * @param cls our "struct GetNextContext*"
1340  * @param nc FIXME
1341  * @return GNUNET_YES if there are more results,
1342  *         GNUNET_NO if there are no more results,
1343  *         GNUNET_SYSERR on internal error
1344  */
1345 static int
1346 get_next_prepare (void *cls,
1347                   struct NextContext *nc)
1348 {
1349   struct GetNextContext *gnc = cls;
1350   int sqoff;
1351   int ret;
1352   int limit_off;
1353
1354   if (nc == NULL)
1355     {
1356       sqlite3_finalize (gnc->stmt);
1357       return GNUNET_SYSERR;
1358     }
1359   if (nc->count == gnc->total)
1360     return GNUNET_NO;
1361   if (nc->count + gnc->off == gnc->total)
1362     nc->last_rowid = 0;
1363   if (nc->count == 0)
1364     limit_off = gnc->off;
1365   else
1366     limit_off = 0;
1367   sqoff = 1;
1368   sqlite3_reset (nc->stmt);
1369   ret = sqlite3_bind_blob (nc->stmt,
1370                            sqoff++,
1371                            &gnc->key,
1372                            sizeof (GNUNET_HashCode),
1373                            SQLITE_TRANSIENT);
1374   if ((gnc->have_vhash) && (ret == SQLITE_OK))
1375     ret = sqlite3_bind_blob (nc->stmt,
1376                              sqoff++,
1377                              &gnc->vhash,
1378                              sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
1379   if ((gnc->type != 0) && (ret == SQLITE_OK))
1380     ret = sqlite3_bind_int (nc->stmt, sqoff++, gnc->type);
1381   if (ret == SQLITE_OK)
1382     ret = sqlite3_bind_int64 (nc->stmt, sqoff++, nc->last_rowid + 1);
1383   if (ret == SQLITE_OK)
1384     ret = sqlite3_bind_int (nc->stmt, sqoff++, limit_off);
1385   if (ret != SQLITE_OK)
1386     return GNUNET_SYSERR;
1387   if (SQLITE_ROW != sqlite3_step (nc->stmt))
1388     return GNUNET_NO;
1389   return GNUNET_OK;
1390 }
1391
1392
1393 /**
1394  * Iterate over the results for a particular key
1395  * in the datastore.
1396  *
1397  * @param cls closure
1398  * @param key maybe NULL (to match all entries)
1399  * @param vhash hash of the value, maybe NULL (to
1400  *        match all values that have the right key).
1401  *        Note that for DBlocks there is no difference
1402  *        betwen key and vhash, but for other blocks
1403  *        there may be!
1404  * @param type entries of which type are relevant?
1405  *     Use 0 for any type.
1406  * @param iter function to call on each matching value;
1407  *        will be called once with a NULL value at the end
1408  * @param iter_cls closure for iter
1409  */
1410 static void
1411 sqlite_plugin_get (void *cls,
1412                    const GNUNET_HashCode * key,
1413                    const GNUNET_HashCode * vhash,
1414                    uint32_t type,
1415                    PluginIterator iter, void *iter_cls)
1416 {
1417   struct Plugin *plugin = cls;
1418   struct GetNextContext *gpc;
1419   struct NextContext *nc;
1420   int ret;
1421   int total;
1422   sqlite3_stmt *stmt;
1423   char scratch[256];
1424   int sqoff;
1425
1426   GNUNET_assert (iter != NULL);
1427   if (key == NULL)
1428     {
1429       sqlite_plugin_iter_low_priority (cls, type, iter, iter_cls);
1430       return;
1431     }
1432   GNUNET_snprintf (scratch, 256,
1433                    "SELECT count(*) FROM gn080 WHERE hash=:1%s%s",
1434                    vhash == NULL ? "" : " AND vhash=:2",
1435                    type == 0 ? "" : (vhash ==
1436                                      NULL) ? " AND type=:2" : " AND type=:3");
1437   if (sq_prepare (plugin->dbh, scratch, &stmt) != SQLITE_OK)
1438     {
1439       LOG_SQLITE (plugin, NULL,
1440                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite_prepare");
1441       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1442       return;
1443     }
1444   sqoff = 1;
1445   ret = sqlite3_bind_blob (stmt,
1446                            sqoff++,
1447                            key, sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
1448   if ((vhash != NULL) && (ret == SQLITE_OK))
1449     ret = sqlite3_bind_blob (stmt,
1450                              sqoff++,
1451                              vhash,
1452                              sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
1453   if ((type != 0) && (ret == SQLITE_OK))
1454     ret = sqlite3_bind_int (stmt, sqoff++, type);
1455   if (SQLITE_OK != ret)
1456     {
1457       LOG_SQLITE (plugin, NULL,
1458                   GNUNET_ERROR_TYPE_ERROR, "sqlite_bind");
1459       sqlite3_reset (stmt);
1460       sqlite3_finalize (stmt);
1461       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1462       return;
1463     }
1464   ret = sqlite3_step (stmt);
1465   if (ret != SQLITE_ROW)
1466     {
1467       LOG_SQLITE (plugin, NULL,
1468                   GNUNET_ERROR_TYPE_ERROR| GNUNET_ERROR_TYPE_BULK,
1469                   "sqlite_step");
1470       sqlite3_reset (stmt);
1471       sqlite3_finalize (stmt);
1472       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1473       return;
1474     }
1475   total = sqlite3_column_int (stmt, 0);
1476   sqlite3_reset (stmt);
1477   sqlite3_finalize (stmt);
1478   if (0 == total)
1479     {
1480       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1481       return;
1482     }
1483
1484   GNUNET_snprintf (scratch, 256,
1485                    "SELECT size, type, prio, anonLevel, expire, hash, value, _ROWID_ "
1486                    "FROM gn080 WHERE hash=:1%s%s AND _ROWID_ >= :%d "
1487                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET :d",
1488                    vhash == NULL ? "" : " AND vhash=:2",
1489                    type == 0 ? "" : (vhash ==
1490                                      NULL) ? " AND type=:2" : " AND type=:3",
1491                    sqoff, sqoff + 1);
1492   if (sq_prepare (plugin->dbh, scratch, &stmt) != SQLITE_OK)
1493     {
1494       LOG_SQLITE (plugin, NULL,
1495                   GNUNET_ERROR_TYPE_ERROR |
1496                   GNUNET_ERROR_TYPE_BULK, "sqlite_prepare");
1497       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1498       return;
1499     }
1500   nc = GNUNET_malloc (sizeof(struct NextContext) +
1501                       sizeof(struct GetNextContext));
1502   nc->plugin = plugin;
1503   nc->iter = iter;
1504   nc->iter_cls = iter_cls;
1505   nc->stmt = stmt;
1506   gpc = (struct GetNextContext*) &nc[1];
1507   gpc->total = total;
1508   gpc->type = type;
1509   gpc->key = *key;
1510   gpc->stmt = stmt; /* alias used for freeing at the end! */
1511   if (NULL != vhash)
1512     {
1513       gpc->have_vhash = GNUNET_YES;
1514       gpc->vhash = *vhash;
1515     }
1516   gpc->off = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, total);
1517   nc->prep = &get_next_prepare;
1518   nc->prep_cls = gpc;
1519   sqlite_next_request (nc, GNUNET_NO);
1520 }
1521
1522
1523 /**
1524  * Drop database.
1525  *
1526  * @param cls our plugin context
1527  */
1528 static void
1529 sqlite_plugin_drop (void *cls)
1530 {
1531   struct Plugin *plugin = cls;
1532   plugin->drop_on_shutdown = GNUNET_YES;
1533 }
1534
1535
1536 /**
1537  * Callback function to process statistic values.
1538  *
1539  * @param cls closure
1540  * @param subsystem name of subsystem that created the statistic
1541  * @param name the name of the datum
1542  * @param value the current value
1543  * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
1544  * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
1545  */
1546 static int
1547 process_stat_in (void *cls,
1548                  const char *subsystem,
1549                  const char *name,
1550                  uint64_t value,
1551                  int is_persistent)
1552 {
1553   struct Plugin *plugin = cls;
1554   plugin->payload += value;
1555 #if DEBUG_SQLITE
1556   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1557                    "sqlite",
1558                    "Notification from statistics about existing payload (%llu), new payload is %llu\n",
1559                    value,
1560                    (unsigned long long) plugin->payload);
1561 #endif
1562   return GNUNET_OK;
1563 }
1564
1565
1566 /**
1567  * Entry point for the plugin.
1568  *
1569  * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
1570  * @return NULL on error, othrewise the plugin context
1571  */
1572 void *
1573 libgnunet_plugin_datastore_sqlite_init (void *cls)
1574 {
1575   static struct Plugin plugin;
1576   struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
1577   struct GNUNET_DATASTORE_PluginFunctions *api;
1578
1579   if (plugin.env != NULL)
1580     return NULL; /* can only initialize once! */
1581   memset (&plugin, 0, sizeof(struct Plugin));
1582   plugin.env = env;
1583   GNUNET_ARM_start_services (env->cfg, env->sched, "statistics", NULL);
1584   plugin.statistics = GNUNET_STATISTICS_create (env->sched,
1585                                                 "sqlite",
1586                                                 env->cfg);
1587   GNUNET_STATISTICS_get (plugin.statistics,
1588                          "sqlite",
1589                          QUOTA_STAT_NAME,
1590                          GNUNET_TIME_UNIT_MINUTES,
1591                          NULL,
1592                          &process_stat_in,
1593                          &plugin);
1594   if (GNUNET_OK !=
1595       database_setup (env->cfg, &plugin))
1596     {
1597       database_shutdown (&plugin);
1598       GNUNET_ARM_stop_services (env->cfg, env->sched, "statistics", NULL);
1599       return NULL;
1600     }
1601   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
1602   api->cls = &plugin;
1603   api->get_size = &sqlite_plugin_get_size;
1604   api->put = &sqlite_plugin_put;
1605   api->next_request = &sqlite_next_request;
1606   api->get = &sqlite_plugin_get;
1607   api->update = &sqlite_plugin_update;
1608   api->iter_low_priority = &sqlite_plugin_iter_low_priority;
1609   api->iter_zero_anonymity = &sqlite_plugin_iter_zero_anonymity;
1610   api->iter_ascending_expiration = &sqlite_plugin_iter_ascending_expiration;
1611   api->iter_migration_order = &sqlite_plugin_iter_migration_order;
1612   api->iter_all_now = &sqlite_plugin_iter_all_now;
1613   api->drop = &sqlite_plugin_drop;
1614   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
1615                    "sqlite", _("Sqlite database running\n"));
1616   return api;
1617 }
1618
1619
1620 /**
1621  * Exit point from the plugin.
1622  *
1623  * @param cls the plugin context (as returned by "init")
1624  * @return always NULL
1625  */
1626 void *
1627 libgnunet_plugin_datastore_sqlite_done (void *cls)
1628 {
1629   char *fn;
1630   struct GNUNET_DATASTORE_PluginFunctions *api = cls;
1631   struct Plugin *plugin = api->cls;
1632
1633   fn = NULL;
1634   if (plugin->drop_on_shutdown)
1635     fn = GNUNET_strdup (plugin->fn);
1636   database_shutdown (plugin);
1637   plugin->env = NULL;
1638   plugin->payload = 0;
1639   GNUNET_STATISTICS_destroy (plugin->statistics);
1640   GNUNET_ARM_stop_services (plugin->env->cfg, plugin->env->sched, "statistics", NULL);
1641   GNUNET_free (api);
1642   if (fn != NULL)
1643     {
1644       if (0 != UNLINK(fn))
1645         GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING,
1646                                   "unlink",
1647                                   fn);
1648       GNUNET_free (fn);
1649     }
1650   return NULL;
1651 }
1652
1653 /* end of plugin_datastore_sqlite.c */