sqlite.c

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

    {
      LOG_SQLITE (handle,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                  GNUNET_GE_BULK, "sqlite_step");
      sqlite3_reset (stmt);
      sqlite3_finalize (stmt);
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;

    }
  total = sqlite3_column_int (stmt, 0);
  sqlite3_reset (stmt);
  sqlite3_finalize (stmt);
  if ((iter == NULL) || (total == 0))
    {
      GNUNET_mutex_unlock (lock);
      return total;
    }

  GNUNET_snprintf (scratch, 256,
                   "SELECT size, type, prio, anonLevel, expire, hash, value, _ROWID_ "
                   "FROM gn080 WHERE hash=:1%s%s AND _ROWID_ >= :%d "
                   "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET :d",
                   vhash == NULL ? "" : " AND vhash=:2",
                   type == 0 ? "" : (vhash ==
                                     NULL) ? " AND type=:2" : " AND type=:3",
                   sqoff, sqoff + 1);
  if (sq_prepare (dbh, scratch, &stmt) != SQLITE_OK)
    {
      LOG_SQLITE (handle,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                  GNUNET_GE_BULK, "sqlite_prepare");
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }
  count = 0;
  last_rowid = 0;
  off = GNUNET_random_u32 (GNUNET_RANDOM_QUALITY_WEAK, total);
  while (1)
    {
      if (count == 0)
        limit_off = off;
      else
        limit_off = 0;
      sqoff = 1;
      ret = sqlite3_bind_blob (stmt,
                               sqoff++,
                               key, sizeof (GNUNET_HashCode),
                               SQLITE_TRANSIENT);
      if ((vhash != NULL) && (ret == SQLITE_OK))
        ret = sqlite3_bind_blob (stmt,
                                 sqoff++,
                                 vhash,
                                 sizeof (GNUNET_HashCode), SQLITE_TRANSIENT);
      if ((type != 0) && (ret == SQLITE_OK))
        ret = sqlite3_bind_int (stmt, sqoff++, type);
      if (ret == SQLITE_OK)
        ret = sqlite3_bind_int64 (stmt, sqoff++, last_rowid);
      if (ret == SQLITE_OK)
        ret = sqlite3_bind_int (stmt, sqoff++, limit_off);
      if (ret == SQLITE_OK)
        {
          ret = sqlite3_step (stmt);
          if (ret != SQLITE_ROW)
            break;
          datum = assembleDatum (handle, stmt, &rkey, &rowid);
          last_rowid = rowid + 1;
          sqlite3_reset (stmt);
          if (datum == NULL)
            continue;
          if ((key != NULL) &&
              (0 != memcmp (&rkey, key, sizeof (GNUNET_HashCode))))
            {
              GNUNET_GE_BREAK (NULL, 0);
              GNUNET_free (datum);
              continue;
            }
          GNUNET_mutex_unlock (lock);
          count++;
          ret = iter (&rkey, datum, closure, rowid);
          GNUNET_mutex_lock (lock);
          if (ret == GNUNET_SYSERR)
            {
              GNUNET_free (datum);
              ret = SQLITE_DONE;
              break;
            }
          if (ret == GNUNET_NO)
            {
              payload -= getContentDatastoreSize (datum);
              delete_by_rowid (handle, rowid);
            }
          GNUNET_free (datum);
        }
      if (count + off == total)
        last_rowid = 0;         /* back to start */
      if (count == total)
        break;
    }
  sqlite3_reset (stmt);
  sqlite3_finalize (stmt);
  GNUNET_mutex_unlock (lock);
  return count;
}

/**
 * Write content to the db.  Always adds a new record
 * (does NOT overwrite existing data).
 *
 * @return GNUNET_SYSERR on error, GNUNET_NO on temporary error, GNUNET_OK if ok.
 */
