From a04bb65f70dafdf462e0478ad19e6de56df89bfc Mon Sep 17 00:00:00 2001 From: Robert Haas Date: Fri, 17 Jul 2015 09:12:03 -0400 Subject: [PATCH] Add new function pg_notification_queue_usage. This tells you what fraction of NOTIFY's queue is currently filled. Brendan Jurd, reviewed by Merlin Moncure and Gurjeet Singh. A few further tweaks by me. --- doc/src/sgml/func.sgml | 19 +++++++- doc/src/sgml/ref/notify.sgml | 5 ++ src/backend/commands/async.c | 50 +++++++++++++++----- src/include/catalog/catversion.h | 2 +- src/include/catalog/pg_proc.h | 4 ++ src/include/commands/async.h | 1 + src/test/isolation/expected/async-notify.out | 17 +++++++ src/test/isolation/specs/async-notify.spec | 14 ++++++ src/test/regress/expected/async.out | 8 ++++ src/test/regress/sql/async.sql | 4 ++ 10 files changed, 110 insertions(+), 14 deletions(-) create mode 100644 src/test/isolation/expected/async-notify.out create mode 100644 src/test/isolation/specs/async-notify.spec diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml index 76f77cb0fc..91fa4ca3b7 100644 --- a/doc/src/sgml/func.sgml +++ b/doc/src/sgml/func.sgml @@ -14805,6 +14805,12 @@ SELECT * FROM pg_ls_dir('.') WITH ORDINALITY AS t(ls,n); channel names that the session is currently listening on + + pg_notification_queue_usage() + double + fraction of the asynchronous notification queue currently occupied (0-1) + + pg_my_temp_schema() oid @@ -14945,10 +14951,19 @@ SET search_path TO schema , schema, .. pg_listening_channels + + pg_notification_queue_usage + + pg_listening_channels returns a set of names of - channels that the current session is listening to. See for more information. + asynchronous notification channels that the current session is listening + to. pg_notification_queue_usage returns the + fraction of the total available space for notifications currently + occupied by notifications that are waiting to be processed, as a + double in the range 0-1. + See and + for more information. diff --git a/doc/src/sgml/ref/notify.sgml b/doc/src/sgml/ref/notify.sgml index ad574e9ea0..4dd560838b 100644 --- a/doc/src/sgml/ref/notify.sgml +++ b/doc/src/sgml/ref/notify.sgml @@ -165,6 +165,11 @@ NOTIFY channel [ , + + The function pg_notification_queue_usage returns the + fraction of the queue that is currently occupied by pending notifications. + See for more information. + A transaction that has executed NOTIFY cannot be prepared for two-phase commit. diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index 2826b7e43c..3b71174b82 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -371,6 +371,7 @@ static bool asyncQueueIsFull(void); static bool asyncQueueAdvance(volatile QueuePosition *position, int entryLength); static void asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe); static ListCell *asyncQueueAddEntries(ListCell *nextNotify); +static double asyncQueueUsage(void); static void asyncQueueFillWarning(void); static bool SignalBackends(void); static void asyncQueueReadAllNotifications(void); @@ -1362,26 +1363,37 @@ asyncQueueAddEntries(ListCell *nextNotify) } /* - * Check whether the queue is at least half full, and emit a warning if so. - * - * This is unlikely given the size of the queue, but possible. - * The warnings show up at most once every QUEUE_FULL_WARN_INTERVAL. + * SQL function to return the fraction of the notification queue currently + * occupied. + */ +Datum +pg_notification_queue_usage(PG_FUNCTION_ARGS) +{ + double usage; + + LWLockAcquire(AsyncQueueLock, LW_SHARED); + usage = asyncQueueUsage(); + LWLockRelease(AsyncQueueLock); + + PG_RETURN_FLOAT8(usage); +} + +/* + * Return the fraction of the queue that is currently occupied. * - * Caller must hold exclusive AsyncQueueLock. + * The caller must hold AysncQueueLock in (at least) shared mode. */ -static void -asyncQueueFillWarning(void) +static double +asyncQueueUsage(void) { int headPage = QUEUE_POS_PAGE(QUEUE_HEAD); int tailPage = QUEUE_POS_PAGE(QUEUE_TAIL); int occupied; - double fillDegree; - TimestampTz t; occupied = headPage - tailPage; if (occupied == 0) - return; /* fast exit for common case */ + return (double) 0; /* fast exit for common case */ if (occupied < 0) { @@ -1389,8 +1401,24 @@ asyncQueueFillWarning(void) occupied += QUEUE_MAX_PAGE + 1; } - fillDegree = (double) occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2); + return (double) occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2); +} + +/* + * Check whether the queue is at least half full, and emit a warning if so. + * + * This is unlikely given the size of the queue, but possible. + * The warnings show up at most once every QUEUE_FULL_WARN_INTERVAL. + * + * Caller must hold exclusive AsyncQueueLock. + */ +static void +asyncQueueFillWarning(void) +{ + double fillDegree; + TimestampTz t; + fillDegree = asyncQueueUsage(); if (fillDegree < 0.5) return; diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h index 44ce2b3a62..8f6685fd0c 100644 --- a/src/include/catalog/catversion.h +++ b/src/include/catalog/catversion.h @@ -53,6 +53,6 @@ */ /* yyyymmddN */ -#define CATALOG_VERSION_NO 201507021 +#define CATALOG_VERSION_NO 201507171 #endif diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h index 6fd1278d1b..1d68ad7209 100644 --- a/src/include/catalog/pg_proc.h +++ b/src/include/catalog/pg_proc.h @@ -4046,10 +4046,14 @@ DATA(insert OID = 2856 ( pg_timezone_names PGNSP PGUID 12 1 1000 0 0 f f f f t DESCR("get the available time zone names"); DATA(insert OID = 2730 ( pg_get_triggerdef PGNSP PGUID 12 1 0 0 0 f f f f t f s 2 0 25 "26 16" _null_ _null_ _null_ _null_ _null_ pg_get_triggerdef_ext _null_ _null_ _null_ )); DESCR("trigger description with pretty-print option"); + +/* asynchronous notifications */ DATA(insert OID = 3035 ( pg_listening_channels PGNSP PGUID 12 1 10 0 0 f f f f t t s 0 0 25 "" _null_ _null_ _null_ _null_ _null_ pg_listening_channels _null_ _null_ _null_ )); DESCR("get the channels that the current backend listens to"); DATA(insert OID = 3036 ( pg_notify PGNSP PGUID 12 1 0 0 0 f f f f f f v 2 0 2278 "25 25" _null_ _null_ _null_ _null_ _null_ pg_notify _null_ _null_ _null_ )); DESCR("send a notification event"); +DATA(insert OID = 3296 ( pg_notification_queue_usage PGNSP PGUID 12 1 0 0 0 f f f f t f v 0 0 701 "" _null_ _null_ _null_ _null_ _null_ pg_notification_queue_usage _null_ _null_ _null_ )); +DESCR("get the fraction of the asynchronous notification queue currently in use"); /* non-persistent series generator */ DATA(insert OID = 1066 ( generate_series PGNSP PGUID 12 1 1000 0 0 f f f f t t i 3 0 23 "23 23 23" _null_ _null_ _null_ _null_ _null_ generate_series_step_int4 _null_ _null_ _null_ )); diff --git a/src/include/commands/async.h b/src/include/commands/async.h index 8491f4736f..677dcaa3d0 100644 --- a/src/include/commands/async.h +++ b/src/include/commands/async.h @@ -37,6 +37,7 @@ extern void Async_UnlistenAll(void); /* notify-related SQL functions */ extern Datum pg_listening_channels(PG_FUNCTION_ARGS); extern Datum pg_notify(PG_FUNCTION_ARGS); +extern Datum pg_notification_queue_usage(PG_FUNCTION_ARGS); /* perform (or cancel) outbound notify processing at transaction commit */ extern void PreCommit_Notify(void); diff --git a/src/test/isolation/expected/async-notify.out b/src/test/isolation/expected/async-notify.out new file mode 100644 index 0000000000..92d281a7d1 --- /dev/null +++ b/src/test/isolation/expected/async-notify.out @@ -0,0 +1,17 @@ +Parsed test spec with 2 sessions + +starting permutation: listen begin check notify check +step listen: LISTEN a; +step begin: BEGIN; +step check: SELECT pg_notification_queue_usage() > 0 AS nonzero; +nonzero + +f +step notify: SELECT count(pg_notify('a', s::text)) FROM generate_series(1, 1000) s; +count + +1000 +step check: SELECT pg_notification_queue_usage() > 0 AS nonzero; +nonzero + +t diff --git a/src/test/isolation/specs/async-notify.spec b/src/test/isolation/specs/async-notify.spec new file mode 100644 index 0000000000..7f451b18a1 --- /dev/null +++ b/src/test/isolation/specs/async-notify.spec @@ -0,0 +1,14 @@ +# Verify that pg_notification_queue_usage correctly reports a non-zero result, +# after submitting notifications while another connection is listening for +# those notifications and waiting inside an active transaction. + +session "listener" +step "listen" { LISTEN a; } +step "begin" { BEGIN; } +teardown { ROLLBACK; } + +session "notifier" +step "check" { SELECT pg_notification_queue_usage() > 0 AS nonzero; } +step "notify" { SELECT count(pg_notify('a', s::text)) FROM generate_series(1, 1000) s; } + +permutation "listen" "begin" "check" "notify" "check" diff --git a/src/test/regress/expected/async.out b/src/test/regress/expected/async.out index ae0d5df3b7..19cbe38e63 100644 --- a/src/test/regress/expected/async.out +++ b/src/test/regress/expected/async.out @@ -32,3 +32,11 @@ NOTIFY notify_async2; LISTEN notify_async2; UNLISTEN notify_async2; UNLISTEN *; +-- Should return zero while there are no pending notifications. +-- src/test/isolation/specs/async-notify.spec tests for actual usage. +SELECT pg_notification_queue_usage(); + pg_notification_queue_usage +----------------------------- + 0 +(1 row) + diff --git a/src/test/regress/sql/async.sql b/src/test/regress/sql/async.sql index af3a904e0f..40f6e01538 100644 --- a/src/test/regress/sql/async.sql +++ b/src/test/regress/sql/async.sql @@ -17,3 +17,7 @@ NOTIFY notify_async2; LISTEN notify_async2; UNLISTEN notify_async2; UNLISTEN *; + +-- Should return zero while there are no pending notifications. +-- src/test/isolation/specs/async-notify.spec tests for actual usage. +SELECT pg_notification_queue_usage(); -- 2.40.0