From 0266efed64df149fd11a73e575be18cf03f36d25 Mon Sep 17 00:00:00 2001 From: Christian Grothoff Date: Fri, 2 Jun 2017 23:12:31 +0200 Subject: [PATCH] more work on pq-ification of datacache --- src/datacache/Makefile.am | 1 + src/datacache/plugin_datacache_postgres.c | 463 ++++++++++------------ 2 files changed, 221 insertions(+), 243 deletions(-) diff --git a/src/datacache/Makefile.am b/src/datacache/Makefile.am index 670a64926..602b1d437 100644 --- a/src/datacache/Makefile.am +++ b/src/datacache/Makefile.am @@ -72,6 +72,7 @@ libgnunet_plugin_datacache_postgres_la_SOURCES = \ plugin_datacache_postgres.c libgnunet_plugin_datacache_postgres_la_LIBADD = \ $(top_builddir)/src/postgres/libgnunetpostgres.la \ + $(top_builddir)/src/pq/libgnunetpq.la \ $(top_builddir)/src/statistics/libgnunetstatistics.la \ $(top_builddir)/src/util/libgnunetutil.la \ $(GN_PLUGIN_LDFLAGS) -lpq diff --git a/src/datacache/plugin_datacache_postgres.c b/src/datacache/plugin_datacache_postgres.c index 8f5cdbde1..e69f2b2c6 100644 --- a/src/datacache/plugin_datacache_postgres.c +++ b/src/datacache/plugin_datacache_postgres.c @@ -26,6 +26,7 @@ #include "platform.h" #include "gnunet_util_lib.h" #include "gnunet_postgres_lib.h" +#include "gnunet_pq_lib.h" #include "gnunet_datacache_plugin.h" #define LOG(kind,...) GNUNET_log_from (kind, "datacache-postgres", __VA_ARGS__) @@ -66,115 +67,67 @@ struct Plugin static int init_connection (struct Plugin *plugin) { - PGresult *ret; + struct GNUNET_PQ_ExecuteStatement es[] = { + GNUNET_PQ_make_execute ("CREATE TEMPORARY TABLE IF NOT EXISTS gn090dc (" + " type INTEGER NOT NULL DEFAULT 0," + " discard_time BIGINT NOT NULL DEFAULT 0," + " key BYTEA NOT NULL DEFAULT ''," + " value BYTEA NOT NULL DEFAULT ''," + " path BYTEA DEFAULT '')" + "WITH OIDS"), + GNUNET_PQ_make_try_execute ("CREATE INDEX IF NOT EXISTS idx_key ON gn090dc (key)"), + GNUNET_PQ_make_try_execute ("CREATE INDEX IF NOT EXISTS idx_dt ON gn090dc (discard_time)"), + GNUNET_PQ_make_execute ("ALTER TABLE gn090dc ALTER value SET STORAGE EXTERNAL"), + GNUNET_PQ_make_execute ("ALTER TABLE gn090dc ALTER key SET STORAGE PLAIN"), + GNUNET_PQ_EXECUTE_STATEMENT_END + }; + struct GNUNET_PQ_PreparedStatement ps[] = { + GNUNET_PQ_make_prepare ("getkt", + "SELECT discard_time,type,value,path FROM gn090dc " + "WHERE key=$1 AND type=$2", + 2), + GNUNET_PQ_make_prepare ("getk", + "SELECT discard_time,type,value,path FROM gn090dc " + "WHERE key=$1", + 1), + GNUNET_PQ_make_prepare ("getm", + "SELECT length(value) AS len,oid,key FROM gn090dc " + "ORDER BY discard_time ASC LIMIT 1", + 0), + GNUNET_PQ_make_prepare ("get_random", + "SELECT discard_time,type,value,path,key FROM gn090dc " + "ORDER BY key ASC LIMIT 1 OFFSET $1", + 1), + GNUNET_PQ_make_prepare ("get_closest", + "SELECT discard_time,type,value,path,key FROM gn090dc " + "WHERE key>=$1 ORDER BY key ASC LIMIT $2", + 1), + GNUNET_PQ_make_prepare ("delrow", + "DELETE FROM gn090dc WHERE oid=$1", + 1), + GNUNET_PQ_make_prepare ("put", + "INSERT INTO gn090dc (type, discard_time, key, value, path) " + "VALUES ($1, $2, $3, $4, $5)", + 5), + GNUNET_PQ_PREPARED_STATEMENT_END + }; plugin->dbh = GNUNET_PQ_connect_with_cfg (plugin->env->cfg, "datacache-postgres"); if (NULL == plugin->dbh) return GNUNET_SYSERR; - ret = - PQexec (plugin->dbh, - "CREATE TEMPORARY TABLE IF NOT EXISTS gn090dc (" - " type INTEGER NOT NULL DEFAULT 0," - " discard_time BIGINT NOT NULL DEFAULT 0," - " key BYTEA NOT NULL DEFAULT ''," - " value BYTEA NOT NULL DEFAULT ''," - " path BYTEA DEFAULT '')" - "WITH OIDS"); - if ( (ret == NULL) || - ((PQresultStatus (ret) != PGRES_COMMAND_OK) && - (0 != strcmp ("42P07", /* duplicate table */ - PQresultErrorField - (ret, - PG_DIAG_SQLSTATE))))) - { - (void) GNUNET_POSTGRES_check_result (plugin->dbh, ret, - PGRES_COMMAND_OK, - "CREATE TABLE", - "gn090dc"); - PQfinish (plugin->dbh); - plugin->dbh = NULL; - return GNUNET_SYSERR; - } - if (PQresultStatus (ret) == PGRES_COMMAND_OK) - { - if ((GNUNET_OK != - GNUNET_POSTGRES_exec (plugin->dbh, - "CREATE INDEX IF NOT EXISTS idx_key ON gn090dc (key)")) || - (GNUNET_OK != - GNUNET_POSTGRES_exec (plugin->dbh, - "CREATE INDEX IF NOT EXISTS idx_dt ON gn090dc (discard_time)"))) - { - PQclear (ret); - PQfinish (plugin->dbh); - plugin->dbh = NULL; - return GNUNET_SYSERR; - } - } - PQclear (ret); - ret = - PQexec (plugin->dbh, - "ALTER TABLE gn090dc ALTER value SET STORAGE EXTERNAL"); if (GNUNET_OK != - GNUNET_POSTGRES_check_result (plugin->dbh, - ret, - PGRES_COMMAND_OK, - "ALTER TABLE", - "gn090dc")) + GNUNET_PQ_exec_statements (plugin->dbh, + es)) { PQfinish (plugin->dbh); plugin->dbh = NULL; return GNUNET_SYSERR; } - PQclear (ret); - ret = PQexec (plugin->dbh, - "ALTER TABLE gn090dc ALTER key SET STORAGE PLAIN"); + if (GNUNET_OK != - GNUNET_POSTGRES_check_result (plugin->dbh, - ret, - PGRES_COMMAND_OK, - "ALTER TABLE", - "gn090dc")) - { - PQfinish (plugin->dbh); - plugin->dbh = NULL; - return GNUNET_SYSERR; - } - PQclear (ret); - if ((GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "getkt", - "SELECT discard_time,type,value,path FROM gn090dc " - "WHERE key=$1 AND type=$2 ", 2)) || - (GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "getk", - "SELECT discard_time,type,value,path FROM gn090dc " - "WHERE key=$1", 1)) || - (GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "getm", - "SELECT length(value),oid,key FROM gn090dc " - "ORDER BY discard_time ASC LIMIT 1", 0)) || - (GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "get_random", - "SELECT discard_time,type,value,path,key FROM gn090dc " - "ORDER BY key ASC LIMIT 1 OFFSET $1", 1)) || - (GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "get_closest", - "SELECT discard_time,type,value,path,key FROM gn090dc " - "WHERE key>=$1 ORDER BY key ASC LIMIT $2", 1)) || - (GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "delrow", - "DELETE FROM gn090dc WHERE oid=$1", 1)) || - (GNUNET_OK != - GNUNET_POSTGRES_prepare (plugin->dbh, - "put", - "INSERT INTO gn090dc (type, discard_time, key, value, path) " - "VALUES ($1, $2, $3, $4, $5)", 5))) + GNUNET_PQ_prepare_statements (plugin->dbh, + ps)) { PQfinish (plugin->dbh); plugin->dbh = NULL; @@ -189,7 +142,7 @@ init_connection (struct Plugin *plugin) * * @param cls closure (our `struct Plugin`) * @param key key to store @a data under - * @param size number of bytes in @a data + * @param data_size number of bytes in @a data * @param data data to store * @param type type of the value * @param discard_time when to discard the value in any case @@ -200,7 +153,7 @@ init_connection (struct Plugin *plugin) static ssize_t postgres_plugin_put (void *cls, const struct GNUNET_HashCode *key, - size_t size, + size_t data_size, const char *data, enum GNUNET_BLOCK_Type type, struct GNUNET_TIME_Absolute discard_time, @@ -208,36 +161,125 @@ postgres_plugin_put (void *cls, const struct GNUNET_PeerIdentity *path_info) { struct Plugin *plugin = cls; - PGresult *ret; - uint32_t btype = htonl (type); - uint64_t bexpi = GNUNET_TIME_absolute_hton (discard_time).abs_value_us__; - - const char *paramValues[] = { - (const char *) &btype, - (const char *) &bexpi, - (const char *) key, - (const char *) data, - (const char *) path_info - }; - int paramLengths[] = { - sizeof (btype), - sizeof (bexpi), - sizeof (struct GNUNET_HashCode), - size, - path_info_len * sizeof (struct GNUNET_PeerIdentity) + uint32_t type32 = (uint32_t) type; + struct GNUNET_PQ_QueryParam params[] = { + GNUNET_PQ_query_param_uint32 (&type32), + GNUNET_PQ_query_param_absolute_time (&discard_time), + GNUNET_PQ_query_param_auto_from_type (key), + GNUNET_PQ_query_param_fixed_size (data, data_size), + GNUNET_PQ_query_param_fixed_size (path_info, + path_info_len * sizeof (struct GNUNET_PeerIdentity)), + GNUNET_PQ_query_param_end }; - const int paramFormats[] = { 1, 1, 1, 1, 1 }; + enum GNUNET_PQ_QueryStatus ret; - ret = - PQexecPrepared (plugin->dbh, "put", 5, paramValues, paramLengths, - paramFormats, 1); - if (GNUNET_OK != - GNUNET_POSTGRES_check_result (plugin->dbh, ret, - PGRES_COMMAND_OK, "PQexecPrepared", "put")) + ret = GNUNET_PQ_eval_prepared_non_select (plugin->dbh, + "put", + params); + if (0 > ret) return -1; plugin->num_items++; - PQclear (ret); - return size + OVERHEAD; + return data_size + OVERHEAD; +} + + +/** + * Closure for #handle_results. + */ +struct HandleResultContext +{ + + /** + * Function to call on each result, may be NULL. + */ + GNUNET_DATACACHE_Iterator iter; + + /** + * Closure for @e iter. + */ + void *iter_cls; + + /** + * Key used. + */ + const struct GNUNET_HashCode *key; +}; + + +/** + * Function to be called with the results of a SELECT statement + * that has returned @a num_results results. Parse the result + * and call the callback given in @a cls + * + * @param cls closure of type `struct HandleResultContext` + * @param result the postgres result + * @param num_result the number of results in @a result + */ +static void +handle_results (void *cls, + PGresult *result, + unsigned int num_results) +{ + struct HandleResultContext *hrc = cls; + + for (unsigned int i=0;iiter) && + (GNUNET_SYSERR == + hrc->iter (hrc->iter_cls, + hrc->key, + data_size, + data, + (enum GNUNET_BLOCK_Type) type, + expiration_time, + path_len, + path)) ) + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Ending iteration (client error)\n"); + GNUNET_PQ_cleanup_result (rs); + return; + } + GNUNET_PQ_cleanup_result (rs); + } } @@ -260,94 +302,30 @@ postgres_plugin_get (void *cls, void *iter_cls) { struct Plugin *plugin = cls; - uint32_t btype = htonl (type); - - const char *paramValues[] = { - (const char *) key, - (const char *) &btype + uint32_t type32 = (uint32_t) type; + struct GNUNET_PQ_QueryParam paramk[] = { + GNUNET_PQ_query_param_auto_from_type (key), + GNUNET_PQ_query_param_end }; - int paramLengths[] = { - sizeof (struct GNUNET_HashCode), - sizeof (btype) + struct GNUNET_PQ_QueryParam paramkt[] = { + GNUNET_PQ_query_param_auto_from_type (key), + GNUNET_PQ_query_param_uint32 (&type32), + GNUNET_PQ_query_param_end }; - const int paramFormats[] = { 1, 1 }; - struct GNUNET_TIME_Absolute expiration_time; - uint32_t size; - unsigned int cnt; - unsigned int i; - unsigned int path_len; - const struct GNUNET_PeerIdentity *path; - PGresult *res; - - res = - PQexecPrepared (plugin->dbh, (type == 0) ? "getk" : "getkt", - (type == 0) ? 1 : 2, paramValues, paramLengths, - paramFormats, 1); - if (GNUNET_OK != - GNUNET_POSTGRES_check_result (plugin->dbh, - res, - PGRES_TUPLES_OK, - "PQexecPrepared", - (type == 0) ? "getk" : "getkt")) - { - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Ending iteration (postgres error)\n"); - return 0; - } - - if (0 == (cnt = PQntuples (res))) - { - /* no result */ - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Ending iteration (no more results)\n"); - PQclear (res); - return 0; - } - if (iter == NULL) - { - PQclear (res); - return cnt; - } - if ( (4 != PQnfields (res)) || - (sizeof (uint64_t) != PQfsize (res, 0)) || - (sizeof (uint32_t) != PQfsize (res, 1))) - { - GNUNET_break (0); - PQclear (res); + enum GNUNET_PQ_QueryStatus res; + struct HandleResultContext hr_ctx; + + hr_ctx.iter = iter; + hr_ctx.iter_cls = iter_cls; + hr_ctx.key = key; + res = GNUNET_PQ_eval_prepared_multi_select (plugin->dbh, + (0 == type) ? "getk" : "getkt", + (0 == type) ? paramk : paramkt, + &handle_results, + &hr_ctx); + if (res < 0) return 0; - } - for (i = 0; i < cnt; i++) - { - expiration_time.abs_value_us = - GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 0)); - type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1)); - size = PQgetlength (res, i, 2); - path_len = PQgetlength (res, i, 3); - if (0 != (path_len % sizeof (struct GNUNET_PeerIdentity))) - { - GNUNET_break (0); - path_len = 0; - } - path_len %= sizeof (struct GNUNET_PeerIdentity); - path = (const struct GNUNET_PeerIdentity *) PQgetvalue (res, i, 3); - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Found result of size %u bytes and type %u in database\n", - (unsigned int) size, (unsigned int) type); - if (GNUNET_SYSERR == - iter (iter_cls, key, size, PQgetvalue (res, i, 2), - (enum GNUNET_BLOCK_Type) type, - expiration_time, - path_len, - path)) - { - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Ending iteration (client error)\n"); - PQclear (res); - return cnt; - } - } - PQclear (res); - return cnt; + return res; } @@ -362,54 +340,53 @@ static int postgres_plugin_del (void *cls) { struct Plugin *plugin = cls; + struct GNUNET_PQ_QueryParam pempty[] = { + GNUNET_PQ_query_param_end + }; uint32_t size; uint32_t oid; struct GNUNET_HashCode key; - PGresult *res; + struct GNUNET_PQ_ResultSpec rs[] = { + GNUNET_PQ_result_spec_uint32 ("len", + &size), + GNUNET_PQ_result_spec_uint32 ("oid", + &oid), + GNUNET_PQ_result_spec_auto_from_type ("key", + &key), + GNUNET_PQ_result_spec_end + }; + enum GNUNET_PQ_QueryStatus res; + struct GNUNET_PQ_QueryParam dparam[] = { + GNUNET_PQ_query_param_uint32 (&oid), + GNUNET_PQ_query_param_end + }; - res = PQexecPrepared (plugin->dbh, - "getm", - 0, NULL, NULL, NULL, 1); - if (GNUNET_OK != - GNUNET_POSTGRES_check_result (plugin->dbh, - res, - PGRES_TUPLES_OK, - "PQexecPrepared", - "getm")) - { - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Ending iteration (postgres error)\n"); - return 0; - } - if (0 == PQntuples (res)) + res = GNUNET_PQ_eval_prepared_singleton_select (plugin->dbh, + "getm", + pempty, + rs); + if (0 > res) + return GNUNET_SYSERR; + if (GNUNET_PQ_STATUS_SUCCESS_NO_RESULTS == res) { /* no result */ LOG (GNUNET_ERROR_TYPE_DEBUG, "Ending iteration (no more results)\n"); - PQclear (res); - return GNUNET_SYSERR; - } - if ((3 != PQnfields (res)) || (sizeof (size) != PQfsize (res, 0)) || - (sizeof (oid) != PQfsize (res, 1)) || - (sizeof (struct GNUNET_HashCode) != PQgetlength (res, 0, 2))) - { - GNUNET_break (0); - PQclear (res); return 0; } - size = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0)); - oid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1)); - GNUNET_memcpy (&key, PQgetvalue (res, 0, 2), sizeof (struct GNUNET_HashCode)); - PQclear (res); - if (GNUNET_OK != - GNUNET_POSTGRES_delete_by_rowid (plugin->dbh, - "delrow", - oid)) + res = GNUNET_PQ_eval_prepared_non_select (plugin->dbh, + "delrow", + dparam); + if (0 > res) + { + GNUNET_PQ_cleanup_result (rs); return GNUNET_SYSERR; + } plugin->num_items--; plugin->env->delete_notify (plugin->env->cls, &key, size + OVERHEAD); + GNUNET_PQ_cleanup_result (rs); return GNUNET_OK; } -- 2.25.1