src/datacache/plugin_datacache_postgres.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2006, 2009, 2010, 2012 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_postgres_lib.h"
  29 #include "gnunet_datacache_plugin.h"
  30
  31 #define LOG(kind,...) GNUNET_log_from (kind, "datacache-postgres", __VA_ARGS__)
  32
  33 /**
  34  * Per-entry overhead estimate
  35  */
  36 #define OVERHEAD (sizeof(struct 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  * @brief Get a database handle
  58  *
  59  * @param plugin global context
  60  * @return GNUNET_OK on success, GNUNET_SYSERR on error
  61  */
  62 static int
  63 init_connection (struct Plugin *plugin)
  64 {
  65   PGresult *ret;
  66
  67   plugin->dbh = GNUNET_POSTGRES_connect (plugin->env->cfg,
  68                                          "datacache-postgres");
  69   if (NULL == plugin->dbh)
  70     return GNUNET_SYSERR;
  71   ret =
  72       PQexec (plugin->dbh,
  73               "CREATE TEMPORARY TABLE gn090dc ("
  74               "  type INTEGER NOT NULL DEFAULT 0,"
  75               "  discard_time BIGINT NOT NULL DEFAULT 0,"
  76               "  key BYTEA NOT NULL DEFAULT '',"
  77               "  value BYTEA NOT NULL DEFAULT '',"
  78               "  path BYTEA DEFAULT '')"
  79               "WITH OIDS");
  80   if ( (ret == NULL) ||
  81        ((PQresultStatus (ret) != PGRES_COMMAND_OK) &&
  82         (0 != strcmp ("42P07",    /* duplicate table */
  83                       PQresultErrorField
  84                       (ret,
  85                        PG_DIAG_SQLSTATE)))))
  86   {
  87     (void) GNUNET_POSTGRES_check_result (plugin->dbh, ret,
  88                                          PGRES_COMMAND_OK, "CREATE TABLE",
  89                                          "gn090dc");
  90     PQfinish (plugin->dbh);
  91     plugin->dbh = NULL;
  92     return GNUNET_SYSERR;
  93   }
  94   if (PQresultStatus (ret) == PGRES_COMMAND_OK)
  95   {
  96     if ((GNUNET_OK !=
  97          GNUNET_POSTGRES_exec (plugin->dbh, "CREATE INDEX idx_key ON gn090dc (key)")) ||
  98         (GNUNET_OK !=
  99          GNUNET_POSTGRES_exec (plugin->dbh, "CREATE INDEX idx_dt ON gn090dc (discard_time)")))
 100     {
 101       PQclear (ret);
 102       PQfinish (plugin->dbh);
 103       plugin->dbh = NULL;
 104       return GNUNET_SYSERR;
 105     }
 106   }
 107   PQclear (ret);
 108   ret =
 109       PQexec (plugin->dbh,
 110               "ALTER TABLE gn090dc ALTER value SET STORAGE EXTERNAL");
 111   if (GNUNET_OK !=
 112       GNUNET_POSTGRES_check_result (plugin->dbh, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090dc"))
 113   {
 114     PQfinish (plugin->dbh);
 115     plugin->dbh = NULL;
 116     return GNUNET_SYSERR;
 117   }
 118   PQclear (ret);
 119   ret = PQexec (plugin->dbh, "ALTER TABLE gn090dc ALTER key SET STORAGE PLAIN");
 120   if (GNUNET_OK !=
 121       GNUNET_POSTGRES_check_result (plugin->dbh, ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090dc"))
 122   {
 123     PQfinish (plugin->dbh);
 124     plugin->dbh = NULL;
 125     return GNUNET_SYSERR;
 126   }
 127   PQclear (ret);
 128   if ((GNUNET_OK !=
 129        GNUNET_POSTGRES_prepare (plugin->dbh, "getkt",
 130                    "SELECT discard_time,type,value,path FROM gn090dc "
 131                    "WHERE key=$1 AND type=$2 ", 2)) ||
 132       (GNUNET_OK !=
 133        GNUNET_POSTGRES_prepare (plugin->dbh, "getk",
 134                    "SELECT discard_time,type,value,path FROM gn090dc "
 135                    "WHERE key=$1", 1)) ||
 136       (GNUNET_OK !=
 137        GNUNET_POSTGRES_prepare (plugin->dbh, "getm",
 138                    "SELECT length(value),oid,key FROM gn090dc "
 139                    "ORDER BY discard_time ASC LIMIT 1", 0)) ||
 140       (GNUNET_OK !=
 141        GNUNET_POSTGRES_prepare (plugin->dbh, "delrow", "DELETE FROM gn090dc WHERE oid=$1", 1)) ||
 142       (GNUNET_OK !=
 143        GNUNET_POSTGRES_prepare (plugin->dbh, "put",
 144                    "INSERT INTO gn090dc (type, discard_time, key, value, path) "
 145                    "VALUES ($1, $2, $3, $4, $5)", 5)))
 146   {
 147     PQfinish (plugin->dbh);
 148     plugin->dbh = NULL;
 149     return GNUNET_SYSERR;
 150   }
 151   return GNUNET_OK;
 152 }
 153
 154
 155 /**
 156  * Store an item in the datastore.
 157  *
 158  * @param cls closure (our "struct Plugin")
 159  * @param key key to store data under
 160  * @param size number of bytes in data
 161  * @param data data to store
 162  * @param type type of the value
 163  * @param discard_time when to discard the value in any case
 164  * @param path_info_len number of entries in 'path_info'
 165  * @param path_info a path through the network
 166  * @return 0 if duplicate, -1 on error, number of bytes used otherwise
 167  */
 168 static ssize_t
 169 postgres_plugin_put (void *cls, const struct GNUNET_HashCode * key, size_t size,
 170                      const char *data, enum GNUNET_BLOCK_Type type,
 171                      struct GNUNET_TIME_Absolute discard_time,
 172                      unsigned int path_info_len,
 173                      const struct GNUNET_PeerIdentity *path_info)
 174 {
 175   struct Plugin *plugin = cls;
 176   PGresult *ret;
 177   uint32_t btype = htonl (type);
 178   uint64_t bexpi = GNUNET_TIME_absolute_hton (discard_time).abs_value_us__;
 179
 180   const char *paramValues[] = {
 181     (const char *) &btype,
 182     (const char *) &bexpi,
 183     (const char *) key,
 184     (const char *) data,
 185     (const char *) path_info
 186   };
 187   int paramLengths[] = {
 188     sizeof (btype),
 189     sizeof (bexpi),
 190     sizeof (struct GNUNET_HashCode),
 191     size,
 192     path_info_len * sizeof (struct GNUNET_PeerIdentity)
 193   };
 194   const int paramFormats[] = { 1, 1, 1, 1, 1 };
 195
 196   ret =
 197       PQexecPrepared (plugin->dbh, "put", 5, paramValues, paramLengths,
 198                       paramFormats, 1);
 199   if (GNUNET_OK !=
 200       GNUNET_POSTGRES_check_result (plugin->dbh, ret,
 201                                     PGRES_COMMAND_OK, "PQexecPrepared", "put"))
 202     return -1;
 203   PQclear (ret);
 204   return size + OVERHEAD;
 205 }
 206
 207
 208 /**
 209  * Iterate over the results for a particular key
 210  * in the datastore.
 211  *
 212  * @param cls closure (our "struct Plugin")
 213  * @param key
 214  * @param type entries of which type are relevant?
 215  * @param iter maybe NULL (to just count)
 216  * @param iter_cls closure for iter
 217  * @return the number of results found
 218  */
 219 static unsigned int
 220 postgres_plugin_get (void *cls, const struct GNUNET_HashCode * key,
 221                      enum GNUNET_BLOCK_Type type,
 222                      GNUNET_DATACACHE_Iterator iter, void *iter_cls)
 223 {
 224   struct Plugin *plugin = cls;
 225   uint32_t btype = htonl (type);
 226
 227   const char *paramValues[] = {
 228     (const char *) key,
 229     (const char *) &btype,
 230   };
 231   int paramLengths[] = {
 232     sizeof (struct GNUNET_HashCode),
 233     sizeof (btype),
 234   };
 235   const int paramFormats[] = { 1, 1 };
 236   struct GNUNET_TIME_Absolute expiration_time;
 237   uint32_t size;
 238   unsigned int cnt;
 239   unsigned int i;
 240   unsigned int path_len;
 241   const struct GNUNET_PeerIdentity *path;
 242   PGresult *res;
 243
 244   res =
 245       PQexecPrepared (plugin->dbh, (type == 0) ? "getk" : "getkt",
 246                       (type == 0) ? 1 : 2, paramValues, paramLengths,
 247                       paramFormats, 1);
 248   if (GNUNET_OK !=
 249       GNUNET_POSTGRES_check_result (plugin->dbh, res, PGRES_TUPLES_OK, "PQexecPrepared",
 250                                     (type == 0) ? "getk" : "getkt"))
 251   {
 252     LOG (GNUNET_ERROR_TYPE_DEBUG,
 253          "Ending iteration (postgres error)\n");
 254     return 0;
 255   }
 256
 257   if (0 == (cnt = PQntuples (res)))
 258   {
 259     /* no result */
 260     LOG (GNUNET_ERROR_TYPE_DEBUG,
 261          "Ending iteration (no more results)\n");
 262     PQclear (res);
 263     return 0;
 264   }
 265   if (iter == NULL)
 266   {
 267     PQclear (res);
 268     return cnt;
 269   }
 270   if ((4 != PQnfields (res)) || (sizeof (uint64_t) != PQfsize (res, 0)) ||
 271       (sizeof (uint32_t) != PQfsize (res, 1)))
 272   {
 273     GNUNET_break (0);
 274     PQclear (res);
 275     return 0;
 276   }
 277   for (i = 0; i < cnt; i++)
 278   {
 279     expiration_time.abs_value_us =
 280         GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 0));
 281     type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1));
 282     size = PQgetlength (res, i, 2);
 283     path_len = PQgetlength (res, i, 3);
 284     if (0 != (path_len % sizeof (struct GNUNET_PeerIdentity)))
 285     {
 286       GNUNET_break (0);
 287       path_len = 0;
 288     }
 289     path_len %= sizeof (struct GNUNET_PeerIdentity);
 290     path = (const struct GNUNET_PeerIdentity *) PQgetvalue (res, i, 3);
 291     LOG (GNUNET_ERROR_TYPE_DEBUG,
 292          "Found result of size %u bytes and type %u in database\n",
 293          (unsigned int) size, (unsigned int) type);
 294     if (GNUNET_SYSERR ==
 295         iter (iter_cls, key, size, PQgetvalue (res, i, 2),
 296               (enum GNUNET_BLOCK_Type) type,
 297               expiration_time,
 298               path_len,
 299               path))
 300     {
 301       LOG (GNUNET_ERROR_TYPE_DEBUG,
 302            "Ending iteration (client error)\n");
 303       PQclear (res);
 304       return cnt;
 305     }
 306   }
 307   PQclear (res);
 308   return cnt;
 309 }
 310
 311
 312 /**
 313  * Delete the entry with the lowest expiration value
 314  * from the datacache right now.
 315  *
 316  * @param cls closure (our "struct Plugin")
 317  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 318  */
 319 static int
 320 postgres_plugin_del (void *cls)
 321 {
 322   struct Plugin *plugin = cls;
 323   uint32_t size;
 324   uint32_t oid;
 325   struct GNUNET_HashCode key;
 326   PGresult *res;
 327
 328   res = PQexecPrepared (plugin->dbh, "getm", 0, NULL, NULL, NULL, 1);
 329   if (GNUNET_OK !=
 330       GNUNET_POSTGRES_check_result (plugin->dbh, res, PGRES_TUPLES_OK, "PQexecPrepared", "getm"))
 331   {
 332     LOG (GNUNET_ERROR_TYPE_DEBUG,
 333          "Ending iteration (postgres error)\n");
 334     return 0;
 335   }
 336   if (0 == PQntuples (res))
 337   {
 338     /* no result */
 339     LOG (GNUNET_ERROR_TYPE_DEBUG,
 340          "Ending iteration (no more results)\n");
 341     PQclear (res);
 342     return GNUNET_SYSERR;
 343   }
 344   if ((3 != PQnfields (res)) || (sizeof (size) != PQfsize (res, 0)) ||
 345       (sizeof (oid) != PQfsize (res, 1)) ||
 346       (sizeof (struct GNUNET_HashCode) != PQgetlength (res, 0, 2)))
 347   {
 348     GNUNET_break (0);
 349     PQclear (res);
 350     return 0;
 351   }
 352   size = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
 353   oid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
 354   memcpy (&key, PQgetvalue (res, 0, 2), sizeof (struct GNUNET_HashCode));
 355   PQclear (res);
 356   if (GNUNET_OK != GNUNET_POSTGRES_delete_by_rowid (plugin->dbh, "delrow", oid))
 357     return GNUNET_SYSERR;
 358   plugin->env->delete_notify (plugin->env->cls, &key, size + OVERHEAD);
 359   return GNUNET_OK;
 360 }
 361
 362
 363 /**
 364  * Entry point for the plugin.
 365  *
 366  * @param cls closure (the "struct GNUNET_DATACACHE_PluginEnvironmnet")
 367  * @return the plugin's closure (our "struct Plugin")
 368  */
 369 void *
 370 libgnunet_plugin_datacache_postgres_init (void *cls)
 371 {
 372   struct GNUNET_DATACACHE_PluginEnvironment *env = cls;
 373   struct GNUNET_DATACACHE_PluginFunctions *api;
 374   struct Plugin *plugin;
 375
 376   plugin = GNUNET_new (struct Plugin);
 377   plugin->env = env;
 378
 379   if (GNUNET_OK != init_connection (plugin))
 380   {
 381     GNUNET_free (plugin);
 382     return NULL;
 383   }
 384
 385   api = GNUNET_new (struct GNUNET_DATACACHE_PluginFunctions);
 386   api->cls = plugin;
 387   api->get = &postgres_plugin_get;
 388   api->put = &postgres_plugin_put;
 389   api->del = &postgres_plugin_del;
 390   LOG (GNUNET_ERROR_TYPE_INFO,
 391        _("Postgres datacache running\n"));
 392   return api;
 393 }
 394
 395
 396 /**
 397  * Exit point from the plugin.
 398  *
 399  * @param cls closure (our "struct Plugin")
 400  * @return NULL
 401  */
 402 void *
 403 libgnunet_plugin_datacache_postgres_done (void *cls)
 404 {
 405   struct GNUNET_DATACACHE_PluginFunctions *api = cls;
 406   struct Plugin *plugin = api->cls;
 407
 408   PQfinish (plugin->dbh);
 409   GNUNET_free (plugin);
 410   GNUNET_free (api);
 411   return NULL;
 412 }
 413
 414
 415
 416 /* end of plugin_datacache_postgres.c */