[PATCH 0/2] thunderbolt / net: Allow batching of descriptors

[PATCH 0/2] thunderbolt / net: Allow batching of descriptors

[Mika Westerberg]

Hi all,

This series makes the Thunderbolt/USB4 networking driver batch the
desriptors to avoid writing the doorbell register for each separately.
This improves throughput.

I can take these through Thunderbolt/USB4 tree as the 2/2 depends on the
1/2.

Mika Westerberg (2):
  thunderbolt: Allow batching of descriptors
  net: thunderbolt: Update ring indices only after all frames are queued

 drivers/net/thunderbolt/main.c |  6 ++--
 drivers/thunderbolt/nhi.c      | 58 +++++++++++++++++++++++++++++-----
 include/linux/thunderbolt.h    | 42 ++++++++++++++++++++++--
 3 files changed, 93 insertions(+), 13 deletions(-)

-- 
2.50.1

[PATCH 1/2] thunderbolt: Allow batching of descriptors

[Mika Westerberg]

The hardware allows queueing descriptors ahead of updating producer/consumer
fields. This way it is possible to avoid unnecessary register writes on
hot-paths such as when transferring networking packets. For this reason
introduce an API that allows Thunderbolt service drivers to opt-in for
this and take advantage of batching.

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

diff --git a/drivers/thunderbolt/nhi.c b/drivers/thunderbolt/nhi.c
index 5827c498c628..a5ecfcf127e9 100644
--- a/drivers/thunderbolt/nhi.c
+++ b/drivers/thunderbolt/nhi.c
@@ -227,12 +227,37 @@ static bool ring_empty(struct tb_ring *ring)
 	return ring->head == ring->tail;
 }
 
+static void __ring_notify(struct tb_ring *ring)
+{
+	lockdep_assert_held(&ring->lock);
+
+	if (ring->notify_pending) {
+		/*
+		 * The doorbell carries the absolute index of the head
+		 * so a single write covers all the descriptors posted
+		 * since the previous one.
+		 */
+		if (ring->is_tx)
+			ring_iowrite_prod(ring, ring->head);
+		else
+			ring_iowrite_cons(ring, ring->head);
+	}
+	ring->notify_pending = false;
+}
+
 /*
  * ring_write_descriptors() - post frames from ring->queue to the controller
+ * @ring: Ring to post the frames to
+ * @notify: Notify the controller about the posted descriptors
+ *
+ * Unless @notify is %true the controller is not notified about the posted
+ * descriptors and the caller is expected to call tb_ring_notify() once it
+ * is done queuing frames. This allows batching of frames before
+ * updating the producer/consumer indices.
  *
  * ring->lock is held.
  */
-static void ring_write_descriptors(struct tb_ring *ring)
+static void ring_write_descriptors(struct tb_ring *ring, bool notify)
 {
 	struct ring_frame *frame, *n;
 	struct ring_desc *descriptor;
@@ -256,11 +281,11 @@ static void ring_write_descriptors(struct tb_ring *ring)
 			descriptor->sof = frame->sof;
 		}
 		ring->head = (ring->head + 1) % ring->size;
-		if (ring->is_tx)
-			ring_iowrite_prod(ring, ring->head);
-		else
-			ring_iowrite_cons(ring, ring->head);
+		ring->notify_pending = true;
 	}
+
+	if (notify)
+		__ring_notify(ring);
 }
 
 /*
@@ -305,7 +330,7 @@ static void ring_work(struct work_struct *work)
 		}
 		ring->tail = (ring->tail + 1) % ring->size;
 	}
-	ring_write_descriptors(ring);
+	ring_write_descriptors(ring, true);
 
 invoke_callback:
 	/* allow callbacks to schedule new work */
@@ -324,7 +349,7 @@ static void ring_work(struct work_struct *work)
 	wake_up(&ring->wait);
 }
 
-int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame)
+int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame, bool more)
 {
 	unsigned long flags;
 	int ret = 0;
@@ -332,7 +357,7 @@ int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame)
 	spin_lock_irqsave(&ring->lock, flags);
 	if (ring->running) {
 		list_add_tail(&frame->list, &ring->queue);
-		ring_write_descriptors(ring);
+		ring_write_descriptors(ring, !more);
 	} else {
 		ret = -ESHUTDOWN;
 	}
@@ -341,6 +366,22 @@ int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame)
 }
 EXPORT_SYMBOL_GPL(__tb_ring_enqueue);
 
