two-separate-counters.patch

diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
index 02a5df8da9..7e1a555094 100644
--- a/src/backend/storage/ipc/shm_mq.c
+++ b/src/backend/storage/ipc/shm_mq.c
@@ -107,6 +107,10 @@ struct shm_mq
  * locally by copying the chunks into a backend-local buffer.  mqh_buffer is
  * the buffer, and mqh_buflen is the number of bytes allocated for it.
  *
+ * mqh_increment_pending is the number of bytes that we have locally consumed
+ * but not yet marked added to mq_bytes_read, while mqh_signal_pending is the
+ * number of bytes we have consumed without signalling the sender.
+ *
  * mqh_partial_bytes, mqh_expected_bytes, and mqh_length_word_complete
  * are used to track the state of non-blocking operations.  When the caller
  * attempts a non-blocking operation that returns SHM_MQ_WOULD_BLOCK, they
@@ -134,7 +138,8 @@ struct shm_mq_handle
 	BackgroundWorkerHandle *mqh_handle;
 	char	   *mqh_buffer;
 	Size		mqh_buflen;
-	Size		mqh_consume_pending;
+	Size		mqh_increment_pending;
+	Size		mqh_signal_pending;
 	Size		mqh_partial_bytes;
 	Size		mqh_expected_bytes;
 	bool		mqh_length_word_complete;
@@ -293,7 +298,8 @@ shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
 	mqh->mqh_handle = handle;
 	mqh->mqh_buffer = NULL;
 	mqh->mqh_buflen = 0;
-	mqh->mqh_consume_pending = 0;
+	mqh->mqh_increment_pending = 0;
+	mqh->mqh_signal_pending = 0;
 	mqh->mqh_partial_bytes = 0;
 	mqh->mqh_expected_bytes = 0;
 	mqh->mqh_length_word_complete = false;
@@ -594,10 +600,15 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
 	 * because SetLatch() is fairly expensive and we don't want to do it too
 	 * often.
 	 */
-	if (mqh->mqh_consume_pending > mq->mq_ring_size / 4)
+	if (mqh->mqh_increment_pending > 0)
+	{
+		shm_mq_inc_bytes_read(mq, mqh->mqh_increment_pending);
+		mqh->mqh_increment_pending = 0;
+	}
+	if (mqh->mqh_signal_pending > mq->mq_ring_size / 4)
 	{
-		shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
-		mqh->mqh_consume_pending = 0;
+		SetLatch(&mq->mq_sender->procLatch);
+		mqh->mqh_signal_pending = 0;
 	}
 
 	/* Try to read, or finish reading, the length word from the buffer. */
@@ -625,7 +636,8 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
 			needed = MAXALIGN(sizeof(Size)) + MAXALIGN(nbytes);
 			if (rb >= needed)
 			{
-				mqh->mqh_consume_pending += needed;
+				mqh->mqh_increment_pending += needed;
+				mqh->mqh_signal_pending += needed;
 				*nbytesp = nbytes;
 				*datap = ((char *) rawdata) + MAXALIGN(sizeof(Size));
 				return SHM_MQ_SUCCESS;
@@ -637,7 +649,8 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
 			 */
 			mqh->mqh_expected_bytes = nbytes;
 			mqh->mqh_length_word_complete = true;
-			mqh->mqh_consume_pending += MAXALIGN(sizeof(Size));
+			mqh->mqh_increment_pending += MAXALIGN(sizeof(Size));
+			mqh->mqh_signal_pending += MAXALIGN(sizeof(Size));
 			rb -= MAXALIGN(sizeof(Size));
 		}
 		else
@@ -664,7 +677,8 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
 			memcpy(&mqh->mqh_buffer[mqh->mqh_partial_bytes], rawdata,
 				   lengthbytes);
 			mqh->mqh_partial_bytes += lengthbytes;
