plugin_datastore_mysql.c

Go to the documentation of this file.

/*
     This file is part of GNUnet
     Copyright (C) 2009, 2010, 2011 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"
#include "gnunet_util_lib.h"
#include "gnunet_mysql_lib.h"
#include "gnunet_my_lib.h"

#define MAX_DATUM_SIZE 65536


struct Plugin {
  struct GNUNET_DATASTORE_PluginEnvironment *env;

  struct GNUNET_MYSQL_Context *mc;

#define INSERT_ENTRY "INSERT INTO gn090 (repl,type,prio,anonLevel,expire,rvalue,hash,vhash,value) VALUES (?,?,?,?,?,?,?,?,?)"
  struct GNUNET_MYSQL_StatementHandle *insert_entry;

#define DELETE_ENTRY_BY_UID "DELETE FROM gn090 WHERE uid=?"
  struct GNUNET_MYSQL_StatementHandle *delete_entry_by_uid;

#define DELETE_ENTRY_BY_HASH_VALUE "DELETE FROM gn090 " \
  "WHERE hash = ? AND " \
  "value = ? " \
  "LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *delete_entry_by_hash_value;

#define RESULT_COLUMNS "repl, type, prio, anonLevel, expire, hash, value, uid"

#define SELECT_ENTRY "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "WHERE uid >= ? AND " \
  "(rvalue >= ? OR 0 = ?) " \
  "ORDER BY uid LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *select_entry;

#define SELECT_ENTRY_BY_HASH "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "FORCE INDEX (idx_hash_type_uid) " \
  "WHERE hash=? AND " \
  "uid >= ? AND " \
  "(rvalue >= ? OR 0 = ?) " \
  "ORDER BY uid LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *select_entry_by_hash;

#define SELECT_ENTRY_BY_HASH_AND_TYPE "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "FORCE INDEX (idx_hash_type_uid) " \
  "WHERE hash = ? AND " \
  "type = ? AND " \
  "uid >= ? AND " \
  "(rvalue >= ? OR 0 = ?) " \
  "ORDER BY uid LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *select_entry_by_hash_and_type;

#define UPDATE_ENTRY "UPDATE gn090 SET " \
  "prio = prio + ?, " \
  "repl = repl + ?, " \
  "expire = GREATEST(expire, ?) " \
  "WHERE hash = ? AND vhash = ?"
  struct GNUNET_MYSQL_StatementHandle *update_entry;

#define DEC_REPL "UPDATE gn090 SET repl=GREATEST (1, repl) - 1 WHERE uid=?"
  struct GNUNET_MYSQL_StatementHandle *dec_repl;

#define SELECT_SIZE "SELECT SUM(LENGTH(value)+256) FROM gn090"
  struct GNUNET_MYSQL_StatementHandle *get_size;

#define SELECT_IT_NON_ANONYMOUS "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "FORCE INDEX (idx_anonLevel_type_rvalue) " \
  "WHERE anonLevel=0 AND " \
  "type=? AND " \
  "uid >= ? " \
  "ORDER BY uid LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *zero_iter;

#define SELECT_IT_EXPIRATION "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "FORCE INDEX (idx_expire) " \
  "WHERE expire < ? " \
  "ORDER BY expire ASC LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *select_expiration;

#define SELECT_IT_PRIORITY "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "FORCE INDEX (idx_prio) " \
  "ORDER BY prio ASC LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *select_priority;

#define SELECT_IT_REPLICATION "SELECT " RESULT_COLUMNS " FROM gn090 " \
  "FORCE INDEX (idx_repl_rvalue) " \
  "WHERE repl=? AND " \
  " (rvalue>=? OR" \
  "  NOT EXISTS (SELECT 1 FROM gn090 FORCE INDEX (idx_repl_rvalue) WHERE repl=? AND rvalue>=?)) " \
  "ORDER BY rvalue ASC " \
  "LIMIT 1"
  struct GNUNET_MYSQL_StatementHandle *select_replication;

#define SELECT_MAX_REPL "SELECT MAX(repl) FROM gn090"
  struct GNUNET_MYSQL_StatementHandle *max_repl;

#define GET_ALL_KEYS "SELECT hash from gn090"
  struct GNUNET_MYSQL_StatementHandle *get_all_keys;
};

#define MAX_PARAM 16

static int
do_delete_entry(struct Plugin *plugin,
                unsigned long long uid)
{
  int ret;
  uint64_t uid64 = (uint64_t)uid;
  struct GNUNET_MY_QueryParam params_delete[] = {
    GNUNET_MY_query_param_uint64(&uid64),
    GNUNET_MY_query_param_end
  };

  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Deleting value %llu from gn090 table\n",
             uid);
  ret = GNUNET_MY_exec_prepared(plugin->mc,
                                plugin->delete_entry_by_uid,
                                params_delete);
  if (ret >= 0)
    {
      return GNUNET_OK;
    }
  GNUNET_log(GNUNET_ERROR_TYPE_WARNING,
             "Deleting value %llu from gn090 table failed\n",
             (unsigned long long)uid);
  return ret;
}


static void
mysql_plugin_estimate_size(void *cls,
                           unsigned long long *estimate)
{
  struct Plugin *plugin = cls;
  uint64_t total;
  int ret;
  struct GNUNET_MY_QueryParam params_get[] = {
    GNUNET_MY_query_param_end
  };
  struct GNUNET_MY_ResultSpec results_get[] = {
    GNUNET_MY_result_spec_uint64(&total),
    GNUNET_MY_result_spec_end
  };

  ret = GNUNET_MY_exec_prepared(plugin->mc,
                                plugin->get_size,
                                params_get);
  *estimate = 0;
  total = UINT64_MAX;
  if ((GNUNET_OK == ret) &&
      (GNUNET_OK ==
       GNUNET_MY_extract_result(plugin->get_size,
                                results_get)))
    {
      *estimate = (unsigned long long)total;
      GNUNET_log(GNUNET_ERROR_TYPE_INFO,
                 "Size estimate for MySQL payload is %lld\n",
                 (long long)total);
      GNUNET_assert(UINT64_MAX != total);
      GNUNET_break(GNUNET_NO ==
                   GNUNET_MY_extract_result(plugin->get_size,
                                            NULL));
    }
}


static void
mysql_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;
  uint64_t lexpiration = expiration.abs_value_us;
  struct GNUNET_HashCode vhash;

  GNUNET_CRYPTO_hash(data,
                     size,
                     &vhash);
  if (!absent)
    {
      struct GNUNET_MY_QueryParam params_update[] = {
        GNUNET_MY_query_param_uint32(&priority),
        GNUNET_MY_query_param_uint32(&replication),
        GNUNET_MY_query_param_uint64(&lexpiration),
        GNUNET_MY_query_param_auto_from_type(key),
        GNUNET_MY_query_param_auto_from_type(&vhash),
        GNUNET_MY_query_param_end
      };

      if (GNUNET_OK !=
          GNUNET_MY_exec_prepared(plugin->mc,
                                  plugin->update_entry,
                                  params_update))
        {
          cont(cont_cls,
               key,
               size,
               GNUNET_SYSERR,
               _("MySQL statement run failure"));
          return;
        }

      MYSQL_STMT *stmt = GNUNET_MYSQL_statement_get_stmt(plugin->update_entry);
      my_ulonglong rows = mysql_stmt_affected_rows(stmt);

      GNUNET_break(GNUNET_NO ==
                   GNUNET_MY_extract_result(plugin->update_entry,
                                            NULL));
      if (0 != rows)
        {
          cont(cont_cls,
               key,
               size,
               GNUNET_NO,
               NULL);
          return;
        }
    }

  uint64_t lrvalue = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK,
                                              UINT64_MAX);
  struct GNUNET_MY_QueryParam params_insert[] = {
    GNUNET_MY_query_param_uint32(&replication),
    GNUNET_MY_query_param_uint32(&type),
    GNUNET_MY_query_param_uint32(&priority),
    GNUNET_MY_query_param_uint32(&anonymity),
    GNUNET_MY_query_param_uint64(&lexpiration),
    GNUNET_MY_query_param_uint64(&lrvalue),
    GNUNET_MY_query_param_auto_from_type(key),
    GNUNET_MY_query_param_auto_from_type(&vhash),
    GNUNET_MY_query_param_fixed_size(data, size),
    GNUNET_MY_query_param_end
  };

  if (size > MAX_DATUM_SIZE)
    {
      GNUNET_break(0);
      cont(cont_cls, key, size, GNUNET_SYSERR, _("Data too large"));
      return;
    }

  if (GNUNET_OK !=
      GNUNET_MY_exec_prepared(plugin->mc,
                              plugin->insert_entry,
                              params_insert))
    {
      cont(cont_cls,
           key,
           size,
           GNUNET_SYSERR,
           _("MySQL statement run failure"));
      return;
    }
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Inserted value `%s' with size %u into gn090 table\n",
             GNUNET_h2s(key),
             (unsigned int)size);
  if (size > 0)
    plugin->env->duc(plugin->env->cls,
                     size);
  GNUNET_break(GNUNET_NO ==
               GNUNET_MY_extract_result(plugin->insert_entry,
                                        NULL));
  cont(cont_cls,
       key,
       size,
       GNUNET_OK,
       NULL);
}


static void
execute_select(struct Plugin *plugin,
               struct GNUNET_MYSQL_StatementHandle *stmt,
               PluginDatumProcessor proc,
               void *proc_cls,
               struct GNUNET_MY_QueryParam *params_select)
{
  int ret;
  uint32_t replication;
  uint32_t type;
  uint32_t priority;
  uint32_t anonymity;
  uint64_t uid;
  size_t value_size;
  void *value;
  struct GNUNET_HashCode key;
  struct GNUNET_TIME_Absolute expiration;
  struct GNUNET_MY_ResultSpec results_select[] = {
    GNUNET_MY_result_spec_uint32(&replication),
    GNUNET_MY_result_spec_uint32(&type),
    GNUNET_MY_result_spec_uint32(&priority),
    GNUNET_MY_result_spec_uint32(&anonymity),
    GNUNET_MY_result_spec_absolute_time(&expiration),
    GNUNET_MY_result_spec_auto_from_type(&key),
    GNUNET_MY_result_spec_variable_size(&value, &value_size),
    GNUNET_MY_result_spec_uint64(&uid),
    GNUNET_MY_result_spec_end
  };

  ret = GNUNET_MY_exec_prepared(plugin->mc,
                                stmt,
                                params_select);
  if (GNUNET_OK != ret)
    {
      proc(proc_cls,
           NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }

  ret = GNUNET_MY_extract_result(stmt,
                                 results_select);
  if (GNUNET_OK != ret)
    {
      proc(proc_cls,
           NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }

  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Found %u-byte value under key `%s' with prio %u, anon %u, expire %s selecting from gn090 table\n",
             (unsigned int)value_size,
             GNUNET_h2s(&key),
             (unsigned int)priority,
             (unsigned int)anonymity,
             GNUNET_STRINGS_absolute_time_to_string(expiration));
  GNUNET_assert(value_size < MAX_DATUM_SIZE);
  GNUNET_break(GNUNET_NO ==
               GNUNET_MY_extract_result(stmt,
                                        NULL));
  ret = proc(proc_cls,
             &key,
             value_size,
             value,
             type,
             priority,
             anonymity,
             replication,
             expiration,
             uid);
  GNUNET_MY_cleanup_result(results_select);
  if (GNUNET_NO == ret)
    {
      do_delete_entry(plugin, uid);
      if (0 != value_size)
        plugin->env->duc(plugin->env->cls,
                         -value_size);
    }
}


static void
mysql_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;
  uint64_t rvalue;

  if (random)
    {
      rvalue = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK,
                                        UINT64_MAX);
      next_uid = 0;
    }
  else
    rvalue = 0;

  if (NULL == key)
    {
      struct GNUNET_MY_QueryParam params_select[] = {
        GNUNET_MY_query_param_uint64(&next_uid),
        GNUNET_MY_query_param_uint64(&rvalue),
        GNUNET_MY_query_param_uint64(&rvalue),
        GNUNET_MY_query_param_end
      };

      execute_select(plugin,
                     plugin->select_entry,
                     proc,
                     proc_cls,
                     params_select);
    }
  else if (type != GNUNET_BLOCK_TYPE_ANY)
    {
      struct GNUNET_MY_QueryParam params_select[] = {
        GNUNET_MY_query_param_auto_from_type(key),
        GNUNET_MY_query_param_uint32(&type),
        GNUNET_MY_query_param_uint64(&next_uid),
        GNUNET_MY_query_param_uint64(&rvalue),
        GNUNET_MY_query_param_uint64(&rvalue),
        GNUNET_MY_query_param_end
      };

      execute_select(plugin,
                     plugin->select_entry_by_hash_and_type,
                     proc,
                     proc_cls,
                     params_select);
    }
  else
    {
      struct GNUNET_MY_QueryParam params_select[] = {
        GNUNET_MY_query_param_auto_from_type(key),
        GNUNET_MY_query_param_uint64(&next_uid),
        GNUNET_MY_query_param_uint64(&rvalue),
        GNUNET_MY_query_param_uint64(&rvalue),
        GNUNET_MY_query_param_end
      };

      execute_select(plugin,
                     plugin->select_entry_by_hash,
                     proc,
                     proc_cls,
                     params_select);
    }
}


static void
mysql_plugin_get_zero_anonymity(void *cls,
                                uint64_t next_uid,
                                enum GNUNET_BLOCK_Type type,
                                PluginDatumProcessor proc,
                                void *proc_cls)
{
  struct Plugin *plugin = cls;
  uint32_t typei = (uint32_t)type;

  struct GNUNET_MY_QueryParam params_zero_iter[] = {
    GNUNET_MY_query_param_uint32(&typei),
    GNUNET_MY_query_param_uint64(&next_uid),
    GNUNET_MY_query_param_end
  };

  execute_select(plugin,
                 plugin->zero_iter,
                 proc,
                 proc_cls,
                 params_zero_iter);
}


struct ReplCtx {
  struct Plugin *plugin;

  PluginDatumProcessor proc;

  void *proc_cls;
};


static int
repl_proc(void *cls,
          const struct GNUNET_HashCode *key,
          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,
          uint64_t uid)
{
  struct ReplCtx *rc = cls;
  struct Plugin *plugin = rc->plugin;
  int ret;
  int iret;

  ret = rc->proc(rc->proc_cls,
                 key,
                 size,
                 data,
                 type,
                 priority,
                 anonymity,
                 replication,
                 expiration,
                 uid);
  if (NULL != key)
    {
      struct GNUNET_MY_QueryParam params_proc[] = {
        GNUNET_MY_query_param_uint64(&uid),
        GNUNET_MY_query_param_end
      };

      iret = GNUNET_MY_exec_prepared(plugin->mc,
                                     plugin->dec_repl,
                                     params_proc);
      if (GNUNET_SYSERR == iret)
        {
          GNUNET_log(GNUNET_ERROR_TYPE_WARNING,
                     "Failed to reduce replication counter\n");
          return GNUNET_SYSERR;
        }
    }
  return ret;
}


static void
mysql_plugin_get_replication(void *cls,
                             PluginDatumProcessor proc,
                             void *proc_cls)
{
  struct Plugin *plugin = cls;
  uint64_t rvalue;
  uint32_t repl;
  struct ReplCtx rc;
  struct GNUNET_MY_QueryParam params_get[] = {
    GNUNET_MY_query_param_end
  };
  struct GNUNET_MY_ResultSpec results_get[] = {
    GNUNET_MY_result_spec_uint32(&repl),
    GNUNET_MY_result_spec_end
  };
  struct GNUNET_MY_QueryParam params_select[] = {
    GNUNET_MY_query_param_uint32(&repl),
    GNUNET_MY_query_param_uint64(&rvalue),
    GNUNET_MY_query_param_uint32(&repl),
    GNUNET_MY_query_param_uint64(&rvalue),
    GNUNET_MY_query_param_end
  };

  rc.plugin = plugin;
  rc.proc = proc;
  rc.proc_cls = proc_cls;

  if (1 !=
      GNUNET_MY_exec_prepared(plugin->mc,
                              plugin->max_repl,
                              params_get))
    {
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }

  if (GNUNET_OK !=
      GNUNET_MY_extract_result(plugin->max_repl,
                               results_get))
    {
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }
  GNUNET_break(GNUNET_NO ==
               GNUNET_MY_extract_result(plugin->max_repl,
                                        NULL));
  rvalue = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK,
                                    UINT64_MAX);

  execute_select(plugin,
                 plugin->select_replication,
                 &repl_proc,
                 &rc,
                 params_select);
}


static void
mysql_plugin_get_keys(void *cls,
                      PluginKeyProcessor proc,
                      void *proc_cls)
{
  struct Plugin *plugin = cls;
  int ret;
  MYSQL_STMT *statement;
  unsigned int cnt;
  struct GNUNET_HashCode key;
  struct GNUNET_HashCode last;
  struct GNUNET_MY_QueryParam params_select[] = {
    GNUNET_MY_query_param_end
  };
  struct GNUNET_MY_ResultSpec results_select[] = {
    GNUNET_MY_result_spec_auto_from_type(&key),
    GNUNET_MY_result_spec_end
  };

  GNUNET_assert(NULL != proc);
  statement = GNUNET_MYSQL_statement_get_stmt(plugin->get_all_keys);
  if (GNUNET_OK !=
      GNUNET_MY_exec_prepared(plugin->mc,
                              plugin->get_all_keys,
                              params_select))
    {
      GNUNET_log(GNUNET_ERROR_TYPE_ERROR,
                 _("`%s' for `%s' failed at %s:%d with error: %s\n"),
                 "mysql_stmt_execute",
                 GET_ALL_KEYS,
                 __FILE__,
                 __LINE__,
                 mysql_stmt_error(statement));
      GNUNET_MYSQL_statements_invalidate(plugin->mc);
      proc(proc_cls, NULL, 0);
      return;
    }
  memset(&last, 0, sizeof(last));   /* make static analysis happy */
  ret = GNUNET_YES;
  cnt = 0;
  while (ret == GNUNET_YES)
    {
      ret = GNUNET_MY_extract_result(plugin->get_all_keys,
                                     results_select);
      if (0 != GNUNET_memcmp(&last,
                             &key))
        {
          if (0 != cnt)
            proc(proc_cls,
                 &last,
                 cnt);
          cnt = 1;
          last = key;
        }
      else
        {
          cnt++;
        }
    }
  if (0 != cnt)
    proc(proc_cls,
         &last,
         cnt);
  /* finally, let app know we are done */
  proc(proc_cls,
       NULL,
       0);
  if (GNUNET_SYSERR == ret)
    {
      GNUNET_log(GNUNET_ERROR_TYPE_ERROR,
                 _("`%s' failed at %s:%d with error: %s\n"),
                 "mysql_stmt_fetch",
                 __FILE__,
                 __LINE__,
                 mysql_stmt_error(statement));
      GNUNET_MYSQL_statements_invalidate(plugin->mc);
      return;
    }
}


