[GNUnet-SVN] r12745 - gnunet/src/datacache

[gnunet]

Author: grothoff
Date: 2010-08-26 15:32:21 +0200 (Thu, 26 Aug 2010)
New Revision: 12745

Added:
   gnunet/src/datacache/plugin_datacache_postgres.c
Log:
towards datacache

Added: gnunet/src/datacache/plugin_datacache_postgres.c
===================================================================
--- gnunet/src/datacache/plugin_datacache_postgres.c                            
(rev 0)
+++ gnunet/src/datacache/plugin_datacache_postgres.c    2010-08-26 13:32:21 UTC 
(rev 12745)
@@ -0,0 +1,519 @@
+/*
+     This file is part of GNUnet
+     (C) 2006, 2009, 2010 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 3, 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 datacache/plugin_datacache_postgres.c
+ * @brief postgres for an implementation of a database backend for the 
datacache
+ * @author Christian Grothoff
+ */
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "plugin_datacache.h"
+#include <postgresql/libpq-fe.h>
+
+#define DEBUG_POSTGRES GNUNET_NO
+
+
+
+/**
+ * Context for all functions in this plugin.
+ */
+struct Plugin 
+{
+  /**
+   * Our execution environment.
+   */
+  struct GNUNET_DATACACHE_PluginEnvironment *env;
+
+  /**
+   * Native Postgres database handle.
+   */
+  PGconn *dbh;
+
+};
+
+
+/**
+ * Check if the result obtained from Postgres has
+ * the desired status code.  If not, log an error, clear the
+ * result and return GNUNET_SYSERR.
+ * 
+ * @return GNUNET_OK if the result is acceptable
+ */
+static int
+check_result (struct Plugin *plugin,
+             PGresult * ret,
+              int expected_status,
+              const char *command, const char *args, int line)
+{
+  if (ret == NULL)
+    {
+      GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                      "datastore-postgres",
+                      "Postgres failed to allocate result for `%s:%s' at %d\n",
+                      command, args, line);
+      return GNUNET_SYSERR;
+    }
+  if (PQresultStatus (ret) != expected_status)
+    {
+      GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                      "datastore-postgres",
+                      _("`%s:%s' failed at %s:%d with error: %s"),
+                      command, args, __FILE__, line, PQerrorMessage 
(plugin->dbh));
+      PQclear (ret);
+      return GNUNET_SYSERR;
+    }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Run simple SQL statement (without results).
+ */
+static int
+pq_exec (struct Plugin *plugin,
+        const char *sql, int line)
+{
+  PGresult *ret;
+  ret = PQexec (plugin->dbh, sql);
+  if (GNUNET_OK != check_result (plugin,
+                                ret, 
+                                PGRES_COMMAND_OK, "PQexec", sql, line))
+    return GNUNET_SYSERR;
+  PQclear (ret);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Prepare SQL statement.
+ */
+static int
+pq_prepare (struct Plugin *plugin,
+           const char *name, const char *sql, int nparms, int line)
+{
+  PGresult *ret;
+  ret = PQprepare (plugin->dbh, name, sql, nparms, NULL);
+  if (GNUNET_OK !=
+      check_result (plugin, 
+                   ret, PGRES_COMMAND_OK, "PQprepare", sql, line))
+    return GNUNET_SYSERR;
+  PQclear (ret);
+  return GNUNET_OK;
+}
+
+
+/**
+ * @brief Get a database handle
+ * @return GNUNET_OK on success, GNUNET_SYSERR on error
+ */
+static int
+init_connection (struct Plugin *plugin)
+{
+  char *conninfo;
+  PGresult *ret;
+
+  /* Open database and precompile statements */
+  conninfo = NULL;
+  GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
+                                        "datacache-postgres",
+                                        "CONFIG",
+                                        &conninfo);
+  plugin->dbh = PQconnectdb (conninfo == NULL ? "" : conninfo);
+  GNUNET_free_non_null (conninfo);
+  if (NULL == plugin->dbh)
+    {
+      /* FIXME: warn about out-of-memory? */
+      return GNUNET_SYSERR;
+    }
+  if (PQstatus (plugin->dbh) != CONNECTION_OK)
+    {
+      GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
+                      "datacache-postgres",
+                      _("Unable to initialize Postgres: %s"),
+                      PQerrorMessage (plugin->dbh));
+      PQfinish (plugin->dbh);
+      plugin->dbh = NULL;
+      return GNUNET_SYSERR;
+    }
+  ret = PQexec (plugin->dbh,
+                "CREATE TEMPORARY `TABLE gn090dc ("
+                "  type INTEGER NOT NULL DEFAULT 0,"
+                "  discard_time BIGINT NOT NULL DEFAULT 0,"
+                "  key BYTEA NOT NULL DEFAULT '',"
+                "  value BYTEA NOT NULL DEFAULT '')" "WITH OIDS");
+  if ( (ret == NULL) || 
+       ( (PQresultStatus (ret) != PGRES_COMMAND_OK) && 
+        (0 != strcmp ("42P07",    /* duplicate table */
+                      PQresultErrorField
+                      (ret,
+                       PG_DIAG_SQLSTATE)))))
+    {
+      check_result (plugin,
+                   ret, PGRES_COMMAND_OK, "CREATE TABLE", "gn090dc", __LINE__);
+      PQfinish (plugin->dbh);
+      plugin->dbh = NULL;
+      return GNUNET_SYSERR;
+    }
+  if (PQresultStatus (ret) == PGRES_COMMAND_OK)
+    {
+      if ((GNUNET_OK !=
+           pq_exec (plugin, "CREATE INDEX idx_key ON gn090dc (key)", 
__LINE__)) ||
+          (GNUNET_OK !=
+           pq_exec (plugin, "CREATE INDEX idx_dt ON gn090 (discard_time)",
+                    __LINE__)) )
+        {
+          PQclear (ret);
+          PQfinish (plugin->dbh);
+          plugin->dbh = NULL;
+          return GNUNET_SYSERR;
+        }
+    }
+  PQclear (ret);
+#if 1
+  ret = PQexec (plugin->dbh,
+                "ALTER TABLE gn090dc ALTER value SET STORAGE EXTERNAL");
+  if (GNUNET_OK != 
+      check_result (plugin,
+                   ret, PGRES_COMMAND_OK,
+                   "ALTER TABLE", "gn090dc", __LINE__))
+    {
+      PQfinish (plugin->dbh);
+      plugin->dbh = NULL;
+      return GNUNET_SYSERR;
+    }
+  PQclear (ret);
+  ret = PQexec (plugin->dbh,
+                "ALTER TABLE gn090dc ALTER key SET STORAGE PLAIN");
+  if (GNUNET_OK !=
+      check_result (plugin,
+                   ret, PGRES_COMMAND_OK,
+                   "ALTER TABLE", "gn090dc", __LINE__))
+    {
+      PQfinish (plugin->dbh);
+      plugin->dbh = NULL;
+      return GNUNET_SYSERR;
+    }
+  PQclear (ret);
+#endif
+  if ((GNUNET_OK !=
+       pq_prepare (plugin,
+                  "getkt",
+                   "SELECT discard_time,type,value FROM gn090dc "
+                   "WHERE hash=$1 type=$2 ",
+                   2,
+                   __LINE__)) ||
+      (GNUNET_OK !=
+       pq_prepare (plugin,
+                  "getk",
+                   "SELECT discard_time,type,value FROM gn090dc "
+                   "WHERE hash=$1",
+                   1,
+                   __LINE__)) ||
+      (GNUNET_OK !=
+       pq_prepare (plugin,
+                  "getm",
+                   "SELECT length(value),oid FROM gn090dc"
+                   "ORDER BY discard_time ASC LIMIT 1",
+                   0,
+                   __LINE__)) ||
+      (GNUNET_OK !=
+       pq_prepare (plugin,
+                  "delrow",
+                   "DELETE FROM gn090dc WHERE oid=$1",
+                   1,
+                   __LINE__)) ||
+      (GNUNET_OK !=
+       pq_prepare (plugin,
+                  "put",
+                   "INSERT INTO gn090dc (type, discard_time, key, value) "
+                   "VALUES ($1, $2, $3, $4)",
+                   4,
+                   __LINE__)) )
+    {
+      PQfinish (plugin->dbh);
+      plugin->dbh = NULL;
+      return GNUNET_SYSERR;
+    }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Delete the row identified by the given rowid (qid
+ * in postgres).
+ *
+ * @return GNUNET_OK on success
+ */
+static int
+delete_by_rowid (struct Plugin *plugin,
+                unsigned int rowid)
+{
+  const char *paramValues[] = { (const char *) &rowid };
+  int paramLengths[] = { sizeof (rowid) };
+  const int paramFormats[] = { 1 };
+  PGresult *ret;
+
+  ret = PQexecPrepared (plugin->dbh,
+                        "delrow",
+                        1, paramValues, paramLengths, paramFormats, 1);
+  if (GNUNET_OK !=
+      check_result (plugin,
+                   ret, PGRES_COMMAND_OK, "PQexecPrepared", "delrow",
+                    __LINE__))
+    {
+      return GNUNET_SYSERR;
+    }
+  PQclear (ret);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Store an item in the datastore.
+ *
+ * @param cls closure (our "struct Plugin")
+ * @param key key to store data under
+ * @param size number of bytes in data
+ * @param data data to store
+ * @param type type of the value
+ * @param discard_time when to discard the value in any case
+ * @return 0 on error, number of bytes used otherwise
+ */
+static uint32_t 
+postgres_plugin_put (void *cls,
+                    const GNUNET_HashCode * key,
+                    uint32_t size,
+                    const char *data,
+                    enum GNUNET_BLOCK_Type type,
+                    struct GNUNET_TIME_Absolute discard_time)
+{
+  struct Plugin *plugin = cls;
+  PGresult *ret;
+  uint32_t btype = htonl (type);
+  uint64_t bexpi = GNUNET_TIME_absolute_hton (discard_time).value__;
+  const char *paramValues[] = {
+    (const char *) &btype,
+    (const char *) &bexpi,
+    (const char *) key,
+    (const char *) data
+  };
+  int paramLengths[] = {
+    sizeof (btype),
+    sizeof (bexpi),
+    sizeof (GNUNET_HashCode),
+    size
+  };
+  const int paramFormats[] = { 1, 1, 1, 1 };
+
+  ret = PQexecPrepared (plugin->dbh,
+                        "put", 4, paramValues, paramLengths, paramFormats, 1);
+  if (GNUNET_OK != check_result (plugin, ret,
+                                 PGRES_COMMAND_OK,
+                                 "PQexecPrepared", "put", __LINE__))
+    return GNUNET_SYSERR;
+  PQclear (ret);
+  return size;
+}
+
+
+/**
+ * Iterate over the results for a particular key
+ * in the datastore.
+ *
+ * @param cls closure (our "struct Plugin")
+ * @param key
+ * @param type entries of which type are relevant?
+ * @param iter maybe NULL (to just count)
+ * @param iter_cls closure for iter
+ * @return the number of results found
+ */
+static unsigned int 
+postgres_plugin_get (void *cls,
+                    const GNUNET_HashCode * key,
+                    enum GNUNET_BLOCK_Type type,
+                    GNUNET_DATACACHE_Iterator iter,
+                    void *iter_cls)
+{
+  struct Plugin *plugin = cls;
+  uint32_t btype = htonl (type);
+  const char *paramValues[] = {
+    (const char *) key,
+    (const char *) &btype,
+  };
+  int paramLengths[] = {
+    sizeof (GNUNET_HashCode),
+    sizeof (btype),
+  };
+  const int paramFormats[] = { 1, 1 };
+  struct GNUNET_TIME_Absolute expiration_time;
+  uint32_t size;
+  unsigned int cnt;
+  unsigned int i;
+  PGresult *res;
+
+  cnt = 0;
+  res = PQexecPrepared (plugin->dbh,
+                       (type == 0) ? "getk" : "getkt",
+                       (type == 0) ? 1 : 2,
+                       paramValues, 
+                       paramLengths,
+                       paramFormats,
+                       1);
+  if (GNUNET_OK != check_result (plugin,
+                                res,
+                                PGRES_TUPLES_OK,
+                                "PQexecPrepared",
+                                (type == 0) ? "getk" : "getkt",
+                                __LINE__))
+    {
+#if DEBUG_POSTGRES
+      GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+                      "datacache-postgres",
+                      "Ending iteration (postgres error)\n");
+#endif
+      return 0;
+    }
+
+  if (0 == (cnt = PQntuples (res)))
+    {
+      /* no result */
+#if DEBUG_POSTGRES
+      GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+                      "datacache-postgres",
+                      "Ending iteration (no more results)\n");
+#endif
+      PQclear (res);
+      return 0; 
+    }
+  if ( (3 != PQnfields (res)) ||
+       (sizeof (uint64_t) != PQfsize (res, 0)) ||
+       (sizeof (uint32_t) != PQfsize (res, 1)))
+    {
+      GNUNET_break (0);
+      PQclear (res);
+      return 0;
+    }
+  for (i=0;i<cnt;i++)
+    {
+      expiration_time.value = GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 
0));
+      type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1));
+      size = PQgetlength (res, i, 2);
+#if DEBUG_POSTGRES
+      GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+                      "datacache-postgres",
+                      "Found result of size %u bytes and type %u in 
database\n",
+                      (unsigned int) size,
+                      (unsigned int) type);
+#endif
+      if (GNUNET_SYSERR ==
+         iter (iter_cls,
+               expiration_time,
+               key,
+               size,
+               PQgetvalue (res, i, 2),
+               (enum GNUNET_BLOCK_Type) type))
+       {
+#if DEBUG_POSTGRES
+         GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+                          "datacache-postgres",
+                          "Ending iteration (client error)\n");
+#endif
+         PQclear (res);
+         return cnt;
+       }      
+    }      
+  PQclear (res);
+  return cnt;
+}
+
+
+/**
+ * Delete the entry with the lowest expiration value
+ * from the datacache right now.
+ * 
+ * @param cls closure (our "struct Plugin")
+ * @return GNUNET_OK on success, GNUNET_SYSERR on error
+ */ 
+static int 
+postgres_plugin_del (void *cls)
+{
+  
+  GNUNET_break (0);
+  return GNUNET_SYSERR;
+}
+
+
+/**
+ * Entry point for the plugin.
+ *
+ * @param cls closure (the "struct GNUNET_DATACACHE_PluginEnvironmnet")
+ * @return the plugin's closure (our "struct Plugin")
+ */
+void *
+libgnunet_plugin_datacache_postgres_init (void *cls)
+{
+  struct GNUNET_DATACACHE_PluginEnvironment *env = cls;
+  struct GNUNET_DATACACHE_PluginFunctions *api;
+  struct Plugin *plugin;
+
+  plugin = GNUNET_malloc (sizeof (struct Plugin));
+  plugin->env = env;
+
+  if (GNUNET_OK !=
+      init_connection (plugin))
+    {
+      GNUNET_free (plugin);
+      return NULL;
+    }
+
+  api = GNUNET_malloc (sizeof (struct GNUNET_DATACACHE_PluginFunctions));
+  api->cls = plugin;
+  api->get = &postgres_plugin_get;
+  api->put = &postgres_plugin_put;
+  api->del = &postgres_plugin_del;
+  GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
+                   "datacache-postgres",
+                  _("Postgres datacache running\n"));
+  return api;
+}
+
+
+/**
+ * Exit point from the plugin.
+ *
+ * @param cls closure (our "struct Plugin")
+ * @return NULL
+ */
+void *
+libgnunet_plugin_datacache_postgres_done (void *cls)
+{
+  struct GNUNET_DATACACHE_PluginFunctions *api = cls;
+  struct Plugin *plugin = api->cls;
+
+  GNUNET_free (plugin);
+  GNUNET_free (api);
+  return NULL;
+}
+
+
+
+/* end of plugin_datacache_postgres.c */
+