src/datastore/plugin_datastore_postgres.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2009, 2010, 2011 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 datastore/plugin_datastore_postgres.c
  23  * @brief postgres-based datastore backend
  24  * @author Christian Grothoff
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_datastore_plugin.h"
  29 #include <postgresql/libpq-fe.h>
  30
  31 #define DEBUG_POSTGRES GNUNET_EXTRA_LOGGING
  32
  33 /**
  34  * After how many ms "busy" should a DB operation fail for good?
  35  * A low value makes sure that we are more responsive to requests
  36  * (especially PUTs).  A high value guarantees a higher success
  37  * rate (SELECTs in iterate can take several seconds despite LIMIT=1).
  38  *
  39  * The default value of 1s should ensure that users do not experience
  40  * huge latencies while at the same time allowing operations to succeed
  41  * with reasonable probability.
  42  */
  43 #define BUSY_TIMEOUT GNUNET_TIME_UNIT_SECONDS
  44
  45
  46 /**
  47  * Context for all functions in this plugin.
  48  */
  49 struct Plugin
  50 {
  51   /**
  52    * Our execution environment.
  53    */
  54   struct GNUNET_DATASTORE_PluginEnvironment *env;
  55
  56   /**
  57    * Native Postgres database handle.
  58    */
  59   PGconn *dbh;
  60
  61 };
  62
  63
  64 /**
  65  * Check if the result obtained from Postgres has
  66  * the desired status code.  If not, log an error, clear the
  67  * result and return GNUNET_SYSERR.
  68  *
  69  * @param plugin global context
  70  * @param ret result to check
  71  * @param expected_status expected return value
  72  * @param command name of SQL command that was run
  73  * @param args arguments to SQL command
  74  * @param line line number for error reporting
  75  * @return GNUNET_OK if the result is acceptable
  76  */
  77 static int
  78 check_result (struct Plugin *plugin, PGresult * ret, int expected_status,
  79               const char *command, const char *args, int line)
  80 {
  81   if (ret == NULL)
  82   {
  83     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
  84                      "datastore-postgres",
  85                      "Postgres failed to allocate result for `%s:%s' at %d\n",
  86                      command, args, line);
  87     return GNUNET_SYSERR;
  88   }
  89   if (PQresultStatus (ret) != expected_status)
  90   {
  91     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
  92                      "datastore-postgres",
  93                      _("`%s:%s' failed at %s:%d with error: %s"), command, args,
  94                      __FILE__, line, PQerrorMessage (plugin->dbh));
  95     PQclear (ret);
  96     return GNUNET_SYSERR;
  97   }
  98   return GNUNET_OK;
  99 }
 100
 101 /**
 102  * Run simple SQL statement (without results).
 103  *
 104  * @param plugin global context
 105  * @param sql statement to run
 106  * @param line code line for error reporting
 107  */
 108 static int
 109 pq_exec (struct Plugin *plugin, const char *sql, int line)
 110 {
 111   PGresult *ret;
 112
 113   ret = PQexec (plugin->dbh, sql);
 114   if (GNUNET_OK !=
 115       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexec", sql, line))
 116     return GNUNET_SYSERR;
 117   PQclear (ret);
 118   return GNUNET_OK;
 119 }
 120
 121 /**
 122  * Prepare SQL statement.
 123  *
 124  * @param plugin global context
 125  * @param name name for the prepared SQL statement
 126  * @param sql SQL code to prepare
 127  * @param nparams number of parameters in sql
 128  * @param line code line for error reporting
 129  * @return GNUNET_OK on success
 130  */
 131 static int
 132 pq_prepare (struct Plugin *plugin, const char *name, const char *sql,
 133             int nparams, int line)
 134 {
 135   PGresult *ret;
 136
 137   ret = PQprepare (plugin->dbh, name, sql, nparams, NULL);
 138   if (GNUNET_OK !=
 139       check_result (plugin, ret, PGRES_COMMAND_OK, "PQprepare", sql, line))
 140     return GNUNET_SYSERR;
 141   PQclear (ret);
 142   return GNUNET_OK;
 143 }
 144
 145 /**
 146  * @brief Get a database handle
 147  *
 148  * @param plugin global context
 149  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 150  */
 151 static int
 152 init_connection (struct Plugin *plugin)
 153 {
 154   char *conninfo;
 155   PGresult *ret;
 156
 157   /* Open database and precompile statements */
 158   conninfo = NULL;
 159   (void) GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 160                                                 "datastore-postgres", "CONFIG",
 161                                                 &conninfo);
 162   plugin->dbh = PQconnectdb (conninfo == NULL ? "" : conninfo);
 163   if (NULL == plugin->dbh)
 164   {
 165     /* FIXME: warn about out-of-memory? */
 166     GNUNET_free_non_null (conninfo);
 167     return GNUNET_SYSERR;
 168   }
 169   if (PQstatus (plugin->dbh) != CONNECTION_OK)
 170   {
 171     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR, "datastore-postgres",
 172                      _
 173                      ("Unable to initialize Postgres with configuration `%s': %s"),
 174                      conninfo, PQerrorMessage (plugin->dbh));
 175     PQfinish (plugin->dbh);
 176     plugin->dbh = NULL;
 177     GNUNET_free_non_null (conninfo);
 178     return GNUNET_SYSERR;
 179   }
 180   GNUNET_free_non_null (conninfo);
 181   ret =
 182       PQexec (plugin->dbh,
 183               "CREATE TABLE gn090 (" "  repl INTEGER NOT NULL DEFAULT 0,"
 184               "  type INTEGER NOT NULL DEFAULT 0,"
 185               "  prio INTEGER NOT NULL DEFAULT 0,"
 186               "  anonLevel INTEGER NOT NULL DEFAULT 0,"
 187               "  expire BIGINT NOT NULL DEFAULT 0,"
 188               "  rvalue BIGINT NOT NULL DEFAULT 0,"
 189               "  hash BYTEA NOT NULL DEFAULT '',"
 190               "  vhash BYTEA NOT NULL DEFAULT '',"
 191               "  value BYTEA NOT NULL DEFAULT '')" "WITH OIDS");
 192   if ((ret == NULL) || ((PQresultStatus (ret) != PGRES_COMMAND_OK) && (0 != strcmp ("42P07",    /* duplicate table */
 193                                                                                     PQresultErrorField
 194                                                                                     (ret,
 195                                                                                      PG_DIAG_SQLSTATE)))))
 196   {
 197     (void) check_result (plugin, ret, PGRES_COMMAND_OK, "CREATE TABLE", "gn090",
 198                          __LINE__);
 199     PQfinish (plugin->dbh);
 200     plugin->dbh = NULL;
 201     return GNUNET_SYSERR;
 202   }
 203   if (PQresultStatus (ret) == PGRES_COMMAND_OK)
 204   {
 205     if ((GNUNET_OK !=
 206          pq_exec (plugin, "CREATE INDEX idx_hash ON gn090 (hash)", __LINE__)) ||
 207         (GNUNET_OK !=
 208          pq_exec (plugin, "CREATE INDEX idx_hash_vhash ON gn090 (hash,vhash)",
 209                   __LINE__)) ||
 210         (GNUNET_OK !=
 211          pq_exec (plugin, "CREATE INDEX idx_prio ON gn090 (prio)", __LINE__)) ||
 212         (GNUNET_OK !=
 213          pq_exec (plugin, "CREATE INDEX idx_expire ON gn090 (expire)",
 214                   __LINE__)) ||
 215         (GNUNET_OK !=
 216          pq_exec (plugin,
 217                   "CREATE INDEX idx_prio_anon ON gn090 (prio,anonLevel)",
 218                   __LINE__)) ||
 219         (GNUNET_OK !=
 220          pq_exec (plugin,
 221                   "CREATE INDEX idx_prio_hash_anon ON gn090 (prio,hash,anonLevel)",
 222                   __LINE__)) ||
 223         (GNUNET_OK !=
 224          pq_exec (plugin, "CREATE INDEX idx_repl_rvalue ON gn090 (repl,rvalue)",
 225                   __LINE__)) ||
 226         (GNUNET_OK !=
 227          pq_exec (plugin, "CREATE INDEX idx_expire_hash ON gn090 (expire,hash)",
 228                   __LINE__)))
 229     {
 230       PQclear (ret);
 231       PQfinish (plugin->dbh);
 232       plugin->dbh = NULL;
 233       return GNUNET_SYSERR;
 234     }
 235   }
 236   PQclear (ret);
 237   ret =
 238       PQexec (plugin->dbh,
 239               "ALTER TABLE gn090 ALTER value SET STORAGE EXTERNAL");
 240   if (GNUNET_OK !=
 241       check_result (plugin, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090",
 242                     __LINE__))
 243   {
 244     PQfinish (plugin->dbh);
 245     plugin->dbh = NULL;
 246     return GNUNET_SYSERR;
 247   }
 248   PQclear (ret);
 249   ret = PQexec (plugin->dbh, "ALTER TABLE gn090 ALTER hash SET STORAGE PLAIN");
 250   if (GNUNET_OK !=
 251       check_result (plugin, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090",
 252                     __LINE__))
 253   {
 254     PQfinish (plugin->dbh);
 255     plugin->dbh = NULL;
 256     return GNUNET_SYSERR;
 257   }
 258   PQclear (ret);
 259   ret = PQexec (plugin->dbh, "ALTER TABLE gn090 ALTER vhash SET STORAGE PLAIN");
 260   if (GNUNET_OK !=
 261       check_result (plugin, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090",
 262                     __LINE__))
 263   {
 264     PQfinish (plugin->dbh);
 265     plugin->dbh = NULL;
 266     return GNUNET_SYSERR;
 267   }
 268   PQclear (ret);
 269   if ((GNUNET_OK !=
 270        pq_prepare (plugin, "getvt",
 271                    "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 272                    "WHERE hash=$1 AND vhash=$2 AND type=$3 "
 273                    "ORDER BY oid ASC LIMIT 1 OFFSET $4", 4, __LINE__)) ||
 274       (GNUNET_OK !=
 275        pq_prepare (plugin, "gett",
 276                    "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 277                    "WHERE hash=$1 AND type=$2 "
 278                    "ORDER BY oid ASC LIMIT 1 OFFSET $3", 3, __LINE__)) ||
 279       (GNUNET_OK !=
 280        pq_prepare (plugin, "getv",
 281                    "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 282                    "WHERE hash=$1 AND vhash=$2 "
 283                    "ORDER BY oid ASC LIMIT 1 OFFSET $3", 3, __LINE__)) ||
 284       (GNUNET_OK !=
 285        pq_prepare (plugin, "get",
 286                    "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 287                    "WHERE hash=$1 " "ORDER BY oid ASC LIMIT 1 OFFSET $2", 2,
 288                    __LINE__)) ||
 289       (GNUNET_OK !=
 290        pq_prepare (plugin, "put",
 291                    "INSERT INTO gn090 (repl, type, prio, anonLevel, expire, rvalue, hash, vhash, value) "
 292                    "VALUES ($1, $2, $3, $4, $5, RANDOM(), $6, $7, $8)", 9,
 293                    __LINE__)) ||
 294       (GNUNET_OK !=
 295        pq_prepare (plugin, "update",
 296                    "UPDATE gn090 SET prio = prio + $1, expire = CASE WHEN expire < $2 THEN $2 ELSE expire END "
 297                    "WHERE oid = $3", 3, __LINE__)) ||
 298       (GNUNET_OK !=
 299        pq_prepare (plugin, "decrepl",
 300                    "UPDATE gn090 SET repl = GREATEST (repl - 1, 0) "
 301                    "WHERE oid = $1", 1, __LINE__)) ||
 302       (GNUNET_OK !=
 303        pq_prepare (plugin, "select_non_anonymous",
 304                    "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 305                    "WHERE anonLevel = 0 AND type = $1 ORDER BY oid DESC LIMIT 1 OFFSET $2",
 306                    1, __LINE__)) ||
 307       (GNUNET_OK !=
 308        pq_prepare (plugin, "select_expiration_order",
 309                    "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 310                    "WHERE expire < $1 ORDER BY prio ASC LIMIT 1) " "UNION "
 311                    "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 312                    "ORDER BY prio ASC LIMIT 1) " "ORDER BY expire ASC LIMIT 1",
 313                    1, __LINE__)) ||
 314       (GNUNET_OK !=
 315        pq_prepare (plugin, "select_replication_order",
 316                    "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
 317                    "ORDER BY repl DESC,RANDOM() LIMIT 1", 0, __LINE__)) ||
 318       (GNUNET_OK !=
 319        pq_prepare (plugin, "delrow", "DELETE FROM gn090 " "WHERE oid=$1", 1,
 320                    __LINE__)))
 321   {
 322     PQfinish (plugin->dbh);
 323     plugin->dbh = NULL;
 324     return GNUNET_SYSERR;
 325   }
 326   return GNUNET_OK;
 327 }
 328
 329
 330 /**
 331  * Delete the row identified by the given rowid (qid
 332  * in postgres).
 333  *
 334  * @param plugin global context
 335  * @param rowid which row to delete
 336  * @return GNUNET_OK on success
 337  */
 338 static int
 339 delete_by_rowid (struct Plugin *plugin, unsigned int rowid)
 340 {
 341   uint32_t browid;
 342   const char *paramValues[] = { (const char *) &browid };
 343   int paramLengths[] = { sizeof (browid) };
 344   const int paramFormats[] = { 1 };
 345   PGresult *ret;
 346
 347   browid = htonl (rowid);
 348   ret =
 349       PQexecPrepared (plugin->dbh, "delrow", 1, paramValues, paramLengths,
 350                       paramFormats, 1);
 351   if (GNUNET_OK !=
 352       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexecPrepared", "delrow",
 353                     __LINE__))
 354   {
 355     return GNUNET_SYSERR;
 356   }
 357   PQclear (ret);
 358   return GNUNET_OK;
 359 }
 360
 361
 362 /**
 363  * Get an estimate of how much space the database is
 364  * currently using.
 365  *
 366  * @param cls our "struct Plugin*"
 367  * @return number of bytes used on disk
 368  */
 369 static unsigned long long
 370 postgres_plugin_estimate_size (void *cls)
 371 {
 372   struct Plugin *plugin = cls;
 373   unsigned long long total;
 374   PGresult *ret;
 375
 376   ret =
 377       PQexecParams (plugin->dbh,
 378                     "SELECT SUM(LENGTH(value))+256*COUNT(*) FROM gn090", 0,
 379                     NULL, NULL, NULL, NULL, 1);
 380   if (GNUNET_OK !=
 381       check_result (plugin, ret, PGRES_TUPLES_OK, "PQexecParams", "get_size",
 382                     __LINE__))
 383   {
 384     return 0;
 385   }
 386   if ((PQntuples (ret) != 1) || (PQnfields (ret) != 1) ||
 387       (PQgetlength (ret, 0, 0) != sizeof (unsigned long long)))
 388   {
 389     GNUNET_break (0);
 390     PQclear (ret);
 391     return 0;
 392   }
 393   total = GNUNET_ntohll (*(const unsigned long long *) PQgetvalue (ret, 0, 0));
 394   PQclear (ret);
 395   return total;
 396 }
 397
 398
 399 /**
 400  * Store an item in the datastore.
 401  *
 402  * @param cls closure
 403  * @param key key for the item
 404  * @param size number of bytes in data
 405  * @param data content stored
 406  * @param type type of the content
 407  * @param priority priority of the content
 408  * @param anonymity anonymity-level for the content
 409  * @param replication replication-level for the content
 410  * @param expiration expiration time for the content
 411  * @param msg set to error message
 412  * @return GNUNET_OK on success
 413  */
 414 static int
 415 postgres_plugin_put (void *cls, const GNUNET_HashCode * key, uint32_t size,
 416                      const void *data, enum GNUNET_BLOCK_Type type,
 417                      uint32_t priority, uint32_t anonymity,
 418                      uint32_t replication,
 419                      struct GNUNET_TIME_Absolute expiration, char **msg)
 420 {
 421   struct Plugin *plugin = cls;
 422   GNUNET_HashCode vhash;
 423   PGresult *ret;
 424   uint32_t btype = htonl (type);
 425   uint32_t bprio = htonl (priority);
 426   uint32_t banon = htonl (anonymity);
 427   uint32_t brepl = htonl (replication);
 428   uint64_t bexpi = GNUNET_TIME_absolute_hton (expiration).abs_value__;
 429
 430   const char *paramValues[] = {
 431     (const char *) &brepl,
 432     (const char *) &btype,
 433     (const char *) &bprio,
 434     (const char *) &banon,
 435     (const char *) &bexpi,
 436     (const char *) key,
 437     (const char *) &vhash,
 438     (const char *) data
 439   };
 440   int paramLengths[] = {
 441     sizeof (brepl),
 442     sizeof (btype),
 443     sizeof (bprio),
 444     sizeof (banon),
 445     sizeof (bexpi),
 446     sizeof (GNUNET_HashCode),
 447     sizeof (GNUNET_HashCode),
 448     size
 449   };
 450   const int paramFormats[] = { 1, 1, 1, 1, 1, 1, 1, 1 };
 451
 452   GNUNET_CRYPTO_hash (data, size, &vhash);
 453   ret =
 454       PQexecPrepared (plugin->dbh, "put", 8, paramValues, paramLengths,
 455                       paramFormats, 1);
 456   if (GNUNET_OK !=
 457       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexecPrepared", "put",
 458                     __LINE__))
 459     return GNUNET_SYSERR;
 460   PQclear (ret);
 461   plugin->env->duc (plugin->env->cls, size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 462 #if DEBUG_POSTGRES
 463   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datastore-postgres",
 464                    "Stored %u bytes in database\n", (unsigned int) size);
 465 #endif
 466   return GNUNET_OK;
 467 }
 468
 469
 470 /**
 471  * Function invoked to process the result and call
 472  * the processor.
 473  *
 474  * @param plugin global plugin data
 475  * @param proc function to call the value (once only).
 476  * @param proc_cls closure for proc
 477  * @param res result from exec
 478  * @param line line number for error messages
 479  */
 480 static void
 481 process_result (struct Plugin *plugin, PluginDatumProcessor proc,
 482                 void *proc_cls, PGresult * res, int line)
 483 {
 484   int iret;
 485   enum GNUNET_BLOCK_Type type;
 486   uint32_t anonymity;
 487   uint32_t priority;
 488   uint32_t size;
 489   unsigned int rowid;
 490   struct GNUNET_TIME_Absolute expiration_time;
 491   GNUNET_HashCode key;
 492
 493   if (GNUNET_OK !=
 494       check_result (plugin, res, PGRES_TUPLES_OK, "PQexecPrepared", "select",
 495                     line))
 496   {
 497 #if DEBUG_POSTGRES
 498     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datastore-postgres",
 499                      "Ending iteration (postgres error)\n");
 500 #endif
 501     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 502     return;
 503   }
 504
 505   if (0 == PQntuples (res))
 506   {
 507     /* no result */
 508 #if DEBUG_POSTGRES
 509     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datastore-postgres",
 510                      "Ending iteration (no more results)\n");
 511 #endif
 512     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 513     PQclear (res);
 514     return;
 515   }
 516   if ((1 != PQntuples (res)) || (7 != PQnfields (res)) ||
 517       (sizeof (uint32_t) != PQfsize (res, 0)) ||
 518       (sizeof (uint32_t) != PQfsize (res, 6)))
 519   {
 520     GNUNET_break (0);
 521     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 522     PQclear (res);
 523     return;
 524   }
 525   rowid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 6));
 526   if ((sizeof (uint32_t) != PQfsize (res, 0)) ||
 527       (sizeof (uint32_t) != PQfsize (res, 1)) ||
 528       (sizeof (uint32_t) != PQfsize (res, 2)) ||
 529       (sizeof (uint64_t) != PQfsize (res, 3)) ||
 530       (sizeof (GNUNET_HashCode) != PQgetlength (res, 0, 4)))
 531   {
 532     GNUNET_break (0);
 533     PQclear (res);
 534     delete_by_rowid (plugin, rowid);
 535     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 536     return;
 537   }
 538
 539   type = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
 540   priority = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
 541   anonymity = ntohl (*(uint32_t *) PQgetvalue (res, 0, 2));
 542   expiration_time.abs_value =
 543       GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, 0, 3));
 544   memcpy (&key, PQgetvalue (res, 0, 4), sizeof (GNUNET_HashCode));
 545   size = PQgetlength (res, 0, 5);
 546 #if DEBUG_POSTGRES
 547   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datastore-postgres",
 548                    "Found result of size %u bytes and type %u in database\n",
 549                    (unsigned int) size, (unsigned int) type);
 550 #endif
 551   iret =
 552       proc (proc_cls, &key, size, PQgetvalue (res, 0, 5),
 553             (enum GNUNET_BLOCK_Type) type, priority, anonymity, expiration_time,
 554             rowid);
 555   PQclear (res);
 556   if (iret == GNUNET_NO)
 557   {
 558 #if DEBUG_POSTGRES
 559     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 560                 "Processor asked for item %u to be removed.\n", rowid);
 561 #endif
 562     if (GNUNET_OK == delete_by_rowid (plugin, rowid))
 563     {
 564 #if DEBUG_POSTGRES
 565       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datastore-postgres",
 566                        "Deleting %u bytes from database\n",
 567                        (unsigned int) size);
 568 #endif
 569       plugin->env->duc (plugin->env->cls,
 570                         -(size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
 571 #if DEBUG_POSTGRES
 572       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "datastore-postgres",
 573                        "Deleted %u bytes from database\n", (unsigned int) size);
 574 #endif
 575     }
 576   }
 577 }
 578
 579
 580 /**
 581  * Iterate over the results for a particular key
 582  * in the datastore.
 583  *
 584  * @param cls closure
 585  * @param offset offset of the result (modulo num-results);
 586  *        specific ordering does not matter for the offset
 587  * @param key maybe NULL (to match all entries)
 588  * @param vhash hash of the value, maybe NULL (to
 589  *        match all values that have the right key).
 590  *        Note that for DBlocks there is no difference
 591  *        betwen key and vhash, but for other blocks
 592  *        there may be!
 593  * @param type entries of which type are relevant?
 594  *     Use 0 for any type.
 595  * @param proc function to call on the matching value;
 596  *        will be called once with a NULL if no value matches
 597  * @param proc_cls closure for iter
 598  */
 599 static void
 600 postgres_plugin_get_key (void *cls, uint64_t offset,
 601                          const GNUNET_HashCode * key,
 602                          const GNUNET_HashCode * vhash,
 603                          enum GNUNET_BLOCK_Type type, PluginDatumProcessor proc,
 604                          void *proc_cls)
 605 {
 606   struct Plugin *plugin = cls;
 607   const int paramFormats[] = { 1, 1, 1, 1, 1 };
 608   int paramLengths[4];
 609   const char *paramValues[4];
 610   int nparams;
 611   const char *pname;
 612   PGresult *ret;
 613   uint64_t total;
 614   uint64_t blimit_off;
 615   uint32_t btype;
 616
 617   GNUNET_assert (key != NULL);
 618   paramValues[0] = (const char *) key;
 619   paramLengths[0] = sizeof (GNUNET_HashCode);
 620   btype = htonl (type);
 621   if (type != 0)
 622   {
 623     if (vhash != NULL)
 624     {
 625       paramValues[1] = (const char *) vhash;
 626       paramLengths[1] = sizeof (GNUNET_HashCode);
 627       paramValues[2] = (const char *) &btype;
 628       paramLengths[2] = sizeof (btype);
 629       paramValues[3] = (const char *) &blimit_off;
 630       paramLengths[3] = sizeof (blimit_off);
 631       nparams = 4;
 632       pname = "getvt";
 633       ret =
 634           PQexecParams (plugin->dbh,
 635                         "SELECT count(*) FROM gn090 WHERE hash=$1 AND vhash=$2 AND type=$3",
 636                         3, NULL, paramValues, paramLengths, paramFormats, 1);
 637     }
 638     else
 639     {
 640       paramValues[1] = (const char *) &btype;
 641       paramLengths[1] = sizeof (btype);
 642       paramValues[2] = (const char *) &blimit_off;
 643       paramLengths[2] = sizeof (blimit_off);
 644       nparams = 3;
 645       pname = "gett";
 646       ret =
 647           PQexecParams (plugin->dbh,
 648                         "SELECT count(*) FROM gn090 WHERE hash=$1 AND type=$2",
 649                         2, NULL, paramValues, paramLengths, paramFormats, 1);
 650     }
 651   }
 652   else
 653   {
 654     if (vhash != NULL)
 655     {
 656       paramValues[1] = (const char *) vhash;
 657       paramLengths[1] = sizeof (GNUNET_HashCode);
 658       paramValues[2] = (const char *) &blimit_off;
 659       paramLengths[2] = sizeof (blimit_off);
 660       nparams = 3;
 661       pname = "getv";
 662       ret =
 663           PQexecParams (plugin->dbh,
 664                         "SELECT count(*) FROM gn090 WHERE hash=$1 AND vhash=$2",
 665                         2, NULL, paramValues, paramLengths, paramFormats, 1);
 666     }
 667     else
 668     {
 669       paramValues[1] = (const char *) &blimit_off;
 670       paramLengths[1] = sizeof (blimit_off);
 671       nparams = 2;
 672       pname = "get";
 673       ret =
 674           PQexecParams (plugin->dbh, "SELECT count(*) FROM gn090 WHERE hash=$1",
 675                         1, NULL, paramValues, paramLengths, paramFormats, 1);
 676     }
 677   }
 678   if (GNUNET_OK !=
 679       check_result (plugin, ret, PGRES_TUPLES_OK, "PQexecParams", pname,
 680                     __LINE__))
 681   {
 682     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 683     return;
 684   }
 685   if ((PQntuples (ret) != 1) || (PQnfields (ret) != 1) ||
 686       (PQgetlength (ret, 0, 0) != sizeof (unsigned long long)))
 687   {
 688     GNUNET_break (0);
 689     PQclear (ret);
 690     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 691     return;
 692   }
 693   total = GNUNET_ntohll (*(const unsigned long long *) PQgetvalue (ret, 0, 0));
 694   PQclear (ret);
 695   if (total == 0)
 696   {
 697     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 698     return;
 699   }
 700   blimit_off = GNUNET_htonll (offset % total);
 701   ret =
 702       PQexecPrepared (plugin->dbh, pname, nparams, paramValues, paramLengths,
 703                       paramFormats, 1);
 704   process_result (plugin, proc, proc_cls, ret, __LINE__);
 705 }
 706
 707
 708 /**
 709  * Select a subset of the items in the datastore and call
 710  * the given iterator for each of them.
 711  *
 712  * @param cls our "struct Plugin*"
 713  * @param offset offset of the result (modulo num-results);
 714  *        specific ordering does not matter for the offset
 715  * @param type entries of which type should be considered?
 716  *        Use 0 for any type.
 717  * @param proc function to call on the matching value;
 718  *        will be called with a NULL if no value matches
 719  * @param proc_cls closure for proc
 720  */
 721 static void
 722 postgres_plugin_get_zero_anonymity (void *cls, uint64_t offset,
 723                                     enum GNUNET_BLOCK_Type type,
 724                                     PluginDatumProcessor proc, void *proc_cls)
 725 {
 726   struct Plugin *plugin = cls;
 727   uint32_t btype;
 728   uint64_t boff;
 729   const int paramFormats[] = { 1, 1 };
 730   int paramLengths[] = { sizeof (btype), sizeof (boff) };
 731   const char *paramValues[] = { (const char *) &btype, (const char *) &boff };
 732   PGresult *ret;
 733
 734   btype = htonl ((uint32_t) type);
 735   boff = GNUNET_htonll (offset);
 736   ret =
 737       PQexecPrepared (plugin->dbh, "select_non_anonymous", 2, paramValues,
 738                       paramLengths, paramFormats, 1);
 739   process_result (plugin, proc, proc_cls, ret, __LINE__);
 740 }
 741
 742
 743 /**
 744  * Context for 'repl_iter' function.
 745  */
 746 struct ReplCtx
 747 {
 748
 749   /**
 750    * Plugin handle.
 751    */
 752   struct Plugin *plugin;
 753
 754   /**
 755    * Function to call for the result (or the NULL).
 756    */
 757   PluginDatumProcessor proc;
 758
 759   /**
 760    * Closure for proc.
 761    */
 762   void *proc_cls;
 763 };
 764
 765
 766 /**
 767  * Wrapper for the iterator for 'sqlite_plugin_replication_get'.
 768  * Decrements the replication counter and calls the original
 769  * iterator.
 770  *
 771  * @param cls closure
 772  * @param key key for the content
 773  * @param size number of bytes in data
 774  * @param data content stored
 775  * @param type type of the content
 776  * @param priority priority of the content
 777  * @param anonymity anonymity-level for the content
 778  * @param expiration expiration time for the content
 779  * @param uid unique identifier for the datum;
 780  *        maybe 0 if no unique identifier is available
 781  *
 782  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
 783  *         (continue on call to "next", of course),
 784  *         GNUNET_NO to delete the item and continue (if supported)
 785  */
 786 static int
 787 repl_proc (void *cls, const GNUNET_HashCode * key, uint32_t size,
 788            const void *data, enum GNUNET_BLOCK_Type type, uint32_t priority,
 789            uint32_t anonymity, struct GNUNET_TIME_Absolute expiration,
 790            uint64_t uid)
 791 {
 792   struct ReplCtx *rc = cls;
 793   struct Plugin *plugin = rc->plugin;
 794   int ret;
 795   PGresult *qret;
 796   uint32_t boid;
 797
 798   ret =
 799       rc->proc (rc->proc_cls, key, size, data, type, priority, anonymity,
 800                 expiration, uid);
 801   if (NULL != key)
 802   {
 803     boid = htonl ((uint32_t) uid);
 804     const char *paramValues[] = {
 805       (const char *) &boid,
 806     };
 807     int paramLengths[] = {
 808       sizeof (boid),
 809     };
 810     const int paramFormats[] = { 1 };
 811     qret =
 812         PQexecPrepared (plugin->dbh, "decrepl", 1, paramValues, paramLengths,
 813                         paramFormats, 1);
 814     if (GNUNET_OK !=
 815         check_result (plugin, qret, PGRES_COMMAND_OK, "PQexecPrepared",
 816                       "decrepl", __LINE__))
 817       return GNUNET_SYSERR;
 818     PQclear (qret);
 819   }
 820   return ret;
 821 }
 822
 823
 824 /**
 825  * Get a random item for replication.  Returns a single, not expired, random item
 826  * from those with the highest replication counters.  The item's
 827  * replication counter is decremented by one IF it was positive before.
 828  * Call 'proc' with all values ZERO or NULL if the datastore is empty.
 829  *
 830  * @param cls closure
 831  * @param proc function to call the value (once only).
 832  * @param proc_cls closure for proc
 833  */
 834 static void
 835 postgres_plugin_get_replication (void *cls, PluginDatumProcessor proc,
 836                                  void *proc_cls)
 837 {
 838   struct Plugin *plugin = cls;
 839   struct ReplCtx rc;
 840   PGresult *ret;
 841
 842   rc.plugin = plugin;
 843   rc.proc = proc;
 844   rc.proc_cls = proc_cls;
 845   ret =
 846       PQexecPrepared (plugin->dbh, "select_replication_order", 0, NULL, NULL,
 847                       NULL, 1);
 848   process_result (plugin, &repl_proc, &rc, ret, __LINE__);
 849 }
 850
 851
 852 /**
 853  * Get a random item for expiration.
 854  * Call 'proc' with all values ZERO or NULL if the datastore is empty.
 855  *
 856  * @param cls closure
 857  * @param proc function to call the value (once only).
 858  * @param proc_cls closure for proc
 859  */
 860 static void
 861 postgres_plugin_get_expiration (void *cls, PluginDatumProcessor proc,
 862                                 void *proc_cls)
 863 {
 864   struct Plugin *plugin = cls;
 865   uint64_t btime;
 866   const int paramFormats[] = { 1 };
 867   int paramLengths[] = { sizeof (btime) };
 868   const char *paramValues[] = { (const char *) &btime };
 869   PGresult *ret;
 870
 871   btime = GNUNET_htonll (GNUNET_TIME_absolute_get ().abs_value);
 872   ret =
 873       PQexecPrepared (plugin->dbh, "select_expiration_order", 1, paramValues,
 874                       paramLengths, paramFormats, 1);
 875   process_result (plugin, proc, proc_cls, ret, __LINE__);
 876 }
 877
 878
 879 /**
 880  * Update the priority for a particular key in the datastore.  If
 881  * the expiration time in value is different than the time found in
 882  * the datastore, the higher value should be kept.  For the
 883  * anonymity level, the lower value is to be used.  The specified
 884  * priority should be added to the existing priority, ignoring the
 885  * priority in value.
 886  *
 887  * Note that it is possible for multiple values to match this put.
 888  * In that case, all of the respective values are updated.
 889  *
 890  * @param cls our "struct Plugin*"
 891  * @param uid unique identifier of the datum
 892  * @param delta by how much should the priority
 893  *     change?  If priority + delta < 0 the
 894  *     priority should be set to 0 (never go
 895  *     negative).
 896  * @param expire new expiration time should be the
 897  *     MAX of any existing expiration time and
 898  *     this value
 899  * @param msg set to error message
 900  * @return GNUNET_OK on success
 901  */
 902 static int
 903 postgres_plugin_update (void *cls, uint64_t uid, int delta,
 904                         struct GNUNET_TIME_Absolute expire, char **msg)
 905 {
 906   struct Plugin *plugin = cls;
 907   PGresult *ret;
 908   int32_t bdelta = (int32_t) htonl ((uint32_t) delta);
 909   uint32_t boid = htonl ((uint32_t) uid);
 910   uint64_t bexpire = GNUNET_TIME_absolute_hton (expire).abs_value__;
 911
 912   const char *paramValues[] = {
 913     (const char *) &bdelta,
 914     (const char *) &bexpire,
 915     (const char *) &boid,
 916   };
 917   int paramLengths[] = {
 918     sizeof (bdelta),
 919     sizeof (bexpire),
 920     sizeof (boid),
 921   };
 922   const int paramFormats[] = { 1, 1, 1 };
 923
 924   ret =
 925       PQexecPrepared (plugin->dbh, "update", 3, paramValues, paramLengths,
 926                       paramFormats, 1);
 927   if (GNUNET_OK !=
 928       check_result (plugin, ret, PGRES_COMMAND_OK, "PQexecPrepared", "update",
 929                     __LINE__))
 930     return GNUNET_SYSERR;
 931   PQclear (ret);
 932   return GNUNET_OK;
 933 }
 934
 935
 936 /**
 937  * Drop database.
 938  */
 939 static void
 940 postgres_plugin_drop (void *cls)
 941 {
 942   struct Plugin *plugin = cls;
 943
 944   pq_exec (plugin, "DROP TABLE gn090", __LINE__);
 945 }
 946
 947
 948 /**
 949  * Entry point for the plugin.
 950  *
 951  * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
 952  * @return our "struct Plugin*"
 953  */
 954 void *
 955 libgnunet_plugin_datastore_postgres_init (void *cls)
 956 {
 957   struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
 958   struct GNUNET_DATASTORE_PluginFunctions *api;
 959   struct Plugin *plugin;
 960
 961   plugin = GNUNET_malloc (sizeof (struct Plugin));
 962   plugin->env = env;
 963   if (GNUNET_OK != init_connection (plugin))
 964   {
 965     GNUNET_free (plugin);
 966     return NULL;
 967   }
 968   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
 969   api->cls = plugin;
 970   api->estimate_size = &postgres_plugin_estimate_size;
 971   api->put = &postgres_plugin_put;
 972   api->update = &postgres_plugin_update;
 973   api->get_key = &postgres_plugin_get_key;
 974   api->get_replication = &postgres_plugin_get_replication;
 975   api->get_expiration = &postgres_plugin_get_expiration;
 976   api->get_zero_anonymity = &postgres_plugin_get_zero_anonymity;
 977   api->drop = &postgres_plugin_drop;
 978   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO, "datastore-postgres",
 979                    _("Postgres database running\n"));
 980   return api;
 981 }
 982
 983
 984 /**
 985  * Exit point from the plugin.
 986  * @param cls our "struct Plugin*"
 987  * @return always NULL
 988  */
 989 void *
 990 libgnunet_plugin_datastore_postgres_done (void *cls)
 991 {
 992   struct GNUNET_DATASTORE_PluginFunctions *api = cls;
 993   struct Plugin *plugin = api->cls;
 994
 995   PQfinish (plugin->dbh);
 996   GNUNET_free (plugin);
 997   GNUNET_free (api);
 998   return NULL;
 999 }
1000
1001 /* end of plugin_datastore_postgres.c */