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