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
@@ -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 ad574e9ea03..4dd560838b1 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 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);
@@ -1361,6 +1362,48 @@ asyncQueueAddEntries(ListCell *nextNotify)
return nextNotify;
}
+/*
+ * 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.
+ *
+ * The caller must hold AysncQueueLock in (at least) shared mode.
+ */
+static double
+asyncQueueUsage(void)
+{
+ int headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
+ int tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
+ int occupied;
+
+ occupied = headPage - tailPage;
+
+ if (occupied == 0)
+ return (double) 0; /* fast exit for common case */
+
+ if (occupied < 0)
+ {
+ /* head has wrapped around, tail not yet */
+ occupied += QUEUE_MAX_PAGE + 1;
+ }
+
+ return (double) occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2);
+}
+
/*
* Check whether the queue is at least half full, and emit a warning if so.
*
@@ -1372,25 +1415,10 @@ asyncQueueAddEntries(ListCell *nextNotify)
static void
asyncQueueFillWarning(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 */
-
- if (occupied < 0)
- {
- /* head has wrapped around, tail not yet */
- occupied += QUEUE_MAX_PAGE + 1;
- }
-
- fillDegree = (double) occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2);
-
+ 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();