diff options
| -rw-r--r-- | doc/src/sgml/func.sgml | 19 | ||||
| -rw-r--r-- | doc/src/sgml/ref/notify.sgml | 5 | ||||
| -rw-r--r-- | src/backend/commands/async.c | 50 | ||||
| -rw-r--r-- | src/include/catalog/catversion.h | 2 | ||||
| -rw-r--r-- | src/include/catalog/pg_proc.h | 4 | ||||
| -rw-r--r-- | src/include/commands/async.h | 1 | ||||
| -rw-r--r-- | src/test/isolation/expected/async-notify.out | 17 | ||||
| -rw-r--r-- | src/test/isolation/specs/async-notify.spec | 14 | ||||
| -rw-r--r-- | src/test/regress/expected/async.out | 8 | ||||
| -rw-r--r-- | src/test/regress/sql/async.sql | 4 |
10 files changed, 110 insertions, 14 deletions
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml index 76f77cb0fcf..91fa4ca3b7c 100644 --- a/doc/src/sgml/func.sgml +++ b/doc/src/sgml/func.sgml @@ -14806,6 +14806,12 @@ SELECT * FROM pg_ls_dir('.') WITH ORDINALITY AS t(ls,n); </row> <row> + <entry><literal><function>pg_notification_queue_usage()</function></literal></entry> + <entry><type>double</type></entry> + <entry>fraction of the asynchronous notification queue currently occupied (0-1)</entry> + </row> + + <row> <entry><literal><function>pg_my_temp_schema()</function></literal></entry> <entry><type>oid</type></entry> <entry>OID of session's temporary schema, or 0 if none</entry> @@ -14945,10 +14951,19 @@ SET search_path TO <replaceable>schema</> <optional>, <replaceable>schema</>, .. <primary>pg_listening_channels</primary> </indexterm> + <indexterm> + <primary>pg_notification_queue_usage</primary> + </indexterm> + <para> <function>pg_listening_channels</function> returns a set of names of - channels that the current session is listening to. See <xref - linkend="sql-listen"> for more information. + asynchronous notification channels that the current session is listening + to. <function>pg_notification_queue_usage</function> returns the + fraction of the total available space for notifications currently + occupied by notifications that are waiting to be processed, as a + <type>double</type> in the range 0-1. + See <xref linkend="sql-listen"> and <xref linkend="sql-notify"> + for more information. </para> <indexterm> diff --git a/doc/src/sgml/ref/notify.sgml b/doc/src/sgml/ref/notify.sgml index ad574e9ea03..4dd560838b1 100644 --- a/doc/src/sgml/ref/notify.sgml +++ b/doc/src/sgml/ref/notify.sgml @@ -166,6 +166,11 @@ NOTIFY <replaceable class="PARAMETER">channel</replaceable> [ , <replaceable cla current transaction so that cleanup can proceed. </para> <para> + The function <function>pg_notification_queue_usage</function> returns the + fraction of the queue that is currently occupied by pending notifications. + See <xref linkend="functions-info"> for more information. + </para> + <para> A transaction that has executed <command>NOTIFY</command> cannot be prepared for two-phase commit. </para> diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index 2826b7e43c4..3b71174b826 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 44ce2b3a620..8f6685fd0cc 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 6fd1278d1b2..1d68ad7209e 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 8491f4736f7..677dcaa3d03 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 00000000000..92d281a7d1f --- /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 00000000000..7f451b18a15 --- /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 ae0d5df3b78..19cbe38e636 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 af3a904e0f5..40f6e015387 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(); |