plugin_datastore_heap.c

Go to the documentation of this file.

/*
     This file is part of GNUnet
     Copyright (C) 2012 GNUnet e.V.
 
     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     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
     Affero General Public License for more details.
 
     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
     SPDX-License-Identifier: AGPL3.0-or-later
 */
 
#include "platform.h"
#include "gnunet_datastore_plugin.h"
 
 
struct Value
{
  struct GNUNET_HashCode key;
 
  const void *data;
 
  struct GNUNET_CONTAINER_HeapNode *expire_heap;
 
  struct GNUNET_CONTAINER_HeapNode *replication_heap;
 
  struct GNUNET_TIME_Absolute expiration;
 
  unsigned int zero_anon_offset;
 
  uint32_t size;
 
  uint32_t priority;
 
  uint32_t anonymity;
 
  uint32_t replication;
 
  enum GNUNET_BLOCK_Type type;
};
 
 
struct ZeroAnonByType
{
  struct ZeroAnonByType *next;
 
  struct ZeroAnonByType *prev;
 
  struct Value **array;
 
  unsigned int array_size;
 
  unsigned int array_pos;
 
  enum GNUNET_BLOCK_Type type;
};
 
 
struct Plugin
{
  struct GNUNET_DATASTORE_PluginEnvironment *env;
 
  struct GNUNET_CONTAINER_MultiHashMap *keyvalue;
 
  struct GNUNET_CONTAINER_Heap *by_expiration;
 
  struct GNUNET_CONTAINER_Heap *by_replication;
 
  struct ZeroAnonByType *zero_head;
 
  struct ZeroAnonByType *zero_tail;
 
  unsigned long long size;
};
 
 
static void
heap_plugin_estimate_size (void *cls, unsigned long long *estimate)
{
  struct Plugin *plugin = cls;
 
  if (NULL != estimate)
    *estimate = plugin->size;
}
 
 
struct UpdateContext
{
  uint32_t size;
 
  const void *data;
 
  uint32_t priority;
 
  uint32_t replication;
 
  struct GNUNET_TIME_Absolute expiration;
 
