[PATCH 0/4] thunderbolt: Couple of improvements for USB4STREAM

[PATCH 0/4] thunderbolt: Couple of improvements for USB4STREAM

[Mika Westerberg]

Hi all,

This includes a few cleanups and then two improvements for USB4STREAM,
useful when it is in the default "interrupt" mode. First one is to avoid
read-modify-write on the interrupt mask path and second one is to use the
same ->start_poll() mechanism that the Thunderbolt/USB4 networking driver
uses.

Mika Westerberg (4):
  thunderbolt: Do not WARN about already disabled interrupt on polled rings
  thunderbolt: Clean up ring interrupt register indexing
  thunderbolt: Use shadow copy for ring interrupt mask
  thunderbolt: stream: Use polling with RX ring

 drivers/thunderbolt/nhi.c    | 104 ++++++++++++++++++++--------------
 drivers/thunderbolt/stream.c | 105 ++++++++++++++++++++++++-----------
 include/linux/thunderbolt.h  |   2 +
 3 files changed, 138 insertions(+), 73 deletions(-)

-- 
2.50.1

[PATCH 1/4] thunderbolt: Do not WARN about already disabled interrupt on polled rings

[Mika Westerberg]

With ->start_poll it is possible that the ring is actually already
disabled before tb_ring_stop() is called. This is expected behaviour so
there is no point warning about that.

Assisted-by: LLM
Signed-off-by: Mika Westerberg <mika.westerberg@linux.intel.com>
---
 drivers/thunderbolt/nhi.c | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/drivers/thunderbolt/nhi.c b/drivers/thunderbolt/nhi.c
index a5ecfcf127e9..985b53399561 100644
--- a/drivers/thunderbolt/nhi.c
+++ b/drivers/thunderbolt/nhi.c
@@ -139,10 +139,19 @@ static void ring_interrupt_active(struct tb_ring *ring, bool active)
 		"%s interrupt at register %#x bit %d (%#x -> %#x)\n",
 		active ? "enabling" : "disabling", reg, interrupt_bit, old, new);
 
-	if (new == old)
-		dev_WARN(ring->nhi->dev, "interrupt for %s %d is already %s\n",
-			 RING_TYPE(ring), ring->hop,
-			 str_enabled_disabled(active));
+	if (new == old) {
+		/*
+		 * Rings that are polled mask the interrupt using while
+		 * the completions are being advanced (see
+		 * __ring_interrupt()) so for those it can already be
+		 * disabled by the time the ring is stopped.
+		 */
+		if (active || !ring->start_poll)
+			dev_WARN(ring->nhi->dev,
+				 "interrupt for %s %d is already %s\n",
+				 RING_TYPE(ring), ring->hop,
+				 str_enabled_disabled(active));
+	}
 
 	if (active)
 		iowrite32(new, ring->nhi->iobase + reg);
-- 
2.50.1

[PATCH 2/4] thunderbolt: Clean up ring interrupt register indexing

[Mika Westerberg]

nhi_mask_interrupt() and nhi_clear_interrupt() take "ring" as parameter
but in fact it is not an actual ring but a byte offset to the interrupt
register. Make this less confusing and name the paramers what it really
is and calculate the offset where it is actually needed.

In addition ring_interrupt_active() has two variables called "index"
with different meanings, and the second one shadows the first one
open-coding ring_interrupt_index() as well. Drop that and rename the
remaining what they actually hold.

While there, make the mask variable u32 and use BIT() to avoid signed
shifting.

No functional changes intended.

Assisted-by: LLM
Signed-off-by: Mika Westerberg <mika.westerberg@linux.intel.com>
---
 drivers/thunderbolt/nhi.c | 44 +++++++++++++++++++--------------------
 1 file changed, 22 insertions(+), 22 deletions(-)

diff --git a/drivers/thunderbolt/nhi.c b/drivers/thunderbolt/nhi.c
index 985b53399561..aaaf9e7966c8 100644
--- a/drivers/thunderbolt/nhi.c
+++ b/drivers/thunderbolt/nhi.c
@@ -41,6 +41,7 @@ static bool host_reset = true;
 module_param(host_reset, bool, 0444);
 MODULE_PARM_DESC(host_reset, "reset USB4 host router (default: true)");
 
+/* Returns absolute bit number of the ring in the interrupt registers */
 static int ring_interrupt_index(const struct tb_ring *ring)
 {
 	int bit = ring->hop;
@@ -49,24 +50,28 @@ static int ring_interrupt_index(const struct tb_ring *ring)
 	return bit;
 }
 
