[PATCH 0/4] thunderbolt: More improvements to USB4STREAM

[PATCH 0/4] thunderbolt: More improvements to USB4STREAM

[Mika Westerberg]

Hi all,

Alan noticed that in certain workloads with concurrent reader and writer
the stream could stall for a while. These below improvements should make it
behave better in these cases.

Mika Westerberg (4):
  thunderbolt: Write descriptors in tb_ring_poll()
  thunderbolt: Add tb_ring_poll_pending()
  thunderbolt: stream: Do not hold the lock while busy polling
  thunderbolt: stream: Check the Rx ring before task starts sleeping

 drivers/thunderbolt/nhi.c    | 26 ++++++++++++++++++++++++++
 drivers/thunderbolt/stream.c | 36 ++++++++++++++++++++++++++++--------
 include/linux/thunderbolt.h  |  1 +
 3 files changed, 55 insertions(+), 8 deletions(-)

-- 
2.50.1

[PATCH 1/4] thunderbolt: Write descriptors in tb_ring_poll()

[Mika Westerberg]

When the ring is not in interrupt mode, it depends on something
enqueuing new frames to write the descriptors from ring->queue to the
hardware, and that can take a long time. Since we know that we just
released one slot, we can push the next frame directly from
tb_ring_poll().

Reported-by: Alan Borzeszkowski <alan.borzeszkowski@linux.intel.com>
Assisted-by: LLM
Signed-off-by: Mika Westerberg <mika.westerberg@linux.intel.com>
---
 drivers/thunderbolt/nhi.c | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/drivers/thunderbolt/nhi.c b/drivers/thunderbolt/nhi.c
index a7e6184cdfe1..371ee155d98f 100644
--- a/drivers/thunderbolt/nhi.c
+++ b/drivers/thunderbolt/nhi.c
@@ -440,6 +440,13 @@ struct ring_frame *tb_ring_poll(struct tb_ring *ring)
 		}
 
 		ring->tail = (ring->tail + 1) % ring->size;
+
+		/*
+		 * There is one more slot available so we can push next
+		 * descriptor to the ring. This is needed when the ring
+		 * is created with %RING_FLAG_NO_INTERRUPT.
+		 */
+		ring_write_descriptors(ring, true);
 	}
 
 unlock:
-- 
2.50.1

[PATCH 2/4] thunderbolt: Add tb_ring_poll_pending()

[Mika Westerberg]

This helper can be used to figure out whether tb_ring_poll() has a frame
to be returned. This can be used without the caller needing to take its
own lock.

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

diff --git a/drivers/thunderbolt/nhi.c b/drivers/thunderbolt/nhi.c
index 371ee155d98f..020db5a1f029 100644
--- a/drivers/thunderbolt/nhi.c
+++ b/drivers/thunderbolt/nhi.c
@@ -455,6 +455,25 @@ struct ring_frame *tb_ring_poll(struct tb_ring *ring)
 }
 EXPORT_SYMBOL_GPL(tb_ring_poll);
 
