src/datastore/plugin_datastore_sqlite.c

   1  /*
   2   * This file is part of GNUnet
   3   * (C) 2009, 2011 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 3, 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_datastore_plugin.h"
  29 #include <sqlite3.h>
  30
  31
  32 /**
  33  * We allocate items on the stack at times.  To prevent a stack
  34  * overflow, we impose a limit on the maximum size for the data per
  35  * item.  64k should be enough.
  36  */
  37 #define MAX_ITEM_SIZE 65536
  38
  39 /**
  40  * After how many ms "busy" should a DB operation fail for good?
  41  * A low value makes sure that we are more responsive to requests
  42  * (especially PUTs).  A high value guarantees a higher success
  43  * rate (SELECTs in iterate can take several seconds despite LIMIT=1).
  44  *
  45  * The default value of 250ms should ensure that users do not experience
  46  * huge latencies while at the same time allowing operations to succeed
  47  * with reasonable probability.
  48  */
  49 #define BUSY_TIMEOUT_MS 250
  50
  51
  52 /**
  53  * Log an error message at log-level 'level' that indicates
  54  * a failure of the command 'cmd' on file 'filename'
  55  * with the message given by strerror(errno).
  56  */
  57 #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 at %s:%u with error: %s"), cmd, __FILE__, __LINE__, sqlite3_errmsg(db->dbh)); } while(0)
  58
  59
  60
  61 /**
  62  * Context for all functions in this plugin.
  63  */
  64 struct Plugin
  65 {
  66   /**
  67    * Our execution environment.
  68    */
  69   struct GNUNET_DATASTORE_PluginEnvironment *env;
  70
  71   /**
  72    * Database filename.
  73    */
  74   char *fn;
  75
  76   /**
  77    * Native SQLite database handle.
  78    */
  79   sqlite3 *dbh;
  80
  81   /**
  82    * Precompiled SQL for deletion.
  83    */
  84   sqlite3_stmt *delRow;
  85
  86   /**
  87    * Precompiled SQL for update.
  88    */
  89   sqlite3_stmt *updPrio;
  90
  91   /**
  92    * Get maximum repl value in database.
  93    */
  94   sqlite3_stmt *maxRepl;
  95
  96   /**
  97    * Precompiled SQL for replication decrement.
  98    */
  99   sqlite3_stmt *updRepl;
 100
 101   /**
 102    * Precompiled SQL for replication selection.
 103    */
 104   sqlite3_stmt *selRepl;
 105
 106   /**
 107    * Precompiled SQL for expiration selection.
 108    */
 109   sqlite3_stmt *selExpi;
 110
 111   /**
 112    * Precompiled SQL for expiration selection.
 113    */
 114   sqlite3_stmt *selZeroAnon;
 115
 116   /**
 117    * Precompiled SQL for insertion.
 118    */
 119   sqlite3_stmt *insertContent;
 120
 121   /**
 122    * Should the database be dropped on shutdown?
 123    */
 124   int drop_on_shutdown;
 125
 126 };
 127
 128
 129 /**
 130  * @brief Prepare a SQL statement
 131  *
 132  * @param dbh handle to the database
 133  * @param zSql SQL statement, UTF-8 encoded
 134  * @param ppStmt set to the prepared statement
 135  * @return 0 on success
 136  */
 137 static int
 138 sq_prepare (sqlite3 * dbh, const char *zSql, sqlite3_stmt ** ppStmt)
 139 {
 140   char *dummy;
 141   int result;
 142
 143   result =
 144       sqlite3_prepare_v2 (dbh, zSql, strlen (zSql), ppStmt,
 145                           (const char **) &dummy);
 146   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
 147                    "Prepared `%s' / %p: %d\n", zSql, *ppStmt, result);
 148   return result;
 149 }
 150
 151
 152 /**
 153  * Create our database indices.
 154  *
 155  * @param dbh handle to the database
 156  */
 157 static void
 158 create_indices (sqlite3 * dbh)
 159 {
 160   /* create indices */
 161   if ((SQLITE_OK !=
 162        sqlite3_exec (dbh, "CREATE INDEX IF NOT EXISTS idx_hash ON gn090 (hash)",
 163                      NULL, NULL, NULL)) ||
 164       (SQLITE_OK !=
 165        sqlite3_exec (dbh,
 166                      "CREATE INDEX IF NOT EXISTS idx_hash_vhash ON gn090 (hash,vhash)",
 167                      NULL, NULL, NULL)) ||
 168       (SQLITE_OK !=
 169        sqlite3_exec (dbh,
 170                      "CREATE INDEX IF NOT EXISTS idx_expire_repl ON gn090 (expire ASC,repl DESC)",
 171                      NULL, NULL, NULL)) ||
 172       (SQLITE_OK !=
 173        sqlite3_exec (dbh,
 174                      "CREATE INDEX IF NOT EXISTS idx_comb ON gn090 (anonLevel ASC,expire ASC,prio,type,hash)",
 175                      NULL, NULL, NULL)) ||
 176       (SQLITE_OK !=
 177        sqlite3_exec (dbh,
 178                      "CREATE INDEX IF NOT EXISTS idx_anon_type_hash ON gn090 (anonLevel ASC,type,hash)",
 179                      NULL, NULL, NULL)) ||
 180       (SQLITE_OK !=
 181        sqlite3_exec (dbh,
 182                      "CREATE INDEX IF NOT EXISTS idx_expire ON gn090 (expire ASC)",
 183                      NULL, NULL, NULL)) ||
 184       (SQLITE_OK !=
 185        sqlite3_exec (dbh,
 186                      "CREATE INDEX IF NOT EXISTS idx_repl_rvalue ON gn090 (repl,rvalue)",
 187                      NULL, NULL, NULL)) ||
 188       (SQLITE_OK !=
 189        sqlite3_exec (dbh,
 190                      "CREATE INDEX IF NOT EXISTS idx_repl ON gn090 (repl DESC)",
 191                      NULL, NULL, NULL)))
 192     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR, "sqlite",
 193                      "Failed to create indices: %s\n", sqlite3_errmsg (dbh));
 194 }
 195
 196
 197 #if 0
 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_ERROR, "%s\n", e); sqlite3_free(e); }
 204 #endif
 205
 206
 207 /**
 208  * Initialize the database connections and associated
 209  * data structures (create tables and indices
 210  * as needed as well).
 211  *
 212  * @param cfg our configuration
 213  * @param plugin the plugin context (state for this module)
 214  * @return GNUNET_OK on success
 215  */
 216 static int
 217 database_setup (const struct GNUNET_CONFIGURATION_Handle *cfg,
 218                 struct Plugin *plugin)
 219 {
 220   sqlite3_stmt *stmt;
 221   char *afsdir;
 222
 223 #if ENULL_DEFINED
 224   char *e;
 225 #endif
 226
 227   if (GNUNET_OK !=
 228       GNUNET_CONFIGURATION_get_value_filename (cfg, "datastore-sqlite",
 229                                                "FILENAME", &afsdir))
 230   {
 231     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR, "sqlite",
 232                      _
 233                      ("Option `%s' in section `%s' missing in configuration!\n"),
 234                      "FILENAME", "datastore-sqlite");
 235     return GNUNET_SYSERR;
 236   }
 237   if (GNUNET_OK != GNUNET_DISK_file_test (afsdir))
 238   {
 239     if (GNUNET_OK != GNUNET_DISK_directory_create_for_file (afsdir))
 240     {
 241       GNUNET_break (0);
 242       GNUNET_free (afsdir);
 243       return GNUNET_SYSERR;
 244     }
 245     /* database is new or got deleted, reset payload to zero! */
 246     plugin->env->duc (plugin->env->cls, 0);
 247   }
 248 #ifdef ENABLE_NLS
 249   plugin->fn =
 250       GNUNET_STRINGS_to_utf8 (afsdir, strlen (afsdir), nl_langinfo (CODESET));
 251 #else
 252   plugin->fn = GNUNET_STRINGS_to_utf8 (afsdir, strlen (afsdir), "UTF-8");       /* good luck */
 253 #endif
 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, "sqlite",
 260                      _("Unable to initialize SQLite: %s.\n"),
 261                      sqlite3_errmsg (plugin->dbh));
 262     return GNUNET_SYSERR;
 263   }
 264   CHECK (SQLITE_OK ==
 265          sqlite3_exec (plugin->dbh, "PRAGMA temp_store=MEMORY", NULL, NULL,
 266                        ENULL));
 267   CHECK (SQLITE_OK ==
 268          sqlite3_exec (plugin->dbh, "PRAGMA synchronous=OFF", NULL, NULL,
 269                        ENULL));
 270   CHECK (SQLITE_OK ==
 271          sqlite3_exec (plugin->dbh, "PRAGMA legacy_file_format=OFF", NULL, NULL,
 272                        ENULL));
 273   CHECK (SQLITE_OK ==
 274          sqlite3_exec (plugin->dbh, "PRAGMA auto_vacuum=INCREMENTAL", NULL,
 275                        NULL, ENULL));
 276   CHECK (SQLITE_OK ==
 277          sqlite3_exec (plugin->dbh, "PRAGMA locking_mode=EXCLUSIVE", NULL, NULL,
 278                        ENULL));
 279   CHECK (SQLITE_OK ==
 280          sqlite3_exec (plugin->dbh, "PRAGMA count_changes=OFF", NULL, NULL,
 281                        ENULL));
 282   CHECK (SQLITE_OK ==
 283          sqlite3_exec (plugin->dbh, "PRAGMA page_size=4092", NULL, NULL,
 284                        ENULL));
 285
 286   CHECK (SQLITE_OK == sqlite3_busy_timeout (plugin->dbh, BUSY_TIMEOUT_MS));
 287
 288
 289   /* We have to do it here, because otherwise precompiling SQL might fail */
 290   CHECK (SQLITE_OK ==
 291          sq_prepare (plugin->dbh,
 292                      "SELECT 1 FROM sqlite_master WHERE tbl_name = 'gn090'",
 293                      &stmt));
 294   if ((sqlite3_step (stmt) == SQLITE_DONE) &&
 295       (sqlite3_exec
 296        (plugin->dbh,
 297         "CREATE TABLE gn090 (" "  repl INT4 NOT NULL DEFAULT 0,"
 298         "  type INT4 NOT NULL DEFAULT 0," "  prio INT4 NOT NULL DEFAULT 0,"
 299         "  anonLevel INT4 NOT NULL DEFAULT 0,"
 300         "  expire INT8 NOT NULL DEFAULT 0," "  rvalue INT8 NOT NULL,"
 301         "  hash TEXT NOT NULL DEFAULT ''," "  vhash TEXT NOT NULL DEFAULT '',"
 302         "  value BLOB NOT NULL DEFAULT '')", NULL, NULL, NULL) != SQLITE_OK))
 303   {
 304     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR, "sqlite3_exec");
 305     sqlite3_finalize (stmt);
 306     return GNUNET_SYSERR;
 307   }
 308   sqlite3_finalize (stmt);
 309   create_indices (plugin->dbh);
 310
 311   if ((sq_prepare
 312        (plugin->dbh,
 313         "UPDATE gn090 "
 314         "SET prio = prio + ?, expire = MAX(expire,?) WHERE _ROWID_ = ?",
 315         &plugin->updPrio) != SQLITE_OK) ||
 316       (sq_prepare
 317        (plugin->dbh,
 318         "UPDATE gn090 " "SET repl = MAX (0, repl - 1) WHERE _ROWID_ = ?",
 319         &plugin->updRepl) != SQLITE_OK) ||
 320       (sq_prepare
 321        (plugin->dbh,
 322         "SELECT type,prio,anonLevel,expire,hash,value,_ROWID_ " "FROM gn090 "
 323 #if SQLITE_VERSION_NUMBER >= 3007000
 324         "INDEXED BY idx_repl_rvalue "
 325 #endif
 326         "WHERE repl=?2 AND " " (rvalue>=?1 OR "
 327         "  NOT EXISTS (SELECT 1 FROM gn090 "
 328 #if SQLITE_VERSION_NUMBER >= 3007000
 329         "INDEXED BY idx_repl_rvalue "
 330 #endif
 331         "WHERE repl=?2 AND rvalue>=?1 LIMIT 1) ) "
 332         "ORDER BY rvalue ASC LIMIT 1", &plugin->selRepl) != SQLITE_OK) ||
 333       (sq_prepare (plugin->dbh, "SELECT MAX(repl) FROM gn090"
 334 #if SQLITE_VERSION_NUMBER >= 3007000
 335                    " INDEXED BY idx_repl_rvalue"
 336 #endif
 337                    "", &plugin->maxRepl) != SQLITE_OK) ||
 338       (sq_prepare
 339        (plugin->dbh,
 340         "SELECT type,prio,anonLevel,expire,hash,value,_ROWID_ " "FROM gn090 "
 341 #if SQLITE_VERSION_NUMBER >= 3007000
 342         "INDEXED BY idx_expire "
 343 #endif
 344         "WHERE NOT EXISTS (SELECT 1 FROM gn090 WHERE expire < ?1 LIMIT 1) OR (expire < ?1) "
 345         "ORDER BY expire ASC LIMIT 1", &plugin->selExpi) != SQLITE_OK) ||
 346       (sq_prepare
 347        (plugin->dbh,
 348         "SELECT type,prio,anonLevel,expire,hash,value,_ROWID_ " "FROM gn090 "
 349 #if SQLITE_VERSION_NUMBER >= 3007000
 350         "INDEXED BY idx_anon_type_hash "
 351 #endif
 352         "WHERE (anonLevel = 0 AND type=?1) "
 353         "ORDER BY hash DESC LIMIT 1 OFFSET ?2",
 354         &plugin->selZeroAnon) != SQLITE_OK) ||
 355       (sq_prepare
 356        (plugin->dbh,
 357         "INSERT INTO gn090 (repl, type, prio, anonLevel, expire, rvalue, hash, vhash, value) "
 358         "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
 359         &plugin->insertContent) != SQLITE_OK) ||
 360       (sq_prepare
 361        (plugin->dbh, "DELETE FROM gn090 WHERE _ROWID_ = ?",
 362         &plugin->delRow) != SQLITE_OK))
 363   {
 364     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR, "precompiling");
 365     return GNUNET_SYSERR;
 366   }
 367
 368   return GNUNET_OK;
 369 }
 370
 371
 372 /**
 373  * Shutdown database connection and associate data
 374  * structures.
 375  * @param plugin the plugin context (state for this module)
 376  */
 377 static void
 378 database_shutdown (struct Plugin *plugin)
 379 {
 380   int result;
 381
 382 #if SQLITE_VERSION_NUMBER >= 3007000
 383   sqlite3_stmt *stmt;
 384 #endif
 385
 386   if (plugin->delRow != NULL)
 387     sqlite3_finalize (plugin->delRow);
 388   if (plugin->updPrio != NULL)
 389     sqlite3_finalize (plugin->updPrio);
 390   if (plugin->updRepl != NULL)
 391     sqlite3_finalize (plugin->updRepl);
 392   if (plugin->selRepl != NULL)
 393     sqlite3_finalize (plugin->selRepl);
 394   if (plugin->maxRepl != NULL)
 395     sqlite3_finalize (plugin->maxRepl);
 396   if (plugin->selExpi != NULL)
 397     sqlite3_finalize (plugin->selExpi);
 398   if (plugin->selZeroAnon != NULL)
 399     sqlite3_finalize (plugin->selZeroAnon);
 400   if (plugin->insertContent != NULL)
 401     sqlite3_finalize (plugin->insertContent);
 402   result = sqlite3_close (plugin->dbh);
 403 #if SQLITE_VERSION_NUMBER >= 3007000
 404   if (result == SQLITE_BUSY)
 405   {
 406     GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING, "sqlite",
 407                      _
 408                      ("Tried to close sqlite without finalizing all prepared statements.\n"));
 409     stmt = sqlite3_next_stmt (plugin->dbh, NULL);
 410     while (stmt != NULL)
 411     {
 412       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
 413                        "Closing statement %p\n", stmt);
 414       result = sqlite3_finalize (stmt);
 415       if (result != SQLITE_OK)
 416         GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING, "sqlite",
 417                          "Failed to close statement %p: %d\n", stmt, result);
 418       stmt = sqlite3_next_stmt (plugin->dbh, NULL);
 419     }
 420     result = sqlite3_close (plugin->dbh);
 421   }
 422 #endif
 423   if (SQLITE_OK != result)
 424     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close");
 425
 426   GNUNET_free_non_null (plugin->fn);
 427 }
 428
 429
 430 /**
 431  * Delete the database entry with the given
 432  * row identifier.
 433  *
 434  * @param plugin the plugin context (state for this module)
 435  * @param rid the ID of the row to delete
 436  */
 437 static int
 438 delete_by_rowid (struct Plugin *plugin, unsigned long long rid)
 439 {
 440   if (SQLITE_OK != sqlite3_bind_int64 (plugin->delRow, 1, rid))
 441   {
 442     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 443                 "sqlite3_bind_XXXX");
 444     if (SQLITE_OK != sqlite3_reset (plugin->delRow))
 445       LOG_SQLITE (plugin, NULL,
 446                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 447                   "sqlite3_reset");
 448     return GNUNET_SYSERR;
 449   }
 450   if (SQLITE_DONE != sqlite3_step (plugin->delRow))
 451   {
 452     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 453                 "sqlite3_step");
 454     if (SQLITE_OK != sqlite3_reset (plugin->delRow))
 455       LOG_SQLITE (plugin, NULL,
 456                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 457                   "sqlite3_reset");
 458     return GNUNET_SYSERR;
 459   }
 460   if (SQLITE_OK != sqlite3_reset (plugin->delRow))
 461     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 462                 "sqlite3_reset");
 463   return GNUNET_OK;
 464 }
 465
 466
 467 /**
 468  * Store an item in the datastore.
 469  *
 470  * @param cls closure
 471  * @param key key for the item
 472  * @param size number of bytes in data
 473  * @param data content stored
 474  * @param type type of the content
 475  * @param priority priority of the content
 476  * @param anonymity anonymity-level for the content
 477  * @param replication replication-level for the content
 478  * @param expiration expiration time for the content
 479  * @param msg set to an error message
 480  * @return GNUNET_OK on success
 481  */
 482 static int
 483 sqlite_plugin_put (void *cls, const GNUNET_HashCode * key, uint32_t size,
 484                    const void *data, enum GNUNET_BLOCK_Type type,
 485                    uint32_t priority, uint32_t anonymity, uint32_t replication,
 486                    struct GNUNET_TIME_Absolute expiration, char **msg)
 487 {
 488   struct Plugin *plugin = cls;
 489   int n;
 490   int ret;
 491   sqlite3_stmt *stmt;
 492   GNUNET_HashCode vhash;
 493   uint64_t rvalue;
 494
 495   if (size > MAX_ITEM_SIZE)
 496     return GNUNET_SYSERR;
 497   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
 498                    "Storing in database block with type %u/key `%s'/priority %u/expiration in %llu ms (%lld).\n",
 499                    type, GNUNET_h2s (key), priority,
 500                    (unsigned long long)
 501                    GNUNET_TIME_absolute_get_remaining (expiration).rel_value,
 502                    (long long) expiration.abs_value);
 503   GNUNET_CRYPTO_hash (data, size, &vhash);
 504   stmt = plugin->insertContent;
 505   rvalue = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
 506   if ((SQLITE_OK != sqlite3_bind_int (stmt, 1, replication)) ||
 507       (SQLITE_OK != sqlite3_bind_int (stmt, 2, type)) ||
 508       (SQLITE_OK != sqlite3_bind_int (stmt, 3, priority)) ||
 509       (SQLITE_OK != sqlite3_bind_int (stmt, 4, anonymity)) ||
 510       (SQLITE_OK != sqlite3_bind_int64 (stmt, 5, expiration.abs_value)) ||
 511       (SQLITE_OK != sqlite3_bind_int64 (stmt, 6, rvalue)) ||
 512       (SQLITE_OK !=
 513        sqlite3_bind_blob (stmt, 7, key, sizeof (GNUNET_HashCode),
 514                           SQLITE_TRANSIENT)) ||
 515       (SQLITE_OK !=
 516        sqlite3_bind_blob (stmt, 8, &vhash, sizeof (GNUNET_HashCode),
 517                           SQLITE_TRANSIENT)) ||
 518       (SQLITE_OK != sqlite3_bind_blob (stmt, 9, data, size, SQLITE_TRANSIENT)))
 519   {
 520     LOG_SQLITE (plugin, msg, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 521                 "sqlite3_bind_XXXX");
 522     if (SQLITE_OK != sqlite3_reset (stmt))
 523       LOG_SQLITE (plugin, NULL,
 524                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 525                   "sqlite3_reset");
 526     return GNUNET_SYSERR;
 527   }
 528   n = sqlite3_step (stmt);
 529   switch (n)
 530   {
 531   case SQLITE_DONE:
 532     plugin->env->duc (plugin->env->cls, size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 533     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
 534                      "Stored new entry (%u bytes)\n",
 535                      size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 536     ret = GNUNET_OK;
 537     break;
 538   case SQLITE_BUSY:
 539     GNUNET_break (0);
 540     LOG_SQLITE (plugin, msg, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 541                 "sqlite3_step");
 542     ret = GNUNET_SYSERR;
 543     break;
 544   default:
 545     LOG_SQLITE (plugin, msg, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 546                 "sqlite3_step");
 547     if (SQLITE_OK != sqlite3_reset (stmt))
 548       LOG_SQLITE (plugin, NULL,
 549                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 550                   "sqlite3_reset");
 551     database_shutdown (plugin);
 552     database_setup (plugin->env->cfg, plugin);
 553     return GNUNET_SYSERR;
 554   }
 555   if (SQLITE_OK != sqlite3_reset (stmt))
 556     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 557                 "sqlite3_reset");
 558   return ret;
 559 }
 560
 561
 562 /**
 563  * Update the priority for a particular key in the datastore.  If
 564  * the expiration time in value is different than the time found in
 565  * the datastore, the higher value should be kept.  For the
 566  * anonymity level, the lower value is to be used.  The specified
 567  * priority should be added to the existing priority, ignoring the
 568  * priority in value.
 569  *
 570  * Note that it is possible for multiple values to match this put.
 571  * In that case, all of the respective values are updated.
 572  *
 573  * @param cls the plugin context (state for this module)
 574  * @param uid unique identifier of the datum
 575  * @param delta by how much should the priority
 576  *     change?  If priority + delta < 0 the
 577  *     priority should be set to 0 (never go
 578  *     negative).
 579  * @param expire new expiration time should be the
 580  *     MAX of any existing expiration time and
 581  *     this value
 582  * @param msg set to an error message
 583  * @return GNUNET_OK on success
 584  */
 585 static int
 586 sqlite_plugin_update (void *cls, uint64_t uid, int delta,
 587                       struct GNUNET_TIME_Absolute expire, char **msg)
 588 {
 589   struct Plugin *plugin = cls;
 590   int n;
 591
 592   if ((SQLITE_OK != sqlite3_bind_int (plugin->updPrio, 1, delta)) ||
 593       (SQLITE_OK != sqlite3_bind_int64 (plugin->updPrio, 2, expire.abs_value))
 594       || (SQLITE_OK != sqlite3_bind_int64 (plugin->updPrio, 3, uid)))
 595   {
 596     LOG_SQLITE (plugin, msg, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 597                 "sqlite3_bind_XXXX");
 598     if (SQLITE_OK != sqlite3_reset (plugin->updPrio))
 599       LOG_SQLITE (plugin, NULL,
 600                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 601                   "sqlite3_reset");
 602     return GNUNET_SYSERR;
 603
 604   }
 605   n = sqlite3_step (plugin->updPrio);
 606   if (SQLITE_OK != sqlite3_reset (plugin->updPrio))
 607     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 608                 "sqlite3_reset");
 609   switch (n)
 610   {
 611   case SQLITE_DONE:
 612     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite", "Block updated\n");
 613     return GNUNET_OK;
 614   case SQLITE_BUSY:
 615     LOG_SQLITE (plugin, msg, GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
 616                 "sqlite3_step");
 617     return GNUNET_NO;
 618   default:
 619     LOG_SQLITE (plugin, msg, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 620                 "sqlite3_step");
 621     return GNUNET_SYSERR;
 622   }
 623 }
 624
 625
 626 /**
 627  * Execute statement that gets a row and call the callback
 628  * with the result.  Resets the statement afterwards.
 629  *
 630  * @param plugin the plugin
 631  * @param stmt the statement
 632  * @param proc processor to call
 633  * @param proc_cls closure for 'proc'
 634  */
 635 static void
 636 execute_get (struct Plugin *plugin, sqlite3_stmt * stmt,
 637              PluginDatumProcessor proc, void *proc_cls)
 638 {
 639   int n;
 640   struct GNUNET_TIME_Absolute expiration;
 641   unsigned long long rowid;
 642   unsigned int size;
 643   int ret;
 644
 645   n = sqlite3_step (stmt);
 646   switch (n)
 647   {
 648   case SQLITE_ROW:
 649     size = sqlite3_column_bytes (stmt, 5);
 650     rowid = sqlite3_column_int64 (stmt, 6);
 651     if (sqlite3_column_bytes (stmt, 4) != sizeof (GNUNET_HashCode))
 652     {
 653       GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING, "sqlite",
 654                        _
 655                        ("Invalid data in database.  Trying to fix (by deletion).\n"));
 656       if (SQLITE_OK != sqlite3_reset (stmt))
 657         LOG_SQLITE (plugin, NULL,
 658                     GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 659                     "sqlite3_reset");
 660       if (GNUNET_OK == delete_by_rowid (plugin, rowid))
 661         plugin->env->duc (plugin->env->cls,
 662                           -(size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
 663       break;
 664     }
 665     expiration.abs_value = sqlite3_column_int64 (stmt, 3);
 666     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
 667                      "Found reply in database with expiration %llu\n",
 668                      (unsigned long long) expiration.abs_value);
 669     ret = proc (proc_cls, sqlite3_column_blob (stmt, 4) /* key */ ,
 670                 size, sqlite3_column_blob (stmt, 5) /* data */ ,
 671                 sqlite3_column_int (stmt, 0) /* type */ ,
 672                 sqlite3_column_int (stmt, 1) /* priority */ ,
 673                 sqlite3_column_int (stmt, 2) /* anonymity */ ,
 674                 expiration, rowid);
 675     if (SQLITE_OK != sqlite3_reset (stmt))
 676       LOG_SQLITE (plugin, NULL,
 677                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 678                   "sqlite3_reset");
 679     if ((GNUNET_NO == ret) && (GNUNET_OK == delete_by_rowid (plugin, rowid)))
 680       plugin->env->duc (plugin->env->cls,
 681                         -(size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
 682     return;
 683   case SQLITE_DONE:
 684     /* database must be empty */
 685     if (SQLITE_OK != sqlite3_reset (stmt))
 686       LOG_SQLITE (plugin, NULL,
 687                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 688                   "sqlite3_reset");
 689     break;
 690   case SQLITE_BUSY:
 691   case SQLITE_ERROR:
 692   case SQLITE_MISUSE:
 693   default:
 694     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 695                 "sqlite3_step");
 696     if (SQLITE_OK != sqlite3_reset (stmt))
 697       LOG_SQLITE (plugin, NULL,
 698                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 699                   "sqlite3_reset");
 700     GNUNET_break (0);
 701     database_shutdown (plugin);
 702     database_setup (plugin->env->cfg, plugin);
 703     break;
 704   }
 705   if (SQLITE_OK != sqlite3_reset (stmt))
 706     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 707                 "sqlite3_reset");
 708   proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 709 }
 710
 711
 712
 713 /**
 714  * Select a subset of the items in the datastore and call
 715  * the given processor for the item.
 716  *
 717  * @param cls our plugin context
 718  * @param offset offset of the result (modulo num-results);
 719  *               specific ordering does not matter for the offset
 720  * @param type entries of which type should be considered?
 721  *        Use 0 for any type.
 722  * @param proc function to call on each matching value;
 723  *        will be called once with a NULL value at the end
 724  * @param proc_cls closure for proc
 725  */
 726 static void
 727 sqlite_plugin_get_zero_anonymity (void *cls, uint64_t offset,
 728                                   enum GNUNET_BLOCK_Type type,
 729                                   PluginDatumProcessor proc, void *proc_cls)
 730 {
 731   struct Plugin *plugin = cls;
 732   sqlite3_stmt *stmt;
 733
 734   GNUNET_assert (type != GNUNET_BLOCK_TYPE_ANY);
 735   stmt = plugin->selZeroAnon;
 736   if ((SQLITE_OK != sqlite3_bind_int (stmt, 1, type)) ||
 737       (SQLITE_OK != sqlite3_bind_int64 (stmt, 2, offset)))
 738   {
 739     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 740                 "sqlite3_bind_XXXX");
 741     if (SQLITE_OK != sqlite3_reset (stmt))
 742       LOG_SQLITE (plugin, NULL,
 743                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 744                   "sqlite3_reset");
 745     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 746     return;
 747   }
 748   execute_get (plugin, stmt, proc, proc_cls);
 749 }
 750
 751
 752
 753 /**
 754  * Get results for a particular key in the datastore.
 755  *
 756  * @param cls closure
 757  * @param offset offset (mod count).
 758  * @param key key to match, never NULL
 759  * @param vhash hash of the value, maybe NULL (to
 760  *        match all values that have the right key).
 761  *        Note that for DBlocks there is no difference
 762  *        betwen key and vhash, but for other blocks
 763  *        there may be!
 764  * @param type entries of which type are relevant?
 765  *     Use 0 for any type.
 766  * @param proc function to call on each matching value;
 767  *        will be called once with a NULL value at the end
 768  * @param proc_cls closure for proc
 769  */
 770 static void
 771 sqlite_plugin_get_key (void *cls, uint64_t offset, const GNUNET_HashCode * key,
 772                        const GNUNET_HashCode * vhash,
 773                        enum GNUNET_BLOCK_Type type, PluginDatumProcessor proc,
 774                        void *proc_cls)
 775 {
 776   struct Plugin *plugin = cls;
 777   int ret;
 778   int total;
 779   int limit_off;
 780   unsigned int sqoff;
 781   sqlite3_stmt *stmt;
 782   char scratch[256];
 783
 784   GNUNET_assert (proc != NULL);
 785   GNUNET_assert (key != NULL);
 786   GNUNET_snprintf (scratch, sizeof (scratch),
 787                    "SELECT count(*) FROM gn090 WHERE hash=?%s%s",
 788                    vhash == NULL ? "" : " AND vhash=?",
 789                    type == 0 ? "" : " AND type=?");
 790   if (sq_prepare (plugin->dbh, scratch, &stmt) != SQLITE_OK)
 791   {
 792     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 793                 "sqlite_prepare");
 794     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 795     return;
 796   }
 797   sqoff = 1;
 798   ret =
 799       sqlite3_bind_blob (stmt, sqoff++, key, sizeof (GNUNET_HashCode),
 800                          SQLITE_TRANSIENT);
 801   if ((vhash != NULL) && (ret == SQLITE_OK))
 802     ret =
 803         sqlite3_bind_blob (stmt, sqoff++, vhash, sizeof (GNUNET_HashCode),
 804                            SQLITE_TRANSIENT);
 805   if ((type != 0) && (ret == SQLITE_OK))
 806     ret = sqlite3_bind_int (stmt, sqoff++, type);
 807   if (SQLITE_OK != ret)
 808   {
 809     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR, "sqlite_bind");
 810     sqlite3_finalize (stmt);
 811     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 812     return;
 813   }
 814   ret = sqlite3_step (stmt);
 815   if (ret != SQLITE_ROW)
 816   {
 817     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 818                 "sqlite_step");
 819     sqlite3_finalize (stmt);
 820     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 821     return;
 822   }
 823   total = sqlite3_column_int (stmt, 0);
 824   sqlite3_finalize (stmt);
 825   if (0 == total)
 826   {
 827     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 828     return;
 829   }
 830   limit_off = (int) (offset % total);
 831   if (limit_off < 0)
 832     limit_off += total;
 833   GNUNET_snprintf (scratch, sizeof (scratch),
 834                    "SELECT type, prio, anonLevel, expire, hash, value, _ROWID_ "
 835                    "FROM gn090 WHERE hash=?%s%s "
 836                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET ?",
 837                    vhash == NULL ? "" : " AND vhash=?",
 838                    type == 0 ? "" : " AND type=?");
 839   if (sq_prepare (plugin->dbh, scratch, &stmt) != SQLITE_OK)
 840   {
 841     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 842                 "sqlite_prepare");
 843     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 844     return;
 845   }
 846   sqoff = 1;
 847   ret =
 848       sqlite3_bind_blob (stmt, sqoff++, key, sizeof (GNUNET_HashCode),
 849                          SQLITE_TRANSIENT);
 850   if ((vhash != NULL) && (ret == SQLITE_OK))
 851     ret =
 852         sqlite3_bind_blob (stmt, sqoff++, vhash, sizeof (GNUNET_HashCode),
 853                            SQLITE_TRANSIENT);
 854   if ((type != 0) && (ret == SQLITE_OK))
 855     ret = sqlite3_bind_int (stmt, sqoff++, type);
 856   if (ret == SQLITE_OK)
 857     ret = sqlite3_bind_int64 (stmt, sqoff++, limit_off);
 858   if (ret != SQLITE_OK)
 859   {
 860     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 861                 "sqlite_bind");
 862     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 863     return;
 864   }
 865   execute_get (plugin, stmt, proc, proc_cls);
 866   sqlite3_finalize (stmt);
 867 }
 868
 869
 870
 871 /**
 872  * Context for 'repl_proc' function.
 873  */
 874 struct ReplCtx
 875 {
 876
 877   /**
 878    * Function to call for the result (or the NULL).
 879    */
 880   PluginDatumProcessor proc;
 881
 882   /**
 883    * Closure for proc.
 884    */
 885   void *proc_cls;
 886
 887   /**
 888    * UID to use.
 889    */
 890   uint64_t uid;
 891
 892   /**
 893    * Yes if UID was set.
 894    */
 895   int have_uid;
 896 };
 897
 898
 899 /**
 900  * Wrapper for the processor for 'sqlite_plugin_replication_get'.
 901  * Decrements the replication counter and calls the original
 902  * processor.
 903  *
 904  * @param cls closure
 905  * @param key key for the content
 906  * @param size number of bytes in data
 907  * @param data content stored
 908  * @param type type of the content
 909  * @param priority priority of the content
 910  * @param anonymity anonymity-level for the content
 911  * @param expiration expiration time for the content
 912  * @param uid unique identifier for the datum;
 913  *        maybe 0 if no unique identifier is available
 914  *
 915  * @return GNUNET_OK for normal return,
 916  *         GNUNET_NO to delete the item
 917  */
 918 static int
 919 repl_proc (void *cls, const GNUNET_HashCode * key, uint32_t size,
 920            const void *data, enum GNUNET_BLOCK_Type type, uint32_t priority,
 921            uint32_t anonymity, struct GNUNET_TIME_Absolute expiration,
 922            uint64_t uid)
 923 {
 924   struct ReplCtx *rc = cls;
 925   int ret;
 926
 927   ret =
 928       rc->proc (rc->proc_cls, key, size, data, type, priority, anonymity,
 929                 expiration, uid);
 930   if (key != NULL)
 931   {
 932     rc->uid = uid;
 933     rc->have_uid = GNUNET_YES;
 934   }
 935   return ret;
 936 }
 937
 938
 939 /**
 940  * Get a random item for replication.  Returns a single random item
 941  * from those with the highest replication counters.  The item's
 942  * replication counter is decremented by one IF it was positive before.
 943  * Call 'proc' with all values ZERO or NULL if the datastore is empty.
 944  *
 945  * @param cls closure
 946  * @param proc function to call the value (once only).
 947  * @param proc_cls closure for proc
 948  */
 949 static void
 950 sqlite_plugin_get_replication (void *cls, PluginDatumProcessor proc,
 951                                void *proc_cls)
 952 {
 953   struct Plugin *plugin = cls;
 954   struct ReplCtx rc;
 955   uint64_t rvalue;
 956   uint32_t repl;
 957   sqlite3_stmt *stmt;
 958
 959   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
 960                    "Getting random block based on replication order.\n");
 961   rc.have_uid = GNUNET_NO;
 962   rc.proc = proc;
 963   rc.proc_cls = proc_cls;
 964   stmt = plugin->maxRepl;
 965   if (SQLITE_ROW != sqlite3_step (stmt))
 966   {
 967     if (SQLITE_OK != sqlite3_reset (stmt))
 968       LOG_SQLITE (plugin, NULL,
 969                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 970                   "sqlite3_reset");
 971     /* DB empty */
 972     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 973     return;
 974   }
 975   repl = sqlite3_column_int (stmt, 0);
 976   if (SQLITE_OK != sqlite3_reset (stmt))
 977     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 978                 "sqlite3_reset");
 979   stmt = plugin->selRepl;
 980   rvalue = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
 981   if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, rvalue))
 982   {
 983     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 984                 "sqlite3_bind_XXXX");
 985     if (SQLITE_OK != sqlite3_reset (stmt))
 986       LOG_SQLITE (plugin, NULL,
 987                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 988                   "sqlite3_reset");
 989     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 990     return;
 991   }
 992   if (SQLITE_OK != sqlite3_bind_int (stmt, 2, repl))
 993   {
 994     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 995                 "sqlite3_bind_XXXX");
 996     if (SQLITE_OK != sqlite3_reset (stmt))
 997       LOG_SQLITE (plugin, NULL,
 998                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 999                   "sqlite3_reset");