-static void nhi_mask_interrupt(struct tb_nhi *nhi, int mask, int ring)
+static void nhi_mask_interrupt(struct tb_nhi *nhi, u32 mask, int reg_index)
 {
+	int offset = reg_index * 4;
+
 	if (nhi->quirks & QUIRK_AUTO_CLEAR_INT) {
 		u32 val;
 
-		val = ioread32(nhi->iobase + REG_RING_INTERRUPT_BASE + ring);
-		iowrite32(val & ~mask, nhi->iobase + REG_RING_INTERRUPT_BASE + ring);
+		val = ioread32(nhi->iobase + REG_RING_INTERRUPT_BASE + offset);
+		iowrite32(val & ~mask, nhi->iobase + REG_RING_INTERRUPT_BASE + offset);
 	} else {
-		iowrite32(mask, nhi->iobase + REG_RING_INTERRUPT_MASK_CLEAR_BASE + ring);
+		iowrite32(mask, nhi->iobase + REG_RING_INTERRUPT_MASK_CLEAR_BASE + offset);
 	}
 }
 
-static void nhi_clear_interrupt(struct tb_nhi *nhi, int ring)
+static void nhi_clear_interrupt(struct tb_nhi *nhi, int reg_index)
 {
+	int offset = reg_index * 4;
+
 	if (nhi->quirks & QUIRK_AUTO_CLEAR_INT)
-		ioread32(nhi->iobase + REG_RING_NOTIFY_BASE + ring);
+		ioread32(nhi->iobase + REG_RING_NOTIFY_BASE + offset);
 	else
-		iowrite32(~0, nhi->iobase + REG_RING_INT_CLEAR + ring);
+		iowrite32(~0, nhi->iobase + REG_RING_INT_CLEAR + offset);
 }
 
 /*
@@ -76,22 +81,17 @@ static void nhi_clear_interrupt(struct tb_nhi *nhi, int ring)
  */
 static void ring_interrupt_active(struct tb_ring *ring, bool active)
 {
-	int index = ring_interrupt_index(ring) / 32 * 4;
-	int reg = REG_RING_INTERRUPT_BASE + index;
-	int interrupt_bit = ring_interrupt_index(ring) & 31;
-	int mask = 1 << interrupt_bit;
+	int interrupt_index = ring_interrupt_index(ring);
+	int reg_index = interrupt_index / 32;
+	int reg = REG_RING_INTERRUPT_BASE + reg_index * 4;
+	int interrupt_bit = interrupt_index % 32;
+	u32 mask = BIT(interrupt_bit);
 	u32 old, new;
 
 	if (ring->irq > 0) {
 		u32 step, shift, ivr, misc, itr;
 		void __iomem *ivr_base;
 		int auto_clear_bit;
-		int index;
-
-		if (ring->is_tx)
-			index = ring->hop;
-		else
-			index = ring->hop + ring->nhi->hop_count;
 
 		/*
 		 * Intel routers support a bit that isn't part of
@@ -114,8 +114,8 @@ static void ring_interrupt_active(struct tb_ring *ring, bool active)
 				  ring->nhi->iobase + REG_DMA_MISC);
 
 		ivr_base = ring->nhi->iobase + REG_INT_VEC_ALLOC_BASE;
-		step = index / REG_INT_VEC_ALLOC_REGS * REG_INT_VEC_ALLOC_BITS;
-		shift = index % REG_INT_VEC_ALLOC_REGS * REG_INT_VEC_ALLOC_BITS;
+		step = interrupt_index / REG_INT_VEC_ALLOC_REGS * REG_INT_VEC_ALLOC_BITS;
+		shift = interrupt_index % REG_INT_VEC_ALLOC_REGS * REG_INT_VEC_ALLOC_BITS;
 		ivr = ioread32(ivr_base + step);
 		ivr &= ~(REG_INT_VEC_ALLOC_MASK << shift);
 		if (active)
@@ -156,7 +156,7 @@ static void ring_interrupt_active(struct tb_ring *ring, bool active)
 	if (active)
 		iowrite32(new, ring->nhi->iobase + reg);
 	else
-		nhi_mask_interrupt(ring->nhi, mask, index);
+		nhi_mask_interrupt(ring->nhi, mask, reg_index);
 }
 
 /*
@@ -169,11 +169,11 @@ void nhi_disable_interrupts(struct tb_nhi *nhi)
 	int i = 0;
 	/* disable interrupts */
 	for (i = 0; i < RING_INTERRUPT_REG_COUNT(nhi); i++)
-		nhi_mask_interrupt(nhi, ~0, 4 * i);
+		nhi_mask_interrupt(nhi, ~0, i);
 
 	/* clear interrupt status bits */
 	for (i = 0; i < RING_NOTIFY_REG_COUNT(nhi); i++)
-		nhi_clear_interrupt(nhi, 4 * i);
+		nhi_clear_interrupt(nhi, i);
 }
 
 /* ring helper methods */
-- 
2.50.1

[PATCH 3/4] thunderbolt: Use shadow copy for ring interrupt mask

[Mika Westerberg]

Each time we mask or unmask ring interrupt we do a read-modify-write and
that is not without a cost when on a hot path. We can get rid of that
read by making a shadow copy of the interrupt mask register and using
that in place of a register read.

Assisted-by: LLM
Signed-off-by: Mika Westerberg <mika.westerberg@linux.intel.com>
---
 drivers/thunderbolt/nhi.c   | 55 +++++++++++++++++++++++--------------
 include/linux/thunderbolt.h |  2 ++
 2 files changed, 36 insertions(+), 21 deletions(-)

diff --git a/drivers/thunderbolt/nhi.c b/drivers/thunderbolt/nhi.c
index aaaf9e7966c8..a7e6184cdfe1 100644
--- a/drivers/thunderbolt/nhi.c
+++ b/drivers/thunderbolt/nhi.c
@@ -53,15 +53,28 @@ static int ring_interrupt_index(const struct tb_ring *ring)
 static void nhi_mask_interrupt(struct tb_nhi *nhi, u32 mask, int reg_index)
 {
 	int offset = reg_index * 4;
+	u32 val;
 
-	if (nhi->quirks & QUIRK_AUTO_CLEAR_INT) {
-		u32 val;
+	/* Use shadow copy instead of reading the register */
+	val = nhi->interrupt_mask[reg_index] & ~mask;
+	nhi->interrupt_mask[reg_index] = val;
 
-		val = ioread32(nhi->iobase + REG_RING_INTERRUPT_BASE + offset);
-		iowrite32(val & ~mask, nhi->iobase + REG_RING_INTERRUPT_BASE + offset);
-	} else {
+	if (nhi->quirks & QUIRK_AUTO_CLEAR_INT)
+		iowrite32(val, nhi->iobase + REG_RING_INTERRUPT_BASE + offset);
+	else
 		iowrite32(mask, nhi->iobase + REG_RING_INTERRUPT_MASK_CLEAR_BASE + offset);
-	}
+}
+
+static void nhi_unmask_interrupt(struct tb_nhi *nhi, u32 mask, int reg_index)
+{
+	int offset = reg_index * 4;
+	u32 val;
+
+	/* Use shadow copy instead of reading the register */
+	val = nhi->interrupt_mask[reg_index] | mask;
+	nhi->interrupt_mask[reg_index] = val;
+
+	iowrite32(val, nhi->iobase + REG_RING_INTERRUPT_BASE + offset);
 }
 
 static void nhi_clear_interrupt(struct tb_nhi *nhi, int reg_index)
@@ -129,7 +142,7 @@ static void ring_interrupt_active(struct tb_ring *ring, bool active)
 			  ring->vector * 4);
 	}
 
