Introduce and use infrastructure for interrupt processing during client reads.

Up to now large swathes of backend code ran inside signal handlers
while reading commands from the client, to allow for speedy reaction to
asynchronous events. Most prominently shared invalidation and NOTIFY
handling. That means that complex code like the starting/stopping of
transactions is run in signal handlers...  The required code was
fragile and verbose, and is likely to contain bugs.

That approach also severely limited what could be done while
communicating with the client. As the read might be from within
openssl it wasn't safely possible to trigger an error, e.g. to cancel
a backend in idle-in-transaction state. We did that in some cases,
namely fatal errors, nonetheless.

Now that FE/BE communication in the backend employs non-blocking
sockets and latches to block, we can quite simply interrupt reads from
signal handlers by setting the latch. That allows us to signal an
interrupted read, which is supposed to be retried after returning from
within the ssl library.

As signal handlers now only need to set the latch to guarantee timely
interrupt processing, remove a fair amount of complicated & fragile
code from async.c and sinval.c.

We could now actually start to process some kinds of interrupts, like
sinval ones, more often that before, but that seems better done
separately.

This work will hopefully allow to handle cases like being blocked by
sending data, interrupting idle transactions and similar to be
implemented without too much effort.  In addition to allowing getting
rid of ImmediateInterruptOK, that is.

Author: Andres Freund
Reviewed-By: Heikki Linnakangas

1 files changed, 37 insertions, 6 deletions

diff --git a/src/backend/libpq/be-secure.c b/src/backend/libpq/be-secure.c
index c9a8f6df4fe..b90ab0ea86f 100644
--- a/src/backend/libpq/be-secure.c
+++ b/src/backend/libpq/be-secure.c
@@ -128,6 +128,7 @@ secure_read(Port *port, void *ptr, size_t len)
 {
 	ssize_t		n;
 
+retry:
 #ifdef USE_SSL
 	if (port->ssl_in_use)
 	{
@@ -139,6 +140,14 @@ secure_read(Port *port, void *ptr, size_t len)
 		n = secure_raw_read(port, ptr, len);
 	}
 
+	/* Process interrupts that happened while (or before) receiving. */
+	ProcessClientReadInterrupt(); /* preserves errno */
+
+	/* retry after processing interrupts */
+	if (n < 0 && errno == EINTR)
+	{
+		goto retry;
+	}
 	return n;
 }
 
@@ -147,8 +156,6 @@ secure_raw_read(Port *port, void *ptr, size_t len)
 {
 	ssize_t		n;
 
-	prepare_for_client_read();
-
 	/*
 	 * Try to read from the socket without blocking. If it succeeds we're
 	 * done, otherwise we'll wait for the socket using the latch mechanism.
@@ -168,10 +175,20 @@ rloop:
 		int		save_errno = errno;
 
 		w = WaitLatchOrSocket(MyLatch,
-							  WL_SOCKET_READABLE,
+							  WL_LATCH_SET | WL_SOCKET_READABLE,
 							  port->sock, 0);
 
-		if (w & WL_SOCKET_READABLE)
+		if (w & WL_LATCH_SET)
+		{
+			ResetLatch(MyLatch);
+			/*
+			 * Force a return, so interrupts can be processed when not
+			 * (possibly) underneath a ssl library.
+			 */
+			errno = EINTR;
+			return -1;
+		}
+		else if (w & WL_SOCKET_READABLE)
 		{
 			goto rloop;
 		}
@@ -183,8 +200,6 @@ rloop:
 		errno = save_errno;
 	}
 
-	client_read_ended();
-
 	return n;
 }
 
@@ -197,6 +212,7 @@ secure_write(Port *port, void *ptr, size_t len)
 {
 	ssize_t		n;
 
+retry:
 #ifdef USE_SSL
 	if (port->ssl_in_use)
 	{
@@ -208,6 +224,21 @@ secure_write(Port *port, void *ptr, size_t len)
 		n = secure_raw_write(port, ptr, len);
 	}
 
+	/*
+	 * XXX: We'll, at some later point, likely want to add interrupt
+	 * processing here.
+	 */
+
+	/*
+	 * Retry after processing interrupts. This can be triggered even though we
+	 * don't check for latch set's during writing yet, because SSL
+	 * renegotiations might have required reading from the socket.
+	 */
+	if (n < 0 && errno == EINTR)
+	{
+		goto retry;
+	}
+
 	return n;
 }