src/datastore/plugin_datastore_sqlite.c

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