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