Test commit

4 files changed, 56 insertions, 17 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 258a16f..077e45e 100644
--- a/src/mpmc_ptr_ring.h
+++ b/src/mpmc_ptr_ring.h
@@ -78,21 +78,29 @@ 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);
 
 	for (;;) {
-		smp_rmb();	 /* TODO */
+		/*
+		 * The snapshot of producer must be up to date with respect to
+		 * consumer.
+		 */
+		smp_rmb();
 		c = atomic_read(&r->consumer_head);
 
-		if ((p - c) < mask) { /* fast path */
-			if (atomic_cmpxchg(&r->producer_head, p, p + 1) == p)
+		if (likely((p - c) < mask)) {
+			if (atomic_try_cmpxchg_relaxed(&r->producer_head, &p, p + 1))
 				break;
 		} else {
 			unsigned int new_p;
 
+			/*
+			 * Either the buffer is full or our copy of
+			 * producer_head is stale.
+			 */
 			smp_rmb();
 			new_p = atomic_read(&r->producer_head);
 
@@ -121,7 +129,7 @@ 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;
 
@@ -134,16 +142,16 @@ static inline void *mpmc_ptr_ring_consume(struct mpmc_ptr_ring *r)
 		p = atomic_read(&r->producer_tail);
 
 		/* Is the ring empty? */
-		if (p == c)
+		if (unlikely(p == c))
 			return NULL;
 
 		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)
+		/*
+		 * Stores to consumer head must be completed before we update
+		 * the head, so we use *_release.
+		 */
+		if (atomic_try_cmpxchg_release(&r->consumer_head, &c, c + 1))
 			break;
 	}
 
diff --git a/src/selftest/mpmc_ring.h b/src/selftest/mpmc_ring.h
index 1cea5bf..f5f8fcf 100644
--- a/src/selftest/mpmc_ring.h
+++ b/src/selftest/mpmc_ring.h
@@ -10,9 +10,9 @@
 #include <linux/workqueue.h>
 #include <linux/wait.h>
 
-#define THREADS_PRODUCER 16
-#define THREADS_CONSUMER 16
-#define ELEMENT_COUNT 1000000LL /* divisible by threads_{consumer,producer} */
+#define THREADS_PRODUCER 2
+#define THREADS_CONSUMER 2
+#define ELEMENT_COUNT 100000000LL /* divisible by threads_{consumer,producer} */
 #define QUEUE_SIZE 1024
 
 #define EXPECTED_TOTAL ((ELEMENT_COUNT * (ELEMENT_COUNT + 1)) / 2)
@@ -40,8 +40,10 @@ static __init void producer_function(struct work_struct *work)
 	uintptr_t count = (td->thread_num * PER_PRODUCER) + 1;
 
 	for (; count <= (td->thread_num + 1) * PER_PRODUCER; ++count) {
-		while (mpmc_ptr_ring_produce(ring, (void *) count))
+		while (mpmc_ptr_ring_produce(ring, (void *) count)) {
 			schedule();
+			/*pr_info("We have awoken (producer)");*/
+		}
 	}
 }
 
@@ -52,9 +54,11 @@ static __init void consumer_function(struct work_struct *work)
 
 	for (i = 0; i < PER_CONSUMER; ++i) {
 		uintptr_t value;
-
-		while (!(value = (uintptr_t) mpmc_ptr_ring_consume(ring)))
+		while (!(value = (uintptr_t) mpmc_ptr_ring_consume(ring))) {
 			schedule();
+			/*cpu_relax();*/
+			/*pr_info("We have awoken (consumer)");*/
+		}
 
 		td->total += value;
 		++(td->count);