sqlite.c

GNUnet是一个安全的点对点网络框架

/*
     This file is part of GNUnet.
     (C) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008 Christian Grothoff (and other contributing authors)

     GNUnet is free software; you can redistribute it and/or modify
     it under the terms of the GNU General Public License as published
     by the Free Software Foundation; either version 2, or (at your
     option) any later version.

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     General Public License for more details.

     You should have received a copy of the GNU General Public License
     along with GNUnet; see the file COPYING.  If not, write to the
     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
     Boston, MA 02111-1307, USA.
*/

/**
 * @file applications/sqstore_sqlite/sqlite.c
 * @brief SQLite based implementation of the sqstore service
 * @author Nils Durner
 * @author Christian Grothoff
 *
 * Database: SQLite
 */

#include "platform.h"
#include "gnunet_directories.h"
#include "gnunet_util.h"
#include "gnunet_sqstore_service.h"
#include "gnunet_protocols.h"
#include "gnunet_stats_service.h"
#include <sqlite3.h>

#define DEBUG_SQLITE GNUNET_NO

/**
 * Die with an error message that indicates
 * a failure of the command 'cmd' with the message given
 * by strerror(errno).
 */
#define DIE_SQLITE(db, cmd) do { GNUNET_GE_LOG(ectx, GNUNET_GE_FATAL | GNUNET_GE_IMMEDIATE | GNUNET_GE_ADMIN, _("`%s' failed at %s:%d with error: %s\n"), cmd, __FILE__, __LINE__, sqlite3_errmsg(db->dbh)); abort(); } while(0)

/**
 * Log an error message at log-level 'level' that indicates
 * a failure of the command 'cmd' on file 'filename'
 * with the message given by strerror(errno).
 */
#define LOG_SQLITE(db, level, cmd) do { GNUNET_GE_LOG(ectx, level, _("`%s' failed at %s:%d with error: %s\n"), cmd, __FILE__, __LINE__, sqlite3_errmsg(db->dbh)); } while(0)

#define SELECT_IT_LOW_PRIORITY_1 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio = ? AND hash > ?) "\
  "ORDER BY hash ASC LIMIT 1"

#define SELECT_IT_LOW_PRIORITY_2 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio > ?) "\
  "ORDER BY prio ASC, hash ASC LIMIT 1"

#define SELECT_IT_NON_ANONYMOUS_1 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio = ? AND hash < ? AND anonLevel = 0) "\
  " ORDER BY hash DESC LIMIT 1"

#define SELECT_IT_NON_ANONYMOUS_2 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (prio < ? AND anonLevel = 0)"\
  " ORDER BY prio DESC, hash DESC LIMIT 1"

#define SELECT_IT_EXPIRATION_TIME_1 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire = ? AND hash > ?) "\
  " ORDER BY hash ASC LIMIT 1"

#define SELECT_IT_EXPIRATION_TIME_2 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire > ?) "\
  " ORDER BY expire ASC, hash ASC LIMIT 1"

#define SELECT_IT_MIGRATION_ORDER_1 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire = ? AND hash < ?) "\
  " ORDER BY hash DESC LIMIT 1"

#define SELECT_IT_MIGRATION_ORDER_2 \
  "SELECT size,type,prio,anonLevel,expire,hash,value,_ROWID_ FROM gn080 WHERE (expire < ?) "\
  " ORDER BY expire DESC, hash DESC LIMIT 1"

/**
 * After how many ms "busy" should a DB operation fail for good?
 * A low value makes sure that we are more responsive to requests
 * (especially PUTs).  A high value guarantees a higher success
 * rate (SELECTs in iterate can take several seconds despite LIMIT=1).
 *
 * The default value of 250ms should ensure that users do not experience
 * huge latencies while at the same time allowing operations to succeed
 * with reasonable probability.
 */
#define BUSY_TIMEOUT_MS 250

/**
 * @brief Wrapper for SQLite
 */
typedef struct
{

  /**
   * Native SQLite database handle - may not be shared between threads!
   */
  sqlite3 *dbh;

  /**
   * Thread ID owning this handle
   */
  struct GNUNET_ThreadHandle *tid;

  /**
   * Precompiled SQL
   */
  sqlite3_stmt *updPrio;

  sqlite3_stmt *insertContent;
} sqliteHandle;

static GNUNET_Stats_ServiceAPI *stats;

static GNUNET_CoreAPIForPlugins *coreAPI;

static unsigned int stat_size;

static struct GNUNET_GE_Context *ectx;

static struct GNUNET_Mutex *lock;

static char *fn;

static unsigned long long payload;

