src/datacache/plugin_datacache_postgres.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2006, 2009, 2010 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_datacache_plugin.h"
  29 #include <postgresql/libpq-fe.h>
  30
  31 #define DEBUG_POSTGRES GNUNET_EXTRA_LOGGING
  32
  33 /**
  34  * Per-entry overhead estimate
  35  */
  36 #define OVERHEAD (sizeof(GNUNET_HashCode) + 24)
  37
  38 /**
  39  * Context for all functions in this plugin.
  40  */
  41 struct Plugin
  42 {
  43   /**
  44    * Our execution environment.
  45    */
  46   struct GNUNET_DATACACHE_PluginEnvironment *env;
  47
  48   /**
  49    * Native Postgres database handle.
  50    */
  51   PGconn *dbh;
  52
  53 };
  54
  55
  56 /**
  57  * Check if the result obtained from Postgres has
  58  * the desired status code.  If not, log an error, clear the
  59  * result and return GNUNET_SYSERR.
  60  *
  61  * @return GNUNET_OK if the result is acceptable
  62  */
  63 static int
  64 check_result (struct Plugin *plugin, PGresult * ret, int expected_status,
  65               const char *command, const char *args, int line)
  66 {
  67   if (ret == NULL)
  68   {
  69     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
  70                      "datastore-postgres",
  71                      "Postgres failed to allocate result for `%s:%s' at %d\n",
  72                      command, args, line);
  73     return GNUNET_SYSERR;
  74   }
  75   if (PQresultStatus (ret) != expected_status)
  76   {
  77     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
  78                      "datastore-postgres",
  79                      _("`%s:%s' failed at %s:%d with error: %s"), command, args,
  80                      __FILE__, line, PQerrorMessage (plugin->dbh));
  81     PQclear (ret);
  82     return GNUNET_SYSERR;
  83   }
  84   return GNUNET_OK;
  85 }
  86
  87
  88 /**
  89  * Run simple SQL statement (without results).
  90  */
  91 static int
  92 pq_exec (struct Plugin *plugin, const char *sql, int line)
  93 {
  94   PGresult *ret;
  95
  96   ret = PQexec (plugin->dbh, sql);
  97   if (GNUNET_OK !=
  98       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexec", sql, line))
  99     return GNUNET_SYSERR;
 100   PQclear (ret);
 101   return GNUNET_OK;
 102 }
 103
 104
 105 /**
 106  * Prepare SQL statement.
 107  */
 108 static int
 109 pq_prepare (struct Plugin *plugin, const char *name, const char *sql,
 110             int nparms, int line)
 111 {
 112   PGresult *ret;
 113
 114   ret = PQprepare (plugin->dbh, name, sql, nparms, NULL);
 115   if (GNUNET_OK !=
 116       check_result (plugin, ret, PGRES_COMMAND_OK, "PQprepare", sql, line))
 117     return GNUNET_SYSERR;
 118   PQclear (ret);
 119   return GNUNET_OK;
 120 }
 121
 122
 123 /**
 124  * @brief Get a database handle
 125  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 126  */
 127 static int
 128 init_connection (struct Plugin *plugin)
 129 {
 130   char *conninfo;
 131   PGresult *ret;
 132
 133   /* Open database and precompile statements */
 134   if (GNUNET_OK !=
 135       GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 136                                              "datacache-postgres", "CONFIG",
 137                                              &conninfo))
 138     conninfo = NULL;
 139   plugin->dbh = PQconnectdb (conninfo == NULL ? "" : conninfo);
 140   GNUNET_free_non_null (conninfo);
 141   if (NULL == plugin->dbh)
 142   {
 143     /* FIXME: warn about out-of-memory? */
 144     return GNUNET_SYSERR;
 145   }
 146   if (PQstatus (plugin->dbh) != CONNECTION_OK)
 147   {
 148     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR, "datacache-postgres",
 149                      _("Unable to initialize Postgres: %s"),
 150                      PQerrorMessage (plugin->dbh));
 151     PQfinish (plugin->dbh);
 152     plugin->dbh = NULL;
 153     return GNUNET_SYSERR;
 154   }
 155   ret =
 156       PQexec (plugin->dbh,
 157               "CREATE TEMPORARY TABLE gn090dc ("
 158               "  type INTEGER NOT NULL DEFAULT 0,"
 159               "  discard_time BIGINT NOT NULL DEFAULT 0,"
 160               "  key BYTEA NOT NULL DEFAULT '',"
 161               "  value BYTEA NOT NULL DEFAULT '')" "WITH OIDS");
 162   if ((ret == NULL) || ((PQresultStatus (ret) != PGRES_COMMAND_OK) && (0 != strcmp ("42P07",    /* duplicate table */
 163                                                                                     PQresultErrorField
 164                                                                                     (ret,
 165                                                                                      PG_DIAG_SQLSTATE)))))
 166   {
 167     (void) check_result (plugin, ret, PGRES_COMMAND_OK, "CREATE TABLE",
 168                          "gn090dc", __LINE__);
 169     PQfinish (plugin->dbh);
 170     plugin->dbh = NULL;
 171     return GNUNET_SYSERR;
 172   }
 173   if (PQresultStatus (ret) == PGRES_COMMAND_OK)
 174   {
 175     if ((GNUNET_OK !=
 176          pq_exec (plugin, "CREATE INDEX idx_key ON gn090dc (key)", __LINE__)) ||
 177         (GNUNET_OK !=
 178          pq_exec (plugin, "CREATE INDEX idx_dt ON gn090dc (discard_time)",
 179                   __LINE__)))
 180     {
 181       PQclear (ret);
 182       PQfinish (plugin->dbh);
 183       plugin->dbh = NULL;
 184       return GNUNET_SYSERR;
 185     }
 186   }
 187   PQclear (ret);
 188 #if 1
 189   ret =
 190       PQexec (plugin->dbh,
 191               "ALTER TABLE gn090dc ALTER value SET STORAGE EXTERNAL");
 192   if (GNUNET_OK !=
 193       check_result (plugin, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090dc",
 194                     __LINE__))
 195   {
 196     PQfinish (plugin->dbh);
 197     plugin->dbh = NULL;
 198     return GNUNET_SYSERR;
 199   }
 200   PQclear (ret);
 201   ret = PQexec (plugin->dbh, "ALTER TABLE gn090dc ALTER key SET STORAGE PLAIN");
 202   if (GNUNET_OK !=
 203       check_result (plugin, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090dc",
 204                     __LINE__))
 205   {
 206     PQfinish (plugin->dbh);
 207     plugin->dbh = NULL;
 208     return GNUNET_SYSERR;
 209   }
 210   PQclear (ret);
 211 #endif
 212   if ((GNUNET_OK !=
 213        pq_prepare (plugin, "getkt",
 214                    "SELECT discard_time,type,value FROM gn090dc "
 215                    "WHERE key=$1 AND type=$2 ", 2, __LINE__)) ||
 216       (GNUNET_OK !=
 217        pq_prepare (plugin, "getk",
 218                    "SELECT discard_time,type,value FROM gn090dc "
 219                    "WHERE key=$1", 1, __LINE__)) ||
 220       (GNUNET_OK !=
 221        pq_prepare (plugin, "getm",
 222                    "SELECT length(value),oid,key FROM gn090dc "
 223                    "ORDER BY discard_time ASC LIMIT 1", 0, __LINE__)) ||
 224       (GNUNET_OK !=
 225        pq_prepare (plugin, "delrow", "DELETE FROM gn090dc WHERE oid=$1", 1,
 226                    __LINE__)) ||
 227       (GNUNET_OK !=
 228        pq_prepare (plugin, "put",
 229                    "INSERT INTO gn090dc (type, discard_time, key, value) "
 230                    "VALUES ($1, $2, $3, $4)", 4, __LINE__)))
 231   {
 232     PQfinish (plugin->dbh);
 233     plugin->dbh = NULL;
 234     return GNUNET_SYSERR;
 235   }
 236   return GNUNET_OK;
 237 }
 238
 239
 240 /**
 241  * Delete the row identified by the given rowid (qid
 242  * in postgres).
 243  *
 244  * @return GNUNET_OK on success
 245  */
 246 static int
 247 delete_by_rowid (struct Plugin *plugin, uint32_t rowid)
 248 {
 249   uint32_t brow = htonl (rowid);
 250   const char *paramValues[] = { (const char *) &brow };
 251   int paramLengths[] = { sizeof (brow) };
 252   const int paramFormats[] = { 1 };
 253   PGresult *ret;
 254
 255   ret =
 256       PQexecPrepared (plugin->dbh, "delrow", 1, paramValues, paramLengths,
 257                       paramFormats, 1);
 258   if (GNUNET_OK !=
 259       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexecPrepared", "delrow",
 260                     __LINE__))
 261   {
 262     return GNUNET_SYSERR;
 263   }
 264   PQclear (ret);
 265   return GNUNET_OK;
 266 }
 267
 268
 269 /**
 270  * Store an item in the datastore.
 271  *
 272  * @param cls closure (our "struct Plugin")
 273  * @param key key to store data under
 274  * @param size number of bytes in data
 275  * @param data data to store
 276  * @param type type of the value
 277  * @param discard_time when to discard the value in any case
 278  * @return 0 on error, number of bytes used otherwise
 279  */
 280 static size_t
 281 postgres_plugin_put (void *cls, const GNUNET_HashCode * key, size_t size,
 282                      const char *data, enum GNUNET_BLOCK_Type type,
 283                      struct GNUNET_TIME_Absolute discard_time)
 284 {
 285   struct Plugin *plugin = cls;
 286   PGresult *ret;
 287   uint32_t btype = htonl (type);
 288   uint64_t bexpi = GNUNET_TIME_absolute_hton (discard_time).abs_value__;
 289
 290   const char *paramValues[] = {
 291     (const char *) &btype,
 292     (const char *) &bexpi,
 293     (const char *) key,
 294     (const char *) data
 295   };
 296   int paramLengths[] = {
 297     sizeof (btype),
 298     sizeof (bexpi),
 299     sizeof (GNUNET_HashCode),
 300     size
 301   };
 302   const int paramFormats[] = { 1, 1, 1, 1 };
 303
 304   ret =
 305       PQexecPrepared (plugin->dbh, "put", 4, paramValues, paramLengths,
 306                       paramFormats, 1);
 307   if (GNUNET_OK !=
 308       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexecPrepared", "put",
 309                     __LINE__))
 310     return GNUNET_SYSERR;
 311   PQclear (ret);
 312   return size + OVERHEAD;
 313 }
 314
 315
 316 /**
 317  * Iterate over the results for a particular key
 318  * in the datastore.
 319  *
 320  * @param cls closure (our "struct Plugin")
 321  * @param key
 322  * @param type entries of which type are relevant?
 323  * @param iter maybe NULL (to just count)
 324  * @param iter_cls closure for iter
 325  * @return the number of results found
 326  */
 327 static unsigned int
 328 postgres_plugin_get (void *cls, const GNUNET_HashCode * key,
 329                      enum GNUNET_BLOCK_Type type,
 330                      GNUNET_DATACACHE_Iterator iter, void *iter_cls)
 331 {
 332   struct Plugin *plugin = cls;
 333   uint32_t btype = htonl (type);
 334
 335   const char *paramValues[] = {
 336     (const char *) key,
 337     (const char *) &btype,
 338   };
 339   int paramLengths[] = {
 340     sizeof (GNUNET_HashCode),
 341     sizeof (btype),
 342   };
 343   const int paramFormats[] = { 1, 1 };
 344   struct GNUNET_TIME_Absolute expiration_time;
 345   uint32_t size;
 346   unsigned int cnt;
 347   unsigned int i;
 348   PGresult *res;
 349
 350   res =
 351       PQexecPrepared (plugin->dbh, (type == 0) ? "getk" : "getkt",
 352                       (type == 0) ? 1 : 2, paramValues, paramLengths,
 353                       paramFormats, 1);
 354   if (GNUNET_OK !=
 355       check_result (plugin, res, PGRES_TUPLES_OK, "PQexecPrepared",
 356                     (type == 0) ? "getk" : "getkt", __LINE__))
 357   {
 358 #if DEBUG_POSTGRES
 359     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 360                      "Ending iteration (postgres error)\n");
 361 #endif
 362     return 0;
 363   }
 364
 365   if (0 == (cnt = PQntuples (res)))
 366   {
 367     /* no result */
 368 #if DEBUG_POSTGRES
 369     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 370                      "Ending iteration (no more results)\n");
 371 #endif
 372     PQclear (res);
 373     return 0;
 374   }
 375   if (iter == NULL)
 376   {
 377     PQclear (res);
 378     return cnt;
 379   }
 380   if ((3 != PQnfields (res)) || (sizeof (uint64_t) != PQfsize (res, 0)) ||
 381       (sizeof (uint32_t) != PQfsize (res, 1)))
 382   {
 383     GNUNET_break (0);
 384     PQclear (res);
 385     return 0;
 386   }
 387   for (i = 0; i < cnt; i++)
 388   {
 389     expiration_time.abs_value =
 390         GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 0));
 391     type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1));
 392     size = PQgetlength (res, i, 2);
 393 #if DEBUG_POSTGRES
 394     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 395                      "Found result of size %u bytes and type %u in database\n",
 396                      (unsigned int) size, (unsigned int) type);
 397 #endif
 398     if (GNUNET_SYSERR ==
 399         iter (iter_cls, expiration_time, key, size, PQgetvalue (res, i, 2),
 400               (enum GNUNET_BLOCK_Type) type))
 401     {
 402 #if DEBUG_POSTGRES
 403       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 404                        "Ending iteration (client error)\n");
 405 #endif
 406       PQclear (res);
 407       return cnt;
 408     }
 409   }
 410   PQclear (res);
 411   return cnt;
 412 }
 413
 414
 415 /**
 416  * Delete the entry with the lowest expiration value
 417  * from the datacache right now.
 418  *
 419  * @param cls closure (our "struct Plugin")
 420  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 421  */
 422 static int
 423 postgres_plugin_del (void *cls)
 424 {
 425   struct Plugin *plugin = cls;
 426   uint32_t size;
 427   uint32_t oid;
 428   GNUNET_HashCode key;
 429   PGresult *res;
 430
 431   res = PQexecPrepared (plugin->dbh, "getm", 0, NULL, NULL, NULL, 1);
 432   if (GNUNET_OK !=
 433       check_result (plugin, res, PGRES_TUPLES_OK, "PQexecPrepared", "getm",
 434                     __LINE__))
 435   {
 436 #if DEBUG_POSTGRES
 437     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 438                      "Ending iteration (postgres error)\n");
 439 #endif
 440     return 0;
 441   }
 442   if (0 == PQntuples (res))
 443   {
 444     /* no result */
 445 #if DEBUG_POSTGRES
 446     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 447                      "Ending iteration (no more results)\n");
 448 #endif
 449     PQclear (res);
 450     return GNUNET_SYSERR;
 451   }
 452   if ((3 != PQnfields (res)) || (sizeof (size) != PQfsize (res, 0)) ||
 453       (sizeof (oid) != PQfsize (res, 1)) ||
 454       (sizeof (GNUNET_HashCode) != PQgetlength (res, 0, 2)))
 455   {
 456     GNUNET_break (0);
 457     PQclear (res);
 458     return 0;
 459   }
 460   size = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
 461   oid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
 462   memcpy (&key, PQgetvalue (res, 0, 2), sizeof (GNUNET_HashCode));
 463   PQclear (res);
 464   if (GNUNET_OK != delete_by_rowid (plugin, oid))
 465     return GNUNET_SYSERR;
 466   plugin->env->delete_notify (plugin->env->cls, &key, size + OVERHEAD);
 467   return GNUNET_OK;
 468 }
 469
 470
 471 /**
 472  * Entry point for the plugin.
 473  *
 474  * @param cls closure (the "struct GNUNET_DATACACHE_PluginEnvironmnet")
 475  * @return the plugin's closure (our "struct Plugin")
 476  */
 477 void *
 478 libgnunet_plugin_datacache_postgres_init (void *cls)
 479 {
 480   struct GNUNET_DATACACHE_PluginEnvironment *env = cls;
 481   struct GNUNET_DATACACHE_PluginFunctions *api;
 482   struct Plugin *plugin;
 483
 484   plugin = GNUNET_malloc (sizeof (struct Plugin));
 485   plugin->env = env;
 486
 487   if (GNUNET_OK != init_connection (plugin))
 488   {
 489     GNUNET_free (plugin);
 490     return NULL;
 491   }
 492
 493   api = GNUNET_malloc (sizeof (struct GNUNET_DATACACHE_PluginFunctions));
 494   api->cls = plugin;
 495   api->get = &postgres_plugin_get;
 496   api->put = &postgres_plugin_put;
 497   api->del = &postgres_plugin_del;
 498   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO, "datacache-postgres",
 499                    _("Postgres datacache running\n"));
 500   return api;
 501 }
 502
 503
 504 /**
 505  * Exit point from the plugin.
 506  *
 507  * @param cls closure (our "struct Plugin")
 508  * @return NULL
 509  */
 510 void *
 511 libgnunet_plugin_datacache_postgres_done (void *cls)
 512 {
 513   struct GNUNET_DATACACHE_PluginFunctions *api = cls;
 514   struct Plugin *plugin = api->cls;
 515
 516   PQfinish (plugin->dbh);
 517   GNUNET_free (plugin);
 518   GNUNET_free (api);
 519   return NULL;
 520 }
 521
 522
 523
 524 /* end of plugin_datacache_postgres.c */