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.

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;
 

src/include/catalog/catversion.h

@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201507021
+#define CATALOG_VERSION_NO	201507171
 
 #endif

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_ ));

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);

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              

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"

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)
+

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();