static unsigned int lastSync;

static unsigned int handle_count;

static sqliteHandle **handles;

/**
 * @brief Prepare a SQL statement
 */
static int
sq_prepare (sqlite3 * dbh, const char *zSql,    /* SQL statement, UTF-8 encoded */
            sqlite3_stmt ** ppStmt)
{                               /* OUT: Statement handle */
  char *dummy;
  return sqlite3_prepare (dbh,
                          zSql,
                          strlen (zSql), ppStmt, (const char **) &dummy);
}

#if 1
#define CHECK(a) GNUNET_GE_BREAK(ectx, a)
#define ENULL NULL
#else
#define ENULL &e
#define ENULL_DEFINED 1
#define CHECK(a) if (! a) { fprintf(stderr, "%s\n", e); sqlite3_free(e); }
#endif

static void
createIndices (sqlite3 * dbh)
{
  /* create indices */
  sqlite3_exec (dbh,
                "CREATE INDEX idx_hash ON gn080 (hash)", NULL, NULL, ENULL);
  sqlite3_exec (dbh,
                "CREATE INDEX idx_hash_vhash ON gn080 (hash,vhash)", NULL,
                NULL, ENULL);
  sqlite3_exec (dbh, "CREATE INDEX idx_prio ON gn080 (prio)", NULL, NULL,
                ENULL);
  sqlite3_exec (dbh, "CREATE INDEX idx_expire ON gn080 (expire)", NULL, NULL,
                ENULL);
  sqlite3_exec (dbh, "CREATE INDEX idx_comb3 ON gn080 (prio,anonLevel)", NULL,
                NULL, ENULL);
  sqlite3_exec (dbh, "CREATE INDEX idx_comb4 ON gn080 (prio,hash,anonLevel)",
                NULL, NULL, ENULL);
  sqlite3_exec (dbh, "CREATE INDEX idx_comb7 ON gn080 (expire,hash)", NULL,
                NULL, ENULL);
}

/**
 * @brief Get a database handle for this thread.
 * @note SQLite handles may no be shared between threads - see
 *        http://permalink.gmane.org/gmane.network.gnunet.devel/1377
 *       We therefore (re)open the database in each thread.
 * @return the native SQLite database handle
 */
