src/datacache/plugin_datacache_postgres.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2006, 2009, 2010, 2012 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file datacache/plugin_datacache_postgres.c
  23  * @brief postgres for an implementation of a database backend for the datacache
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_postgres_lib.h"
  29 #include "gnunet_datacache_plugin.h"
  30 #include <postgresql/libpq-fe.h>
  31
  32 #define LOG(kind,...) GNUNET_log_from (kind, "datacache-postgres", __VA_ARGS__)
  33
  34 /**
  35  * Per-entry overhead estimate
  36  */
  37 #define OVERHEAD (sizeof(struct GNUNET_HashCode) + 24)
  38
  39 /**
  40  * Context for all functions in this plugin.
  41  */
  42 struct Plugin
  43 {
  44   /**
  45    * Our execution environment.
  46    */
  47   struct GNUNET_DATACACHE_PluginEnvironment *env;
  48
  49   /**
  50    * Native Postgres database handle.
  51    */
  52   PGconn *dbh;
  53
  54 };
  55
  56
  57 /**
  58  * @brief Get a database handle
  59  *
  60  * @param plugin global context
  61  * @return GNUNET_OK on success, GNUNET_SYSERR on error
  62  */
  63 static int
  64 init_connection (struct Plugin *plugin)
  65 {
  66   PGresult *ret;
  67
  68   plugin->dbh = GNUNET_POSTGRES_connect (plugin->env->cfg,
  69                                          "datacache-postgres");
  70   if (NULL == plugin->dbh)
  71     return GNUNET_SYSERR;
  72   ret =
  73       PQexec (plugin->dbh,
  74               "CREATE TEMPORARY TABLE gn090dc ("
  75               "  type INTEGER NOT NULL DEFAULT 0,"
  76               "  discard_time BIGINT NOT NULL DEFAULT 0,"
  77               "  key BYTEA NOT NULL DEFAULT '',"
  78               "  value BYTEA NOT NULL DEFAULT '',"
  79               "  path BYTEA DEFAULT '')"
  80               "WITH OIDS");
  81   if ( (ret == NULL) ||
  82        ((PQresultStatus (ret) != PGRES_COMMAND_OK) &&
  83         (0 != strcmp ("42P07",    /* duplicate table */
  84                       PQresultErrorField
  85                       (ret,
  86                        PG_DIAG_SQLSTATE)))))
  87   {
  88     (void) GNUNET_POSTGRES_check_result (plugin->dbh, ret,
  89                                          PGRES_COMMAND_OK, "CREATE TABLE",
  90                                          "gn090dc");
  91     PQfinish (plugin->dbh);
  92     plugin->dbh = NULL;
  93     return GNUNET_SYSERR;
  94   }
  95   if (PQresultStatus (ret) == PGRES_COMMAND_OK)
  96   {
  97     if ((GNUNET_OK !=
  98          GNUNET_POSTGRES_exec (plugin->dbh, "CREATE INDEX idx_key ON gn090dc (key)")) ||
  99         (GNUNET_OK !=
 100          GNUNET_POSTGRES_exec (plugin->dbh, "CREATE INDEX idx_dt ON gn090dc (discard_time)")))
 101     {
 102       PQclear (ret);
 103       PQfinish (plugin->dbh);
 104       plugin->dbh = NULL;
 105       return GNUNET_SYSERR;
 106     }
 107   }
 108   PQclear (ret);
 109   ret =
 110       PQexec (plugin->dbh,
 111               "ALTER TABLE gn090dc ALTER value SET STORAGE EXTERNAL");
 112   if (GNUNET_OK !=
 113       GNUNET_POSTGRES_check_result (plugin->dbh, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090dc"))
 114   {
 115     PQfinish (plugin->dbh);
 116     plugin->dbh = NULL;
 117     return GNUNET_SYSERR;
 118   }
 119   PQclear (ret);
 120   ret = PQexec (plugin->dbh, "ALTER TABLE gn090dc ALTER key SET STORAGE PLAIN");
 121   if (GNUNET_OK !=
 122       GNUNET_POSTGRES_check_result (plugin->dbh, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090dc"))
 123   {
 124     PQfinish (plugin->dbh);
 125     plugin->dbh = NULL;
 126     return GNUNET_SYSERR;
 127   }
 128   PQclear (ret);
 129   if ((GNUNET_OK !=
 130        GNUNET_POSTGRES_prepare (plugin->dbh, "getkt",
 131                    "SELECT discard_time,type,value,path FROM gn090dc "
 132                    "WHERE key=$1 AND type=$2 ", 2)) ||
 133       (GNUNET_OK !=
 134        GNUNET_POSTGRES_prepare (plugin->dbh, "getk",
 135                    "SELECT discard_time,type,value,path FROM gn090dc "
 136                    "WHERE key=$1", 1)) ||
 137       (GNUNET_OK !=
 138        GNUNET_POSTGRES_prepare (plugin->dbh, "getm",
 139                    "SELECT length(value),oid,key FROM gn090dc "
 140                    "ORDER BY discard_time ASC LIMIT 1", 0)) ||
 141       (GNUNET_OK !=
 142        GNUNET_POSTGRES_prepare (plugin->dbh, "delrow", "DELETE FROM gn090dc WHERE oid=$1", 1)) ||
 143       (GNUNET_OK !=
 144        GNUNET_POSTGRES_prepare (plugin->dbh, "put",
 145                    "INSERT INTO gn090dc (type, discard_time, key, value, path) "
 146                    "VALUES ($1, $2, $3, $4, $5)", 5)))
 147   {
 148     PQfinish (plugin->dbh);
 149     plugin->dbh = NULL;
 150     return GNUNET_SYSERR;
 151   }
 152   return GNUNET_OK;
 153 }
 154
 155
 156 /**
 157  * Store an item in the datastore.
 158  *
 159  * @param cls closure (our "struct Plugin")
 160  * @param key key to store data under
 161  * @param size number of bytes in data
 162  * @param data data to store
 163  * @param type type of the value
 164  * @param discard_time when to discard the value in any case
 165  * @param path_info_len number of entries in 'path_info'
 166  * @param path_info a path through the network
 167  * @return 0 if duplicate, -1 on error, number of bytes used otherwise
 168  */
 169 static ssize_t
 170 postgres_plugin_put (void *cls, const struct GNUNET_HashCode * key, size_t size,
 171                      const char *data, enum GNUNET_BLOCK_Type type,
 172                      struct GNUNET_TIME_Absolute discard_time,
 173                      unsigned int path_info_len,
 174                      const struct GNUNET_PeerIdentity *path_info)
 175 {
 176   struct Plugin *plugin = cls;
 177   PGresult *ret;
 178   uint32_t btype = htonl (type);
 179   uint64_t bexpi = GNUNET_TIME_absolute_hton (discard_time).abs_value__;
 180
 181   const char *paramValues[] = {
 182     (const char *) &btype,
 183     (const char *) &bexpi,
 184     (const char *) key,
 185     (const char *) data,
 186     (const char *) path_info
 187   };
 188   int paramLengths[] = {
 189     sizeof (btype),
 190     sizeof (bexpi),
 191     sizeof (struct GNUNET_HashCode),
 192     size,
 193     path_info_len * sizeof (struct GNUNET_PeerIdentity)
 194   };
 195   const int paramFormats[] = { 1, 1, 1, 1, 1 };
 196
 197   ret =
 198       PQexecPrepared (plugin->dbh, "put", 5, paramValues, paramLengths,
 199                       paramFormats, 1);
 200   if (GNUNET_OK !=
 201       GNUNET_POSTGRES_check_result (plugin->dbh, ret,
 202                                     PGRES_COMMAND_OK, "PQexecPrepared", "put"))
 203     return -1;
 204   PQclear (ret);
 205   return size + OVERHEAD;
 206 }
 207
 208
 209 /**
 210  * Iterate over the results for a particular key
 211  * in the datastore.
 212  *
 213  * @param cls closure (our "struct Plugin")
 214  * @param key
 215  * @param type entries of which type are relevant?
 216  * @param iter maybe NULL (to just count)
 217  * @param iter_cls closure for iter
 218  * @return the number of results found
 219  */
 220 static unsigned int
 221 postgres_plugin_get (void *cls, const struct GNUNET_HashCode * key,
 222                      enum GNUNET_BLOCK_Type type,
 223                      GNUNET_DATACACHE_Iterator iter, void *iter_cls)
 224 {
 225   struct Plugin *plugin = cls;
 226   uint32_t btype = htonl (type);
 227
 228   const char *paramValues[] = {
 229     (const char *) key,
 230     (const char *) &btype,
 231   };
 232   int paramLengths[] = {
 233     sizeof (struct GNUNET_HashCode),
 234     sizeof (btype),
 235   };
 236   const int paramFormats[] = { 1, 1 };
 237   struct GNUNET_TIME_Absolute expiration_time;
 238   uint32_t size;
 239   unsigned int cnt;
 240   unsigned int i;
 241   unsigned int path_len;
 242   const struct GNUNET_PeerIdentity *path;
 243   PGresult *res;
 244
 245   res =
 246       PQexecPrepared (plugin->dbh, (type == 0) ? "getk" : "getkt",
 247                       (type == 0) ? 1 : 2, paramValues, paramLengths,
 248                       paramFormats, 1);
 249   if (GNUNET_OK !=
 250       GNUNET_POSTGRES_check_result (plugin->dbh, res, PGRES_TUPLES_OK, "PQexecPrepared",
 251                                     (type == 0) ? "getk" : "getkt"))
 252   {
 253     LOG (GNUNET_ERROR_TYPE_DEBUG,
 254          "Ending iteration (postgres error)\n");
 255     return 0;
 256   }
 257
 258   if (0 == (cnt = PQntuples (res)))
 259   {
 260     /* no result */
 261     LOG (GNUNET_ERROR_TYPE_DEBUG,
 262          "Ending iteration (no more results)\n");
 263     PQclear (res);
 264     return 0;
 265   }
 266   if (iter == NULL)
 267   {
 268     PQclear (res);
 269     return cnt;
 270   }
 271   if ((4 != PQnfields (res)) || (sizeof (uint64_t) != PQfsize (res, 0)) ||
 272       (sizeof (uint32_t) != PQfsize (res, 1)))
 273   {
 274     GNUNET_break (0);
 275     PQclear (res);
 276     return 0;
 277   }
 278   for (i = 0; i < cnt; i++)
 279   {
 280     expiration_time.abs_value =
 281         GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 0));
 282     type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1));
 283     size = PQgetlength (res, i, 2);
 284     path_len = PQgetlength (res, i, 3);
 285     if (0 != (path_len % sizeof (struct GNUNET_PeerIdentity)))
 286     {
 287       GNUNET_break (0);
 288       path_len = 0;
 289     }
 290     path_len %= sizeof (struct GNUNET_PeerIdentity);
 291     path = (const struct GNUNET_PeerIdentity *) PQgetvalue (res, i, 3);
 292     LOG (GNUNET_ERROR_TYPE_DEBUG,
 293          "Found result of size %u bytes and type %u in database\n",
 294          (unsigned int) size, (unsigned int) type);
 295     if (GNUNET_SYSERR ==
 296         iter (iter_cls, key, size, PQgetvalue (res, i, 2),
 297               (enum GNUNET_BLOCK_Type) type,
 298               expiration_time,
 299               path_len,
 300               path))
 301     {
 302       LOG (GNUNET_ERROR_TYPE_DEBUG,
 303            "Ending iteration (client error)\n");
 304       PQclear (res);
 305       return cnt;
 306     }
 307   }
 308   PQclear (res);
 309   return cnt;
 310 }
 311
 312
 313 /**
 314  * Delete the entry with the lowest expiration value
 315  * from the datacache right now.
 316  *
 317  * @param cls closure (our "struct Plugin")
 318  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 319  */
 320 static int
 321 postgres_plugin_del (void *cls)
 322 {
 323   struct Plugin *plugin = cls;
 324   uint32_t size;
 325   uint32_t oid;
 326   struct GNUNET_HashCode key;
 327   PGresult *res;
 328
 329   res = PQexecPrepared (plugin->dbh, "getm", 0, NULL, NULL, NULL, 1);
 330   if (GNUNET_OK !=
 331       GNUNET_POSTGRES_check_result (plugin->dbh, res, PGRES_TUPLES_OK, "PQexecPrepared", "getm"))
 332   {
 333     LOG (GNUNET_ERROR_TYPE_DEBUG,
 334          "Ending iteration (postgres error)\n");
 335     return 0;
 336   }
 337   if (0 == PQntuples (res))
 338   {
 339     /* no result */
 340     LOG (GNUNET_ERROR_TYPE_DEBUG,
 341          "Ending iteration (no more results)\n");
 342     PQclear (res);
 343     return GNUNET_SYSERR;
 344   }
 345   if ((3 != PQnfields (res)) || (sizeof (size) != PQfsize (res, 0)) ||
 346       (sizeof (oid) != PQfsize (res, 1)) ||
 347       (sizeof (struct GNUNET_HashCode) != PQgetlength (res, 0, 2)))
 348   {
 349     GNUNET_break (0);
 350     PQclear (res);
 351     return 0;
 352   }
 353   size = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
 354   oid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
 355   memcpy (&key, PQgetvalue (res, 0, 2), sizeof (struct GNUNET_HashCode));
 356   PQclear (res);
 357   if (GNUNET_OK != GNUNET_POSTGRES_delete_by_rowid (plugin->dbh, "delrow", oid))
 358     return GNUNET_SYSERR;
 359   plugin->env->delete_notify (plugin->env->cls, &key, size + OVERHEAD);
 360   return GNUNET_OK;
 361 }
 362
 363
 364 /**
 365  * Entry point for the plugin.
 366  *
 367  * @param cls closure (the "struct GNUNET_DATACACHE_PluginEnvironmnet")
 368  * @return the plugin's closure (our "struct Plugin")
 369  */
 370 void *
 371 libgnunet_plugin_datacache_postgres_init (void *cls)
 372 {
 373   struct GNUNET_DATACACHE_PluginEnvironment *env = cls;
 374   struct GNUNET_DATACACHE_PluginFunctions *api;
 375   struct Plugin *plugin;
 376
 377   plugin = GNUNET_malloc (sizeof (struct Plugin));
 378   plugin->env = env;
 379
 380   if (GNUNET_OK != init_connection (plugin))
 381   {
 382     GNUNET_free (plugin);
 383     return NULL;
 384   }
 385
 386   api = GNUNET_malloc (sizeof (struct GNUNET_DATACACHE_PluginFunctions));
 387   api->cls = plugin;
 388   api->get = &postgres_plugin_get;
 389   api->put = &postgres_plugin_put;
 390   api->del = &postgres_plugin_del;
 391   LOG (GNUNET_ERROR_TYPE_INFO,
 392        _("Postgres datacache running\n"));
 393   return api;
 394 }
 395
 396
 397 /**
 398  * Exit point from the plugin.
 399  *
 400  * @param cls closure (our "struct Plugin")
 401  * @return NULL
 402  */
 403 void *
 404 libgnunet_plugin_datacache_postgres_done (void *cls)
 405 {
 406   struct GNUNET_DATACACHE_PluginFunctions *api = cls;
 407   struct Plugin *plugin = api->cls;
 408
 409   PQfinish (plugin->dbh);
 410   GNUNET_free (plugin);
 411   GNUNET_free (api);
 412   return NULL;
 413 }
 414
 415
 416
 417 /* end of plugin_datacache_postgres.c */