-	old = ioread32(ring->nhi->iobase + reg);
+	old = ring->nhi->interrupt_mask[reg_index];
 	if (active)
 		new = old | mask;
 	else
@@ -141,10 +154,10 @@ static void ring_interrupt_active(struct tb_ring *ring, bool active)
 
 	if (new == old) {
 		/*
-		 * Rings that are polled mask the interrupt using while
-		 * the completions are being advanced (see
-		 * __ring_interrupt()) so for those it can already be
-		 * disabled by the time the ring is stopped.
+		 * Rings that are polled mask the interrupt while the
+		 * completions are being advanced (see __ring_interrupt())
+		 * so for those it can already be disabled by the time
+		 * the ring is stopped.
 		 */
 		if (active || !ring->start_poll)
 			dev_WARN(ring->nhi->dev,
@@ -154,7 +167,7 @@ static void ring_interrupt_active(struct tb_ring *ring, bool active)
 	}
 
 	if (active)
-		iowrite32(new, ring->nhi->iobase + reg);
+		nhi_unmask_interrupt(ring->nhi, mask, reg_index);
 	else
 		nhi_mask_interrupt(ring->nhi, mask, reg_index);
 }
@@ -437,17 +450,14 @@ EXPORT_SYMBOL_GPL(tb_ring_poll);
 
 static void __ring_interrupt_mask(struct tb_ring *ring, bool mask)
 {
-	int idx = ring_interrupt_index(ring);
-	int reg = REG_RING_INTERRUPT_BASE + idx / 32 * 4;
-	int bit = idx % 32;
-	u32 val;
+	int interrupt_index = ring_interrupt_index(ring);
+	int reg_index = interrupt_index / 32;
+	int interrupt_bit = interrupt_index % 32;
 
-	val = ioread32(ring->nhi->iobase + reg);
 	if (mask)
-		val &= ~BIT(bit);
+		nhi_mask_interrupt(ring->nhi, BIT(interrupt_bit), reg_index);
 	else
-		val |= BIT(bit);
-	iowrite32(val, ring->nhi->iobase + reg);
+		nhi_unmask_interrupt(ring->nhi, BIT(interrupt_bit), reg_index);
 }
 
 /* Both @nhi->lock and @ring->lock should be held */
