2 This file is part of GNUnet
3 (C) 2009, 2010 Christian Grothoff (and other contributing authors)
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.
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.
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.
22 * @file datastore/plugin_datastore_postgres.c
23 * @brief postgres-based datastore backend
24 * @author Christian Grothoff
28 #include "plugin_datastore.h"
29 #include <postgresql/libpq-fe.h>
31 #define DEBUG_POSTGRES GNUNET_NO
33 #define SELECT_IT_LOW_PRIORITY "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
34 "WHERE (prio = $1 AND oid > $2) " \
35 "ORDER BY prio ASC,oid ASC LIMIT 1) "\
37 "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
38 "WHERE (prio > $1 AND oid != $2)"\
39 "ORDER BY prio ASC,oid ASC LIMIT 1)"\
40 "ORDER BY prio ASC,oid ASC LIMIT 1"
42 #define SELECT_IT_NON_ANONYMOUS "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
43 "WHERE (prio = $1 AND oid < $2)"\
44 " AND anonLevel=0 ORDER BY prio DESC,oid DESC LIMIT 1) "\
46 "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
47 "WHERE (prio < $1 AND oid != $2)"\
48 " AND anonLevel=0 ORDER BY prio DESC,oid DESC LIMIT 1) "\
49 "ORDER BY prio DESC,oid DESC LIMIT 1"
51 #define SELECT_IT_EXPIRATION_TIME "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
52 "WHERE (expire = $1 AND oid > $2) "\
53 "ORDER BY expire ASC,oid ASC LIMIT 1) "\
55 "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
56 "WHERE (expire > $1 AND oid != $2) " \
57 "ORDER BY expire ASC,oid ASC LIMIT 1)"\
58 "ORDER BY expire ASC,oid ASC LIMIT 1"
61 #define SELECT_IT_MIGRATION_ORDER "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
62 "WHERE (expire = $1 AND oid < $2)"\
63 " AND expire > $3 AND type!=3"\
64 " ORDER BY expire DESC,oid DESC LIMIT 1) "\
66 "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
67 "WHERE (expire < $1 AND oid != $2)" \
68 " AND expire > $3 AND type!=3"\
69 " ORDER BY expire DESC,oid DESC LIMIT 1)"\
70 "ORDER BY expire DESC,oid DESC LIMIT 1"
73 * After how many ms "busy" should a DB operation fail for good?
74 * A low value makes sure that we are more responsive to requests
75 * (especially PUTs). A high value guarantees a higher success
76 * rate (SELECTs in iterate can take several seconds despite LIMIT=1).
78 * The default value of 1s should ensure that users do not experience
79 * huge latencies while at the same time allowing operations to succeed
80 * with reasonable probability.
82 #define BUSY_TIMEOUT GNUNET_TIME_UNIT_SECONDS
86 * Closure for 'postgres_next_request_cont'.
88 struct NextRequestClosure
93 struct Plugin *plugin;
96 * Function to call for each matching entry.
101 * Closure for 'iter'.
106 * Parameters for the prepared statement.
108 const char *paramValues[5];
111 * Name of the prepared statement to run.
116 * Size of values pointed to by paramValues.
121 * Number of paramters in paramValues/paramLengths.
126 * Current time (possible parameter), big-endian.
131 * Key (possible parameter)
136 * Hash of value (possible parameter)
138 GNUNET_HashCode vhash;
141 * Number of entries found so far
146 * Offset this iteration starts at.
151 * Current offset to use in query, big-endian.
156 * Overall number of matching entries.
158 unsigned long long total;
161 * Expiration value of previous result (possible parameter), big-endian.
163 uint64_t blast_expire;
166 * Row ID of last result (possible paramter), big-endian.
168 uint32_t blast_rowid;
171 * Priority of last result (possible parameter), big-endian.
176 * Type of block (possible paramter), big-endian.
181 * Flag set to GNUNET_YES to stop iteration.
188 * Context for all functions in this plugin.
193 * Our execution environment.
195 struct GNUNET_DATASTORE_PluginEnvironment *env;
198 * Native Postgres database handle.
203 * Closure of the 'next_task' (must be freed if 'next_task' is cancelled).
205 struct NextRequestClosure *next_task_nc;
208 * Pending task with scheduler for running the next request.
210 GNUNET_SCHEDULER_TaskIdentifier next_task;
216 * Check if the result obtained from Postgres has
217 * the desired status code. If not, log an error, clear the
218 * result and return GNUNET_SYSERR.
220 * @param plugin global context
221 * @param ret result to check
222 * @param expected_status expected return value
223 * @param command name of SQL command that was run
224 * @param args arguments to SQL command
225 * @param line line number for error reporting
226 * @return GNUNET_OK if the result is acceptable
229 check_result (struct Plugin *plugin,
232 const char *command, const char *args, int line)
236 GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
237 "datastore-postgres",
238 "Postgres failed to allocate result for `%s:%s' at %d\n",
239 command, args, line);
240 return GNUNET_SYSERR;
242 if (PQresultStatus (ret) != expected_status)
244 GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
245 "datastore-postgres",
246 _("`%s:%s' failed at %s:%d with error: %s"),
247 command, args, __FILE__, line, PQerrorMessage (plugin->dbh));
249 return GNUNET_SYSERR;
255 * Run simple SQL statement (without results).
257 * @param plugin global context
258 * @param sql statement to run
259 * @param line code line for error reporting
262 pq_exec (struct Plugin *plugin,
263 const char *sql, int line)
266 ret = PQexec (plugin->dbh, sql);
267 if (GNUNET_OK != check_result (plugin,
269 PGRES_COMMAND_OK, "PQexec", sql, line))
270 return GNUNET_SYSERR;
276 * Prepare SQL statement.
278 * @param plugin global context
279 * @param name name for the prepared SQL statement
280 * @param sql SQL code to prepare
281 * @param nparams number of parameters in sql
282 * @param line code line for error reporting
283 * @return GNUNET_OK on success
286 pq_prepare (struct Plugin *plugin,
287 const char *name, const char *sql, int nparams, int line)
290 ret = PQprepare (plugin->dbh, name, sql, nparams, NULL);
292 check_result (plugin,
293 ret, PGRES_COMMAND_OK, "PQprepare", sql, line))
294 return GNUNET_SYSERR;
300 * @brief Get a database handle
302 * @param plugin global context
303 * @return GNUNET_OK on success, GNUNET_SYSERR on error
306 init_connection (struct Plugin *plugin)
311 /* Open database and precompile statements */
313 GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
314 "datastore-postgres",
317 plugin->dbh = PQconnectdb (conninfo == NULL ? "" : conninfo);
318 GNUNET_free_non_null (conninfo);
319 if (NULL == plugin->dbh)
321 /* FIXME: warn about out-of-memory? */
322 return GNUNET_SYSERR;
324 if (PQstatus (plugin->dbh) != CONNECTION_OK)
326 GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
327 "datastore-postgres",
328 _("Unable to initialize Postgres: %s"),
329 PQerrorMessage (plugin->dbh));
330 PQfinish (plugin->dbh);
332 return GNUNET_SYSERR;
334 ret = PQexec (plugin->dbh,
335 "CREATE TABLE gn090 ("
336 " type INTEGER NOT NULL DEFAULT 0,"
337 " prio INTEGER NOT NULL DEFAULT 0,"
338 " anonLevel INTEGER NOT NULL DEFAULT 0,"
339 " expire BIGINT NOT NULL DEFAULT 0,"
340 " hash BYTEA NOT NULL DEFAULT '',"
341 " vhash BYTEA NOT NULL DEFAULT '',"
342 " value BYTEA NOT NULL DEFAULT '')" "WITH OIDS");
343 if ( (ret == NULL) ||
344 ( (PQresultStatus (ret) != PGRES_COMMAND_OK) &&
345 (0 != strcmp ("42P07", /* duplicate table */
348 PG_DIAG_SQLSTATE)))))
350 check_result (plugin,
351 ret, PGRES_COMMAND_OK, "CREATE TABLE", "gn090", __LINE__);
352 PQfinish (plugin->dbh);
354 return GNUNET_SYSERR;
356 if (PQresultStatus (ret) == PGRES_COMMAND_OK)
359 pq_exec (plugin, "CREATE INDEX idx_hash ON gn090 (hash)", __LINE__)) ||
361 pq_exec (plugin, "CREATE INDEX idx_hash_vhash ON gn090 (hash,vhash)",
364 pq_exec (plugin, "CREATE INDEX idx_prio ON gn090 (prio)", __LINE__))
366 pq_exec (plugin, "CREATE INDEX idx_expire ON gn090 (expire)", __LINE__))
368 pq_exec (plugin, "CREATE INDEX idx_comb3 ON gn090 (prio,anonLevel)",
372 (plugin, "CREATE INDEX idx_comb4 ON gn090 (prio,hash,anonLevel)",
375 pq_exec (plugin, "CREATE INDEX idx_comb7 ON gn090 (expire,hash)",
379 PQfinish (plugin->dbh);
381 return GNUNET_SYSERR;
386 ret = PQexec (plugin->dbh,
387 "ALTER TABLE gn090 ALTER value SET STORAGE EXTERNAL");
389 check_result (plugin,
390 ret, PGRES_COMMAND_OK,
391 "ALTER TABLE", "gn090", __LINE__))
393 PQfinish (plugin->dbh);
395 return GNUNET_SYSERR;
398 ret = PQexec (plugin->dbh,
399 "ALTER TABLE gn090 ALTER hash SET STORAGE PLAIN");
401 check_result (plugin,
402 ret, PGRES_COMMAND_OK,
403 "ALTER TABLE", "gn090", __LINE__))
405 PQfinish (plugin->dbh);
407 return GNUNET_SYSERR;
410 ret = PQexec (plugin->dbh,
411 "ALTER TABLE gn090 ALTER vhash SET STORAGE PLAIN");
413 check_result (plugin,
414 ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090", __LINE__))
416 PQfinish (plugin->dbh);
418 return GNUNET_SYSERR;
425 "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
426 "WHERE hash=$1 AND vhash=$2 AND type=$3 "
427 "AND oid > $4 ORDER BY oid ASC LIMIT 1 OFFSET $5",
433 "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
434 "WHERE hash=$1 AND type=$2"
435 "AND oid > $3 ORDER BY oid ASC LIMIT 1 OFFSET $4",
441 "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
442 "WHERE hash=$1 AND vhash=$2"
443 "AND oid > $3 ORDER BY oid ASC LIMIT 1 OFFSET $4",
449 "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
451 "AND oid > $2 ORDER BY oid ASC LIMIT 1 OFFSET $3",
457 "INSERT INTO gn090 (type, prio, anonLevel, expire, hash, vhash, value) "
458 "VALUES ($1, $2, $3, $4, $5, $6, $7)",
464 "UPDATE gn090 SET prio = prio + $1, expire = CASE WHEN expire < $2 THEN $2 ELSE expire END "
470 "select_low_priority",
471 SELECT_IT_LOW_PRIORITY,
476 "select_non_anonymous",
477 SELECT_IT_NON_ANONYMOUS,
482 "select_expiration_time",
483 SELECT_IT_EXPIRATION_TIME,
488 "select_migration_order",
489 SELECT_IT_MIGRATION_ORDER,
495 "DELETE FROM gn090 " "WHERE oid=$1", 1, __LINE__)))
497 PQfinish (plugin->dbh);
499 return GNUNET_SYSERR;
506 * Delete the row identified by the given rowid (qid
509 * @param plugin global context
510 * @param rowid which row to delete
511 * @return GNUNET_OK on success
514 delete_by_rowid (struct Plugin *plugin,
517 const char *paramValues[] = { (const char *) &rowid };
518 int paramLengths[] = { sizeof (rowid) };
519 const int paramFormats[] = { 1 };
522 ret = PQexecPrepared (plugin->dbh,
524 1, paramValues, paramLengths, paramFormats, 1);
526 check_result (plugin,
527 ret, PGRES_COMMAND_OK, "PQexecPrepared", "delrow",
530 return GNUNET_SYSERR;
538 * Get an estimate of how much space the database is
541 * @param cls our "struct Plugin*"
542 * @return number of bytes used on disk
544 static unsigned long long
545 postgres_plugin_get_size (void *cls)
547 struct Plugin *plugin = cls;
548 unsigned long long total;
551 ret = PQexecParams (plugin->dbh,
552 "SELECT SUM(LENGTH(value))+256*COUNT(*) FROM gn090",
553 0, NULL, NULL, NULL, NULL, 1);
554 if (GNUNET_OK != check_result (plugin,
563 if ((PQntuples (ret) != 1) ||
564 (PQnfields (ret) != 1) ||
565 (PQgetlength (ret, 0, 0) != sizeof (unsigned long long)))
571 total = GNUNET_ntohll (*(const unsigned long long *) PQgetvalue (ret, 0, 0));
578 * Store an item in the datastore.
581 * @param key key for the item
582 * @param size number of bytes in data
583 * @param data content stored
584 * @param type type of the content
585 * @param priority priority of the content
586 * @param anonymity anonymity-level for the content
587 * @param expiration expiration time for the content
588 * @param msg set to error message
589 * @return GNUNET_OK on success
592 postgres_plugin_put (void *cls,
593 const GNUNET_HashCode * key,
596 enum GNUNET_BLOCK_Type type,
599 struct GNUNET_TIME_Absolute expiration,
602 struct Plugin *plugin = cls;
603 GNUNET_HashCode vhash;
605 uint32_t btype = htonl (type);
606 uint32_t bprio = htonl (priority);
607 uint32_t banon = htonl (anonymity);
608 uint64_t bexpi = GNUNET_TIME_absolute_hton (expiration).value__;
609 const char *paramValues[] = {
610 (const char *) &btype,
611 (const char *) &bprio,
612 (const char *) &banon,
613 (const char *) &bexpi,
615 (const char *) &vhash,
618 int paramLengths[] = {
623 sizeof (GNUNET_HashCode),
624 sizeof (GNUNET_HashCode),
627 const int paramFormats[] = { 1, 1, 1, 1, 1, 1, 1 };
629 GNUNET_CRYPTO_hash (data, size, &vhash);
630 ret = PQexecPrepared (plugin->dbh,
631 "put", 7, paramValues, paramLengths, paramFormats, 1);
632 if (GNUNET_OK != check_result (plugin, ret,
634 "PQexecPrepared", "put", __LINE__))
635 return GNUNET_SYSERR;
637 plugin->env->duc (plugin->env->cls, size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
639 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
640 "datastore-postgres",
641 "Stored %u bytes in database\n",
642 (unsigned int) size);
648 * Function invoked on behalf of a "PluginIterator"
649 * asking the database plugin to call the iterator
650 * with the next item.
652 * @param next_cls the 'struct NextRequestClosure'
653 * @param tc scheduler context
656 postgres_next_request_cont (void *next_cls,
657 const struct GNUNET_SCHEDULER_TaskContext *tc)
659 struct NextRequestClosure *nrc = next_cls;
660 struct Plugin *plugin = nrc->plugin;
661 const int paramFormats[] = { 1, 1, 1, 1, 1 };
664 enum GNUNET_BLOCK_Type type;
669 struct GNUNET_TIME_Absolute expiration_time;
672 plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
673 plugin->next_task_nc = NULL;
674 if ( (GNUNET_YES == nrc->end_it) ||
675 (nrc->count == nrc->total) )
678 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
679 "datastore-postgres",
680 "Ending iteration (%s)\n",
681 (GNUNET_YES == nrc->end_it) ? "client requested it" : "completed result set");
683 nrc->iter (nrc->iter_cls,
684 NULL, NULL, 0, NULL, 0, 0, 0,
685 GNUNET_TIME_UNIT_ZERO_ABS, 0);
691 nrc->blimit_off = GNUNET_htonll (nrc->off);
693 nrc->blimit_off = GNUNET_htonll (0);
694 if (nrc->count + nrc->off == nrc->total)
695 nrc->blast_rowid = htonl (0); /* back to start */
697 res = PQexecPrepared (plugin->dbh,
703 if (GNUNET_OK != check_result (plugin,
711 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
712 "datastore-postgres",
713 "Ending iteration (postgres error)\n");
715 nrc->iter (nrc->iter_cls,
716 NULL, NULL, 0, NULL, 0, 0, 0,
717 GNUNET_TIME_UNIT_ZERO_ABS, 0);
722 if (0 == PQntuples (res))
726 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
727 "datastore-postgres",
728 "Ending iteration (no more results)\n");
730 nrc->iter (nrc->iter_cls,
731 NULL, NULL, 0, NULL, 0, 0, 0,
732 GNUNET_TIME_UNIT_ZERO_ABS, 0);
737 if ((1 != PQntuples (res)) ||
738 (7 != PQnfields (res)) ||
739 (sizeof (uint32_t) != PQfsize (res, 0)) ||
740 (sizeof (uint32_t) != PQfsize (res, 6)))
743 nrc->iter (nrc->iter_cls,
744 NULL, NULL, 0, NULL, 0, 0, 0,
745 GNUNET_TIME_UNIT_ZERO_ABS, 0);
750 rowid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 6));
751 if ((sizeof (uint32_t) != PQfsize (res, 0)) ||
752 (sizeof (uint32_t) != PQfsize (res, 1)) ||
753 (sizeof (uint32_t) != PQfsize (res, 2)) ||
754 (sizeof (uint64_t) != PQfsize (res, 3)) ||
755 (sizeof (GNUNET_HashCode) != PQgetlength (res, 0, 4)) )
759 delete_by_rowid (plugin, rowid);
760 nrc->iter (nrc->iter_cls,
761 NULL, NULL, 0, NULL, 0, 0, 0,
762 GNUNET_TIME_UNIT_ZERO_ABS, 0);
767 type = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
768 priority = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
769 anonymity = ntohl ( *(uint32_t *) PQgetvalue (res, 0, 2));
770 expiration_time.value = GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, 0, 3));
771 memcpy (&key, PQgetvalue (res, 0, 4), sizeof (GNUNET_HashCode));
772 size = PQgetlength (res, 0, 5);
774 nrc->blast_prio = htonl (priority);
775 nrc->blast_expire = GNUNET_htonll (expiration_time.value);
776 nrc->blast_rowid = htonl (rowid);
780 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
781 "datastore-postgres",
782 "Found result of size %u bytes and type %u in database\n",
784 (unsigned int) type);
786 iret = nrc->iter (nrc->iter_cls,
790 PQgetvalue (res, 0, 5),
791 (enum GNUNET_BLOCK_Type) type,
797 if (iret == GNUNET_SYSERR)
800 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
801 "datastore-postgres",
802 "Ending iteration (client error)\n");
806 if (iret == GNUNET_NO)
808 if (GNUNET_OK == delete_by_rowid (plugin, rowid))
811 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
812 "datastore-postgres",
813 "Deleting %u bytes from database\n",
814 (unsigned int) size);
816 plugin->env->duc (plugin->env->cls,
817 - (size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
819 GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
820 "datastore-postgres",
821 "Deleted %u bytes from database\n",
822 (unsigned int) size);
830 * Function invoked on behalf of a "PluginIterator"
831 * asking the database plugin to call the iterator
832 * with the next item.
834 * @param next_cls whatever argument was given
835 * to the PluginIterator as "next_cls".
836 * @param end_it set to GNUNET_YES if we
837 * should terminate the iteration early
838 * (iterator should be still called once more
839 * to signal the end of the iteration).
842 postgres_plugin_next_request (void *next_cls,
845 struct NextRequestClosure *nrc = next_cls;
847 if (GNUNET_YES == end_it)
848 nrc->end_it = GNUNET_YES;
849 nrc->plugin->next_task_nc = nrc;
850 nrc->plugin->next_task = GNUNET_SCHEDULER_add_now (nrc->plugin->env->sched,
851 &postgres_next_request_cont,
857 * Update the priority for a particular key in the datastore. If
858 * the expiration time in value is different than the time found in
859 * the datastore, the higher value should be kept. For the
860 * anonymity level, the lower value is to be used. The specified
861 * priority should be added to the existing priority, ignoring the
864 * Note that it is possible for multiple values to match this put.
865 * In that case, all of the respective values are updated.
867 * @param cls our "struct Plugin*"
868 * @param uid unique identifier of the datum
869 * @param delta by how much should the priority
870 * change? If priority + delta < 0 the
871 * priority should be set to 0 (never go
873 * @param expire new expiration time should be the
874 * MAX of any existing expiration time and
876 * @param msg set to error message
877 * @return GNUNET_OK on success
880 postgres_plugin_update (void *cls,
882 int delta, struct GNUNET_TIME_Absolute expire,
885 struct Plugin *plugin = cls;
887 int32_t bdelta = (int32_t) htonl ((uint32_t) delta);
888 uint32_t boid = htonl ( (uint32_t) uid);
889 uint64_t bexpire = GNUNET_TIME_absolute_hton (expire).value__;
890 const char *paramValues[] = {
891 (const char *) &bdelta,
892 (const char *) &bexpire,
893 (const char *) &boid,
895 int paramLengths[] = {
900 const int paramFormats[] = { 1, 1, 1 };
902 ret = PQexecPrepared (plugin->dbh,
904 3, paramValues, paramLengths, paramFormats, 1);
905 if (GNUNET_OK != check_result (plugin,
908 "PQexecPrepared", "update", __LINE__))
909 return GNUNET_SYSERR;
916 * Call a method for each key in the database and
917 * call the callback method on it.
919 * @param plugin global context
920 * @param type entries of which type should be considered?
921 * @param is_asc ascending or descending iteration?
922 * @param iter_select which SELECT method should be used?
923 * @param iter maybe NULL (to just count); iter
924 * should return GNUNET_SYSERR to abort the
925 * iteration, GNUNET_NO to delete the entry and
926 * continue and GNUNET_OK to continue iterating
927 * @param iter_cls closure for 'iter'
930 postgres_iterate (struct Plugin *plugin,
933 unsigned int iter_select,
934 PluginIterator iter, void *iter_cls)
936 struct NextRequestClosure *nrc;
938 nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
939 nrc->count = UINT32_MAX;
940 nrc->plugin = plugin;
942 nrc->iter_cls = iter_cls;
945 nrc->blast_prio = htonl (0);
946 nrc->blast_rowid = htonl (0);
947 nrc->blast_expire = htonl (0);
951 nrc->blast_prio = htonl (0x7FFFFFFFL);
952 nrc->blast_rowid = htonl (0xFFFFFFFF);
953 nrc->blast_expire = GNUNET_htonll (0x7FFFFFFFFFFFFFFFLL);
958 nrc->pname = "select_low_priority";
960 nrc->paramValues[0] = (const char *) &nrc->blast_prio;
961 nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
962 nrc->paramLengths[0] = sizeof (nrc->blast_prio);
963 nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
966 nrc->pname = "select_non_anonymous";
968 nrc->paramValues[0] = (const char *) &nrc->blast_prio;
969 nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
970 nrc->paramLengths[0] = sizeof (nrc->blast_prio);
971 nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
974 nrc->pname = "select_expiration_time";
976 nrc->paramValues[0] = (const char *) &nrc->blast_expire;
977 nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
978 nrc->paramLengths[0] = sizeof (nrc->blast_expire);
979 nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
982 nrc->pname = "select_migration_order";
984 nrc->paramValues[0] = (const char *) &nrc->blast_expire;
985 nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
986 nrc->paramValues[2] = (const char *) &nrc->bnow;
987 nrc->paramLengths[0] = sizeof (nrc->blast_expire);
988 nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
989 nrc->paramLengths[2] = sizeof (nrc->bnow);
994 NULL, NULL, 0, NULL, 0, 0, 0,
995 GNUNET_TIME_UNIT_ZERO_ABS, 0);
999 nrc->bnow = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ()).value__;
1000 postgres_plugin_next_request (nrc,
1006 * Select a subset of the items in the datastore and call
1007 * the given iterator for each of them.
1009 * @param cls our "struct Plugin*"
1010 * @param type entries of which type should be considered?
1011 * Use 0 for any type.
1012 * @param iter function to call on each matching value;
1013 * will be called once with a NULL value at the end
1014 * @param iter_cls closure for iter
1017 postgres_plugin_iter_low_priority (void *cls,
1018 enum GNUNET_BLOCK_Type type,
1019 PluginIterator iter,
1022 struct Plugin *plugin = cls;
1024 postgres_iterate (plugin,
1034 * Iterate over the results for a particular key
1037 * @param cls closure
1038 * @param key maybe NULL (to match all entries)
1039 * @param vhash hash of the value, maybe NULL (to
1040 * match all values that have the right key).
1041 * Note that for DBlocks there is no difference
1042 * betwen key and vhash, but for other blocks
1044 * @param type entries of which type are relevant?
1045 * Use 0 for any type.
1046 * @param iter function to call on each matching value;
1047 * will be called once with a NULL value at the end
1048 * @param iter_cls closure for iter
1051 postgres_plugin_get (void *cls,
1052 const GNUNET_HashCode * key,
1053 const GNUNET_HashCode * vhash,
1054 enum GNUNET_BLOCK_Type type,
1055 PluginIterator iter, void *iter_cls)
1057 struct Plugin *plugin = cls;
1058 struct NextRequestClosure *nrc;
1059 const int paramFormats[] = { 1, 1, 1, 1, 1 };
1064 postgres_plugin_iter_low_priority (plugin, type,
1068 nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
1069 nrc->plugin = plugin;
1071 nrc->iter_cls = iter_cls;
1074 nrc->vhash = *vhash;
1075 nrc->paramValues[0] = (const char*) &nrc->key;
1076 nrc->paramLengths[0] = sizeof (GNUNET_HashCode);
1077 nrc->btype = htonl (type);
1082 nrc->paramValues[1] = (const char *) &nrc->vhash;
1083 nrc->paramLengths[1] = sizeof (nrc->vhash);
1084 nrc->paramValues[2] = (const char *) &nrc->btype;
1085 nrc->paramLengths[2] = sizeof (nrc->btype);
1086 nrc->paramValues[3] = (const char *) &nrc->blast_rowid;
1087 nrc->paramLengths[3] = sizeof (nrc->blast_rowid);
1088 nrc->paramValues[4] = (const char *) &nrc->blimit_off;
1089 nrc->paramLengths[4] = sizeof (nrc->blimit_off);
1091 nrc->pname = "getvt";
1092 ret = PQexecParams (plugin->dbh,
1093 "SELECT count(*) FROM gn090 WHERE hash=$1 AND vhash=$2 AND type=$3",
1102 nrc->paramValues[1] = (const char *) &nrc->btype;
1103 nrc->paramLengths[1] = sizeof (nrc->btype);
1104 nrc->paramValues[2] = (const char *) &nrc->blast_rowid;
1105 nrc->paramLengths[2] = sizeof (nrc->blast_rowid);
1106 nrc->paramValues[3] = (const char *) &nrc->blimit_off;
1107 nrc->paramLengths[3] = sizeof (nrc->blimit_off);
1109 nrc->pname = "gett";
1110 ret = PQexecParams (plugin->dbh,
1111 "SELECT count(*) FROM gn090 WHERE hash=$1 AND type=$2",
1123 nrc->paramValues[1] = (const char *) &nrc->vhash;
1124 nrc->paramLengths[1] = sizeof (nrc->vhash);
1125 nrc->paramValues[2] = (const char *) &nrc->blast_rowid;
1126 nrc->paramLengths[2] = sizeof (nrc->blast_rowid);
1127 nrc->paramValues[3] = (const char *) &nrc->blimit_off;
1128 nrc->paramLengths[3] = sizeof (nrc->blimit_off);
1130 nrc->pname = "getv";
1131 ret = PQexecParams (plugin->dbh,
1132 "SELECT count(*) FROM gn090 WHERE hash=$1 AND vhash=$2",
1141 nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
1142 nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
1143 nrc->paramValues[2] = (const char *) &nrc->blimit_off;
1144 nrc->paramLengths[2] = sizeof (nrc->blimit_off);
1147 ret = PQexecParams (plugin->dbh,
1148 "SELECT count(*) FROM gn090 WHERE hash=$1",
1156 if (GNUNET_OK != check_result (plugin,
1164 NULL, NULL, 0, NULL, 0, 0, 0,
1165 GNUNET_TIME_UNIT_ZERO_ABS, 0);
1169 if ((PQntuples (ret) != 1) ||
1170 (PQnfields (ret) != 1) ||
1171 (PQgetlength (ret, 0, 0) != sizeof (unsigned long long)))
1176 NULL, NULL, 0, NULL, 0, 0, 0,
1177 GNUNET_TIME_UNIT_ZERO_ABS, 0);
1181 nrc->total = GNUNET_ntohll (*(const unsigned long long *) PQgetvalue (ret, 0, 0));
1183 if (nrc->total == 0)
1186 NULL, NULL, 0, NULL, 0, 0, 0,
1187 GNUNET_TIME_UNIT_ZERO_ABS, 0);
1191 nrc->off = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1193 postgres_plugin_next_request (nrc,
1199 * Select a subset of the items in the datastore and call
1200 * the given iterator for each of them.
1202 * @param cls our "struct Plugin*"
1203 * @param type entries of which type should be considered?
1204 * Use 0 for any type.
1205 * @param iter function to call on each matching value;
1206 * will be called once with a NULL value at the end
1207 * @param iter_cls closure for iter
1210 postgres_plugin_iter_zero_anonymity (void *cls,
1211 enum GNUNET_BLOCK_Type type,
1212 PluginIterator iter,
1215 struct Plugin *plugin = cls;
1217 postgres_iterate (plugin,
1224 * Select a subset of the items in the datastore and call
1225 * the given iterator for each of them.
1227 * @param cls our "struct Plugin*"
1228 * @param type entries of which type should be considered?
1229 * Use 0 for any type.
1230 * @param iter function to call on each matching value;
1231 * will be called once with a NULL value at the end
1232 * @param iter_cls closure for iter
1235 postgres_plugin_iter_ascending_expiration (void *cls,
1236 enum GNUNET_BLOCK_Type type,
1237 PluginIterator iter,
1240 struct Plugin *plugin = cls;
1242 postgres_iterate (plugin, type, GNUNET_YES, 2,
1248 * Select a subset of the items in the datastore and call
1249 * the given iterator for each of them.
1251 * @param cls our "struct Plugin*"
1252 * @param type entries of which type should be considered?
1253 * Use 0 for any type.
1254 * @param iter function to call on each matching value;
1255 * will be called once with a NULL value at the end
1256 * @param iter_cls closure for iter
1259 postgres_plugin_iter_migration_order (void *cls,
1260 enum GNUNET_BLOCK_Type type,
1261 PluginIterator iter,
1264 struct Plugin *plugin = cls;
1266 postgres_iterate (plugin, 0, GNUNET_NO, 3,
1272 * Select a subset of the items in the datastore and call
1273 * the given iterator for each of them.
1275 * @param cls our "struct Plugin*"
1276 * @param type entries of which type should be considered?
1277 * Use 0 for any type.
1278 * @param iter function to call on each matching value;
1279 * will be called once with a NULL value at the end
1280 * @param iter_cls closure for iter
1283 postgres_plugin_iter_all_now (void *cls,
1284 enum GNUNET_BLOCK_Type type,
1285 PluginIterator iter,
1288 struct Plugin *plugin = cls;
1290 postgres_iterate (plugin,
1300 postgres_plugin_drop (void *cls)
1302 struct Plugin *plugin = cls;
1304 pq_exec (plugin, "DROP TABLE gn090", __LINE__);
1309 * Entry point for the plugin.
1311 * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
1312 * @return our "struct Plugin*"
1315 libgnunet_plugin_datastore_postgres_init (void *cls)
1317 struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
1318 struct GNUNET_DATASTORE_PluginFunctions *api;
1319 struct Plugin *plugin;
1321 plugin = GNUNET_malloc (sizeof (struct Plugin));
1323 if (GNUNET_OK != init_connection (plugin))
1325 GNUNET_free (plugin);
1328 api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
1330 api->get_size = &postgres_plugin_get_size;
1331 api->put = &postgres_plugin_put;
1332 api->next_request = &postgres_plugin_next_request;
1333 api->get = &postgres_plugin_get;
1334 api->update = &postgres_plugin_update;
1335 api->iter_low_priority = &postgres_plugin_iter_low_priority;
1336 api->iter_zero_anonymity = &postgres_plugin_iter_zero_anonymity;
1337 api->iter_ascending_expiration = &postgres_plugin_iter_ascending_expiration;
1338 api->iter_migration_order = &postgres_plugin_iter_migration_order;
1339 api->iter_all_now = &postgres_plugin_iter_all_now;
1340 api->drop = &postgres_plugin_drop;
1341 GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
1342 "datastore-postgres",
1343 _("Postgres database running\n"));
1349 * Exit point from the plugin.
1350 * @param cls our "struct Plugin*"
1351 * @return always NULL
1354 libgnunet_plugin_datastore_postgres_done (void *cls)
1356 struct GNUNET_DATASTORE_PluginFunctions *api = cls;
1357 struct Plugin *plugin = api->cls;
1359 if (plugin->next_task != GNUNET_SCHEDULER_NO_TASK)
1361 GNUNET_SCHEDULER_cancel (plugin->env->sched,
1363 plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
1364 GNUNET_free (plugin->next_task_nc);
1365 plugin->next_task_nc = NULL;
1367 PQfinish (plugin->dbh);
1368 GNUNET_free (plugin);
1373 /* end of plugin_datastore_postgres.c */