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