@@ -1315,7 +1325,10 @@ int nhi_probe(struct tb_nhi *nhi)
 				     sizeof(*nhi->tx_rings), GFP_KERNEL);
 	nhi->rx_rings = devm_kcalloc(dev, nhi->hop_count,
 				     sizeof(*nhi->rx_rings), GFP_KERNEL);
-	if (!nhi->tx_rings || !nhi->rx_rings)
+	nhi->interrupt_mask = devm_kcalloc(dev, RING_INTERRUPT_REG_COUNT(nhi),
+					   sizeof(*nhi->interrupt_mask),
+					   GFP_KERNEL);
+	if (!nhi->tx_rings || !nhi->rx_rings || !nhi->interrupt_mask)
 		return -ENOMEM;
 
 	nhi_reset(nhi);
diff --git a/include/linux/thunderbolt.h b/include/linux/thunderbolt.h
index 7fb9e7e1aae4..8ca941e8b5d0 100644
--- a/include/linux/thunderbolt.h
+++ b/include/linux/thunderbolt.h
@@ -507,6 +507,7 @@ void tb_service_properties_changed(struct tb_service *svc);
  * @iobase: MMIO space of the NHI
  * @tx_rings: All Tx rings available on this host controller
  * @rx_rings: All Rx rings available on this host controller
+ * @interrupt_mask: Shadow copy of the ring interrupt mask register
  * @going_away: The host controller device is about to disappear so when
  *		this flag is set, avoid touching the hardware anymore.
  * @iommu_dma_protection: An IOMMU will isolate external-facing ports.