struct ExpiCtx {
  struct Plugin *plugin;

  PluginDatumProcessor proc;

  void *proc_cls;
};



static int
expi_proc(void *cls,
          const struct GNUNET_HashCode *key,
          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,
          uint64_t uid)
{
  struct ExpiCtx *rc = cls;
  struct Plugin *plugin = rc->plugin;
  struct GNUNET_MY_QueryParam params_select[] = {
    GNUNET_MY_query_param_end
  };

  if (NULL == key)
    {
      execute_select(plugin,
                     plugin->select_priority,
                     rc->proc,
                     rc->proc_cls,
                     params_select);
      return GNUNET_SYSERR;
    }
  return rc->proc(rc->proc_cls,
                  key,
                  size,
                  data,
                  type,
                  priority,
                  anonymity,
                  replication,
                  expiration,
                  uid);
}


static void
mysql_plugin_get_expiration(void *cls,
                            PluginDatumProcessor proc,
                            void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct GNUNET_TIME_Absolute now;
  struct GNUNET_MY_QueryParam params_select[] = {
    GNUNET_MY_query_param_absolute_time(&now),
    GNUNET_MY_query_param_end
  };
  struct ExpiCtx rc;

  rc.plugin = plugin;
  rc.proc = proc;
  rc.proc_cls = proc_cls;
  now = GNUNET_TIME_absolute_get();
  execute_select(plugin,
                 plugin->select_expiration,
                 expi_proc,
                 &rc,
                 params_select);
}