-			mqh->mqh_consume_pending += MAXALIGN(lengthbytes);
+			mqh->mqh_increment_pending += MAXALIGN(lengthbytes);
+			mqh->mqh_signal_pending += MAXALIGN(lengthbytes);
 			rb -= lengthbytes;
 
 			/* If we now have the whole word, we're ready to read payload. */
@@ -692,7 +706,8 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
 		if (rb >= nbytes)
 		{
 			mqh->mqh_length_word_complete = false;
-			mqh->mqh_consume_pending += MAXALIGN(nbytes);
+			mqh->mqh_increment_pending += MAXALIGN(nbytes);
+			mqh->mqh_signal_pending += MAXALIGN(nbytes);
 			*nbytesp = nbytes;
 			*datap = rawdata;
 			return SHM_MQ_SUCCESS;
@@ -738,7 +753,8 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
 		 * a multiple of MAXIMUM_ALIGNOF, and each read and write is as well.
 		 */
 		Assert(mqh->mqh_partial_bytes == nbytes || rb == MAXALIGN(rb));
-		mqh->mqh_consume_pending += MAXALIGN(rb);
+		mqh->mqh_increment_pending += MAXALIGN(rb);
+		mqh->mqh_signal_pending += MAXALIGN(rb);
 
 		/* If we got all the data, exit the loop. */
 		if (mqh->mqh_partial_bytes >= nbytes)
@@ -1034,7 +1050,7 @@ shm_mq_receive_bytes(shm_mq_handle *mqh, Size bytes_needed, bool nowait,
 		 * consumed.
 		 */
 		read = pg_atomic_read_u64(&mq->mq_bytes_read) +
-			mqh->mqh_consume_pending;
+			mqh->mqh_increment_pending;
 		used = written - read;
 		Assert(used <= ringsize);
 		offset = read % (uint64) ringsize;
@@ -1067,12 +1083,18 @@ shm_mq_receive_bytes(shm_mq_handle *mqh, Size bytes_needed, bool nowait,
 
 		/*
 		 * We didn't get enough data to satisfy the request, so mark any data
-		 * previously-consumed as read to make more buffer space.
+		 * previously-consumed as read to make more buffer space, and signal
+		 * the sender as necessary to make sure they're not waiting for us.
 		 */
-		if (mqh->mqh_consume_pending > 0)
+		if (mqh->mqh_increment_pending > 0)
+		{
+			shm_mq_inc_bytes_read(mq, mqh->mqh_increment_pending);
+			mqh->mqh_increment_pending = 0;
+		}
+		if (mqh->mqh_signal_pending > 0)
 		{
-			shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
-			mqh->mqh_consume_pending = 0;
+			SetLatch(&mq->mq_sender->procLatch);
+			mqh->mqh_signal_pending = 0;
 		}
 
 		/* Skip manipulation of our latch if nowait = true. */
@@ -1193,8 +1215,6 @@ shm_mq_wait_internal(volatile shm_mq *mq, PGPROC *volatile *ptr,
 static void
 shm_mq_inc_bytes_read(volatile shm_mq *mq, Size n)
 {
-	PGPROC	   *sender;
-
 	/*
 	 * Separate prior reads of mq_ring from the increment of mq_bytes_read
 	 * which follows.  Pairs with the full barrier in shm_mq_send_bytes(). We
@@ -1209,14 +1229,6 @@ shm_mq_inc_bytes_read(volatile shm_mq *mq, Size n)
 	 */
 	pg_atomic_write_u64(&mq->mq_bytes_read,
 						pg_atomic_read_u64(&mq->mq_bytes_read) + n);
-
-	/*
-	 * We shouldn't have any bytes to read without a sender, so we can read
-	 * mq_sender here without a lock.  Once it's initialized, it can't change.
-	 */
-	sender = mq->mq_sender;
-	Assert(sender != NULL);
-	SetLatch(&sender->procLatch);
 }
 
 /*