  bool updated;
};
 
 
static int
update_iterator (void *cls,
                 const struct GNUNET_HashCode *key,
                 void *val)
{
  struct UpdateContext *uc = cls;
  struct Value *value = val;
 
  if (value->size != uc->size)
    return GNUNET_YES;
  if (0 != memcmp (value->data, uc->data, uc->size))
    return GNUNET_YES;
  uc->expiration = GNUNET_TIME_absolute_max (value->expiration,
                                             uc->expiration);
  if (value->expiration.abs_value_us != uc->expiration.abs_value_us)
  {
    value->expiration = uc->expiration;
    GNUNET_CONTAINER_heap_update_cost (value->expire_heap,
                                       value->expiration.abs_value_us);
  }
  /* Saturating adds, don't overflow */
  if (value->priority > UINT32_MAX - uc->priority)
    value->priority = UINT32_MAX;
  else
    value->priority += uc->priority;
  if (value->replication > UINT32_MAX - uc->replication)
    value->replication = UINT32_MAX;
  else
    value->replication += uc->replication;
  uc->updated = true;
  return GNUNET_NO;
}
 
 
static void
heap_plugin_put (void *cls,
                 const struct GNUNET_HashCode *key,
                 bool absent,
                 uint32_t size,
                 const void *data,
                 enum GNUNET_BLOCK_Type type,
                 uint32_t priority,
                 uint32_t anonymity,
                 uint32_t replication,
                 struct GNUNET_TIME_Absolute expiration,
                 PluginPutCont cont,
                 void *cont_cls)
{
  struct Plugin *plugin = cls;
  struct Value *value;
 
  if (! absent)
  {
    struct UpdateContext uc;
 
    uc.size = size;
    uc.data = data;
    uc.priority = priority;
    uc.replication = replication;
    uc.expiration = expiration;
    uc.updated = false;
    GNUNET_CONTAINER_multihashmap_get_multiple (plugin->keyvalue,
                                                key,
                                                &update_iterator,
                                                &uc);
    if (uc.updated)
    {
      cont (cont_cls, key, size, GNUNET_NO, NULL);
      return;
    }
  }
  value = GNUNET_malloc (sizeof(struct Value) + size);
  value->key = *key;
  value->data = &value[1];
  value->expire_heap = GNUNET_CONTAINER_heap_insert (plugin->by_expiration,
                                                     value,
                                                     expiration.abs_value_us);
  value->replication_heap = GNUNET_CONTAINER_heap_insert (
    plugin->by_replication,
    value,
    replication);
  value->expiration = expiration;
  if (0 == anonymity)
  {
    struct ZeroAnonByType *zabt;
 
    for (zabt = plugin->zero_head; NULL != zabt; zabt = zabt->next)
      if (zabt->type == type)
        break;
    if (NULL == zabt)
    {
      zabt = GNUNET_new (struct ZeroAnonByType);
      zabt->type = type;
      GNUNET_CONTAINER_DLL_insert (plugin->zero_head,
                                   plugin->zero_tail,
                                   zabt);
    }
    if (zabt->array_size == zabt->array_pos)
    {
      GNUNET_array_grow (zabt->array,
                         zabt->array_size,
                         zabt->array_size * 2 + 4);
    }
    value->zero_anon_offset = zabt->array_pos;
    zabt->array[zabt->array_pos++] = value;
  }
  value->size = size;
  value->priority = priority;
  value->anonymity = anonymity;
  value->replication = replication;
  value->type = type;
  GNUNET_memcpy (&value[1], data, size);
  GNUNET_CONTAINER_multihashmap_put (plugin->keyvalue,
                                     &value->key,
                                     value,
                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  plugin->size += size;
  cont (cont_cls, key, size, GNUNET_OK, NULL);
}
 
 
static void
delete_value (struct Plugin *plugin,
              struct Value *value)
{
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multihashmap_remove (plugin->keyvalue,
                                                       &value->key,
                                                       value));
  GNUNET_assert (value == GNUNET_CONTAINER_heap_remove_node (
                   value->expire_heap));
  GNUNET_assert (value == GNUNET_CONTAINER_heap_remove_node (
                   value->replication_heap));
  if (0 == value->anonymity)
  {
    struct ZeroAnonByType *zabt;
 
    for (zabt = plugin->zero_head; NULL != zabt; zabt = zabt->next)
      if (zabt->type == value->type)
        break;
    GNUNET_assert (NULL != zabt);
    zabt->array[value->zero_anon_offset] = zabt->array[--zabt->array_pos];
    zabt->array[value->zero_anon_offset]->zero_anon_offset =
      value->zero_anon_offset;
    if (0 == zabt->array_pos)
    {
      GNUNET_array_grow (zabt->array,
                         zabt->array_size,
                         0);
      GNUNET_CONTAINER_DLL_remove (plugin->zero_head,
                                   plugin->zero_tail,
                                   zabt);
      GNUNET_free (zabt);
    }
  }
  plugin->size -= value->size;
  GNUNET_free (value);
}
 
 
struct GetContext
{
  uint64_t next_uid;
 
  struct Value *value;
 
  enum GNUNET_BLOCK_Type type;
 
