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

@@ -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:

@@ -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 */