commit bc381df64bab015ea1bfc91315b04656a41ea69d
parent cd1e543ce198c60b058758f93767dd8108c16410
Author: Florian Dold <dold@taler.net>
Date: Tue, 1 Sep 2026 19:51:32 +0200
pq: poll queued events after synchronous queries
Diffstat:
5 files changed, 147 insertions(+), 24 deletions(-)
diff --git a/src/lib/pq/pq.c b/src/lib/pq/pq.c
@@ -114,6 +114,7 @@ GNUNET_PQ_exec_prepared (struct GNUNET_PQ_Context *db,
GNUNET_PQ_reconnect_ (db);
res = NULL;
}
+ GNUNET_PQ_event_poll_ (db);
for (off = 0; off < soff; off++)
GNUNET_free (scratch[off]);
diff --git a/src/lib/pq/pq.h b/src/lib/pq/pq.h
@@ -183,4 +183,14 @@ GNUNET_PQ_event_reconnect_ (struct GNUNET_PQ_Context *db,
int fd);
+/**
+ * Schedule polling for notifications that libpq may have queued while
+ * executing a synchronous query.
+ *
+ * @param db the DB handle
+ */
+void
+GNUNET_PQ_event_poll_ (struct GNUNET_PQ_Context *db);
+
+
#endif
diff --git a/src/lib/pq/pq_event.c b/src/lib/pq/pq_event.c
@@ -324,6 +324,18 @@ do_poll (void *cls)
}
+void
+GNUNET_PQ_event_poll_ (struct GNUNET_PQ_Context *db)
+{
+ if ( (NULL == db->conn) ||
+ (0 == GNUNET_CONTAINER_multishortmap_size (db->channel_map)) ||
+ (NULL != db->poller_task) )
+ return;
+ db->poller_task = GNUNET_SCHEDULER_add_now (&do_poll,
+ db);
+}
+
+
/**
* Function called when the Postgres FD changes and we need
* to update the scheduler event loop task.
@@ -604,12 +616,7 @@ GNUNET_PQ_event_notify (struct GNUNET_PQ_Context *db,
Just waiting for the db socket to be readable won't work,
as postgres only queues notifications we triggered for
ourselves in an internal data structure. */
- if (NULL == db->poller_task)
- {
- db->poller_task
- = GNUNET_SCHEDULER_add_now (&do_poll,
- db);
- }
+ GNUNET_PQ_event_poll_ (db);
}
diff --git a/src/lib/pq/pq_exec.c b/src/lib/pq/pq_exec.c
@@ -84,9 +84,11 @@ GNUNET_PQ_exec_statements (struct GNUNET_PQ_Context *db,
PQresStatus (PQresultStatus (result)),
PQerrorMessage (db->conn));
PQclear (result);
+ GNUNET_PQ_event_poll_ (db);
return GNUNET_SYSERR;
}
PQclear (result);
+ GNUNET_PQ_event_poll_ (db);
}
return GNUNET_OK;
}
diff --git a/src/lib/pq/test_pq.c b/src/lib/pq/test_pq.c
@@ -27,6 +27,7 @@
#include "gnunet_pq_lib.h"
#include "gnunet_time_lib.h"
#include "pq.h"
+#include <pthread.h>
/**
* Database handle.
@@ -48,6 +49,16 @@ static struct GNUNET_DB_EventHandler *eh;
*/
static struct GNUNET_SCHEDULER_Task *tt;
+/**
+ * Stage of the event notification test.
+ */
+static unsigned int event_stage;
+
+/**
+ * Event used by the notification tests.
+ */
+static struct GNUNET_DB_EventHeaderP event_spec;
+
/**
* Setup prepared statements.
@@ -154,7 +165,6 @@ postgres_prepare (struct GNUNET_PQ_Context *db_)
",arr_var_bytes"
" FROM test_pq_struct_arrays"
" WHERE id = $1;"),
-
GNUNET_PQ_PREPARED_STATEMENT_END
};
@@ -1604,21 +1614,115 @@ timeout_cb (void *cls)
/**
- * Task called on expected event
+ * Context for the notification thread.
+ */
+struct NotifyThreadContext
+{
+ /**
+ * Postgres notification channel.
+ */
+ char *channel;
+
+ /**
+ * Set to true on success.
+ */
+ bool success;
+};
+
+
+/**
+ * Send a notification while the main connection is busy in a synchronous
+ * query.
*
- * @param cls NULL
+ * @param cls a `struct NotifyThreadContext *`
+ * @return NULL
+ */
+static void *
+notify_during_query (void *cls)
+{
+ struct NotifyThreadContext *ntc = cls;
+ PGconn *conn;
+ PGresult *result;
+ char *sql;
+
+ usleep (50 * 1000);
+ conn = PQconnectdb ("postgres:///gnunetcheck");
+ if (CONNECTION_OK != PQstatus (conn))
+ {
+ PQfinish (conn);
+ return NULL;
+ }
+ GNUNET_asprintf (&sql,
+ "NOTIFY %s",
+ ntc->channel);
+ result = PQexec (conn,
+ sql);
+ ntc->success = (PGRES_COMMAND_OK == PQresultStatus (result));
+ PQclear (result);
+ GNUNET_free (sql);
+ PQfinish (conn);
+ return NULL;
+}
+
+
+/**
+ * Run a synchronous query while another connection sends a notification.
+ *
+ * @param cls a `struct GNUNET_DB_EventHeaderP *`
+ */
+static void
+test_event_during_query (void *cls)
+{
+ const struct GNUNET_DB_EventHeaderP *es = cls;
+ struct NotifyThreadContext ntc;
+ struct GNUNET_PQ_ExecuteStatement statements[] = {
+ GNUNET_PQ_make_execute (
+ "DO $$ BEGIN PERFORM pg_sleep(0.25); END $$;"),
+ GNUNET_PQ_EXECUTE_STATEMENT_END
+ };
+ pthread_t thread;
+
+ ntc.channel = GNUNET_PQ_get_event_notify_channel (es);
+ ntc.success = false;
+ GNUNET_assert (0 ==
+ pthread_create (&thread,
+ NULL,
+ ¬ify_during_query,
+ &ntc));
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_PQ_exec_statements (db,
+ statements));
+ GNUNET_assert (0 ==
+ pthread_join (thread,
+ NULL));
+ GNUNET_assert (ntc.success);
+ GNUNET_free (ntc.channel);
+}
+
+
+/**
+ * Task called on expected event.
+ *
+ * @param cls a `struct GNUNET_DB_EventHeaderP *`
*/
static void
event_sched_cb (void *cls,
const void *extra,
size_t extra_size)
{
- (void) cls;
- GNUNET_assert (5 == extra_size);
- GNUNET_assert (0 ==
- memcmp ("hello",
- extra,
- 5));
+ if (0 == event_stage)
+ {
+ GNUNET_assert (5 == extra_size);
+ GNUNET_assert (0 ==
+ memcmp ("hello",
+ extra,
+ 5));
+ event_stage++;
+ GNUNET_SCHEDULER_add_now (&test_event_during_query,
+ cls);
+ return;
+ }
+ GNUNET_assert (0 == extra_size);
GNUNET_SCHEDULER_shutdown ();
}
@@ -1631,25 +1735,24 @@ event_sched_cb (void *cls,
static void
sched_tests (void *cls)
{
- struct GNUNET_DB_EventHeaderP es = {
- .size = htons (sizeof (es)),
- .type = htons (42)
- };
-
(void) cls;
- tt = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
+ event_spec.size = htons (sizeof (event_spec));
+ event_spec.type = htons (42);
+ tt = GNUNET_SCHEDULER_add_delayed (
+ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
+ 2),
&timeout_cb,
NULL);
eh = GNUNET_PQ_event_listen (db,
- &es,
+ &event_spec,
GNUNET_TIME_UNIT_FOREVER_REL,
&event_sched_cb,
- NULL);
+ &event_spec);
GNUNET_PQ_reconnect_ (db);
GNUNET_SCHEDULER_add_shutdown (&event_end,
NULL);
GNUNET_PQ_event_notify (db,
- &es,
+ &event_spec,
"hello",
5);
}
