mpmc_ptr_ring: use atomic_try_cmpxchg()

3 files changed, 37 insertions, 12 deletions

diff --git a/src/compat/atomic/atomic.h b/src/compat/atomic/atomic.h
new file mode 100644
index 0000000..c8c1c90
--- /dev/null
+++ b/src/compat/atomic/atomic.h
@@ -0,0 +1,22 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+
+#ifndef atomic_try_cmpxchg
+
+#define __atomic_try_cmpxchg(type, _p, _po, _n)				\
+({									\
+	typeof(_po) __po = (_po);					\
+	typeof(*(_po)) __o = *__po;					\
+	*__po = atomic_cmpxchg##type((_p), __o, (_n));			\
+	(*__po == __o);							\
+})
+
+#define atomic_try_cmpxchg(_p, _po, _n)		__atomic_try_cmpxchg(, _p, _po, _n)
+#define atomic_try_cmpxchg_relaxed(_p, _po, _n)	__atomic_try_cmpxchg(_relaxed, _p, _po, _n)
+#define atomic_try_cmpxchg_acquire(_p, _po, _n)	__atomic_try_cmpxchg(_acquire, _p, _po, _n)
+#define atomic_try_cmpxchg_release(_p, _po, _n)	__atomic_try_cmpxchg(_release, _p, _po, _n)
+
+#else /* atomic_try_cmpxchg */
+#define atomic_try_cmpxchg_relaxed	atomic_try_cmpxchg
+#define atomic_try_cmpxchg_acquire	atomic_try_cmpxchg
+#define atomic_try_cmpxchg_release	atomic_try_cmpxchg
+#endif /* atomic_try_cmpxchg */
diff --git a/src/compat/compat.h b/src/compat/compat.h
index 2853c09..5affedf 100644
--- a/src/compat/compat.h
+++ b/src/compat/compat.h
@@ -591,6 +591,11 @@ static inline void *skb_put_data(struct sk_buff *skb, const void *data, unsigned
 }
 #endif
 
+#include <linux/atomic.h>
+#if !defined(atomic_try_cmpxchg)
+#include "atomic/atomic.h"
+#endif
+
 /* https://lkml.org/lkml/2017/6/23/790 */
 #if IS_ENABLED(CONFIG_NF_CONNTRACK)
 #include <linux/ip.h>
diff --git a/src/mpmc_ptr_ring.h b/src/mpmc_ptr_ring.h
index a09e165..f68d709 100644
--- a/src/mpmc_ptr_ring.h
+++ b/src/mpmc_ptr_ring.h
@@ -78,7 +78,7 @@ static inline bool mpmc_ptr_ring_empty(struct mpmc_ptr_ring *r)
 
 static inline int mpmc_ptr_ring_produce(struct mpmc_ptr_ring *r, void *ptr)
 {
-	unsigned int p, c;
+	int p, c;
 	unsigned int mask = r->mask;
 
 	p = atomic_read(&r->producer_head);
@@ -88,10 +88,10 @@ static inline int mpmc_ptr_ring_produce(struct mpmc_ptr_ring *r, void *ptr)
 		c = atomic_read(&r->consumer_head);
 
 		if ((p - c) < mask) { /* fast path */
-			if (atomic_cmpxchg(&r->producer_head, p, p + 1) == p)
+			if (atomic_try_cmpxchg_relaxed(&r->producer_head, &p, p + 1))
 				break;
 		} else {
-			unsigned int new_p;
+			int new_p;
 
 			smp_rmb();
 			new_p = atomic_read(&r->producer_head);
@@ -121,11 +121,11 @@ static inline int mpmc_ptr_ring_produce(struct mpmc_ptr_ring *r, void *ptr)
 
 static inline void *mpmc_ptr_ring_consume(struct mpmc_ptr_ring *r)
 {
-	unsigned int c, p, old_c;
+	int c, p;
 	unsigned int mask = r->mask;
 	void *element;
 
-	for (;;) {
+	do {
 		c = atomic_read(&r->consumer_head);
 
 		/* Fetch consumer_head first. */
@@ -139,13 +139,11 @@ static inline void *mpmc_ptr_ring_consume(struct mpmc_ptr_ring *r)
 
 		element = READ_ONCE(r->queue[c & mask]);
 
-		/* TODO: Why? */
-		smp_rmb();
-
-		old_c = atomic_cmpxchg(&r->consumer_head, c, c + 1);
-		if (old_c == c)
-			break;
-	}
+		/*
+		 * Stores to consumer_head must be completed before we update
+		 * the head, so we use *_release.
+		 */
+	} while (!atomic_try_cmpxchg_release(&r->consumer_head, &c, c + 1));
 
 	return element;
 }