static int
put (const GNUNET_HashCode * key, const GNUNET_DatastoreValue * value)
{
  int n;
  sqlite3_stmt *stmt;
  unsigned int contentSize;
  unsigned int size, type, prio, anon;
  unsigned long long expir;
  GNUNET_HashCode vhash;
  sqliteHandle *dbh;
#if DEBUG_SQLITE
  GNUNET_EncName enc;

  IF_GELOG (ectx, GNUNET_GE_DEBUG | GNUNET_GE_BULK | GNUNET_GE_USER,
            GNUNET_hash_to_enc (key, &enc));
  GNUNET_GE_LOG (ectx, GNUNET_GE_DEBUG | GNUNET_GE_BULK | GNUNET_GE_USER,
                 "Storing in database block with type %u/key `%s'/priority %u/expiration %llu.\n",
                 ntohl (*(int *) &value[1]), &enc, ntohl (value->priority),
                 GNUNET_ntohll (value->expiration_time));
#endif

  if ((ntohl (value->size) < sizeof (GNUNET_DatastoreValue)))
    {
      GNUNET_GE_BREAK (ectx, 0);
      return GNUNET_SYSERR;
    }
  size = ntohl (value->size);
  type = ntohl (value->type);
  prio = ntohl (value->priority);
  anon = ntohl (value->anonymity_level);
  expir = GNUNET_ntohll (value->expiration_time);
  contentSize = size - sizeof (GNUNET_DatastoreValue);
  GNUNET_hash (&value[1], contentSize, &vhash);
  GNUNET_mutex_lock (lock);
  dbh = getDBHandle ();
  if (lastSync > 1000)
    syncStats (dbh);
  stmt = dbh->insertContent;
  if ((SQLITE_OK != sqlite3_bind_int (stmt, 1, size)) ||
      (SQLITE_OK != sqlite3_bind_int (stmt, 2, type)) ||
      (SQLITE_OK != sqlite3_bind_int (stmt, 3, prio)) ||
      (SQLITE_OK != sqlite3_bind_int (stmt, 4, anon)) ||
      (SQLITE_OK != sqlite3_bind_int64 (stmt, 5, expir)) ||
      (SQLITE_OK !=
       sqlite3_bind_blob (stmt, 6, key, sizeof (GNUNET_HashCode),
                          SQLITE_TRANSIENT)) ||
      (SQLITE_OK !=
       sqlite3_bind_blob (stmt, 7, &vhash, sizeof (GNUNET_HashCode),
                          SQLITE_TRANSIENT))
      || (SQLITE_OK !=
          sqlite3_bind_blob (stmt, 8, &value[1], contentSize,
                             SQLITE_TRANSIENT)))
    {
      LOG_SQLITE (dbh,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                  GNUNET_GE_BULK, "sqlite3_bind_XXXX");
      if (SQLITE_OK != sqlite3_reset (stmt))
        LOG_SQLITE (dbh,
                    GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                    GNUNET_GE_BULK, "sqlite3_reset");
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }

  n = sqlite3_step (stmt);
  if (n != SQLITE_DONE)
    {
      if (n == SQLITE_BUSY)
        {
          sqlite3_reset (stmt);
          GNUNET_mutex_unlock (lock);
	  GNUNET_GE_BREAK(NULL, 0);
          return GNUNET_NO;
        }
      LOG_SQLITE (dbh,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                  GNUNET_GE_BULK, "sqlite3_step");
      sqlite3_reset (stmt);
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }
  if (SQLITE_OK != sqlite3_reset (stmt))
    LOG_SQLITE (dbh,
                GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                GNUNET_GE_BULK, "sqlite3_reset");
  lastSync++;
  payload += getContentDatastoreSize (value);
#if DEBUG_SQLITE
  GNUNET_GE_LOG (ectx,
                 GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 "SQLite: done writing content\n");
#endif
  GNUNET_mutex_unlock (lock);
  return GNUNET_OK;
}

/**
 * Update the priority for a particular key
 * in the datastore.
 */
static int
update (unsigned long long uid, int delta, GNUNET_CronTime expire)
{
  int n;
  sqliteHandle *dbh;

  GNUNET_mutex_lock (lock);
  dbh = getDBHandle ();
  sqlite3_bind_int (dbh->updPrio, 1, delta);
  sqlite3_bind_int64 (dbh->updPrio, 2, expire);
  sqlite3_bind_int64 (dbh->updPrio, 3, uid);
  n = sqlite3_step (dbh->updPrio);
  if (n != SQLITE_DONE)
    LOG_SQLITE (dbh,
                GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                GNUNET_GE_BULK, "sqlite3_step");

  sqlite3_reset (dbh->updPrio);

#if DEBUG_SQLITE
  GNUNET_GE_LOG (ectx, GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 "SQLite: block updated\n");
#endif
  GNUNET_mutex_unlock (lock);
  if (n == SQLITE_BUSY)
    return GNUNET_NO;
  return n == SQLITE_OK ? GNUNET_OK : GNUNET_SYSERR;
}

