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

[Mika Westerberg <mika.westerberg@linux.intel.com>]

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