static void
mysql_plugin_drop(void *cls)
{
  struct Plugin *plugin = cls;

  if (GNUNET_OK !=
      GNUNET_MYSQL_statement_run(plugin->mc,
                                 "DROP TABLE gn090"))
    return;                     /* error */
  plugin->env->duc(plugin->env->cls, 0);
}


static void
mysql_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 GNUNET_MY_QueryParam params_delete[] = {
    GNUNET_MY_query_param_auto_from_type(key),
    GNUNET_MY_query_param_fixed_size(data, size),
    GNUNET_MY_query_param_end
  };

  if (GNUNET_OK !=
      GNUNET_MY_exec_prepared(plugin->mc,
                              plugin->delete_entry_by_hash_value,
                              params_delete))
    {
      GNUNET_log(GNUNET_ERROR_TYPE_WARNING,
                 "Removing key `%s' from gn090 table failed\n",
                 GNUNET_h2s(key));
      cont(cont_cls,
           key,
           size,
           GNUNET_SYSERR,
           _("MySQL statement run failure"));
      return;
    }

  MYSQL_STMT *stmt = GNUNET_MYSQL_statement_get_stmt(plugin->delete_entry_by_hash_value);
  my_ulonglong rows = mysql_stmt_affected_rows(stmt);

  if (0 == rows)
    {
      cont(cont_cls,
           key,
           size,
           GNUNET_NO,
           NULL);
      return;
    }
  plugin->env->duc(plugin->env->cls,
                   -size);
  cont(cont_cls,
       key,
       size,
       GNUNET_OK,
       NULL);
}