GNUNET_SQstore_ServiceAPI *
provide_module_sqstore_sqlite (GNUNET_CoreAPIForPlugins * capi)
{
  static GNUNET_SQstore_ServiceAPI api;

  char *dir;
  char *afsdir;
  sqliteHandle *dbh;

  ectx = capi->ectx;
#if DEBUG_SQLITE
  GNUNET_GE_LOG (ectx,
                 GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 "SQLite: initializing database\n");
#endif

  payload = 0;
  lastSync = 0;

  afsdir = NULL;
  GNUNET_GC_get_configuration_value_filename (capi->cfg,
                                              "FS",
                                              "DIR",
                                              GNUNET_DEFAULT_DAEMON_VAR_DIRECTORY
                                              "/data/fs/", &afsdir);
  dir = GNUNET_malloc (strlen (afsdir) + strlen ("/content/gnunet.dat") + 2);
  strcpy (dir, afsdir);
  strcat (dir, "/content/gnunet.dat");
  GNUNET_free (afsdir);
  if (GNUNET_OK != GNUNET_disk_directory_create_for_file (ectx, dir))
    {
      GNUNET_GE_BREAK (ectx, 0);
      GNUNET_free (dir);
      return NULL;
    }
  fn = dir;
  dbh = getDBHandle ();
  if (dbh == NULL)
    {
      GNUNET_GE_BREAK (ectx, 0);
      GNUNET_free (fn);
      fn = NULL;
      return NULL;
    }

  payload = getStat (dbh, "PAYLOAD");
  if (payload == GNUNET_SYSERR)
    {
      GNUNET_GE_BREAK (ectx, 0);
      LOG_SQLITE (dbh,
                  GNUNET_GE_ERROR | GNUNET_GE_ADMIN | GNUNET_GE_USER |
                  GNUNET_GE_BULK, "sqlite_payload");
      GNUNET_mutex_destroy (lock);
      GNUNET_free (fn);
      fn = NULL;
      return NULL;
    }
  lock = GNUNET_mutex_create (GNUNET_NO);
  coreAPI = capi;
  stats = coreAPI->service_request ("stats");
  if (stats)
    stat_size = stats->create (gettext_noop ("# bytes in datastore"));

  api.getSize = &getSize;
  api.put = &put;
  api.get = &get;
  api.iterateLowPriority = &iterateLowPriority;
  api.iterateNonAnonymous = &iterateNonAnonymous;
  api.iterateExpirationTime = &iterateExpirationTime;
  api.iterateMigrationOrder = &iterateMigrationOrder;
  api.iterateAllNow = &iterateAllNow;
  api.drop = &drop;
  api.update = &update;
  return &api;
}

/**
 * Shutdown the module.
 */
void
release_module_sqstore_sqlite ()
{
  if (stats != NULL)
    coreAPI->service_release (stats);
  sqlite_shutdown ();
#if DEBUG_SQLITE
  GNUNET_GE_LOG (ectx,
                 GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 "SQLite: database shutdown\n");
#endif
  GNUNET_mutex_destroy (lock);
  lock = NULL;
  coreAPI = NULL;
}



/**
 * Update sqlite database module.
 *
 * Currently only makes sure that the sqlite indices are created.
 */
void
update_module_sqstore_sqlite (GNUNET_UpdateAPI * uapi)
{
  sqliteHandle *dbh;
  char *dir;
  char *afsdir;

  payload = 0;
  lastSync = 0;
  afsdir = NULL;
  GNUNET_GC_get_configuration_value_filename (uapi->cfg,
                                              "FS",
                                              "DIR",
                                              GNUNET_DEFAULT_DAEMON_VAR_DIRECTORY
                                              "/data/fs/", &afsdir);
  dir = GNUNET_malloc (strlen (afsdir) + 8 + 2);        /* 8 = "content/" */
  strcpy (dir, afsdir);
  strcat (dir, "/content/");
  GNUNET_free (afsdir);
  if (GNUNET_OK != GNUNET_disk_directory_create (ectx, dir))
    {
      GNUNET_free (dir);
      return;
    }
  fn = dir;
  lock = GNUNET_mutex_create (GNUNET_NO);
  dbh = getDBHandle ();
  if (dbh == NULL)
    {
      GNUNET_mutex_destroy (lock);
      GNUNET_free (fn);
      fn = NULL;
      return;
    }
  createIndices (dbh->dbh);
  sqlite_shutdown ();
  GNUNET_mutex_destroy (lock);
}

/* end of sqlite.c */