datastore.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/datastore/datastore.c
 * @brief This module is responsible to manage content, in particular
 *        it needs to decide what content to keep.  This module
 *        also uses the bloomfilter to reduce get operations on the
 *        database.
 * @author Christian Grothoff
 */

#include "platform.h"
#include "gnunet_util.h"
#include "gnunet_directories.h"
#include "gnunet_protocols.h"
#include "gnunet_datastore_service.h"
#include "gnunet_sqstore_service.h"
#include "gnunet_state_service.h"
#include "gnunet_stats_service.h"
#include "filter.h"
#include "prefetch.h"

#define DEBUG_DATASTORE GNUNET_NO

#define MAINTENANCE_FREQUENCY (10 * GNUNET_CRON_SECONDS)

/**
 * SQ-store handle
 */
static GNUNET_SQstore_ServiceAPI *sq;

/**
 * Core API handle.
 */
static GNUNET_CoreAPIForPlugins *coreAPI;

/**
 * Minimum priority in the DB.
 */
static unsigned int minPriority;

/**
 * Available space (maybe negative!)
 */
static long long available;

/**
 * Quota from config file.
 */
static unsigned long long quota;

static struct GNUNET_CronManager *cron;

static struct GNUNET_Mutex *lock;

static GNUNET_Stats_ServiceAPI *stats;

static int stat_filtered;

static int stat_filter_failed;

/**
 * Time at which the database was created (used for
 * content aging).
 */
static GNUNET_Int32Time db_creation_time;

/**
 * Require 1/100th of quota to be 'free' space.
 */
#define MIN_GNUNET_free (quota / 100)

/**
 * One month of database uptime corresponds to one
 * priority point.
 */
static int
comp_priority ()
{
  GNUNET_Int32Time now;
  GNUNET_get_time_int32 (&now);
  if (db_creation_time < now)
    return 1;
  return 1 + (db_creation_time - now) / 60 / 60 / 24 / 30;
}

static unsigned long long
getSize ()
{
  return sq->getSize ();
}

static int
get (const GNUNET_HashCode * query,
     unsigned int type, GNUNET_DatastoreValueIterator iter, void *closure)
{
  int ret = 0;

  if (!testAvailable (query))
    {
#if DEBUG_DATASTORE
      GNUNET_EncName enc;

      IF_GELOG (coreAPI->ectx,
                GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                GNUNET_hash_to_enc (query, &enc));
      GNUNET_GE_LOG (coreAPI->ectx,
                     GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                     "Datastore availability pre-test failed for `%s'.\n",
                     &enc);
#endif
      if (stats != NULL)
        stats->change (stat_filtered, 1);

#if 0
      /* this is just an extra check to validate that 
         the bloomfilter was corret; doing this check
         is very costly -- do not do in production! */
      ret = sq->get (query, NULL, type, iter, closure);
      GNUNET_GE_BREAK (NULL, ret == 0);
#endif
      return ret;
    }
  ret = sq->get (query, NULL, type, iter, closure);
  if ((ret == 0) && (stats != NULL))
    stats->change (stat_filter_failed, 1);
  return ret;
}


static int
deleteCB (const GNUNET_HashCode * key,
          const GNUNET_DatastoreValue * value, void *closure,
          unsigned long long uid)
{
  const GNUNET_DatastoreValue *have = closure;
  if (have == NULL)
    return GNUNET_NO;
  if ((value->size == have->size) &&
      (0 == memcmp (&have[1],
                    &value[1],
                    ntohl (value->size) - sizeof (GNUNET_DatastoreValue))))
    return GNUNET_NO;
  return GNUNET_OK;
}

/**
 * Explicitly remove some content from the database.
 */
