src/datastore/plugin_datastore_sqlite.c

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