mq-threshold-wip.patch

diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
index 2c79a649f46..71733612fb5 100644
--- a/src/backend/storage/ipc/shm_mq.c
+++ b/src/backend/storage/ipc/shm_mq.c
@@ -142,7 +142,9 @@ struct shm_mq_handle
 	char	   *mqh_buffer;
 	Size		mqh_buflen;
 	Size		mqh_consume_pending;
+	Size		mqh_consume_threshold;
 	Size		mqh_send_pending;
+	Size		mqh_send_threshold;
 	Size		mqh_partial_bytes;
 	Size		mqh_expected_bytes;
 	bool		mqh_length_word_complete;
@@ -305,6 +307,10 @@ shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
 	mqh->mqh_counterparty_attached = false;
 	mqh->mqh_context = CurrentMemoryContext;
 
+	/* start at 64B, then gradually ramp up */
+	mqh->mqh_consume_threshold = Min(64, mq->mq_ring_size / 4);
+	mqh->mqh_send_threshold = Min(64, mq->mq_ring_size / 4);
+
 	if (seg != NULL)
 		on_dsm_detach(seg, shm_mq_detach_callback, PointerGetDatum(mq));
 
@@ -535,12 +541,16 @@ shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait,
 	 * 1/4 of the ring size, mark it as written in shared memory and notify
 	 * the receiver.
 	 */
-	if (force_flush || mqh->mqh_send_pending > (mq->mq_ring_size >> 2))
+	if (force_flush || mqh->mqh_send_pending > mqh->mqh_send_threshold)
 	{
 		shm_mq_inc_bytes_written(mq, mqh->mqh_send_pending);
 		if (receiver != NULL)
 			SetLatch(&receiver->procLatch);
 		mqh->mqh_send_pending = 0;
+
+		/* ramp up, up to (mq_ring_size / 4) */
+		mqh->mqh_send_threshold = Min(mqh->mqh_send_threshold * 2,
+									  mq->mq_ring_size / 4);
 	}
 
 	return SHM_MQ_SUCCESS;
@@ -622,10 +632,14 @@ 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_consume_pending > mqh->mqh_consume_threshold)
 	{
 		shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
 		mqh->mqh_consume_pending = 0;
+
+		/* ramp up, up to (mq_ring_size / 4) */
+		mqh->mqh_consume_threshold = Min(mqh->mqh_consume_threshold * 2,
+										 mq->mq_ring_size / 4);
 	}
 
 	/* Try to read, or finish reading, the length word from the buffer. */