+/**
+ * tb_ring_poll_pending() - Does the ring have completed frame
+ * @ring: Ring to check
+ *
+ * Can be used to check whether there is a completed frame in the ring
+ * that next call to tb_ring_poll() returns.
+ *
+ * Return: %true if a completed frame is waiting, %false otherwise.
+ */
+bool tb_ring_poll_pending(struct tb_ring *ring)
+{
+	guard(spinlock_irqsave)(&ring->lock);
+
+	if (!ring->running || ring_empty(ring))
+		return false;
+	return !!(ring->descriptors[ring->tail].flags & RING_DESC_COMPLETED);
+}
+EXPORT_SYMBOL_GPL(tb_ring_poll_pending);
+
 static void __ring_interrupt_mask(struct tb_ring *ring, bool mask)
 {
 	int interrupt_index = ring_interrupt_index(ring);
diff --git a/include/linux/thunderbolt.h b/include/linux/thunderbolt.h
index 8ca941e8b5d0..779381ebeea2 100644
--- a/include/linux/thunderbolt.h
+++ b/include/linux/thunderbolt.h
@@ -757,6 +757,7 @@ static inline int tb_ring_tx_more(struct tb_ring *ring, struct ring_frame *frame
 
 /* Used only when the ring is in polling mode */
 struct ring_frame *tb_ring_poll(struct tb_ring *ring);
+bool tb_ring_poll_pending(struct tb_ring *ring);
 void tb_ring_poll_complete(struct tb_ring *ring);
 
 int tb_ring_throttling(struct tb_ring *ring, unsigned int interval_nsec);
-- 
2.50.1

[PATCH 3/4] thunderbolt: stream: Do not hold the lock while busy polling

[Mika Westerberg]

When in busy poll mode with a reader and writer, and there is nothing
completed on the rings both sides keep taking and releasing the lock
from each other and this can cause stalls. Avoid this by calling the new
API function tb_ring_poll_pending() that only returns true if there is
anything available in which case they can take the lock and know there
is now work to do.

Reported-by: Alan Borzeszkowski <alan.borzeszkowski@linux.intel.com>
Assisted-by: LLM
Signed-off-by: Mika Westerberg <mika.westerberg@linux.intel.com>
---
 drivers/thunderbolt/stream.c | 28 ++++++++++++++++++++++------
 1 file changed, 22 insertions(+), 6 deletions(-)

diff --git a/drivers/thunderbolt/stream.c b/drivers/thunderbolt/stream.c
index 7b618d8a7d82..b4f3b1db9d82 100644
--- a/drivers/thunderbolt/stream.c
+++ b/drivers/thunderbolt/stream.c
@@ -734,6 +734,22 @@ static int tbstream_dev_lock(struct tbstream_dev *sdev, bool nowait)
 	return 0;
 }
 
+/* Must not be called with @sdev->lock held */
+static int tbstream_dev_busy_poll_wait(struct tbstream_dev *sdev,
+				       struct tbstream_ring *ring)
+{
+	for (;;) {
+		if (signal_pending(current))
+			return -ERESTARTSYS;
+		if (tb_ring_poll_pending(ring->ring))
+			return 0;
+		if (tbstream_dev_valid(sdev) != 0 ||
+		    tbstream_dev_closed(sdev) || tbstream_dev_removed(sdev))
+			return 0;
+		cond_resched();
+	}
+}
+
 static ssize_t
 tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to)
 {
@@ -777,9 +793,9 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to)
 			return -EAGAIN;
 
 		if (sdev->busy_poll) {
-			if (signal_pending(current))
-				return -ERESTARTSYS;
-			cond_resched();
+			ret = tbstream_dev_busy_poll_wait(sdev, &sdev->rx_ring);
+			if (ret)
+				return ret;
 		} else {
 			ret = wait_event_interruptible(sdev->wait,
 					READ_ONCE(sdev->rx_pending) ||
@@ -891,9 +907,9 @@ tbstream_dev_fops_write_iter(struct kiocb *kiocb, struct iov_iter *from)
 			return -EAGAIN;
 
 		if (sdev->busy_poll) {
-			if (signal_pending(current))
-				return -ERESTARTSYS;
-			cond_resched();
+			ret = tbstream_dev_busy_poll_wait(sdev, &sdev->tx_ring);
+			if (ret)
+				return ret;
 		} else {
 			ret = wait_event_interruptible(sdev->wait,
 					tbstream_ring_available(&sdev->tx_ring) ||
-- 
2.50.1

[PATCH 4/4] thunderbolt: stream: Check the Rx ring before task starts sleeping

[Mika Westerberg]

In interrupt mode when tb_ring_poll_complete() is called it is possible
that a frame was completed right before the ring interrupt was unmasked
in which case there will be no interrupt and the task starts sleeping.
Avoid this and check if there is anything completed prior putting the
task to sleep. The same can happen with poll() so check the ring there
too before reporting that there is nothing to read.

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

diff --git a/drivers/thunderbolt/stream.c b/drivers/thunderbolt/stream.c
index b4f3b1db9d82..e2b4896dfafb 100644
--- a/drivers/thunderbolt/stream.c
+++ b/drivers/thunderbolt/stream.c
@@ -799,6 +799,7 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to)
 		} else {
 			ret = wait_event_interruptible(sdev->wait,
 					READ_ONCE(sdev->rx_pending) ||
+					tb_ring_poll_pending(sdev->rx_ring.ring) ||
 					tbstream_ring_available(&sdev->rx_ring) ||
 					tbstream_dev_valid(sdev) != 0 ||
 					tbstream_dev_closed(sdev) ||
@@ -977,10 +978,13 @@ tbstream_dev_fops_poll(struct file *file, struct poll_table_struct *wait)
 
 	if (tbstream_ring_available(&sdev->tx_ring))
 		mask |= EPOLLOUT | EPOLLWRNORM;
-	if (tbstream_ring_available(&sdev->rx_ring))
+	if (tbstream_ring_available(&sdev->rx_ring)) {
 		mask |= EPOLLIN | EPOLLRDNORM;
-	else
+	} else {
 		tbstream_dev_complete_rx(sdev);
+		if (tb_ring_poll_pending(sdev->rx_ring.ring))
+			mask |= EPOLLIN | EPOLLRDNORM;
+	}
 
 	return mask;
 }
-- 
2.50.1