  bool random;
};
 
 
static int
get_iterator (void *cls,
              const struct GNUNET_HashCode *key,
              void *val)
{
  struct GetContext *gc = cls;
  struct Value *value = val;
 
  if ((gc->type != GNUNET_BLOCK_TYPE_ANY) &&
      (gc->type != value->type))
    return GNUNET_OK;
  if (gc->random)
  {
    gc->value = value;
    return GNUNET_NO;
  }
  if ((uint64_t) (intptr_t) value < gc->next_uid)
    return GNUNET_OK;
  if ((NULL != gc->value) &&
      (value > gc->value))
    return GNUNET_OK;
  gc->value = value;
  return GNUNET_OK;
}
 
 
static void
heap_plugin_get_key (void *cls,
                     uint64_t next_uid,
                     bool random,
                     const struct GNUNET_HashCode *key,
                     enum GNUNET_BLOCK_Type type,
                     PluginDatumProcessor proc,
                     void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct GetContext gc;
 
  gc.value = NULL;
  gc.next_uid = next_uid;
  gc.random = random;
  gc.type = type;
  if (NULL == key)
  {
    GNUNET_CONTAINER_multihashmap_iterate (plugin->keyvalue,
                                           &get_iterator,
                                           &gc);
  }
  else
  {
    GNUNET_CONTAINER_multihashmap_get_multiple (plugin->keyvalue,
                                                key,
                                                &get_iterator,
                                                &gc);
  }
  if (NULL == gc.value)
  {
    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
    return;
  }
  GNUNET_assert (GNUNET_OK ==
                 proc (proc_cls,
                       &gc.value->key,
                       gc.value->size,
                       &gc.value[1],
                       gc.value->type,
                       gc.value->priority,
                       gc.value->anonymity,
                       gc.value->replication,
                       gc.value->expiration,
                       (uint64_t) (intptr_t) gc.value));
}
 
 
static void
heap_plugin_get_replication (void *cls,
                             PluginDatumProcessor proc,
                             void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct Value *value;
 
  value = GNUNET_CONTAINER_heap_remove_root (plugin->by_replication);
  if (NULL == value)
  {
    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
    return;
  }
  if (value->replication > 0)
  {
    value->replication--;
    value->replication_heap = GNUNET_CONTAINER_heap_insert (
      plugin->by_replication,
      value,
      value->replication);
  }
  else
  {
    /* need a better way to pick a random item, replication level is always 0 */
    value->replication_heap = GNUNET_CONTAINER_heap_insert (
      plugin->by_replication,
      value,
      value->replication);
    value = GNUNET_CONTAINER_heap_walk_get_next (plugin->by_replication);
  }
  GNUNET_assert (GNUNET_OK ==
                 proc (proc_cls,
                       &value->key,
                       value->size,
                       &value[1],
                       value->type,
                       value->priority,
                       value->anonymity,
                       value->replication,
                       value->expiration,
                       (uint64_t) (intptr_t) value));
}
 
 
static void
heap_plugin_get_expiration (void *cls, PluginDatumProcessor proc,
                            void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct Value *value;
 
  value = GNUNET_CONTAINER_heap_peek (plugin->by_expiration);
  if (NULL == value)
  {
    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
    return;
  }
  if (GNUNET_NO ==
      proc (proc_cls,
            &value->key,
            value->size,
            &value[1],
            value->type,
            value->priority,
            value->anonymity,
            value->replication,
            value->expiration,
            (uint64_t) (intptr_t) value))
    delete_value (plugin, value);
}
 
 
static void
heap_plugin_get_zero_anonymity (void *cls, uint64_t next_uid,
                                enum GNUNET_BLOCK_Type type,
                                PluginDatumProcessor proc, void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct ZeroAnonByType *zabt;
  struct Value *value = NULL;
 
  for (zabt = plugin->zero_head; NULL != zabt; zabt = zabt->next)
  {
    if ((type != GNUNET_BLOCK_TYPE_ANY) &&
        (type != zabt->type))
      continue;
    for (int i = 0; i < zabt->array_pos; ++i)
    {
      if ((uint64_t) (intptr_t) zabt->array[i] < next_uid)
        continue;
      if ((NULL != value) &&
          (zabt->array[i] > value))
        continue;
      value = zabt->array[i];
    }
  }
  if (NULL == value)
  {
    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
    return;
  }
  GNUNET_assert (GNUNET_OK ==
                 proc (proc_cls,
                       &value->key,
                       value->size,
                       &value[1],
                       value->type,
                       value->priority,
                       value->anonymity,
                       value->replication,
                       value->expiration,
                       (uint64_t) (intptr_t) value));
}
 
 
static void
heap_plugin_drop (void *cls)
{
  /* nothing needs to be done */
}
 
 
struct GetAllContext
{
  PluginKeyProcessor proc;
 
  void *proc_cls;
};
 
 
static int
return_value (void *cls,
              const struct GNUNET_HashCode *key,
              void *val)
{
  struct GetAllContext *gac = cls;
 
  gac->proc (gac->proc_cls,
             key,
             1);
  return GNUNET_OK;
}
 
 
static void
heap_get_keys (void *cls,
               PluginKeyProcessor proc,
               void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct GetAllContext gac;
 
  gac.proc = proc;
  gac.proc_cls = proc_cls;
  GNUNET_CONTAINER_multihashmap_iterate (plugin->keyvalue,
                                         &return_value,
                                         &gac);
  proc (proc_cls, NULL, 0);
}
 
 
struct RemoveContext
{
  struct Value *value;
 
