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   sqlite3_stmt *stmt;
 363   stmt = NULL;
 364
 365   if (plugin->delRow != NULL)
 366     sqlite3_finalize (plugin->delRow);
 367   if (plugin->updPrio != NULL)
 368     sqlite3_finalize (plugin->updPrio);
 369   if (plugin->insertContent != NULL)
 370     sqlite3_finalize (plugin->insertContent);
 371   result = sqlite3_close(plugin->dbh);
 372 #if SQLITE_VERSION_NUMBER >= 3007000
 373   if (result == SQLITE_BUSY)
 374     {
 375       GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING,
 376                        "sqlite",
 377                        _("Tried to close sqlite without finalizing all prepared statements.\n"));
 378       stmt = sqlite3_next_stmt(plugin->dbh, NULL);
 379       while (stmt != NULL)
 380         {
 381 #if DEBUG_SQLITE
 382           GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 383                      "sqlite", "Closing statement %p\n", stmt);
 384 #endif
 385           result = sqlite3_finalize(stmt);
 386 #if DEBUG_SQLITE
 387           if (result != SQLITE_OK)
 388               GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 389                                "sqlite",
 390                                "Failed to close statement %p: %d\n", stmt, result);
 391 #endif
 392           stmt = sqlite3_next_stmt(plugin->dbh, NULL);
 393         }
 394       result = sqlite3_close(plugin->dbh);
 395     }
 396 #endif
 397   if (SQLITE_OK != result)
 398       LOG_SQLITE (plugin, NULL,
 399                   GNUNET_ERROR_TYPE_ERROR,
 400                   "sqlite3_close");
 401
 402   GNUNET_free_non_null (plugin->fn);
 403 }
 404
 405
 406 /**
 407  * Delete the database entry with the given
 408  * row identifier.
 409  *
 410  * @param plugin the plugin context (state for this module)
 411  * @param rid the ID of the row to delete
 412  */
 413 static int
 414 delete_by_rowid (struct Plugin* plugin,
 415                  unsigned long long rid)
 416 {
 417
 418   sqlite3_bind_int64 (plugin->delRow, 1, rid);
 419   if (SQLITE_DONE != sqlite3_step (plugin->delRow))
 420     {
 421       LOG_SQLITE (plugin, NULL,
 422                   GNUNET_ERROR_TYPE_ERROR |
 423                   GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 424       if (SQLITE_OK != sqlite3_reset (plugin->delRow))
 425           LOG_SQLITE (plugin, NULL,
 426                       GNUNET_ERROR_TYPE_ERROR |
 427                       GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 428       return GNUNET_SYSERR;
 429     }
 430   if (SQLITE_OK != sqlite3_reset (plugin->delRow))
 431       LOG_SQLITE (plugin, NULL,
 432                   GNUNET_ERROR_TYPE_ERROR |
 433                   GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 434   return GNUNET_OK;
 435 }
 436
 437
 438 /**
 439  * Context for the universal iterator.
 440  */
 441 struct NextContext;
 442
 443 /**
 444  * Type of a function that will prepare
 445  * the next iteration.
 446  *
 447  * @param cls closure
 448  * @param nc the next context; NULL for the last
 449  *         call which gives the callback a chance to
 450  *         clean up the closure
 451  * @return GNUNET_OK on success, GNUNET_NO if there are
 452  *         no more values, GNUNET_SYSERR on error
 453  */
 454 typedef int (*PrepareFunction)(void *cls,
 455                                struct NextContext *nc);
 456
 457
 458 /**
 459  * Context we keep for the "next request" callback.
 460  */
 461 struct NextContext
 462 {
 463   /**
 464    * Internal state.
 465    */
 466   struct Plugin *plugin;
 467
 468   /**
 469    * Function to call on the next value.
 470    */
 471   PluginIterator iter;
 472
 473   /**
 474    * Closure for iter.
 475    */
 476   void *iter_cls;
 477
 478   /**
 479    * Function to call to prepare the next
 480    * iteration.
 481    */
 482   PrepareFunction prep;
 483
 484   /**
 485    * Closure for prep.
 486    */
 487   void *prep_cls;
 488
 489   /**
 490    * Statement that the iterator will get the data
 491    * from (updated or set by prep).
 492    */
 493   sqlite3_stmt *stmt;
 494
 495   /**
 496    * Row ID of the last result.
 497    */
 498   unsigned long long last_rowid;
 499
 500   /**
 501    * Key of the last result.
 502    */
 503   GNUNET_HashCode lastKey;
 504
 505   /**
 506    * Expiration time of the last value visited.
 507    */
 508   struct GNUNET_TIME_Absolute lastExpiration;
 509
 510   /**
 511    * Priority of the last value visited.
 512    */
 513   unsigned int lastPriority;
 514
 515   /**
 516    * Number of results processed so far.
 517    */
 518   unsigned int count;
 519
 520   /**
 521    * Set to GNUNET_YES if we must stop now.
 522    */
 523   int end_it;
 524 };
 525
 526
 527 /**
 528  * Continuation of "sqlite_next_request".
 529  *
 530  * @param cls the next context
 531  * @param tc the task context (unused)
 532  */
 533 static void
 534 sqlite_next_request_cont (void *cls,
 535                           const struct GNUNET_SCHEDULER_TaskContext *tc)
 536 {
 537   struct NextContext * nc = cls;
 538   struct Plugin *plugin;
 539   unsigned long long rowid;
 540   sqlite3_stmt *stmtd;
 541   int ret;
 542   unsigned int type;
 543   unsigned int size;
 544   unsigned int priority;
 545   unsigned int anonymity;
 546   struct GNUNET_TIME_Absolute expiration;
 547   const GNUNET_HashCode *key;
 548   const void *data;
 549
 550   plugin = nc->plugin;
 551   plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
 552   plugin->next_task_nc = NULL;
 553   if ( (GNUNET_YES == nc->end_it) ||
 554        (GNUNET_OK != (nc->prep(nc->prep_cls,
 555                                nc))) )
 556     {
 557     END:
 558       nc->iter (nc->iter_cls,
 559                 NULL, NULL, 0, NULL, 0, 0, 0,
 560                 GNUNET_TIME_UNIT_ZERO_ABS, 0);
 561       nc->prep (nc->prep_cls, NULL);
 562       GNUNET_free (nc);
 563       return;
 564     }
 565
 566   rowid = sqlite3_column_int64 (nc->stmt, 7);
 567   nc->last_rowid = rowid;
 568   type = sqlite3_column_int (nc->stmt, 1);
 569   size = sqlite3_column_bytes (nc->stmt, 6);
 570   if (sqlite3_column_bytes (nc->stmt, 5) != sizeof (GNUNET_HashCode))
 571     {
 572       GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING,
 573                        "sqlite",
 574                        _("Invalid data in database.  Trying to fix (by deletion).\n"));
 575       if (SQLITE_OK != sqlite3_reset (nc->stmt))
 576         LOG_SQLITE (nc->plugin, NULL,
 577                     GNUNET_ERROR_TYPE_ERROR |
 578                     GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 579       if (sq_prepare
 580           (nc->plugin->dbh,
 581            "DELETE FROM gn080 WHERE NOT LENGTH(hash) = ?",
 582            &stmtd) != SQLITE_OK)
 583         {
 584           LOG_SQLITE (nc->plugin, NULL,
 585                       GNUNET_ERROR_TYPE_ERROR |
 586                       GNUNET_ERROR_TYPE_BULK,
 587                       "sq_prepare");
 588           goto END;
 589         }
 590
 591       if (SQLITE_OK != sqlite3_bind_int (stmtd, 1, sizeof (GNUNET_HashCode)))
 592         LOG_SQLITE (nc->plugin, NULL,
 593                     GNUNET_ERROR_TYPE_ERROR |
 594                     GNUNET_ERROR_TYPE_BULK, "sqlite3_bind_int");
 595       if (SQLITE_DONE != sqlite3_step (stmtd))
 596         LOG_SQLITE (nc->plugin, NULL,
 597                     GNUNET_ERROR_TYPE_ERROR |
 598                     GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 599       if (SQLITE_OK != sqlite3_finalize (stmtd))
 600         LOG_SQLITE (nc->plugin, NULL,
 601                     GNUNET_ERROR_TYPE_ERROR |
 602                     GNUNET_ERROR_TYPE_BULK, "sqlite3_finalize");
 603       goto END;
 604     }
 605
 606   priority = sqlite3_column_int (nc->stmt, 2);
 607   anonymity = sqlite3_column_int (nc->stmt, 3);
 608   expiration.abs_value = sqlite3_column_int64 (nc->stmt, 4);
 609   key = sqlite3_column_blob (nc->stmt, 5);
 610   nc->lastPriority = priority;
 611   nc->lastExpiration = expiration;
 612   memcpy (&nc->lastKey, key, sizeof(GNUNET_HashCode));
 613   data = sqlite3_column_blob (nc->stmt, 6);
 614   nc->count++;
 615   ret = nc->iter (nc->iter_cls,
 616                   nc,
 617                   key,
 618                   size,
 619                   data,
 620                   type,
 621                   priority,
 622                   anonymity,
 623                   expiration,
 624                   rowid);
 625   if (ret == GNUNET_SYSERR)
 626     {
 627       nc->end_it = GNUNET_YES;
 628       return;
 629     }
 630 #if DEBUG_SQLITE
 631   if (ret == GNUNET_NO)
 632     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 633                      "sqlite",
 634                      "Asked to remove entry %llu (%u bytes)\n",
 635                      (unsigned long long) rowid,
 636                      size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 637 #endif
 638   if ( (ret == GNUNET_NO) &&
 639        (GNUNET_OK == delete_by_rowid (plugin, rowid)) )
 640     {
 641       plugin->env->duc (plugin->env->cls,
 642                         - (size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
 643 #if DEBUG_SQLITE
 644       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 645                        "sqlite",
 646                        "Removed entry %llu (%u bytes)\n",
 647                        (unsigned long long) rowid,
 648                        size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 649 #endif
 650     }
 651 }
 652
 653
 654 /**
 655  * Function invoked on behalf of a "PluginIterator"
 656  * asking the database plugin to call the iterator
 657  * with the next item.
 658  *
 659  * @param next_cls whatever argument was given
 660  *        to the PluginIterator as "next_cls".
 661  * @param end_it set to GNUNET_YES if we
 662  *        should terminate the iteration early
 663  *        (iterator should be still called once more
 664  *         to signal the end of the iteration).
 665  */
 666 static void
 667 sqlite_next_request (void *next_cls,
 668                      int end_it)
 669 {
 670   struct NextContext * nc= next_cls;
 671
 672   if (GNUNET_YES == end_it)
 673     nc->end_it = GNUNET_YES;
 674   nc->plugin->next_task_nc = nc;
 675   nc->plugin->next_task = GNUNET_SCHEDULER_add_now (&sqlite_next_request_cont,
 676                                                     nc);
 677 }
 678
 679
 680
 681 /**
 682  * Store an item in the datastore.
 683  *
 684  * @param cls closure
 685  * @param key key for the item
 686  * @param size number of bytes in data
 687  * @param data content stored
 688  * @param type type of the content
 689  * @param priority priority of the content
 690  * @param anonymity anonymity-level for the content
 691  * @param expiration expiration time for the content
 692  * @param msg set to an error message
 693  * @return GNUNET_OK on success
 694  */
 695 static int
 696 sqlite_plugin_put (void *cls,
 697                    const GNUNET_HashCode * key,
 698                    uint32_t size,
 699                    const void *data,
 700                    enum GNUNET_BLOCK_Type type,
 701                    uint32_t priority,
 702                    uint32_t anonymity,
 703                    struct GNUNET_TIME_Absolute expiration,
 704                    char ** msg)
 705 {
 706   struct Plugin *plugin = cls;
 707   int n;
 708   sqlite3_stmt *stmt;
 709   GNUNET_HashCode vhash;
 710
 711 #if DEBUG_SQLITE
 712   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 713                    "sqlite",
 714                    "Storing in database block with type %u/key `%s'/priority %u/expiration %llu (%lld).\n",
 715                    type,
 716                    GNUNET_h2s(key),
 717                    priority,
 718                    (unsigned long long) GNUNET_TIME_absolute_get_remaining (expiration).rel_value,
 719                    (long long) expiration.abs_value);
 720 #endif
 721   GNUNET_CRYPTO_hash (data, size, &vhash);
 722   stmt = plugin->insertContent;
 723   if ((SQLITE_OK != sqlite3_bind_int (stmt, 1, size)) ||
 724       (SQLITE_OK != sqlite3_bind_int (stmt, 2, type)) ||
 725       (SQLITE_OK != sqlite3_bind_int (stmt, 3, priority)) ||
 726       (SQLITE_OK != sqlite3_bind_int (stmt, 4, anonymity)) ||
 727       (SQLITE_OK != sqlite3_bind_int64 (stmt, 5, expiration.abs_value)) ||
 728       (SQLITE_OK !=
 729        sqlite3_bind_blob (stmt, 6, key, sizeof (GNUNET_HashCode),
 730                           SQLITE_TRANSIENT)) ||
 731       (SQLITE_OK !=
 732        sqlite3_bind_blob (stmt, 7, &vhash, sizeof (GNUNET_HashCode),
 733                           SQLITE_TRANSIENT))
 734       || (SQLITE_OK !=
 735           sqlite3_bind_blob (stmt, 8, data, size,
 736                              SQLITE_TRANSIENT)))
 737     {
 738       LOG_SQLITE (plugin,
 739                   msg,
 740                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_bind_XXXX");
 741       if (SQLITE_OK != sqlite3_reset (stmt))
 742         LOG_SQLITE (plugin, NULL,
 743                     GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 744       return GNUNET_SYSERR;
 745     }
 746   n = sqlite3_step (stmt);
 747   if (n != SQLITE_DONE)
 748     {
 749       if (n == SQLITE_BUSY)
 750         {
 751           LOG_SQLITE (plugin, msg,
 752                       GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 753           sqlite3_reset (stmt);
 754           GNUNET_break (0);
 755           return GNUNET_NO;
 756         }
 757       LOG_SQLITE (plugin, msg,
 758                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_step");
 759       sqlite3_reset (stmt);
 760       database_shutdown (plugin);
 761       database_setup (plugin->env->cfg,
 762                       plugin);
 763       return GNUNET_SYSERR;
 764     }
 765   if (SQLITE_OK != sqlite3_reset (stmt))
 766     LOG_SQLITE (plugin, NULL,
 767                 GNUNET_ERROR_TYPE_ERROR |
 768                 GNUNET_ERROR_TYPE_BULK, "sqlite3_reset");
 769   plugin->env->duc (plugin->env->cls,
 770                     size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 771 #if DEBUG_SQLITE
 772   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 773                    "sqlite",
 774                    "Stored new entry (%u bytes)\n",
 775            size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 776 #endif
 777   return GNUNET_OK;
 778 }
 779
 780
 781 /**
 782  * Update the priority for a particular key in the datastore.  If
 783  * the expiration time in value is different than the time found in
 784  * the datastore, the higher value should be kept.  For the
 785  * anonymity level, the lower value is to be used.  The specified
 786  * priority should be added to the existing priority, ignoring the
 787  * priority in value.
 788  *
 789  * Note that it is possible for multiple values to match this put.
 790  * In that case, all of the respective values are updated.
 791  *
 792  * @param cls the plugin context (state for this module)
 793  * @param uid unique identifier of the datum
 794  * @param delta by how much should the priority
 795  *     change?  If priority + delta < 0 the
 796  *     priority should be set to 0 (never go
 797  *     negative).
 798  * @param expire new expiration time should be the
 799  *     MAX of any existing expiration time and
 800  *     this value
 801  * @param msg set to an error message
 802  * @return GNUNET_OK on success
 803  */
 804 static int
 805 sqlite_plugin_update (void *cls,
 806                       uint64_t uid,
 807                       int delta, struct GNUNET_TIME_Absolute expire,
 808                       char **msg)
 809 {
 810   struct Plugin *plugin = cls;
 811   int n;
 812
 813   sqlite3_bind_int (plugin->updPrio, 1, delta);
 814   sqlite3_bind_int64 (plugin->updPrio, 2, expire.abs_value);
 815   sqlite3_bind_int64 (plugin->updPrio, 3, uid);
 816   n = sqlite3_step (plugin->updPrio);
 817   if (n != SQLITE_DONE)
 818     LOG_SQLITE (plugin, msg,
 819                 GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
 820                 "sqlite3_step");
 821 #if DEBUG_SQLITE
 822   else
 823     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
 824                      "sqlite",
 825                      "Block updated\n");
 826 #endif
 827   sqlite3_reset (plugin->updPrio);
 828
 829   if (n == SQLITE_BUSY)
 830     return GNUNET_NO;
 831   return n == SQLITE_DONE ? GNUNET_OK : GNUNET_SYSERR;
 832 }
 833
 834
 835 /**
 836  * Internal context for an iteration.
 837  */
 838 struct IterContext
 839 {
 840   /**
 841    * FIXME.
 842    */
 843   sqlite3_stmt *stmt_1;
 844
 845   /**
 846    * FIXME.
 847    */
 848   sqlite3_stmt *stmt_2;
 849
 850   /**
 851    * FIXME.
 852    */
 853   int is_asc;
 854
 855   /**
 856    * FIXME.
 857    */
 858   int is_prio;
 859
 860   /**
 861    * FIXME.
 862    */
 863   int is_migr;
 864
 865   /**
 866    * FIXME.
 867    */
 868   int limit_nonanonymous;
 869
 870   /**
 871    * Desired type for blocks returned by this iterator.
 872    */
 873   enum GNUNET_BLOCK_Type type;
 874 };
 875
 876
 877 /**
 878  * Prepare our SQL query to obtain the next record from the database.
 879  *
 880  * @param cls our "struct IterContext"
 881  * @param nc NULL to terminate the iteration, otherwise our context for
 882  *           getting the next result.
 883  * @return GNUNET_OK on success, GNUNET_NO if there are no more results,
 884  *         GNUNET_SYSERR on error (or end of iteration)
 885  */
 886 static int
 887 iter_next_prepare (void *cls,
 888                    struct NextContext *nc)
 889 {
 890   struct IterContext *ic = cls;
 891   struct Plugin *plugin;
 892   int ret;
 893
 894   if (nc == NULL)
 895     {
 896 #if DEBUG_SQLITE
 897       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 898                   "Asked to clean up iterator state.\n");
 899 #endif
 900       sqlite3_finalize (ic->stmt_1);
 901       sqlite3_finalize (ic->stmt_2);
 902       return GNUNET_SYSERR;
 903     }
 904   sqlite3_reset (ic->stmt_1);
 905   sqlite3_reset (ic->stmt_2);
 906   plugin = nc->plugin;
 907   if (ic->is_prio)
 908     {
 909 #if DEBUG_SQLITE
 910       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 911                   "Restricting to results larger than the last priority %u\n",
 912                   nc->lastPriority);
 913 #endif
 914       sqlite3_bind_int (ic->stmt_1, 1, nc->lastPriority);
 915       sqlite3_bind_int (ic->stmt_2, 1, nc->lastPriority);
 916     }
 917   else
 918     {
 919 #if DEBUG_SQLITE
 920       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 921                   "Restricting to results larger than the last expiration %llu\n",
 922                   (unsigned long long) nc->lastExpiration.abs_value);
 923 #endif
 924       sqlite3_bind_int64 (ic->stmt_1, 1, nc->lastExpiration.abs_value);
 925       sqlite3_bind_int64 (ic->stmt_2, 1, nc->lastExpiration.abs_value);
 926     }
 927 #if DEBUG_SQLITE
 928   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 929               "Restricting to results larger than the last key `%s'\n",
 930               GNUNET_h2s(&nc->lastKey));
 931 #endif
 932   sqlite3_bind_blob (ic->stmt_1, 2,
 933                      &nc->lastKey,
 934                      sizeof (GNUNET_HashCode),
 935                      SQLITE_TRANSIENT);
 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       return GNUNET_SYSERR;
 952     }
 953   if (SQLITE_OK != sqlite3_reset (ic->stmt_1))
 954     LOG_SQLITE (plugin, NULL,
 955                 GNUNET_ERROR_TYPE_ERROR |
 956                 GNUNET_ERROR_TYPE_BULK,
 957                 "sqlite3_reset");
 958   if (SQLITE_ROW == (ret = sqlite3_step (ic->stmt_2)))
 959     {
 960 #if DEBUG_SQLITE
 961       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 962                   "Result found using iterator 2\n");
 963 #endif
 964       nc->stmt = ic->stmt_2;
 965       return GNUNET_OK;
 966     }
 967   if (ret != SQLITE_DONE)
 968     {
 969       LOG_SQLITE (plugin, NULL,
 970                   GNUNET_ERROR_TYPE_ERROR |
 971                   GNUNET_ERROR_TYPE_BULK,
 972                   "sqlite3_step");
 973       return GNUNET_SYSERR;
 974     }
 975   if (SQLITE_OK != sqlite3_reset (ic->stmt_2))
 976     LOG_SQLITE (plugin, NULL,
 977                 GNUNET_ERROR_TYPE_ERROR |
 978                 GNUNET_ERROR_TYPE_BULK,
 979                 "sqlite3_reset");
 980 #if DEBUG_SQLITE
 981   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 982               "No result found using either iterator\n");
 983 #endif
 984   return GNUNET_NO;
 985 }
 986
 987
 988 /**
 989  * Call a method for each key in the database and
 990  * call the callback method on it.
 991  *
 992  * @param plugin our plugin context
 993  * @param type entries of which type should be considered?
 994  * @param is_asc are we iterating in ascending order?
 995  * @param is_prio are we iterating by priority (otherwise by expiration)
 996  * @param is_migr are we iterating in migration order?
 997  * @param limit_nonanonymous are we restricting results to those with anonymity
 998  *              level zero?
 999  * @param stmt_str_1 first SQL statement to execute
1000  * @param stmt_str_2 SQL statement to execute to get "more" results (inner iteration)
1001  * @param iter function to call on each matching value;
1002  *        will be called once with a NULL value at the end
1003  * @param iter_cls closure for iter
1004  */
1005 static void
1006 basic_iter (struct Plugin *plugin,
1007             enum GNUNET_BLOCK_Type type,
1008             int is_asc,
1009             int is_prio,
1010             int is_migr,
1011             int limit_nonanonymous,
1012             const char *stmt_str_1,
1013             const char *stmt_str_2,
1014             PluginIterator iter,
1015             void *iter_cls)
1016 {
1017   struct NextContext *nc;
1018   struct IterContext *ic;
1019   sqlite3_stmt *stmt_1;
1020   sqlite3_stmt *stmt_2;
1021
1022 #if DEBUG_SQLITE
1023   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1024               "At %llu, using queries `%s' and `%s'\n",
1025               (unsigned long long) GNUNET_TIME_absolute_get ().abs_value,
1026               stmt_str_1,
1027               stmt_str_2);
1028 #endif
1029   if (sq_prepare (plugin->dbh, stmt_str_1, &stmt_1) != SQLITE_OK)
1030     {
1031       LOG_SQLITE (plugin, NULL,
1032                   GNUNET_ERROR_TYPE_ERROR |
1033                   GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare_v2");
1034       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1035       return;
1036     }
1037   if (sq_prepare (plugin->dbh, stmt_str_2, &stmt_2) != SQLITE_OK)
1038     {
1039       LOG_SQLITE (plugin, NULL,
1040                   GNUNET_ERROR_TYPE_ERROR |
1041                   GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare_v2");
1042       sqlite3_finalize (stmt_1);
1043       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1044       return;
1045     }
1046   nc = GNUNET_malloc (sizeof(struct NextContext) +
1047                       sizeof(struct IterContext));
1048   nc->plugin = plugin;
1049   nc->iter = iter;
1050   nc->iter_cls = iter_cls;
1051   nc->stmt = NULL;
1052   ic = (struct IterContext*) &nc[1];
1053   ic->stmt_1 = stmt_1;
1054   ic->stmt_2 = stmt_2;
1055   ic->type = type;
1056   ic->is_asc = is_asc;
1057   ic->is_prio = is_prio;
1058   ic->is_migr = is_migr;
1059   ic->limit_nonanonymous = limit_nonanonymous;
1060   nc->prep = &iter_next_prepare;
1061   nc->prep_cls = ic;
1062   if (is_asc)
1063     {
1064       nc->lastPriority = 0;
1065       nc->lastExpiration.abs_value = 0;
1066       memset (&nc->lastKey, 0, sizeof (GNUNET_HashCode));
1067     }
1068   else
1069     {
1070       nc->lastPriority = 0x7FFFFFFF;
1071       nc->lastExpiration.abs_value = 0x7FFFFFFFFFFFFFFFLL;
1072       memset (&nc->lastKey, 255, sizeof (GNUNET_HashCode));
1073     }
1074   sqlite_next_request (nc, GNUNET_NO);
1075 }
1076
1077
1078 /**
1079  * Select a subset of the items in the datastore and call
1080  * the given iterator for each of them.
1081  *
1082  * @param cls our plugin context
1083  * @param type entries of which type should be considered?
1084  *        Use 0 for any type.
1085  * @param iter function to call on each matching value;
1086  *        will be called once with a NULL value at the end
1087  * @param iter_cls closure for iter
1088  */
1089 static void
1090 sqlite_plugin_iter_low_priority (void *cls,
1091                                  enum GNUNET_BLOCK_Type type,
1092                                  PluginIterator iter,
1093                                  void *iter_cls)
1094 {
1095   basic_iter (cls,
1096               type,
1097               GNUNET_YES, GNUNET_YES,
1098               GNUNET_NO, GNUNET_NO,
1099               SELECT_IT_LOW_PRIORITY_1,
1100               SELECT_IT_LOW_PRIORITY_2,
1101               iter, iter_cls);
1102 }
1103
1104
1105 /**
1106  * Select a subset of the items in the datastore and call
1107  * the given iterator for each of them.
1108  *
1109  * @param cls our plugin context
1110  * @param type entries of which type should be considered?
1111  *        Use 0 for any type.
1112  * @param iter function to call on each matching value;
1113  *        will be called once with a NULL value at the end
1114  * @param iter_cls closure for iter
1115  */
1116 static void
1117 sqlite_plugin_iter_zero_anonymity (void *cls,
1118                                    enum GNUNET_BLOCK_Type type,
1119                                    PluginIterator iter,
1120                                    void *iter_cls)
1121 {
1122   struct GNUNET_TIME_Absolute now;
1123   char *q1;
1124   char *q2;
1125
1126   now = GNUNET_TIME_absolute_get ();
1127   GNUNET_asprintf (&q1, SELECT_IT_NON_ANONYMOUS_1,
1128                    (unsigned long long) now.abs_value);
1129   GNUNET_asprintf (&q2, SELECT_IT_NON_ANONYMOUS_2,
1130                    (unsigned long long) now.abs_value);
1131   basic_iter (cls,
1132               type,
1133               GNUNET_NO, GNUNET_YES,
1134               GNUNET_NO, GNUNET_YES,
1135               q1,
1136               q2,
1137               iter, iter_cls);
1138   GNUNET_free (q1);
1139   GNUNET_free (q2);
1140 }
1141
1142
1143
1144 /**
1145  * Select a subset of the items in the datastore and call
1146  * the given iterator for each of them.
1147  *
1148  * @param cls our plugin context
1149  * @param type entries of which type should be considered?
1150  *        Use 0 for any type.
1151  * @param iter function to call on each matching value;
1152  *        will be called once with a NULL value at the end
1153  * @param iter_cls closure for iter
1154  */
1155 static void
1156 sqlite_plugin_iter_ascending_expiration (void *cls,
1157                                          enum GNUNET_BLOCK_Type type,
1158                                          PluginIterator iter,
1159                                          void *iter_cls)
1160 {
1161   struct GNUNET_TIME_Absolute now;
1162   char *q1;
1163   char *q2;
1164
1165   now = GNUNET_TIME_absolute_get ();
1166   GNUNET_asprintf (&q1, SELECT_IT_EXPIRATION_TIME_1,
1167                    (unsigned long long) 0*now.abs_value);
1168   GNUNET_asprintf (&q2, SELECT_IT_EXPIRATION_TIME_2,
1169                    (unsigned long long) 0*now.abs_value);
1170   basic_iter (cls,
1171               type,
1172               GNUNET_YES, GNUNET_NO,
1173               GNUNET_NO, GNUNET_NO,
1174               q1, q2,
1175               iter, iter_cls);
1176   GNUNET_free (q1);
1177   GNUNET_free (q2);
1178 }
1179
1180
1181 /**
1182  * Select a subset of the items in the datastore and call
1183  * the given iterator for each of them.
1184  *
1185  * @param cls our plugin context
1186  * @param type entries of which type should be considered?
1187  *        Use 0 for any type.
1188  * @param iter function to call on each matching value;
1189  *        will be called once with a NULL value at the end
1190  * @param iter_cls closure for iter
1191  */
1192 static void
1193 sqlite_plugin_iter_migration_order (void *cls,
1194                                     enum GNUNET_BLOCK_Type type,
1195                                     PluginIterator iter,
1196                                     void *iter_cls)
1197 {
1198   struct GNUNET_TIME_Absolute now;
1199   char *q;
1200
1201   now = GNUNET_TIME_absolute_get ();
1202   GNUNET_asprintf (&q, SELECT_IT_MIGRATION_ORDER_2,
1203                    (unsigned long long) now.abs_value);
1204   basic_iter (cls,
1205               type,
1206               GNUNET_NO, GNUNET_NO,
1207               GNUNET_YES, GNUNET_NO,
1208               SELECT_IT_MIGRATION_ORDER_1,
1209               q,
1210               iter, iter_cls);
1211   GNUNET_free (q);
1212 }
1213
1214
1215 /**
1216  * Call sqlite using the already prepared query to get
1217  * the next result.
1218  *
1219  * @param cls context with the prepared query
1220  * @param nc context with the prepared query
1221  * @return GNUNET_OK on success, GNUNET_SYSERR on error, GNUNET_NO if
1222  *        there are no more results
1223  */
1224 static int
1225 all_next_prepare (void *cls,
1226                   struct NextContext *nc)
1227 {
1228   struct Plugin *plugin;
1229   int ret;
1230
1231   if (nc == NULL)
1232     {
1233 #if DEBUG_SQLITE
1234       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1235                   "Asked to clean up iterator state.\n");
1236 #endif
1237       nc = (struct NextContext *)cls;
1238       if (nc->stmt)
1239           sqlite3_finalize (nc->stmt);
1240       nc->stmt = NULL;
1241       return GNUNET_SYSERR;
1242     }
1243   plugin = nc->plugin;
1244   if (SQLITE_ROW == (ret = sqlite3_step (nc->stmt)))
1245     {
1246       return GNUNET_OK;
1247     }
1248   if (ret != SQLITE_DONE)
1249     {
1250       LOG_SQLITE (plugin, NULL,
1251                   GNUNET_ERROR_TYPE_ERROR |
1252                   GNUNET_ERROR_TYPE_BULK,
1253                   "sqlite3_step");
1254       return GNUNET_SYSERR;
1255     }
1256   return GNUNET_NO;
1257 }
1258
1259
1260 /**
1261  * Select a subset of the items in the datastore and call
1262  * the given iterator for each of them.
1263  *
1264  * @param cls our plugin context
1265  * @param type entries of which type should be considered?
1266  *        Use 0 for any type.
1267  * @param iter function to call on each matching value;
1268  *        will be called once with a NULL value at the end
1269  * @param iter_cls closure for iter
1270  */
1271 static void
1272 sqlite_plugin_iter_all_now (void *cls,
1273                             enum GNUNET_BLOCK_Type type,
1274                             PluginIterator iter,
1275                             void *iter_cls)
1276 {
1277   struct Plugin *plugin = cls;
1278   struct NextContext *nc;
1279   sqlite3_stmt *stmt;
1280
1281   if (sq_prepare (plugin->dbh,
1282                   "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080",
1283                   &stmt) != SQLITE_OK)
1284     {
1285       LOG_SQLITE (plugin, NULL,
1286                   GNUNET_ERROR_TYPE_ERROR |
1287                   GNUNET_ERROR_TYPE_BULK, "sqlite3_prepare_v2");
1288       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1289       return;
1290     }
1291   nc = GNUNET_malloc (sizeof(struct NextContext));
1292   nc->plugin = plugin;
1293   nc->iter = iter;
1294   nc->iter_cls = iter_cls;
1295   nc->stmt = stmt;
1296   nc->prep = &all_next_prepare;
1297   nc->prep_cls = nc;
1298   sqlite_next_request (nc, GNUNET_NO);
1299 }
1300
1301
1302 /**
1303  * FIXME.
1304  */
1305 struct GetNextContext
1306 {
1307
1308   /**
1309    * FIXME.
1310    */
1311   int total;
1312
1313   /**
1314    * FIXME.
1315    */
1316   int off;
1317
1318   /**
1319    * FIXME.
1320    */
1321   int have_vhash;
1322
1323   /**
1324    * FIXME.
1325    */
1326   unsigned int type;
1327
1328   /**
1329    * FIXME.
1330    */
1331   sqlite3_stmt *stmt;
1332
1333   /**
1334    * FIXME.
1335    */
1336   GNUNET_HashCode key;
1337
1338   /**
1339    * FIXME.
1340    */
1341   GNUNET_HashCode vhash;
1342 };
1343
1344
1345
1346 /**
1347  * FIXME.
1348  *
1349  * @param cls our "struct GetNextContext*"
1350  * @param nc FIXME
1351  * @return GNUNET_YES if there are more results,
1352  *         GNUNET_NO if there are no more results,
1353  *         GNUNET_SYSERR on internal error
1354  */
1355 static int
1356 get_next_prepare (void *cls,
1357                   struct NextContext *nc)
1358 {
1359   struct GetNextContext *gnc = cls;
1360   int sqoff;
1361   int ret;
1362   int limit_off;
1363
1364   if (nc == NULL)
1365     {
1366       sqlite3_finalize (gnc->stmt);
1367       return GNUNET_SYSERR;
1368     }
1369   if (nc->count == gnc->total)
1370     return GNUNET_NO;
1371   if (nc->count + gnc->off == gnc->total)
1372     nc->last_rowid = 0;
1373   if (nc->count == 0)
1374     limit_off = gnc->off;
1375   else
1376     limit_off = 0;
1377   sqoff = 1;
1378   sqlite3_reset (nc->stmt);
1379   ret = sqlite3_bind_blob (nc->stmt,
1380                            sqoff++,
1381                            &gnc->key,
1382                            sizeof (GNUNET_HashCode),
1383                            SQLITE_TRANSIENT);
1384   if ((gnc->have_vhash) && (ret == SQLITE_OK))
1385     ret = sqlite3_bind_blob (nc->stmt,
1386                              sqoff++,
1387                              &gnc->vhash,
1388                              sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
1389   if ((gnc->type != 0) && (ret == SQLITE_OK))
1390     ret = sqlite3_bind_int (nc->stmt, sqoff++, gnc->type);
1391   if (ret == SQLITE_OK)
1392     ret = sqlite3_bind_int64 (nc->stmt, sqoff++, nc->last_rowid + 1);
1393   if (ret == SQLITE_OK)
1394     ret = sqlite3_bind_int (nc->stmt, sqoff++, limit_off);
1395   if (ret != SQLITE_OK)
1396     return GNUNET_SYSERR;
1397   if (SQLITE_ROW != sqlite3_step (nc->stmt))
1398     return GNUNET_NO;
1399   return GNUNET_OK;
1400 }
1401
1402
1403 /**
1404  * Iterate over the results for a particular key
1405  * in the datastore.
1406  *
1407  * @param cls closure
1408  * @param key maybe NULL (to match all entries)
1409  * @param vhash hash of the value, maybe NULL (to
1410  *        match all values that have the right key).
1411  *        Note that for DBlocks there is no difference
1412  *        betwen key and vhash, but for other blocks
1413  *        there may be!
1414  * @param type entries of which type are relevant?
1415  *     Use 0 for any type.
1416  * @param iter function to call on each matching value;
1417  *        will be called once with a NULL value at the end
1418  * @param iter_cls closure for iter
1419  */
1420 static void
1421 sqlite_plugin_get (void *cls,
1422                    const GNUNET_HashCode * key,
1423                    const GNUNET_HashCode * vhash,
1424                    enum GNUNET_BLOCK_Type type,
1425                    PluginIterator iter, void *iter_cls)
1426 {
1427   struct Plugin *plugin = cls;
1428   struct GetNextContext *gpc;
1429   struct NextContext *nc;
1430   int ret;
1431   int total;
1432   sqlite3_stmt *stmt;
1433   char scratch[256];
1434   int sqoff;
1435
1436   GNUNET_assert (iter != NULL);
1437   if (key == NULL)
1438     {
1439       sqlite_plugin_iter_low_priority (cls, type, iter, iter_cls);
1440       return;
1441     }
1442   GNUNET_snprintf (scratch, sizeof (scratch),
1443                    "SELECT count(*) FROM gn080 WHERE hash=:1%s%s",
1444                    vhash == NULL ? "" : " AND vhash=:2",
1445                    type == 0 ? "" : (vhash ==
1446                                      NULL) ? " AND type=:2" : " AND type=:3");
1447   if (sq_prepare (plugin->dbh, scratch, &stmt) != SQLITE_OK)
1448     {
1449       LOG_SQLITE (plugin, NULL,
1450                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite_prepare");
1451       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1452       return;
1453     }
1454   sqoff = 1;
1455   ret = sqlite3_bind_blob (stmt,
1456                            sqoff++,
1457                            key, sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
1458   if ((vhash != NULL) && (ret == SQLITE_OK))
1459     ret = sqlite3_bind_blob (stmt,
1460                              sqoff++,
1461                              vhash,
1462                              sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
1463   if ((type != 0) && (ret == SQLITE_OK))
1464     ret = sqlite3_bind_int (stmt, sqoff++, type);
1465   if (SQLITE_OK != ret)
1466     {
1467       LOG_SQLITE (plugin, NULL,
1468                   GNUNET_ERROR_TYPE_ERROR, "sqlite_bind");
1469       sqlite3_reset (stmt);
1470       sqlite3_finalize (stmt);
1471       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1472       return;
1473     }
1474   ret = sqlite3_step (stmt);
1475   if (ret != SQLITE_ROW)
1476     {
1477       LOG_SQLITE (plugin, NULL,
1478                   GNUNET_ERROR_TYPE_ERROR| GNUNET_ERROR_TYPE_BULK,
1479                   "sqlite_step");
1480       sqlite3_reset (stmt);
1481       sqlite3_finalize (stmt);
1482       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1483       return;
1484     }
1485   total = sqlite3_column_int (stmt, 0);
1486   sqlite3_reset (stmt);
1487   sqlite3_finalize (stmt);
1488   if (0 == total)
1489     {
1490       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1491       return;
1492     }
1493
1494   GNUNET_snprintf (scratch, sizeof (scratch),
1495                    "SELECT size, type, prio, anonLevel, expire, hash, value, _ROWID_ "
1496                    "FROM gn080 WHERE hash=:1%s%s AND _ROWID_ >= :%d "
1497                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET :d",
1498                    vhash == NULL ? "" : " AND vhash=:2",
1499                    type == 0 ? "" : (vhash ==
1500                                      NULL) ? " AND type=:2" : " AND type=:3",
1501                    sqoff, sqoff + 1);
1502   if (sq_prepare (plugin->dbh, scratch, &stmt) != SQLITE_OK)
1503     {
1504       LOG_SQLITE (plugin, NULL,
1505                   GNUNET_ERROR_TYPE_ERROR |
1506                   GNUNET_ERROR_TYPE_BULK, "sqlite_prepare");
1507       iter (iter_cls, NULL, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1508       return;
1509     }
1510   nc = GNUNET_malloc (sizeof(struct NextContext) +
1511                       sizeof(struct GetNextContext));
1512   nc->plugin = plugin;
1513   nc->iter = iter;
1514   nc->iter_cls = iter_cls;
1515   nc->stmt = stmt;
1516   gpc = (struct GetNextContext*) &nc[1];
1517   gpc->total = total;
1518   gpc->type = type;
1519   gpc->key = *key;
1520   gpc->stmt = stmt; /* alias used for freeing at the end! */
1521   if (NULL != vhash)
1522     {
1523       gpc->have_vhash = GNUNET_YES;
1524       gpc->vhash = *vhash;
1525     }
1526   gpc->off = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, total);
1527   nc->prep = &get_next_prepare;
1528   nc->prep_cls = gpc;
1529   sqlite_next_request (nc, GNUNET_NO);
1530 }
1531
1532
1533 /**
1534  * Drop database.
1535  *
1536  * @param cls our plugin context
1537  */
1538 static void
1539 sqlite_plugin_drop (void *cls)
1540 {
1541   struct Plugin *plugin = cls;
1542   plugin->drop_on_shutdown = GNUNET_YES;
1543 }
1544
1545
1546 static unsigned long long
1547 sqlite_plugin_get_size (void *cls)
1548 {
1549   struct Plugin *plugin = cls;
1550   sqlite3_stmt *stmt;
1551   uint64_t pages;
1552   uint64_t page_size;
1553 #if ENULL_DEFINED
1554   char *e;
1555 #endif
1556
1557   if (SQLITE_VERSION_NUMBER < 3006000)
1558     {
1559       GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING,
1560                        "datastore-sqlite",
1561                        _("sqlite version to old to determine size, assuming zero\n"));
1562       return 0;
1563     }
1564   CHECK (SQLITE_OK ==
1565          sqlite3_exec (plugin->dbh,
1566                        "VACUUM", NULL, NULL, ENULL));
1567   CHECK (SQLITE_OK ==
1568          sqlite3_exec (plugin->dbh,
1569                        "PRAGMA auto_vacuum=INCREMENTAL", NULL, NULL, ENULL));
1570   CHECK (SQLITE_OK ==
1571          sq_prepare (plugin->dbh,
1572                      "PRAGMA page_count",
1573                      &stmt));
1574   if (SQLITE_ROW ==
1575       sqlite3_step (stmt))
1576     pages = sqlite3_column_int64 (stmt, 0);
1577   else
1578     pages = 0;
1579   sqlite3_finalize (stmt);
1580   CHECK (SQLITE_OK ==
1581          sq_prepare (plugin->dbh,
1582                      "PRAGMA page_size",
1583                      &stmt));
1584   CHECK (SQLITE_ROW ==
1585          sqlite3_step (stmt));
1586   page_size = sqlite3_column_int64 (stmt, 0);
1587   sqlite3_finalize (stmt);
1588   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1589               _("Using sqlite page utilization to estimate payload (%llu pages of size %llu bytes)\n"),
1590               (unsigned long long) pages,
1591               (unsigned long long) page_size);
1592   return  pages * page_size;
1593 }
1594
1595
1596 /**
1597  * Entry point for the plugin.
1598  *
1599  * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
1600  * @return NULL on error, othrewise the plugin context
1601  */
1602 void *
1603 libgnunet_plugin_datastore_sqlite_init (void *cls)
1604 {
1605   static struct Plugin plugin;
1606   struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
1607   struct GNUNET_DATASTORE_PluginFunctions *api;
1608
1609   if (plugin.env != NULL)
1610     return NULL; /* can only initialize once! */
1611   memset (&plugin, 0, sizeof(struct Plugin));
1612   plugin.env = env;
1613   if (GNUNET_OK !=
1614       database_setup (env->cfg, &plugin))
1615     {
1616       database_shutdown (&plugin);
1617       return NULL;
1618     }
1619   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
1620   api->cls = &plugin;
1621   api->get_size = &sqlite_plugin_get_size;
1622   api->put = &sqlite_plugin_put;
1623   api->next_request = &sqlite_next_request;
1624   api->get = &sqlite_plugin_get;
1625   api->update = &sqlite_plugin_update;
1626   api->iter_low_priority = &sqlite_plugin_iter_low_priority;
1627   api->iter_zero_anonymity = &sqlite_plugin_iter_zero_anonymity;
1628   api->iter_ascending_expiration = &sqlite_plugin_iter_ascending_expiration;
1629   api->iter_migration_order = &sqlite_plugin_iter_migration_order;
1630   api->iter_all_now = &sqlite_plugin_iter_all_now;
1631   api->drop = &sqlite_plugin_drop;
1632   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
1633                    "sqlite", _("Sqlite database running\n"));
1634   return api;
1635 }
1636
1637
1638 /**
1639  * Exit point from the plugin.
1640  *
1641  * @param cls the plugin context (as returned by "init")
1642  * @return always NULL
1643  */
1644 void *
1645 libgnunet_plugin_datastore_sqlite_done (void *cls)
1646 {
1647   char *fn;
1648   struct GNUNET_DATASTORE_PluginFunctions *api = cls;
1649   struct Plugin *plugin = api->cls;
1650
1651 #if DEBUG_SQLITE
1652   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1653                    "sqlite",
1654                    "sqlite plugin is doneing\n");
1655 #endif
1656
1657   if (plugin->next_task != GNUNET_SCHEDULER_NO_TASK)
1658     {
1659 #if DEBUG_SQLITE
1660       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1661                        "sqlite",
1662                        "Canceling next task\n");
1663 #endif
1664       GNUNET_SCHEDULER_cancel (plugin->next_task);
1665       plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
1666 #if DEBUG_SQLITE
1667       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1668                        "sqlite",
1669                        "Prep'ing next task\n");
1670 #endif
1671       plugin->next_task_nc->prep (plugin->next_task_nc->prep_cls, NULL);
1672       GNUNET_free (plugin->next_task_nc);
1673       plugin->next_task_nc = NULL;
1674     }
1675   fn = NULL;
1676   if (plugin->drop_on_shutdown)
1677     fn = GNUNET_strdup (plugin->fn);
1678 #if DEBUG_SQLITE
1679   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1680                    "sqlite",
1681                    "Shutting down database\n");
1682 #endif
1683   database_shutdown (plugin);
1684   plugin->env = NULL;
1685   GNUNET_free (api);
1686   if (fn != NULL)
1687     {
1688       if (0 != UNLINK(fn))
1689         GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING,
1690                                   "unlink",
1691                                   fn);
1692       GNUNET_free (fn);
1693     }
1694 #if DEBUG_SQLITE
1695   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1696                    "sqlite",
1697                    "sqlite plugin is finished doneing\n");
1698 #endif
1699   return NULL;
1700 }
1701
1702 /* end of plugin_datastore_sqlite.c */