static sqliteHandle *
getDBHandle ()
{
  unsigned int idx;
  sqliteHandle *ret;
  sqlite3_stmt *stmt;
#if ENULL_DEFINED
  char *e;
#endif

  /* Is the DB already open? */
  for (idx = 0; idx < handle_count; idx++)
    if (GNUNET_thread_test_self (handles[idx]->tid))
      return handles[idx];

  /* we haven't opened the DB for this thread yet */
  ret = GNUNET_malloc (sizeof (sqliteHandle));
  /* Open database and precompile statements */
  if (sqlite3_open (fn, &ret->dbh) != SQLITE_OK)
    {
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                     _("Unable to initialize SQLite: %s.\n"),
                     sqlite3_errmsg (ret->dbh));
      sqlite3_close (ret->dbh);
      GNUNET_free (ret);
      return NULL;
    }

  CHECK (SQLITE_OK ==
         sqlite3_exec (ret->dbh,
                       "PRAGMA temp_store=MEMORY", NULL, NULL, ENULL));
  CHECK (SQLITE_OK ==
         sqlite3_exec (ret->dbh,
                       "PRAGMA synchronous=OFF", NULL, NULL, ENULL));
  CHECK (SQLITE_OK ==
         sqlite3_exec (ret->dbh,
                       "PRAGMA count_changes=OFF", NULL, NULL, ENULL));
  CHECK (SQLITE_OK ==
         sqlite3_exec (ret->dbh, "PRAGMA page_size=4092", NULL, NULL, ENULL));

  CHECK (SQLITE_OK == sqlite3_busy_timeout (ret->dbh, BUSY_TIMEOUT_MS));


  /* We have to do it here, because otherwise precompiling SQL might fail */
  CHECK (SQLITE_OK ==
         sq_prepare (ret->dbh,
                     "SELECT 1 FROM sqlite_master WHERE tbl_name = 'gn080'",
                     &stmt));
  if (sqlite3_step (stmt) == SQLITE_DONE)
    {
      if (sqlite3_exec (ret->dbh,
                        "CREATE TABLE gn080 ("
                        "  size INTEGER NOT NULL DEFAULT 0,"
                        "  type INTEGER NOT NULL DEFAULT 0,"
                        "  prio INTEGER NOT NULL DEFAULT 0,"
                        "  anonLevel INTEGER NOT NULL DEFAULT 0,"
                        "  expire INTEGER NOT NULL DEFAULT 0,"
                        "  hash TEXT NOT NULL DEFAULT '',"
                        "  vhash TEXT NOT NULL DEFAULT '',"
                        "  value BLOB NOT NULL DEFAULT '')", NULL, NULL,
                        NULL) != SQLITE_OK)
        {
          LOG_SQLITE (ret,
                      GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                      GNUNET_GE_BULK, "sqlite_create");
          sqlite3_finalize (stmt);
          GNUNET_free (ret);
          return NULL;
        }
    }
  sqlite3_finalize (stmt);
  createIndices (ret->dbh);

  CHECK (SQLITE_OK ==
         sq_prepare (ret->dbh,
                     "SELECT 1 FROM sqlite_master WHERE tbl_name = 'gn071'",
                     &stmt));
  if (sqlite3_step (stmt) == SQLITE_DONE)
    {
      if (sqlite3_exec (ret->dbh,
                        "CREATE TABLE gn071 ("
                        "  key TEXT NOT NULL DEFAULT '',"
                        "  value INTEGER NOT NULL DEFAULT 0)", NULL, NULL,
                        NULL) != SQLITE_OK)
        {
          LOG_SQLITE (ret,
                      GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                      GNUNET_GE_BULK, "sqlite_create");
          sqlite3_finalize (stmt);
          GNUNET_free (ret);
          return NULL;
        }
    }
  sqlite3_finalize (stmt);

  if ((sq_prepare (ret->dbh,
                   "UPDATE gn080 SET prio = prio + ?, expire = MAX(expire,?) WHERE "
                   "_ROWID_ = ?",
                   &ret->updPrio) != SQLITE_OK) ||
      (sq_prepare (ret->dbh,
                   "INSERT INTO gn080 (size, type, prio, "
                   "anonLevel, expire, hash, vhash, value) VALUES "
                   "(?, ?, ?, ?, ?, ?, ?, ?)",
                   &ret->insertContent) != SQLITE_OK))
    {
      LOG_SQLITE (ret,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                  GNUNET_GE_BULK, "precompiling");
      if (ret->updPrio != NULL)
        sqlite3_finalize (ret->updPrio);
      if (ret->insertContent != NULL)
        sqlite3_finalize (ret->insertContent);
      GNUNET_free (ret);
      return NULL;
    }
  ret->tid = GNUNET_thread_get_self ();
  GNUNET_array_append (handles, handle_count, ret);
  return ret;
}

/**
 * @brief Returns the storage needed for the specfied int
 */
static unsigned int
getIntSize (unsigned long long l)
{
  if ((l & 0x7FFFFFFFFFFFLL) == l)
    if ((l & 0x7FFFFFFF) == l)
      if ((l & 0x7FFFFF) == l)
        if ((l & 0x7FFF) == l)
          if ((l & 0x7F) == l)
            return 1;
          else
            return 2;
        else
          return 3;
      else
        return 4;
    else
      return 6;
  else
    return 8;
}

/**
 * Get a (good) estimate of the size of the given
 * value (and its key) in the datastore.<p>
 * <pre>
 * row length = GNUNET_hash length + block length + numbers + column count + estimated index size + 1
 * </pre>
 */
static unsigned int
getContentDatastoreSize (const GNUNET_DatastoreValue * value)
{
  return sizeof (GNUNET_HashCode) * 2 + ntohl (value->size) -
    sizeof (GNUNET_DatastoreValue) + getIntSize (ntohl (value->size)) +
    getIntSize (ntohl (value->type)) + getIntSize (ntohl (value->priority)) +
    getIntSize (ntohl (value->anonymity_level)) +
    getIntSize (GNUNET_ntohll (value->expiration_time)) + 7 + 245 + 1;
}



/**
 * Get the current on-disk size of the SQ store.  Estimates are fine,
 * if that's the only thing available.
 *
 * @return number of bytes used on disk
 */
static unsigned long long
getSize ()
{
  double ret;

  GNUNET_mutex_lock (lock);
  ret = payload;
  if (stats)
    stats->set (stat_size, ret);
  GNUNET_mutex_unlock (lock);
  return (unsigned long long) (ret * 1.13);
  /* benchmarking shows 13% overhead */
}

static int
delete_by_rowid (sqliteHandle * handle, unsigned long long rid)
{
  sqlite3_stmt *stmt;

  if (sq_prepare (handle->dbh,
                  "DELETE FROM gn080 WHERE _ROWID_ = ?", &stmt) != SQLITE_OK)
    {
      LOG_SQLITE (handle,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |