Send NOTIFY signals during CommitTransaction.

Formerly, we sent signals for outgoing NOTIFY messages within
ProcessCompletedNotifies, which was also responsible for sending
relevant ones of those messages to our connected client.  It therefore
had to run during the main-loop processing that occurs just before
going idle.  This arrangement had two big disadvantages:

* Now that procedures allow intra-command COMMITs, it would be
useful to send NOTIFYs to other sessions immediately at COMMIT
(though, for reasons of wire-protocol stability, we still shouldn't
forward them to our client until end of command).

* Background processes such as replication workers would not send
NOTIFYs at all, since they never execute the client communication
loop.  We've had requests to allow triggers running in replication
workers to send NOTIFYs, so that's a problem.

To fix these things, move transmission of outgoing NOTIFY signals
into AtCommit_Notify, where it will happen during CommitTransaction.
Also move the possible call of asyncQueueAdvanceTail there, to
ensure we don't bloat the async SLRU if a background worker sends
many NOTIFYs with no one listening.

We can also drop the call of asyncQueueReadAllNotifications,
allowing ProcessCompletedNotifies to go away entirely.  That's
because commit 790026972 added a call of ProcessNotifyInterrupt
adjacent to PostgresMain's call of ProcessCompletedNotifies,
and that does its own call of asyncQueueReadAllNotifications,
meaning that we were uselessly doing two such calls (inside two
separate transactions) whenever inbound notify signals coincided
with an outbound notify.  We need only set notifyInterruptPending
to ensure that ProcessNotifyInterrupt runs, and we're done.

The existing documentation suggests that custom background workers
should call ProcessCompletedNotifies if they want to send NOTIFY
messages.  To avoid an ABI break in the back branches, reduce it
to an empty routine rather than removing it entirely.  Removal
will occur in v15.

Although the problems mentioned above have existed for awhile,
I don't feel comfortable back-patching this any further than v13.
There was quite a bit of churn in adjacent code between 12 and 13.
At minimum we'd have to also backpatch 51004c717, and a good deal
of other adjustment would also be needed, so the benefit-to-risk
ratio doesn't look attractive.

Per bug #15293 from Michael Powers (and similar gripes from others).

Artur Zakirov and Tom Lane

Discussion: https://postgr.es/m/153243441449.1404.2274116228506175596@wrigleys.postgresql.org

doc/src/sgml/bgworker.sgml

@@ -284,15 +284,13 @@ typedef struct BackgroundWorker
   </para>
 
   <para>
-   If a background worker sends asynchronous notifications with the
+   Background workers can send asynchronous notification messages, either by
-   <command>NOTIFY</command> command via the Server Programming Interface
+   using the <command>NOTIFY</command> command via <acronym>SPI</acronym>,
-   (<acronym>SPI</acronym>), it should call
+   or directly via <function>Async_Notify()</function>.  Such notifications
-   <function>ProcessCompletedNotifies</function> explicitly after committing
+   will be sent at transaction commit.
-   the enclosing transaction so that any notifications can be delivered.  If a
+   Background workers should not register to receive asynchronous
-   background worker registers to receive asynchronous notifications with
+   notifications with the <command>LISTEN</command> command, as there is no
-   the <command>LISTEN</command> through <acronym>SPI</acronym>, the worker
+   infrastructure for a worker to consume such notifications.
-   will log those notifications, but there is no programmatic way for the
-   worker to intercept and respond to those notifications.
   </para>
 
   <para>

src/backend/access/transam/xact.c

@@ -2242,7 +2242,17 @@ CommitTransaction(void)
 	 */
 	smgrDoPendingDeletes(true);
 
+	/*
+	 * Send out notification signals to other backends (and do other
+	 * post-commit NOTIFY cleanup).  This must not happen until after our
+	 * transaction is fully done from the viewpoint of other backends.
+	 */
 	AtCommit_Notify();
+
+	/*
+	 * Everything after this should be purely internal-to-this-backend
+	 * cleanup.
+	 */
 	AtEOXact_GUC(true, 1);
 	AtEOXact_SPI(true);
 	AtEOXact_Enum();

src/backend/commands/async.c

@@ -68,17 +68,27 @@
  *	  CommitTransaction() which will then do the actual transaction commit.
  *
  *	  After commit we are called another time (AtCommit_Notify()). Here we
- *	  make the actual updates to the effective listen state (listenChannels).
+ *	  make any actual updates to the effective listen state (listenChannels).
+ *	  Then we signal any backends that may be interested in our messages
+ *	  (including our own backend, if listening).  This is done by
+ *	  SignalBackends(), which scans the list of listening backends and sends a
+ *	  PROCSIG_NOTIFY_INTERRUPT signal to every listening backend (we don't
+ *	  know which backend is listening on which channel so we must signal them
+ *	  all).  We can exclude backends that are already up to date, though, and
+ *	  we can also exclude backends that are in other databases (unless they
+ *	  are way behind and should be kicked to make them advance their
+ *	  pointers).
  *
- *	  Finally, after we are out of the transaction altogether, we check if
+ *	  Finally, after we are out of the transaction altogether and about to go
- *	  we need to signal listening backends.  In SignalBackends() we scan the
+ *	  idle, we scan the queue for messages that need to be sent to our
- *	  list of listening backends and send a PROCSIG_NOTIFY_INTERRUPT signal
+ *	  frontend (which might be notifies from other backends, or self-notifies
- *	  to every listening backend (we don't know which backend is listening on
+ *	  from our own).  This step is not part of the CommitTransaction sequence
- *	  which channel so we must signal them all). We can exclude backends that
+ *	  for two important reasons.  First, we could get errors while sending
- *	  are already up to date, though, and we can also exclude backends that
+ *	  data to our frontend, and it's really bad for errors to happen in
- *	  are in other databases (unless they are way behind and should be kicked
+ *	  post-commit cleanup.  Second, in cases where a procedure issues commits
- *	  to make them advance their pointers).  We don't bother with a
+ *	  within a single frontend command, we don't want to send notifies to our
- *	  self-signal either, but just process the queue directly.
+ *	  frontend until the command is done; but notifies to other backends
+ *	  should go out immediately after each commit.
  *
  * 5. Upon receipt of a PROCSIG_NOTIFY_INTERRUPT signal, the signal handler
  *	  sets the process's latch, which triggers the event to be processed
@@ -429,11 +439,8 @@ static bool unlistenExitRegistered = false;
 /* True if we're currently registered as a listener in asyncQueueControl */
 static bool amRegisteredListener = false;
 
-/* has this backend sent notifications in the current transaction? */
-static bool backendHasSentNotifications = false;
-
 /* have we advanced to a page that's a multiple of QUEUE_CLEANUP_DELAY? */
-static bool backendTryAdvanceTail = false;
+static bool tryAdvanceTail = false;
 
 /* GUC parameter */
 bool		Trace_notify = false;
@@ -462,7 +469,7 @@ static bool asyncQueueProcessPageEntries(volatile QueuePosition *current,
 										 char *page_buffer,
 										 Snapshot snapshot);
 static void asyncQueueAdvanceTail(void);
-static void ProcessIncomingNotify(void);
+static void ProcessIncomingNotify(bool flush);
 static bool AsyncExistsPendingNotify(Notification *n);
 static void AddEventToPendingNotifies(Notification *n);
 static uint32 notification_hash(const void *key, Size keysize);
@@ -954,8 +961,6 @@ PreCommit_Notify(void)
 						 AccessExclusiveLock);
 
 		/* Now push the notifications into the queue */
-		backendHasSentNotifications = true;
-
 		nextNotify = list_head(pendingNotifies->events);
 		while (nextNotify != NULL)
 		{
@@ -980,6 +985,8 @@ PreCommit_Notify(void)
 			nextNotify = asyncQueueAddEntries(nextNotify);
 			LWLockRelease(NotifyQueueLock);
 		}
+
+		/* Note that we don't clear pendingNotifies; AtCommit_Notify will. */
 	}
 }
 
@@ -989,6 +996,11 @@ PreCommit_Notify(void)
  *		This is called at transaction commit, after committing to clog.
  *
  *		Update listenChannels and clear transaction-local state.
+ *
+ *		If we issued any notifications in the transaction, send signals to
+ *		listening backends (possibly including ourselves) to process them.
+ *		Also, if we filled enough queue pages with new notifies, try to
+ *		advance the queue tail pointer.
  */
 void
 AtCommit_Notify(void)
@@ -1031,6 +1043,29 @@ AtCommit_Notify(void)
 	if (amRegisteredListener && listenChannels == NIL)
 		asyncQueueUnregister();
 
+	/*
+	 * Send signals to listening backends.  We need do this only if there are
+	 * pending notifies, which were previously added to the shared queue by
+	 * PreCommit_Notify().
+	 */
+	if (pendingNotifies != NULL)
+		SignalBackends();
+
+	/*
+	 * If it's time to try to advance the global tail pointer, do that.
+	 *
+	 * (It might seem odd to do this in the sender, when more than likely the
+	 * listeners won't yet have read the messages we just sent.  However,
+	 * there's less contention if only the sender does it, and there is little
+	 * need for urgency in advancing the global tail.  So this typically will
+	 * be clearing out messages that were sent some time ago.)
+	 */
+	if (tryAdvanceTail)
+	{
+		tryAdvanceTail = false;
+		asyncQueueAdvanceTail();
+	}
+
 	/* And clean up */
 	ClearPendingActionsAndNotifies();
 }
@@ -1204,82 +1239,17 @@ Exec_UnlistenAllCommit(void)
 }
 
 /*
- * ProcessCompletedNotifies --- send out signals and self-notifies
+ * ProcessCompletedNotifies --- nowadays this does nothing
  *
- * This is called from postgres.c just before going idle at the completion
+ * This routine used to send signals and handle self-notifies,
- * of a transaction.  If we issued any notifications in the just-completed
+ * but that functionality has been moved elsewhere.
- * transaction, send signals to other backends to process them, and also
+ * We'd delete it entirely, except that the documentation used to instruct
- * process the queue ourselves to send messages to our own frontend.
+ * background-worker authors to call it.  To avoid an ABI break in stable
- * Also, if we filled enough queue pages with new notifies, try to advance
+ * branches, keep it as a no-op routine.
- * the queue tail pointer.
- *
- * The reason that this is not done in AtCommit_Notify is that there is
- * a nonzero chance of errors here (for example, encoding conversion errors
- * while trying to format messages to our frontend).  An error during
- * AtCommit_Notify would be a PANIC condition.  The timing is also arranged
- * to ensure that a transaction's self-notifies are delivered to the frontend
- * before it gets the terminating ReadyForQuery message.
- *
- * Note that we send signals and process the queue even if the transaction
- * eventually aborted.  This is because we need to clean out whatever got
- * added to the queue.
- *
- * NOTE: we are outside of any transaction here.
  */
 void
 ProcessCompletedNotifies(void)
 {
-	MemoryContext caller_context;
-
-	/* Nothing to do if we didn't send any notifications */
-	if (!backendHasSentNotifications)
-		return;
-
-	/*
-	 * We reset the flag immediately; otherwise, if any sort of error occurs
-	 * below, we'd be locked up in an infinite loop, because control will come
-	 * right back here after error cleanup.
-	 */
-	backendHasSentNotifications = false;
-
-	/*
-	 * We must preserve the caller's memory context (probably MessageContext)
-	 * across the transaction we do here.
-	 */
-	caller_context = CurrentMemoryContext;
-
-	if (Trace_notify)
-		elog(DEBUG1, "ProcessCompletedNotifies");
-
-	/*
-	 * We must run asyncQueueReadAllNotifications inside a transaction, else
-	 * bad things happen if it gets an error.
-	 */
-	StartTransactionCommand();
-
-	/* Send signals to other backends */
-	SignalBackends();
-
-	if (listenChannels != NIL)
-	{
-		/* Read the queue ourselves, and send relevant stuff to the frontend */
-		asyncQueueReadAllNotifications();
-	}
-
-	/*
-	 * If it's time to try to advance the global tail pointer, do that.
-	 */
-	if (backendTryAdvanceTail)
-	{
-		backendTryAdvanceTail = false;
-		asyncQueueAdvanceTail();
-	}
-
-	CommitTransactionCommand();
-
-	MemoryContextSwitchTo(caller_context);
-
-	/* We don't need pq_flush() here since postgres.c will do one shortly */
 }
 
 /*
@@ -1547,7 +1517,7 @@ asyncQueueAddEntries(ListCell *nextNotify)
 			 * pointer (we don't want to actually do that right here).
 			 */
 			if (QUEUE_POS_PAGE(queue_head) % QUEUE_CLEANUP_DELAY == 0)
-				backendTryAdvanceTail = true;
+				tryAdvanceTail = true;
 
 			/* And exit the loop */
 			break;
@@ -1662,8 +1632,6 @@ asyncQueueFillWarning(void)
 /*
  * Send signals to listening backends.
  *
- * We never signal our own process; that should be handled by our caller.
- *
  * Normally we signal only backends in our own database, since only those
  * backends could be interested in notifies we send.  However, if there's
  * notify traffic in our database but no traffic in another database that
@@ -1672,6 +1640,9 @@ asyncQueueFillWarning(void)
  * advance their queue position pointers, allowing the global tail to advance.
  *
  * Since we know the BackendId and the Pid the signaling is quite cheap.
+ *
+ * This is called during CommitTransaction(), so it's important for it
+ * to have very low probability of failure.
  */
 static void
 SignalBackends(void)
@@ -1686,8 +1657,7 @@ SignalBackends(void)
 	 * list of target PIDs.
 	 *
 	 * XXX in principle these pallocs could fail, which would be bad. Maybe
-	 * preallocate the arrays?	But in practice this is only run in trivial
+	 * preallocate the arrays?  They're not that large, though.
-	 * transactions, so there should surely be space available.
 	 */
 	pids = (int32 *) palloc(MaxBackends * sizeof(int32));
 	ids = (BackendId *) palloc(MaxBackends * sizeof(BackendId));
@@ -1700,8 +1670,6 @@ SignalBackends(void)
 		QueuePosition pos;
 
 		Assert(pid != InvalidPid);
-		if (pid == MyProcPid)
-			continue;			/* never signal self */
 		pos = QUEUE_BACKEND_POS(i);
 		if (QUEUE_BACKEND_DBOID(i) == MyDatabaseId)
 		{
@@ -1734,6 +1702,16 @@ SignalBackends(void)
 	{
 		int32		pid = pids[i];
 
+		/*
+		 * If we are signaling our own process, no need to involve the kernel;
+		 * just set the flag directly.
+		 */
+		if (pid == MyProcPid)
+		{
+			notifyInterruptPending = true;
+			continue;
+		}
+
 		/*
 		 * Note: assuming things aren't broken, a signal failure here could
 		 * only occur if the target backend exited since we released
@@ -1914,15 +1892,20 @@ HandleNotifyInterrupt(void)
  *		via the process's latch, and this routine will get called.
  *		If we are truly idle (ie, *not* inside a transaction block),
  *		process the incoming notifies.
+ *
+ *		If "flush" is true, force any frontend messages out immediately.
+ *		This can be false when being called at the end of a frontend command,
+ *		since we'll flush after sending ReadyForQuery.
  */
 void
-ProcessNotifyInterrupt(void)
+ProcessNotifyInterrupt(bool flush)
 {
 	if (IsTransactionOrTransactionBlock())
 		return;					/* not really idle */
 
+	/* Loop in case another signal arrives while sending messages */
 	while (notifyInterruptPending)
-		ProcessIncomingNotify();
+		ProcessIncomingNotify(flush);
 }
 
 
@@ -2185,6 +2168,9 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
 /*
  * Advance the shared queue tail variable to the minimum of all the
  * per-backend tail pointers.  Truncate pg_notify space if possible.
+ *
+ * This is (usually) called during CommitTransaction(), so it's important for
+ * it to have very low probability of failure.
  */
 static void
 asyncQueueAdvanceTail(void)
@@ -2258,17 +2244,16 @@ asyncQueueAdvanceTail(void)
 /*
  * ProcessIncomingNotify
  *
- *		Deal with arriving NOTIFYs from other backends as soon as it's safe to
+ *		Scan the queue for arriving notifications and report them to the front
- *		do so. This used to be called from the PROCSIG_NOTIFY_INTERRUPT
+ *		end.  The notifications might be from other sessions, or our own;
- *		signal handler, but isn't anymore.
+ *		there's no need to distinguish here.
  *
- *		Scan the queue for arriving notifications and report them to my front
+ *		If "flush" is true, force any frontend messages out immediately.
- *		end.
  *
  *		NOTE: since we are outside any transaction, we must create our own.
  */
 static void
-ProcessIncomingNotify(void)
+ProcessIncomingNotify(bool flush)
 {
 	/* We *must* reset the flag */
 	notifyInterruptPending = false;
@@ -2293,8 +2278,10 @@ ProcessIncomingNotify(void)
 	CommitTransactionCommand();
 
 	/*
-	 * Must flush the notify messages to ensure frontend gets them promptly.
+	 * If this isn't an end-of-command case, we must flush the notify messages
+	 * to ensure frontend gets them promptly.
 	 */
+	if (flush)
 		pq_flush();
 
 	set_ps_display("idle");
@@ -2321,9 +2308,9 @@ NotifyMyFrontEnd(const char *channel, const char *payload, int32 srcPid)
 		pq_endmessage(&buf);
 
 		/*
-		 * NOTE: we do not do pq_flush() here.  For a self-notify, it will
+		 * NOTE: we do not do pq_flush() here.  Some level of caller will
-		 * happen at the end of the transaction, and for incoming notifies
+		 * handle it later, allowing this message to be combined into a packet
-		 * ProcessIncomingNotify will do it after finding all the notifies.
+		 * with other ones.
 		 */
 	}
 	else

src/backend/tcop/postgres.c

@@ -538,7 +538,7 @@ ProcessClientReadInterrupt(bool blocked)
 
 		/* Process notify interrupts, if any */
 		if (notifyInterruptPending)
-			ProcessNotifyInterrupt();
+			ProcessNotifyInterrupt(true);
 	}
 	else if (ProcDiePending)
 	{
@@ -4243,17 +4243,15 @@ PostgresMain(int argc, char *argv[],
 			}
 			else
 			{
-				/* Send out notify signals and transmit self-notifies */
-				ProcessCompletedNotifies();
-
 				/*
-				 * Also process incoming notifies, if any.  This is mostly to
+				 * Process incoming notifies (including self-notifies), if
-				 * ensure stable behavior in tests: if any notifies were
+				 * any, and send relevant messages to the client.  Doing it
-				 * received during the just-finished transaction, they'll be
+				 * here helps ensure stable behavior in tests: if any notifies
-				 * seen by the client before ReadyForQuery is.
+				 * were received during the just-finished transaction, they'll
+				 * be seen by the client before ReadyForQuery is.
 				 */
 				if (notifyInterruptPending)
-					ProcessNotifyInterrupt();
+					ProcessNotifyInterrupt(false);
 
 				pgstat_report_stat(false);
 

src/include/commands/async.h

@@ -49,6 +49,6 @@ extern void ProcessCompletedNotifies(void);
 extern void HandleNotifyInterrupt(void);
 
 /* process interrupts */
-extern void ProcessNotifyInterrupt(void);
+extern void ProcessNotifyInterrupt(bool flush);
 
 #endif							/* ASYNC_H */