summaryrefslogtreecommitdiffstats
path: root/src/sink.c
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/sink.c14
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);