static int
del (const GNUNET_HashCode * query, const GNUNET_DatastoreValue * value)
{
  int ok;
  int ret;
  GNUNET_EncName enc;
  GNUNET_HashCode vhc;

  if (!testAvailable (query))
    {
      IF_GELOG (coreAPI->ectx,
                GNUNET_GE_WARNING | GNUNET_GE_BULK | GNUNET_GE_USER,
                GNUNET_hash_to_enc (query, &enc));
      GNUNET_GE_LOG (coreAPI->ectx,
                     GNUNET_GE_WARNING | GNUNET_GE_BULK | GNUNET_GE_USER,
                     _("Availability test failed for `%s' at %s:%d.\n"), &enc,
                     __FILE__, __LINE__);
      return GNUNET_NO;
    }
  GNUNET_hash (&value[1],
               ntohl (value->size) - sizeof (GNUNET_DatastoreValue), &vhc);
  ok = sq->get (query, &vhc, ntohl (value->type), &deleteCB, (void *) value);
  if (ok == GNUNET_SYSERR)
    return GNUNET_SYSERR;
  if (ok == 0)
    {
      IF_GELOG (coreAPI->ectx,
                GNUNET_GE_WARNING | GNUNET_GE_BULK | GNUNET_GE_USER,
                GNUNET_hash_to_enc (query, &enc));
      GNUNET_GE_LOG (coreAPI->ectx,
                     GNUNET_GE_WARNING | GNUNET_GE_BULK | GNUNET_GE_USER,
                     _("Availability test failed for `%s' at %s:%d.\n"), &enc,
                     __FILE__, __LINE__);
      return GNUNET_NO;
    }
  ret = ok;
  while (ok-- > 0)
    {
      makeUnavailable (query);  /* update bloom filter! */
      available += ntohl (value->size);
    }
  return ret;
}

typedef struct
{
  int exists;
  const GNUNET_DatastoreValue *value;
  unsigned long long uid;
  GNUNET_CronTime expiration;
} CE;

static int
checkExists (const GNUNET_HashCode * key,
             const GNUNET_DatastoreValue * value, void *cls,
             unsigned long long uid)
{
  CE *ce = cls;

  if ((value->size != ce->value->size) ||
      (0 != memcmp (&value[1],
                    &ce->value[1],
                    ntohl (value->size) - sizeof (GNUNET_DatastoreValue))))
    return GNUNET_OK;           /* found another value, but different content! */
  ce->uid = uid;
  ce->expiration = GNUNET_ntohll (value->expiration_time);
  ce->exists = GNUNET_YES;
  return GNUNET_SYSERR;         /* abort iteration! */
}

/**
 * Store an item in the datastore.  If the item is already present,
 * the priorities are summed up and the higher expiration time and
 * lower anonymity level is used.
 *
 * @return GNUNET_YES on success, GNUNET_NO if the datastore is
 *   full and the priority of the item is not high enough
 *   to justify removing something else, GNUNET_SYSERR on
 *   other serious error (i.e. IO permission denied)
 */
static int
putUpdate (const GNUNET_HashCode * key, const GNUNET_DatastoreValue * value)
{
  CE cls;
  int ok;
  int comp_prio;
  GNUNET_DatastoreValue *nvalue;
  GNUNET_HashCode vhc;

  /* check if it already exists... */
  cls.exists = GNUNET_NO;
  cls.value = value;
  GNUNET_hash (&value[1],
               ntohl (value->size) - sizeof (GNUNET_DatastoreValue), &vhc);
  GNUNET_mutex_lock (lock);
  sq->get (key, &vhc, ntohl (value->type), &checkExists, &cls);
  if ((!cls.exists) && (ntohl (value->type) == GNUNET_ECRS_BLOCKTYPE_DATA))
    sq->get (key, &vhc, GNUNET_ECRS_BLOCKTYPE_ONDEMAND, &checkExists, &cls);
  if (cls.exists)
    {
      if ((ntohl (value->priority) == 0) &&
          (GNUNET_ntohll (value->expiration_time) <= cls.expiration))
        {
          GNUNET_mutex_unlock (lock);
          return GNUNET_OK;
        }
      /* update priority */
      sq->update (cls.uid,
                  ntohl (value->priority),
                  GNUNET_ntohll (value->expiration_time));
      GNUNET_mutex_unlock (lock);
      return GNUNET_OK;
    }
  comp_prio = comp_priority ();
#if DEBUG_DATASTORE
  GNUNET_GE_LOG (coreAPI->ectx,
                 GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 "Migration: available %llu (need %u), min priority %u have %u\n",
                 available, ntohl (value->size), minPriority,
                 ntohl (value->priority) + comp_prio);
#endif
  /* check if we have enough space / priority */
  if ((available < ntohl (value->size)) ||
      (minPriority >= ntohl (value->priority) + comp_prio))
    {
      /* new content either does not fit (for sure)
         or has such a low priority that we should
         not even bother! */
      GNUNET_mutex_unlock (lock);
      return GNUNET_NO;
    }
  if (ntohl (value->priority) + comp_prio < minPriority)
    minPriority = ntohl (value->priority) + comp_prio;
  /* construct new value with comp'ed priority */
  nvalue = GNUNET_malloc (ntohl (value->size));
  memcpy (nvalue, value, ntohl (value->size));
  nvalue->priority = htonl (comp_priority () + ntohl (value->priority));
  /* add the content */
  ok = sq->put (key, nvalue);