@@ -528,6 +529,7 @@ struct tb_nhi {
 	void __iomem *iobase;
 	struct tb_ring **tx_rings;
 	struct tb_ring **rx_rings;
+	u32 *interrupt_mask;
 	bool going_away;
 	bool iommu_dma_protection;
 	struct work_struct interrupt_work;
-- 
2.50.1

[PATCH 4/4] thunderbolt: stream: Use polling with RX ring

[Mika Westerberg]

The ring API already supports polling so that once there is anything in
the RX ring, it will mask the ring interrupt and call the ->start_poll
callback. This is the same mechanism that the USB4/Thunderbolt
networking driver already uses.

We can do the same for the USB4STREAM driver and poll any RX frames
while they are available, until we re-enable the ring interrupt again.
This provides better latency when the stream is in "normal"
(non-busy-polling) mode.

Assisted-by: LLM
Signed-off-by: Mika Westerberg <mika.westerberg@linux.intel.com>
---
 drivers/thunderbolt/stream.c | 105 ++++++++++++++++++++++++-----------
 1 file changed, 73 insertions(+), 32 deletions(-)

diff --git a/drivers/thunderbolt/stream.c b/drivers/thunderbolt/stream.c
index c737dd0ca6e7..56e155750466 100644
--- a/drivers/thunderbolt/stream.c
+++ b/drivers/thunderbolt/stream.c
@@ -131,6 +131,7 @@ struct tbstream_ring {
  * @ring_size: Size of the rings
  * @throttling: Interrupt throttling rate in ns
  * @busy_poll: Instead of interrupts, busy poll the rings
+ * @rx_pending: Receive ring has completions that need to be advanced
  * @users: Number of times @cdev has been opened
  * @closed: CLOSE packet was received
  * @removed: Userspace removed the ConfigFS group underneath.
@@ -151,6 +152,7 @@ struct tbstream_dev {
 	unsigned int ring_size;
 	unsigned int throttling;
 	bool busy_poll;
+	bool rx_pending;
 	int users;
 	bool closed;
 	bool removed;
@@ -278,6 +280,14 @@ static inline bool tbstream_ring_available(const struct tbstream_ring *ring)
 	return ring->prod > ring->cons;
 }
 
+static void tbstream_ring_poll(struct tbstream_ring *ring)
+{
+	struct ring_frame *frame;
+
+	while ((frame = tb_ring_poll(ring->ring)))
+		frame->callback(ring->ring, frame, false);
+}
+
 static inline struct tb_xdomain *tbstream_dev_xdomain(struct tbstream_dev *sdev)
 {
 	if (sdev->stream)
@@ -540,18 +550,6 @@ tbstream_dev_send_data(struct tbstream_dev *sdev, struct iov_iter *from,
 	return tb_ring_tx(sdev->tx_ring.ring, &sf->frame);
 }
 
-static void
-tbstream_dev_poll_ring(struct tbstream_dev *sdev, struct tbstream_ring *ring)
-{
-	struct ring_frame *frame;
-
-	if (!sdev->busy_poll)
-		return;
-
-	while ((frame = tb_ring_poll(ring->ring)))
-		frame->callback(ring->ring, frame, false);
-}
-
 static int tbstream_dev_send_close(struct tbstream_dev *sdev)
 {
 	struct tbstream_frame *sf;
@@ -568,7 +566,7 @@ static int tbstream_dev_send_close(struct tbstream_dev *sdev)
 		do {
 			if (tbstream_ring_available(&sdev->tx_ring))
 				break;
-			tbstream_dev_poll_ring(sdev, &sdev->tx_ring);
+			tbstream_ring_poll(&sdev->tx_ring);
 			fsleep(15);
 		} while (ktime_before(ktime_get(), timeout));
 	}
@@ -579,16 +577,45 @@ static int tbstream_dev_send_close(struct tbstream_dev *sdev)
 	return tb_ring_tx(sdev->tx_ring.ring, &sf->frame);
 }
 
+static void tbstream_dev_start_poll(void *data)
+{
+	struct tbstream_dev *sdev = data;
+
+	WRITE_ONCE(sdev->rx_pending, true);
+	wake_up_interruptible_poll(&sdev->wait, EPOLLIN | EPOLLRDNORM);
+}
+
+/* sdev->lock must be held */
+static void tbstream_dev_advance_rx(struct tbstream_dev *sdev)
+{
+	/*
+	 * Clear before running the completions so that an interrupt
+	 * that arrives while we are doing that is not missed.
+	 */
+	WRITE_ONCE(sdev->rx_pending, false);
+	tbstream_ring_poll(&sdev->rx_ring);
+}
+
+/* sdev->lock must be held */
+static void tbstream_dev_complete_rx(struct tbstream_dev *sdev)
+{
+	if (!sdev->busy_poll)
+		tb_ring_poll_complete(sdev->rx_ring.ring);
+}
+
 static int tbstream_dev_start(struct tbstream_dev *sdev)
 {
 	struct tb_xdomain *xd = tbstream_dev_xdomain(sdev);
 	unsigned int flags = RING_FLAG_FRAME | RING_FLAG_E2E;
+	void (*start_poll)(void *) = NULL;
 	u16 sof_mask, eof_mask;
 	struct tb_ring *ring;
 	int ret, e2e_tx_hop;
 
 	if (sdev->busy_poll)
 		flags |= RING_FLAG_NO_INTERRUPT;
+	else
+		start_poll = tbstream_dev_start_poll;
 
 	ring = tb_ring_alloc_tx(xd->tb->nhi, -1, sdev->ring_size, flags);
 	if (!ring)
@@ -604,7 +631,8 @@ static int tbstream_dev_start(struct tbstream_dev *sdev)
 	eof_mask = BIT(TBSTREAM_DATA) | BIT(TBSTREAM_CLOSE);
 
 	ring = tb_ring_alloc_rx(xd->tb->nhi, -1, sdev->ring_size, flags,
-				e2e_tx_hop, sof_mask, eof_mask, NULL, NULL);
+				e2e_tx_hop, sof_mask, eof_mask, start_poll,
+				sdev);
 	if (!ring) {
 		ret = -ENOMEM;
 		goto err_free_tx_buffers;
@@ -621,6 +649,8 @@ static int tbstream_dev_start(struct tbstream_dev *sdev)
 	tb_ring_throttling(sdev->tx_ring.ring, sdev->throttling);
 	tb_ring_throttling(sdev->rx_ring.ring, sdev->throttling);
 
+	sdev->rx_pending = false;
+
 	tb_ring_start(sdev->tx_ring.ring);
 	tb_ring_start(sdev->rx_ring.ring);
 
@@ -667,19 +697,16 @@ static void tbstream_dev_stop(struct tbstream_dev *sdev)
 		do {
 			if (tbstream_dev_tx_drained(sdev))
 				break;
-			tbstream_dev_poll_ring(sdev, &sdev->tx_ring);
+			tbstream_ring_poll(&sdev->tx_ring);
 			fsleep(15);
 		} while (ktime_before(ktime_get(), timeout));
-
-		tb_ring_stop(sdev->tx_ring.ring);
-		tb_ring_stop(sdev->rx_ring.ring);
 	} else {
 		tb_ring_flush(sdev->tx_ring.ring, 500);
-		tb_ring_stop(sdev->tx_ring.ring);
-		tb_ring_flush(sdev->rx_ring.ring, 500);
-		tb_ring_stop(sdev->rx_ring.ring);
 	}
 
+	tb_ring_stop(sdev->tx_ring.ring);
+	tb_ring_stop(sdev->rx_ring.ring);
+
 	xd = tbstream_dev_xdomain(sdev);
 	if (xd) {
 		tb_xdomain_disable_paths(xd, sdev->out_hopid,
@@ -727,8 +754,8 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to)
 		return ret;
 
 	for (;;) {
-		/* When busy polling, advance any completions manually */
-		tbstream_dev_poll_ring(sdev, &sdev->rx_ring);
+		/* Advance RX completions */
+		tbstream_dev_advance_rx(sdev);
 
 		ret = tbstream_dev_valid(sdev);
 		if (ret) {
@@ -744,6 +771,8 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to)
 		if (tbstream_ring_available(&sdev->rx_ring))
 			break;
 
+		/* Polled all we could. Re-enable the interrupt now. */
+		tbstream_dev_complete_rx(sdev);
 		mutex_unlock(&sdev->lock);
 
 		if (nowait)
@@ -755,6 +784,7 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to)
 			cond_resched();
 		} else {
 			ret = wait_event_interruptible(sdev->wait,
+					READ_ONCE(sdev->rx_pending) ||
 					tbstream_ring_available(&sdev->rx_ring) ||
 					tbstream_dev_valid(sdev) != 0 ||
 					tbstream_dev_closed(sdev) ||
@@ -839,7 +869,9 @@ tbstream_dev_fops_write_iter(struct kiocb *kiocb, struct iov_iter *from)
 		return ret;
 
 	for (;;) {
-		tbstream_dev_poll_ring(sdev, &sdev->tx_ring);
+		/* When busy polling, advance any completions manually */
+		if (sdev->busy_poll)
+			tbstream_ring_poll(&sdev->tx_ring);
 
 		ret = tbstream_dev_valid(sdev);
 		if (ret) {
@@ -919,14 +951,23 @@ tbstream_dev_fops_poll(struct file *file, struct poll_table_struct *wait)
 
 	poll_wait(file, &sdev->wait, wait);
 	guard(mutex)(&sdev->lock);
-	if (tbstream_dev_valid(sdev) != 0) {
-		mask |= EPOLLHUP | EPOLLERR;
-	} else {
-		if (tbstream_ring_available(&sdev->tx_ring))
-			mask |= EPOLLOUT | EPOLLWRNORM;
-		if (tbstream_ring_available(&sdev->rx_ring))
-			mask |= EPOLLIN | EPOLLRDNORM;
-	}
+	if (tbstream_dev_valid(sdev) != 0)
+		return EPOLLHUP | EPOLLERR;
+
+	/*
+	 * The RX completions are only advanced from here and from
+	 * read(2) so do that now, otherwise we would never report
+	 * anything to be available.
+	 */
+	tbstream_dev_advance_rx(sdev);
+
+	if (tbstream_ring_available(&sdev->tx_ring))
+		mask |= EPOLLOUT | EPOLLWRNORM;
+	if (tbstream_ring_available(&sdev->rx_ring))
+		mask |= EPOLLIN | EPOLLRDNORM;
+	else
+		tbstream_dev_complete_rx(sdev);
+
 	return mask;
 }
 
-- 
2.50.1