  uint32_t size;
 
  const void *data;
};
 
 
static int
remove_iterator (void *cls,
                 const struct GNUNET_HashCode *key,
                 void *val)
{
  struct RemoveContext *rc = cls;
  struct Value *value = val;
 
  if (value->size != rc->size)
    return GNUNET_YES;
  if (0 != memcmp (value->data, rc->data, rc->size))
    return GNUNET_YES;
  rc->value = value;
  return GNUNET_NO;
}
 
 
static void
heap_plugin_remove_key (void *cls,
                        const struct GNUNET_HashCode *key,
                        uint32_t size,
                        const void *data,
                        PluginRemoveCont cont,
                        void *cont_cls)
{
  struct Plugin *plugin = cls;
  struct RemoveContext rc;
 
  rc.value = NULL;
  rc.size = size;
  rc.data = data;
  GNUNET_CONTAINER_multihashmap_get_multiple (plugin->keyvalue,
                                              key,
                                              &remove_iterator,
                                              &rc);
  if (NULL == rc.value)
  {
    cont (cont_cls,
          key,
          size,
          GNUNET_NO,
          NULL);
    return;
  }
  delete_value (plugin,
                rc.value);
  cont (cont_cls,
        key,
        size,
        GNUNET_OK,
        NULL);
}
 
void *
libgnunet_plugin_datastore_heap_init (void *cls);
 
void *
libgnunet_plugin_datastore_heap_init (void *cls)
{
  struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
  struct GNUNET_DATASTORE_PluginFunctions *api;
  struct Plugin *plugin;
  unsigned long long esize;
 
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number (env->cfg,
                                             "datastore-heap",
                                             "HASHMAPSIZE",
                                             &esize))
    esize = 128 * 1024;
  plugin = GNUNET_new (struct Plugin);
  plugin->env = env;
  plugin->keyvalue = GNUNET_CONTAINER_multihashmap_create (esize, GNUNET_YES);
  plugin->by_expiration = GNUNET_CONTAINER_heap_create (
    GNUNET_CONTAINER_HEAP_ORDER_MIN);
  plugin->by_replication = GNUNET_CONTAINER_heap_create (
    GNUNET_CONTAINER_HEAP_ORDER_MAX);
  api = GNUNET_new (struct GNUNET_DATASTORE_PluginFunctions);
  api->cls = plugin;
  api->estimate_size = &heap_plugin_estimate_size;
  api->put = &heap_plugin_put;
  api->get_key = &heap_plugin_get_key;
  api->get_replication = &heap_plugin_get_replication;
  api->get_expiration = &heap_plugin_get_expiration;
  api->get_zero_anonymity = &heap_plugin_get_zero_anonymity;
  api->drop = &heap_plugin_drop;
  api->get_keys = &heap_get_keys;
  api->remove_key = &heap_plugin_remove_key;
  GNUNET_log_from (GNUNET_ERROR_TYPE_INFO, "heap",
                   _ ("Heap database running\n"));
  return api;
}
 
 
static int
free_value (void *cls,
            const struct GNUNET_HashCode *key,
            void *val)
{
  struct Plugin *plugin = cls;
  struct Value *value = val;
 
  delete_value (plugin, value);
  return GNUNET_OK;
}
 
void *
libgnunet_plugin_datastore_heap_done (void *cls);
 
void *
libgnunet_plugin_datastore_heap_done (void *cls)
{
  struct GNUNET_DATASTORE_PluginFunctions *api = cls;
  struct Plugin *plugin = api->cls;
 
  GNUNET_CONTAINER_multihashmap_iterate (plugin->keyvalue,
                                         &free_value,
                                         plugin);
  GNUNET_CONTAINER_multihashmap_destroy (plugin->keyvalue);
  GNUNET_CONTAINER_heap_destroy (plugin->by_expiration);
  GNUNET_CONTAINER_heap_destroy (plugin->by_replication);
  GNUNET_free (plugin);
  GNUNET_free (api);
  return NULL;
}
 
 
/* end of plugin_datastore_heap.c */