[PATCH bpf-next] xsk: update rings for load-acquire/store-release semantics

["Björn Töpel" <bjorn.topel@gmail.com>]

From: Björn Töpel <bjorn.topel@intel.com>

Currently, the AF_XDP rings uses fences for the kernel-side
produce/consume functions. By updating rings for
load-acquire/store-release semantics, the full barrier (smp_mb()) on
the consumer side can be replaced.

Signed-off-by: Björn Töpel <bjorn.topel@intel.com>
---
 net/xdp/xsk_queue.h | 31 ++++++++++++++-----------------
 1 file changed, 14 insertions(+), 17 deletions(-)

diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
index bec2af11853a..2fff80576ee1 100644
--- a/net/xdp/xsk_queue.h
+++ b/net/xdp/xsk_queue.h
@@ -39,19 +39,18 @@ struct xsk_queue {
 	u64 invalid_descs;
 };
 
-/* The structure of the shared state of the rings are the same as the
- * ring buffer in kernel/events/ring_buffer.c. For the Rx and completion
- * ring, the kernel is the producer and user space is the consumer. For
- * the Tx and fill rings, the kernel is the consumer and user space is
- * the producer.
+/* The structure of the shared state of the rings are a simple
+ * circular buffer, as outlined in
+ * Documentation/core-api/circular-buffers.rst. For the Rx and
+ * completion ring, the kernel is the producer and user space is the
+ * consumer. For the Tx and fill rings, the kernel is the consumer and
+ * user space is the producer.
  *
  * producer                         consumer
  *
- * if (LOAD ->consumer) {           LOAD ->producer
- *                    (A)           smp_rmb()       (C)
+ * if (LOAD ->consumer) {  (A)      LOAD.acq ->producer  (C)
  *    STORE $data                   LOAD $data
- *    smp_wmb()       (B)           smp_mb()        (D)
- *    STORE ->producer              STORE ->consumer
+ *    STORE.rel ->producer (B)      STORE.rel ->consumer (D)
  * }
  *
  * (A) pairs with (D), and (B) pairs with (C).
@@ -220,15 +219,14 @@ static inline bool xskq_cons_read_desc(struct xsk_queue *q,
 
 static inline void __xskq_cons_release(struct xsk_queue *q)
 {
-	smp_mb(); /* D, matches A */
-	WRITE_ONCE(q->ring->consumer, q->cached_cons);
+	/* D, matches A */
+	smp_store_release(&q->ring->consumer, q->cached_cons);
 }
 
 static inline void __xskq_cons_peek(struct xsk_queue *q)
 {
-	/* Refresh the local pointer */
-	q->cached_prod = READ_ONCE(q->ring->producer);
-	smp_rmb(); /* C, matches B */
+	/* C, matches B */
+	q->cached_prod = smp_load_acquire(&q->ring->producer);
 }
 
 static inline void xskq_cons_get_entries(struct xsk_queue *q)
@@ -340,9 +338,8 @@ static inline int xskq_prod_reserve_desc(struct xsk_queue *q,
 
 static inline void __xskq_prod_submit(struct xsk_queue *q, u32 idx)
 {
-	smp_wmb(); /* B, matches C */
-
-	WRITE_ONCE(q->ring->producer, idx);
+	/* B, matches C */
+	smp_store_release(&q->ring->producer, idx);
 }
 
 static inline void xskq_prod_submit(struct xsk_queue *q)
-- 
2.20.1