src/datastore/plugin_datastore_sqlite.c

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