void *
libgnunet_plugin_datastore_mysql_init(void *cls)
{
  struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
  struct GNUNET_DATASTORE_PluginFunctions *api;
  struct Plugin *plugin;

  plugin = GNUNET_new(struct Plugin);
  plugin->env = env;
  plugin->mc = GNUNET_MYSQL_context_create(env->cfg,
                                           "datastore-mysql");
  if (NULL == plugin->mc)
    {
      GNUNET_free(plugin);
      return NULL;
    }
#define MRUNS(a) (GNUNET_OK != GNUNET_MYSQL_statement_run(plugin->mc, a))
#define PINIT(a, b) (NULL == (a = GNUNET_MYSQL_statement_prepare(plugin->mc, b)))
  if (MRUNS
        ("CREATE TABLE IF NOT EXISTS gn090 ("
        " repl INT(11) UNSIGNED NOT NULL DEFAULT 0,"
        " type INT(11) UNSIGNED NOT NULL DEFAULT 0,"
        " prio INT(11) UNSIGNED NOT NULL DEFAULT 0,"
        " anonLevel INT(11) UNSIGNED NOT NULL DEFAULT 0,"
        " expire BIGINT UNSIGNED NOT NULL DEFAULT 0,"
        " rvalue BIGINT UNSIGNED NOT NULL,"
        " hash BINARY(64) NOT NULL DEFAULT '',"
        " vhash BINARY(64) NOT NULL DEFAULT '',"
        " value BLOB NOT NULL DEFAULT ''," " uid BIGINT NOT NULL AUTO_INCREMENT,"
        " PRIMARY KEY (uid),"
        " INDEX idx_hash_type_uid (hash(64),type,rvalue),"
        " INDEX idx_prio (prio),"
        " INDEX idx_repl_rvalue (repl,rvalue),"
        " INDEX idx_expire (expire),"
        " INDEX idx_anonLevel_type_rvalue (anonLevel,type,rvalue)"
        ") ENGINE=InnoDB") || MRUNS("SET AUTOCOMMIT = 1") ||
      PINIT(plugin->insert_entry, INSERT_ENTRY) ||
      PINIT(plugin->delete_entry_by_uid, DELETE_ENTRY_BY_UID) ||
      PINIT(plugin->delete_entry_by_hash_value, DELETE_ENTRY_BY_HASH_VALUE) ||
      PINIT(plugin->select_entry, SELECT_ENTRY) ||
      PINIT(plugin->select_entry_by_hash, SELECT_ENTRY_BY_HASH) ||
      PINIT(plugin->select_entry_by_hash_and_type,
            SELECT_ENTRY_BY_HASH_AND_TYPE) ||
      PINIT(plugin->get_size, SELECT_SIZE) ||
      PINIT(plugin->update_entry, UPDATE_ENTRY) ||
      PINIT(plugin->dec_repl, DEC_REPL) ||
      PINIT(plugin->zero_iter, SELECT_IT_NON_ANONYMOUS) ||
      PINIT(plugin->select_expiration, SELECT_IT_EXPIRATION) ||
      PINIT(plugin->select_priority, SELECT_IT_PRIORITY) ||
      PINIT(plugin->max_repl, SELECT_MAX_REPL) ||
      PINIT(plugin->get_all_keys, GET_ALL_KEYS) ||
      PINIT(plugin->select_replication, SELECT_IT_REPLICATION) ||
      false)
    {
      GNUNET_MYSQL_context_destroy(plugin->mc);
      GNUNET_free(plugin);
      return NULL;
    }
#undef PINIT
#undef MRUNS

  api = GNUNET_new(struct GNUNET_DATASTORE_PluginFunctions);
  api->cls = plugin;
  api->estimate_size = &mysql_plugin_estimate_size;
  api->put = &mysql_plugin_put;
  api->get_key = &mysql_plugin_get_key;
  api->get_replication = &mysql_plugin_get_replication;
  api->get_expiration = &mysql_plugin_get_expiration;
  api->get_zero_anonymity = &mysql_plugin_get_zero_anonymity;
  api->get_keys = &mysql_plugin_get_keys;
  api->drop = &mysql_plugin_drop;
  api->remove_key = &mysql_plugin_remove_key;
  GNUNET_log_from(GNUNET_ERROR_TYPE_INFO, "mysql",
                  _("Mysql database running\n"));
  return api;
}


void *
libgnunet_plugin_datastore_mysql_done(void *cls)
{
  struct GNUNET_DATASTORE_PluginFunctions *api = cls;
  struct Plugin *plugin = api->cls;

  GNUNET_MYSQL_context_destroy(plugin->mc);
  GNUNET_free(plugin);
  GNUNET_free(api);
  return NULL;
}

/* end of plugin_datastore_mysql.c */