1000     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1001     return;
1002   }
1003   execute_get (plugin, stmt, &repl_proc, &rc);
1004   if (GNUNET_YES == rc.have_uid)
1005   {
1006     if (SQLITE_OK != sqlite3_bind_int64 (plugin->updRepl, 1, rc.uid))
1007     {
1008       LOG_SQLITE (plugin, NULL,
1009                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1010                   "sqlite3_bind_XXXX");
1011       if (SQLITE_OK != sqlite3_reset (plugin->updRepl))
1012         LOG_SQLITE (plugin, NULL,
1013                     GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1014                     "sqlite3_reset");
1015       return;
1016     }
1017     if (SQLITE_DONE != sqlite3_step (plugin->updRepl))
1018       LOG_SQLITE (plugin, NULL,
1019                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1020                   "sqlite3_step");
1021     if (SQLITE_OK != sqlite3_reset (plugin->updRepl))
1022       LOG_SQLITE (plugin, NULL,
1023                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1024                   "sqlite3_reset");
1025   }
1026 }
1027
1028
1029
1030 /**
1031  * Get a random item that has expired or has low priority.
1032  * Call 'proc' with all values ZERO or NULL if the datastore is empty.
1033  *
1034  * @param cls closure
1035  * @param proc function to call the value (once only).
1036  * @param proc_cls closure for proc
1037  */
1038 static void
1039 sqlite_plugin_get_expiration (void *cls, PluginDatumProcessor proc,
1040                               void *proc_cls)
1041 {
1042   struct Plugin *plugin = cls;
1043   sqlite3_stmt *stmt;
1044   struct GNUNET_TIME_Absolute now;
1045
1046   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
1047                    "Getting random block based on expiration and priority order.\n");
1048   now = GNUNET_TIME_absolute_get ();
1049   stmt = plugin->selExpi;
1050   if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, now.abs_value))
1051   {
1052     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1053                 "sqlite3_bind_XXXX");
1054     if (SQLITE_OK != sqlite3_reset (stmt))
1055       LOG_SQLITE (plugin, NULL,
1056                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1057                   "sqlite3_reset");
1058     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1059     return;
1060   }
1061   execute_get (plugin, stmt, proc, proc_cls);
1062 }
1063
1064
1065
1066 /**
1067  * Get all of the keys in the datastore.
1068  *
1069  * @param cls closure
1070  * @param proc function to call on each key
1071  * @param proc_cls closure for proc
1072  */
1073 static void
1074 sqlite_plugin_get_keys (void *cls,
1075                         PluginKeyProcessor proc,
1076                         void *proc_cls)
1077 {
1078   struct Plugin *plugin = cls;
1079   const GNUNET_HashCode *key;
1080   sqlite3_stmt *stmt;
1081   int ret;
1082
1083   GNUNET_assert (proc != NULL);
1084   if (sq_prepare (plugin->dbh, "SELECT hash FROM gn090", &stmt) != SQLITE_OK)
1085   {
1086     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
1087                 "sqlite_prepare");
1088     return;
1089   }
1090   while (SQLITE_ROW == (ret = sqlite3_step (stmt)))
1091   {
1092     key = sqlite3_column_blob (stmt, 1);
1093     if (sizeof (GNUNET_HashCode) == sqlite3_column_bytes (stmt, 1))
1094       proc (proc_cls, key, 1);
1095   }
1096   if (SQLITE_DONE != ret)
1097     LOG_SQLITE (plugin, NULL, GNUNET_ERROR_TYPE_ERROR, "sqlite_step");
1098   sqlite3_finalize (stmt);
1099 }
1100
1101
1102 /**
1103  * Drop database.
1104  *
1105  * @param cls our plugin context
1106  */
1107 static void
1108 sqlite_plugin_drop (void *cls)
1109 {
1110   struct Plugin *plugin = cls;
1111
1112   plugin->drop_on_shutdown = GNUNET_YES;
1113 }
1114
1115
1116 /**
1117  * Get an estimate of how much space the database is
1118  * currently using.
1119  *
1120  * @param cls the 'struct Plugin'
1121  * @return the size of the database on disk (estimate)
1122  */
1123 static unsigned long long
1124 sqlite_plugin_estimate_size (void *cls)
1125 {
1126   struct Plugin *plugin = cls;
1127   sqlite3_stmt *stmt;
1128   uint64_t pages;
1129   uint64_t page_size;
1130
1131 #if ENULL_DEFINED
1132   char *e;
1133 #endif
1134
1135   if (SQLITE_VERSION_NUMBER < 3006000)
1136   {
1137     GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING, "datastore-sqlite",
1138                      _
1139                      ("sqlite version to old to determine size, assuming zero\n"));
1140     return 0;
1141   }
1142   CHECK (SQLITE_OK == sqlite3_exec (plugin->dbh, "VACUUM", NULL, NULL, ENULL));
1143   CHECK (SQLITE_OK ==
1144          sqlite3_exec (plugin->dbh, "PRAGMA auto_vacuum=INCREMENTAL", NULL,
1145                        NULL, ENULL));
1146   CHECK (SQLITE_OK == sq_prepare (plugin->dbh, "PRAGMA page_count", &stmt));
1147   if (SQLITE_ROW == sqlite3_step (stmt))
1148     pages = sqlite3_column_int64 (stmt, 0);
1149   else
1150     pages = 0;
1151   sqlite3_finalize (stmt);
1152   CHECK (SQLITE_OK == sq_prepare (plugin->dbh, "PRAGMA page_size", &stmt));
1153   CHECK (SQLITE_ROW == sqlite3_step (stmt));
1154   page_size = sqlite3_column_int64 (stmt, 0);
1155   sqlite3_finalize (stmt);
1156   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1157               _
1158               ("Using sqlite page utilization to estimate payload (%llu pages of size %llu bytes)\n"),
1159               (unsigned long long) pages, (unsigned long long) page_size);
1160   return pages * page_size;
1161 }
1162
1163
1164 /**
1165  * Entry point for the plugin.
1166  *
1167  * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
1168  * @return NULL on error, othrewise the plugin context
1169  */
1170 void *
1171 libgnunet_plugin_datastore_sqlite_init (void *cls)
1172 {
1173   static struct Plugin plugin;
1174   struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
1175   struct GNUNET_DATASTORE_PluginFunctions *api;
1176
1177   if (plugin.env != NULL)
1178     return NULL;                /* can only initialize once! */
1179   memset (&plugin, 0, sizeof (struct Plugin));
1180   plugin.env = env;
1181   if (GNUNET_OK != database_setup (env->cfg, &plugin))
1182   {
1183     database_shutdown (&plugin);
1184     return NULL;
1185   }
1186   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
1187   api->cls = &plugin;
1188   api->estimate_size = &sqlite_plugin_estimate_size;
1189   api->put = &sqlite_plugin_put;
1190   api->update = &sqlite_plugin_update;
1191   api->get_key = &sqlite_plugin_get_key;
1192   api->get_replication = &sqlite_plugin_get_replication;
1193   api->get_expiration = &sqlite_plugin_get_expiration;
1194   api->get_zero_anonymity = &sqlite_plugin_get_zero_anonymity;
1195   api->get_keys = &sqlite_plugin_get_keys;
1196   api->drop = &sqlite_plugin_drop;
1197   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO, "sqlite",
1198                    _("Sqlite database running\n"));
1199   return api;
1200 }
1201
1202
1203 /**
1204  * Exit point from the plugin.
1205  *
1206  * @param cls the plugin context (as returned by "init")
1207  * @return always NULL
1208  */
1209 void *
1210 libgnunet_plugin_datastore_sqlite_done (void *cls)
1211 {
1212   char *fn;
1213   struct GNUNET_DATASTORE_PluginFunctions *api = cls;
1214   struct Plugin *plugin = api->cls;
1215
1216   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
1217                    "sqlite plugin is done\n");
1218   fn = NULL;
1219   if (plugin->drop_on_shutdown)
1220     fn = GNUNET_strdup (plugin->fn);
1221   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
1222                    "Shutting down database\n");
1223   database_shutdown (plugin);
1224   plugin->env = NULL;
1225   GNUNET_free (api);
1226   if (fn != NULL)
1227   {
1228     if (0 != UNLINK (fn))
1229       GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING, "unlink", fn);
1230     GNUNET_free (fn);
1231   }
1232   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite",
1233                    "sqlite plugin is finished\n");
1234   return NULL;
1235 }
1236
1237 /* end of plugin_datastore_sqlite.c */