+/**
+ * tb_ring_notify() - Notify the controller about the queued frames
+ * @ring: Ring to notify
+ *
+ * Notifies the controller about frames that were enqueued using
+ * tb_ring_tx_more() or tb_ring_rx_more(). Does nothing if there are no
+ * such frames pending.
+ */
+void tb_ring_notify(struct tb_ring *ring)
+{
+	guard(spinlock_irqsave)(&ring->lock);
+	if (ring->running)
+		__ring_notify(ring);
+}
+EXPORT_SYMBOL_GPL(tb_ring_notify);
+
 /**
  * tb_ring_poll() - Poll one completed frame from the ring
  * @ring: Ring to poll
@@ -788,6 +829,7 @@ void tb_ring_stop(struct tb_ring *ring)
 	ring_iowrite32desc(ring, 0, 12);
 	ring->head = 0;
 	ring->tail = 0;
+	ring->notify_pending = false;
 	ring->running = false;
 
 err:
diff --git a/include/linux/thunderbolt.h b/include/linux/thunderbolt.h
index b62dfa52b149..7fb9e7e1aae4 100644
--- a/include/linux/thunderbolt.h
+++ b/include/linux/thunderbolt.h
@@ -553,6 +553,8 @@ struct tb_nhi {
  * @work: Interrupt work structure
  * @is_tx: Is the ring Tx or Rx
  * @running: Is the ring running
+ * @notify_pending: Controller has not been notified about the posted
+ *		    descriptors yet
  * @irq: MSI-X irq number if the ring uses MSI-X. %0 otherwise.
  * @vector: MSI-X vector number the ring uses (only set if @irq is > 0)
  * @flags: Ring specific flags
@@ -582,6 +584,7 @@ struct tb_ring {
 	struct work_struct work;
 	bool is_tx:1;
 	bool running:1;
+	bool notify_pending:1;
 	int irq;
 	u8 vector;
 	unsigned int flags;
@@ -672,7 +675,8 @@ bool tb_ring_flush(struct tb_ring *ring, unsigned int timeout_msec);
 void tb_ring_stop(struct tb_ring *ring);
 void tb_ring_free(struct tb_ring *ring);
 
-int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame);
+int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame, bool more);
+void tb_ring_notify(struct tb_ring *ring);
 
 /**
  * tb_ring_rx() - enqueue a frame on an RX ring
@@ -693,7 +697,24 @@ int __tb_ring_enqueue(struct tb_ring *ring, struct ring_frame *frame);
 static inline int tb_ring_rx(struct tb_ring *ring, struct ring_frame *frame)
 {
 	WARN_ON(ring->is_tx);
-	return __tb_ring_enqueue(ring, frame);
+	return __tb_ring_enqueue(ring, frame, false);
+}
+
+/**
+ * tb_ring_rx_more() - enqueue a frame on an RX ring without notifying
+ * @ring: Ring to enqueue the frame
+ * @frame: Frame to enqueue
+ *
+ * Same as tb_ring_rx() but does not notify the controller about the
+ * enqueued frame. The caller must call tb_ring_notify() once it is done
+ * enqueuing frames.
+ *
+ * Return: %-ESHUTDOWN if tb_ring_stop() has been called, %0 otherwise.
+ */
+static inline int tb_ring_rx_more(struct tb_ring *ring, struct ring_frame *frame)
+{
+	WARN_ON(ring->is_tx);
+	return __tb_ring_enqueue(ring, frame, true);
 }
 
 /**
@@ -714,7 +735,22 @@ static inline int tb_ring_rx(struct tb_ring *ring, struct ring_frame *frame)
 static inline int tb_ring_tx(struct tb_ring *ring, struct ring_frame *frame)
 {
 	WARN_ON(!ring->is_tx);
-	return __tb_ring_enqueue(ring, frame);
+	return __tb_ring_enqueue(ring, frame, false);
+}
+
+/**
+ * tb_ring_tx_more() - enqueue a frame on a TX ring without notifying
+ * @ring: Ring to enqueue the frame
+ * @frame: Frame to enqueue
+ *
+ * Same as tb_ring_rx_more() but for TX ring.
+ *
+ * Return: %-ESHUTDOWN if tb_ring_stop() has been called, %0 otherwise.
+ */
+static inline int tb_ring_tx_more(struct tb_ring *ring, struct ring_frame *frame)
+{
+	WARN_ON(!ring->is_tx);
+	return __tb_ring_enqueue(ring, frame, true);
 }
 
 /* Used only when the ring is in polling mode */
-- 
2.50.1

[PATCH 2/2] net: thunderbolt: Update ring indices only after all frames are queued

[Mika Westerberg]

Take advantage of the new API and update the ring producer/consumer
indices only after the frames have been queued. This reduces the number
of doorbell register writes from one per frame to one per batch.

Measured using iperf3 over a Thunderbolt/USB4 net connection between
Arrow Lake and Panther Lake systems (average of three runs):

  before         after
  ~14.3 Gbit/s   ~15.3 Gbit/s

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

diff --git a/drivers/net/thunderbolt/main.c b/drivers/net/thunderbolt/main.c
index d9fb587a62c5..cf51b9c39f4e 100644
--- a/drivers/net/thunderbolt/main.c
+++ b/drivers/net/thunderbolt/main.c
@@ -540,11 +540,12 @@ static int tbnet_alloc_rx_buffers(struct tbnet *net, unsigned int nbuffers)
 		trace_tbnet_alloc_rx_frame(index, tf->page, dma_addr,
 					   DMA_FROM_DEVICE);
 
-		tb_ring_rx(ring->ring, &tf->frame);
+		tb_ring_rx_more(ring->ring, &tf->frame);
 
 		ring->prod++;
 	}
 
+	tb_ring_notify(ring->ring);
 	return 0;
 
 err_free:
@@ -1243,7 +1244,8 @@ static netdev_tx_t tbnet_start_xmit(struct sk_buff *skb,
 		goto err_drop;
 
 	for (i = 0; i < frame_index + 1; i++)
-		tb_ring_tx(net->tx_ring.ring, &frames[i]->frame);
+		tb_ring_tx_more(net->tx_ring.ring, &frames[i]->frame);
+	tb_ring_notify(net->tx_ring.ring);
 
 	if (net->svc->prtcstns & TBNET_MATCH_FRAGS_ID)
 		atomic_inc(&net->frame_id);
-- 
2.50.1