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   cnt = 0;
 351   res =
 352       PQexecPrepared (plugin->dbh, (type == 0) ? "getk" : "getkt",
 353                       (type == 0) ? 1 : 2, paramValues, paramLengths,
 354                       paramFormats, 1);
 355   if (GNUNET_OK !=
 356       check_result (plugin, res, PGRES_TUPLES_OK, "PQexecPrepared",
 357                     (type == 0) ? "getk" : "getkt", __LINE__))
 358   {
 359 #if DEBUG_POSTGRES
 360     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 361                      "Ending iteration (postgres error)\n");
 362 #endif
 363     return 0;
 364   }
 365
 366   if (0 == (cnt = PQntuples (res)))
 367   {
 368     /* no result */
 369 #if DEBUG_POSTGRES
 370     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 371                      "Ending iteration (no more results)\n");
 372 #endif
 373     PQclear (res);
 374     return 0;
 375   }
 376   if (iter == NULL)
 377   {
 378     PQclear (res);
 379     return cnt;
 380   }
 381   if ((3 != PQnfields (res)) || (sizeof (uint64_t) != PQfsize (res, 0)) ||
 382       (sizeof (uint32_t) != PQfsize (res, 1)))
 383   {
 384     GNUNET_break (0);
 385     PQclear (res);
 386     return 0;
 387   }
 388   for (i = 0; i < cnt; i++)
 389   {
 390     expiration_time.abs_value =
 391         GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 0));
 392     type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1));
 393     size = PQgetlength (res, i, 2);
 394 #if DEBUG_POSTGRES
 395     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 396                      "Found result of size %u bytes and type %u in database\n",
 397                      (unsigned int) size, (unsigned int) type);
 398 #endif
 399     if (GNUNET_SYSERR ==
 400         iter (iter_cls, expiration_time, key, size, PQgetvalue (res, i, 2),
 401               (enum GNUNET_BLOCK_Type) type))
 402     {
 403 #if DEBUG_POSTGRES
 404       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 405                        "Ending iteration (client error)\n");
 406 #endif
 407       PQclear (res);
 408       return cnt;
 409     }
 410   }
 411   PQclear (res);
 412   return cnt;
 413 }
 414
 415
 416 /**
 417  * Delete the entry with the lowest expiration value
 418  * from the datacache right now.
 419  *
 420  * @param cls closure (our "struct Plugin")
 421  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 422  */
 423 static int
 424 postgres_plugin_del (void *cls)
 425 {
 426   struct Plugin *plugin = cls;
 427   uint32_t size;
 428   uint32_t oid;
 429   GNUNET_HashCode key;
 430   PGresult *res;
 431
 432   res = PQexecPrepared (plugin->dbh, "getm", 0, NULL, NULL, NULL, 1);
 433   if (GNUNET_OK !=
 434       check_result (plugin, res, PGRES_TUPLES_OK, "PQexecPrepared", "getm",
 435                     __LINE__))
 436   {
 437 #if DEBUG_POSTGRES
 438     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 439                      "Ending iteration (postgres error)\n");
 440 #endif
 441     return 0;
 442   }
 443   if (0 == PQntuples (res))
 444   {
 445     /* no result */
 446 #if DEBUG_POSTGRES
 447     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datacache-postgres",
 448                      "Ending iteration (no more results)\n");
 449 #endif
 450     PQclear (res);
 451     return GNUNET_SYSERR;
 452   }
 453   if ((3 != PQnfields (res)) || (sizeof (size) != PQfsize (res, 0)) ||
 454       (sizeof (oid) != PQfsize (res, 1)) ||
 455       (sizeof (GNUNET_HashCode) != PQgetlength (res, 0, 2)))
 456   {
 457     GNUNET_break (0);
 458     PQclear (res);
 459     return 0;
 460   }
 461   size = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
 462   oid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
 463   memcpy (&key, PQgetvalue (res, 0, 2), sizeof (GNUNET_HashCode));
 464   PQclear (res);
 465   if (GNUNET_OK != delete_by_rowid (plugin, oid))
 466     return GNUNET_SYSERR;
 467   plugin->env->delete_notify (plugin->env->cls, &key, size + OVERHEAD);
 468   return GNUNET_OK;
 469 }
 470
 471
 472 /**
 473  * Entry point for the plugin.
 474  *
 475  * @param cls closure (the "struct GNUNET_DATACACHE_PluginEnvironmnet")
 476  * @return the plugin's closure (our "struct Plugin")
 477  */
 478 void *
 479 libgnunet_plugin_datacache_postgres_init (void *cls)
 480 {
 481   struct GNUNET_DATACACHE_PluginEnvironment *env = cls;
 482   struct GNUNET_DATACACHE_PluginFunctions *api;
 483   struct Plugin *plugin;
 484
 485   plugin = GNUNET_malloc (sizeof (struct Plugin));
 486   plugin->env = env;
 487
 488   if (GNUNET_OK != init_connection (plugin))
 489   {
 490     GNUNET_free (plugin);
 491     return NULL;
 492   }
 493
 494   api = GNUNET_malloc (sizeof (struct GNUNET_DATACACHE_PluginFunctions));
 495   api->cls = plugin;
 496   api->get = &postgres_plugin_get;
 497   api->put = &postgres_plugin_put;
 498   api->del = &postgres_plugin_del;
 499   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO, "datacache-postgres",
 500                    _("Postgres datacache running\n"));
 501   return api;
 502 }
 503
 504
 505 /**
 506  * Exit point from the plugin.
 507  *
 508  * @param cls closure (our "struct Plugin")
 509  * @return NULL
 510  */
 511 void *
 512 libgnunet_plugin_datacache_postgres_done (void *cls)
 513 {
 514   struct GNUNET_DATACACHE_PluginFunctions *api = cls;
 515   struct Plugin *plugin = api->cls;
 516
 517   PQfinish (plugin->dbh);
 518   GNUNET_free (plugin);
 519   GNUNET_free (api);
 520   return NULL;
 521 }
 522
 523
 524
 525 /* end of plugin_datacache_postgres.c */