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