Adding upstream version 2.9.7.upstream/2.9.7

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 11 insertions, 3 deletions

diff --git a/src/sink.c b/src/sink.c
index 66c2b8c..58adfcd 100644
--- a/src/sink.c
+++ b/src/sink.c
@@ -439,7 +439,7 @@ static void sink_forward_oc_io_handler(struct appctx *appctx)
 	struct ring *ring = sink->ctx.ring;
 	struct buffer *buf = &ring->buf;
 	uint64_t msg_len;
-	size_t len, cnt, ofs;
+	size_t len, cnt, ofs, last_ofs;
 	int ret = 0;
 	char *p;
 
@@ -530,16 +530,24 @@ static void sink_forward_oc_io_handler(struct appctx *appctx)
 	}
 
 	HA_ATOMIC_INC(b_peek(buf, ofs));
+	last_ofs = b_tail_ofs(buf);
 	sft->ofs = b_peek_ofs(buf, ofs);
-
 	HA_RWLOCK_RDUNLOCK(RING_LOCK, &ring->lock);
 
 	if (ret) {
 		/* let's be woken up once new data arrive */
 		HA_RWLOCK_WRLOCK(RING_LOCK, &ring->lock);
 		LIST_APPEND(&ring->waiters, &appctx->wait_entry);
+		ofs = b_tail_ofs(buf);
 		HA_RWLOCK_WRUNLOCK(RING_LOCK, &ring->lock);
-		applet_have_no_more_data(appctx);
+		if (ofs != last_ofs) {
+			/* more data was added into the ring between the
+			 * unlock and the lock, and the writer might not
+			 * have seen us. We need to reschedule a read.
+			 */
+			applet_have_more_data(appctx);
+		} else
+			applet_have_no_more_data(appctx);
 	}
 	HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);