[dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[Jia He]

We watched a rte panic of mbuf_autotest in our qualcomm arm64 server due
to a possible race condition.

To fix this race, there are 2 options as suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release(refer to [2]).
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines.

Already fuctionally tested on the machines as follows:
- on X86
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=n

---
Changelog:
V4: split into small patches
V3: arch specific implementation for enqueue/dequeue barrier
V2: let users choose whether using load_acquire/store_release
V1: rte_smp_rmb() between 2 loads

Jia He (4):
  eal/arm64: remove the braces {} for dmb() and dsb()
  ring: guarantee load/load order in enqueue and dequeue
  ring: introduce new header file to include common functions
  ring: introduce new header file to support C11 memory model

 config/common_armv8a_linuxapp                      |   2 +
 .../common/include/arch/arm/rte_atomic_64.h        |   4 +-
 lib/librte_eventdev/rte_event_ring.h               |   6 +-
 lib/librte_ring/Makefile                           |   4 +-
 lib/librte_ring/rte_ring.h                         | 161 ++---------------
 lib/librte_ring/rte_ring_c11_mem.h                 | 185 ++++++++++++++++++++
 lib/librte_ring/rte_ring_generic.h                 | 194 +++++++++++++++++++++
 7 files changed, 404 insertions(+), 152 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h
 create mode 100644 lib/librte_ring/rte_ring_generic.h

-- 
2.7.4

[dpdk-dev] [PATCH v4 1/4] eal/arm64: remove the braces {} for dmb() and dsb()

[Jia He]

for the code as follows:
if (condition)
	rte_smp_rmb();
else
	rte_smp_wmb();
Without this patch, compiler will report this error:
error: 'else' without a previous 'if'

Fixes: 84733fd0d75e ("eal/arm64: fix memory barrier definition")
Signed-off-by: Jia He <jia.he@hxt-semitech.com>
---
 lib/librte_eal/common/include/arch/arm/rte_atomic_64.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
index 0b70d62..71da29c 100644
--- a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
+++ b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
@@ -43,8 +43,8 @@ extern "C" {
 
 #include "generic/rte_atomic.h"
 
-#define dsb(opt)  { asm volatile("dsb " #opt : : : "memory"); }
-#define dmb(opt)  { asm volatile("dmb " #opt : : : "memory"); }
+#define dsb(opt) asm volatile("dsb " #opt : : : "memory")
+#define dmb(opt) asm volatile("dmb " #opt : : : "memory")
 
 #define rte_mb() dsb(sy)
 
-- 
2.7.4

[dpdk-dev] [PATCH v4 2/4] ring: guarantee load/load order in enqueue and dequeue

[Jia He]

We watched a rte panic of mbuf_autotest in our qualcomm arm64 server.
In __rte_ring_move_cons_head()
...
        do {
                /* Restore n as it may change every loop */
                n = max;

                *old_head = r->cons.head;                //1st load
                const uint32_t prod_tail = r->prod.tail; //2nd load

In weak memory order architectures(powerpc,arm), the 2nd load might be
reodered before the 1st load, that makes *entries is bigger than we wanted.
This nasty reording messed enque/deque up.

cpu1(producer)          cpu2(consumer)          cpu3(consumer)
                        load r->prod.tail
in enqueue:
load r->cons.tail
load r->prod.head

store r->prod.tail

                                                load r->cons.head
                                                load r->prod.tail
                                                ...
                                                store r->cons.{head,tail}
                        load r->cons.head

Then, r->cons.head will be bigger than prod_tail, then make *entries very
big and the consumer will go forward incorrectly.

After this patch, the old cons.head will be recaculated after failure of
rte_atomic32_cmpset

There is no such issue on X86, because X86 is strong memory order model.

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Signed-off-by: jie2.liu@hxt-semitech.com
Signed-off-by: bing.zhao@hxt-semitech.com
---
 lib/librte_ring/rte_ring.h | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 5e9b3b7..3e8085a 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -409,6 +409,11 @@ __rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
 		n = max;
 
 		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86 */
+		rte_smp_rmb();
+
 		const uint32_t cons_tail = r->cons.tail;
 		/*
 		 *  The subtraction is done between two unsigned 32bits value
@@ -517,6 +522,11 @@ __rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
 		n = max;
 
 		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86 */
+		rte_smp_rmb();
+
 		const uint32_t prod_tail = r->prod.tail;
 		/* The subtraction is done between two unsigned 32bits value
 		 * (the result is always modulo 32 bits even if we have
-- 
2.7.4

[dpdk-dev] [PATCH v4 3/4] ring: introduce new header file to include common functions

[Jia He]

move the common part of rte_ring.h into rte_ring_generic.h
move the memory barrier part into update_tail()
no functional changes here

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
Suggested-by: Ananyev, Konstantin <konstantin.ananyev@intel.com>
---
 lib/librte_eventdev/rte_event_ring.h |   6 +-
 lib/librte_ring/Makefile             |   3 +-
 lib/librte_ring/rte_ring.h           | 159 +---------------------------
 lib/librte_ring/rte_ring_generic.h   | 194 +++++++++++++++++++++++++++++++++++
 4 files changed, 202 insertions(+), 160 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_generic.h

diff --git a/lib/librte_eventdev/rte_event_ring.h b/lib/librte_eventdev/rte_event_ring.h
index ea9b688..3e49458 100644
--- a/lib/librte_eventdev/rte_event_ring.h
+++ b/lib/librte_eventdev/rte_event_ring.h
@@ -126,9 +126,8 @@ rte_event_ring_enqueue_burst(struct rte_event_ring *r,
 		goto end;
 
 	ENQUEUE_PTRS(&r->r, &r[1], prod_head, events, n, struct rte_event);
-	rte_smp_wmb();
 
-	update_tail(&r->r.prod, prod_head, prod_next, 1);
+	update_tail(&r->r.prod, prod_head, prod_next, 1, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -168,9 +167,8 @@ rte_event_ring_dequeue_burst(struct rte_event_ring *r,
 		goto end;
 
 	DEQUEUE_PTRS(&r->r, &r[1], cons_head, events, n, struct rte_event);
-	rte_smp_rmb();
 
-	update_tail(&r->r.cons, cons_head, cons_next, 1);
+	update_tail(&r->r.cons, cons_head, cons_next, 1, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index e34d9d9..c959945 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -45,6 +45,7 @@ LIBABIVER := 1
 SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
-SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
+					rte_ring_generic.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 3e8085a..519614c 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,90 +356,8 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-static __rte_always_inline void
-update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
-		uint32_t single)
-{
-	/*
-	 * If there are other enqueues/dequeues in progress that preceded us,
-	 * we need to wait for them to complete
-	 */
-	if (!single)
-		while (unlikely(ht->tail != old_val))
-			rte_pause();
-
-	ht->tail = new_val;
-}
-
-/**
- * @internal This function updates the producer head for enqueue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sp
- *   Indicates whether multi-producer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where enqueue starts
- * @param new_head
- *   Returns the current/new head value i.e. where enqueue finishes
- * @param free_entries
- *   Returns the amount of free space in the ring BEFORE head was moved
- * @return
- *   Actual number of objects enqueued.
- *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *free_entries)
-{
-	const uint32_t capacity = r->capacity;
-	unsigned int max = n;
-	int success;
-
-	do {
-		/* Reset n to the initial burst count */
-		n = max;
-
-		*old_head = r->prod.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86 */
-		rte_smp_rmb();
-
-		const uint32_t cons_tail = r->cons.tail;
-		/*
-		 *  The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * *old_head > cons_tail). So 'free_entries' is always between 0
-		 * and capacity (which is < size).
-		 */
-		*free_entries = (capacity + cons_tail - *old_head);
-
-		/* check that we have enough room in ring */
-		if (unlikely(n > *free_entries))
-			n = (behavior == RTE_RING_QUEUE_FIXED) ?
-					0 : *free_entries;
-
-		if (n == 0)
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sp)
-			r->prod.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->prod.head,
-					*old_head, *new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
+/* Move common functions to generic file */
+#include "rte_ring_generic.h"
 
 /**
  * @internal Enqueue several objects on the ring
@@ -475,9 +393,8 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 		goto end;
 
 	ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);
-	rte_smp_wmb();
 
-	update_tail(&r->prod, prod_head, prod_next, is_sp);
+	update_tail(&r->prod, prod_head, prod_next, is_sp, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -485,73 +402,6 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 }
 
 /**
- * @internal This function updates the consumer head for dequeue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sc
- *   Indicates whether multi-consumer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where dequeue starts
- * @param new_head
- *   Returns the current/new head value i.e. where dequeue finishes
- * @param entries
- *   Returns the number of entries in the ring BEFORE head was moved
- * @return
- *   - Actual number of objects dequeued.
- *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *entries)
-{
-	unsigned int max = n;
-	int success;
-
-	/* move cons.head atomically */
-	do {
-		/* Restore n as it may change every loop */
-		n = max;
-
-		*old_head = r->cons.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86 */
-		rte_smp_rmb();
-
-		const uint32_t prod_tail = r->prod.tail;
-		/* The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * cons_head > prod_tail). So 'entries' is always between 0
-		 * and size(ring)-1. */
-		*entries = (prod_tail - *old_head);
-
-		/* Set the actual entries for dequeue */
-		if (n > *entries)
-			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
-
-		if (unlikely(n == 0))
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sc)
-			r->cons.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
-					*new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
-
-/**
  * @internal Dequeue several objects from the ring
  *
  * @param r
@@ -585,9 +435,8 @@ __rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,
 		goto end;
 
 	DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *);
-	rte_smp_rmb();
 
-	update_tail(&r->cons, cons_head, cons_next, is_sc);
+	update_tail(&r->cons, cons_head, cons_next, is_sc, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/rte_ring_generic.h b/lib/librte_ring/rte_ring_generic.h
new file mode 100644
index 0000000..3f49ac1
--- /dev/null
+++ b/lib/librte_ring/rte_ring_generic.h
@@ -0,0 +1,194 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_GENERIC_H_
+#define _RTE_RING_GENERIC_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	if (enqueue)
+		rte_smp_wmb();
+	else
+		rte_smp_rmb();
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	ht->tail = new_val;
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->prod.head,
+					*old_head, *new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+
+		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1. */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
+					*new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_GENERIC_H_ */
-- 
2.7.4

[dpdk-dev] [PATCH v4 4/4] ring: introduce new header file to support C11 memory model

[Jia He]

To fix the cpu reorder race condition in enque/deque, there are 2 options
suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release(refer to [1]).
for the 2nd option, CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by
default it is "y" only on arm64.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines, refer to [2].

We haven't tested on ppc64. If anyone verifies it, he can add
CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.

[1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
[2] http://dpdk.org/ml/archives/dev/2017-October/080861.html

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
---
 config/common_armv8a_linuxapp      |   2 +
 lib/librte_ring/Makefile           |   3 +-
 lib/librte_ring/rte_ring.h         |  14 ++-
 lib/librte_ring/rte_ring_c11_mem.h | 185 +++++++++++++++++++++++++++++++++++++
 4 files changed, 202 insertions(+), 2 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h

diff --git a/config/common_armv8a_linuxapp b/config/common_armv8a_linuxapp
index 6732d1e..5b7b2eb 100644
--- a/config/common_armv8a_linuxapp
+++ b/config/common_armv8a_linuxapp
@@ -49,3 +49,5 @@ CONFIG_RTE_LIBRTE_SFC_EFX_PMD=n
 CONFIG_RTE_LIBRTE_AVP_PMD=n
 
 CONFIG_RTE_SCHED_VECTOR=n
+
+CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index c959945..a2682e7 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -46,6 +46,7 @@ SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
 SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
-					rte_ring_generic.h
+					rte_ring_generic.h \
+					rte_ring_c11_mem.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 519614c..3343eba 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,8 +356,20 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-/* Move common functions to generic file */
+/* Between load and load. there might be cpu reorder in weak model
+ * (powerpc/arm).
+ * There are 2 choices for the users
+ * 1.use rmb() memory barrier
+ * 2.use one-direcion load_acquire/store_release barrier,defined by
+ * CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
+ * It depends on performance test results.
+ * By default, move common functions to rte_ring_generic.h
+ */
+#ifdef RTE_RING_USE_C11_MEM_MODEL
+#include "rte_ring_c11_mem.h"
+#else
 #include "rte_ring_generic.h"
+#endif
 
 /**
  * @internal Enqueue several objects on the ring
diff --git a/lib/librte_ring/rte_ring_c11_mem.h b/lib/librte_ring/rte_ring_c11_mem.h
new file mode 100644
index 0000000..c167b46
--- /dev/null
+++ b/lib/librte_ring/rte_ring_c11_mem.h
@@ -0,0 +1,185 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_C11_MEM_H_
+#define _RTE_RING_C11_MEM_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	RTE_SET_USED(enqueue);
+
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	__atomic_store_n(&ht->tail, new_val, __ATOMIC_RELEASE);
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = __atomic_load_n(&r->prod.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->prod.head,
+					old_head, *new_head,
+					0, __ATOMIC_ACQUIRE,
+					__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+		*old_head = __atomic_load_n(&r->cons.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1. */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->cons.head,
+							old_head, *new_head,
+							0, __ATOMIC_ACQUIRE,
+							__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_C11_MEM_H_ */
-- 
2.7.4

Re: [dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[Bruce Richardson]

On Wed, Nov 08, 2017 at 09:54:37AM +0000, Jia He wrote:
> We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
> due to a possible race condition.
> 
> To fix this race, there are 2 options as suggested by Jerin: 1. use
> rte_smp_rmb 2. use load_acquire/store_release(refer to [2]).
> CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is
> "y" only on arm64 so far.
> 
> The reason why providing 2 options is due to the performance benchmark
> difference in different arm machines.
> 
> Already fuctionally tested on the machines as follows: - on X86 - on
> arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y - on arm64 with
> CONFIG_RTE_RING_USE_C11_MEM_MODEL=n
> 
> --- Changelog: V4: split into small patches V3: arch specific
> implementation for enqueue/dequeue barrier V2: let users choose
> whether using load_acquire/store_release V1: rte_smp_rmb() between 2
> loads
> 
> Jia He (4): eal/arm64: remove the braces {} for dmb() and dsb() ring:
> guarantee load/load order in enqueue and dequeue ring: introduce new
> header file to include common functions ring: introduce new header
> file to support C11 memory model
> 
I'm wondering what the merge plans are for this set, given we are now
past RC3 in 17.11? As the rings are broken on ARM machines we need to
merge in some fix, but I'm a little concerned about the scope of the
changes from the 3rd and 4th patches. Would it be acceptable to just
merge in patches 1 & 2 in 17.11 and leave the rework and C11 memory
model additions in patches 3 & 4 to 18.02 release?

Regards,
/Bruce

Re: [dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[Jia He]

Hi Bruce


On 11/8/2017 8:15 PM, Bruce Richardson Wrote:
> On Wed, Nov 08, 2017 at 09:54:37AM +0000, Jia He wrote:
>> We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
>> due to a possible race condition.
>>
>> To fix this race, there are 2 options as suggested by Jerin: 1. use
>> rte_smp_rmb 2. use load_acquire/store_release(refer to [2]).
>> CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is
>> "y" only on arm64 so far.
>>
>> The reason why providing 2 options is due to the performance benchmark
>> difference in different arm machines.
>>
>> Already fuctionally tested on the machines as follows: - on X86 - on
>> arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y - on arm64 with
>> CONFIG_RTE_RING_USE_C11_MEM_MODEL=n
>>
>> --- Changelog: V4: split into small patches V3: arch specific
>> implementation for enqueue/dequeue barrier V2: let users choose
>> whether using load_acquire/store_release V1: rte_smp_rmb() between 2
>> loads
>>
>> Jia He (4): eal/arm64: remove the braces {} for dmb() and dsb() ring:
>> guarantee load/load order in enqueue and dequeue ring: introduce new
>> header file to include common functions ring: introduce new header
>> file to support C11 memory model
>>
> I'm wondering what the merge plans are for this set, given we are now
> past RC3 in 17.11? As the rings are broken on ARM machines we need to
> merge in some fix, but I'm a little concerned about the scope of the
> changes from the 3rd and 4th patches. Would it be acceptable to just
> merge in patches 1 & 2 in 17.11 and leave the rework and C11 memory
> model additions in patches 3 & 4 to 18.02 release?
As far as I'm concerned, it is ok.

Cheers,
Jia

Re: [dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[Jerin Jacob]

-----Original Message-----
> Date: Wed, 8 Nov 2017 23:11:32 +0800
> From: Jia He <hejianet@gmail.com>
> To: Bruce Richardson <bruce.richardson@intel.com>
> Cc: jerin.jacob@caviumnetworks.com, dev@dpdk.org, olivier.matz@6wind.com,
>  konstantin.ananyev@intel.com, jianbo.liu@arm.com, hemant.agrawal@nxp.com
> Subject: Re: [PATCH v4 0/4] fix race condition in enqueue/dequeue because
>  of cpu reorder
> User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64; rv:52.0) Gecko/20100101
>  Thunderbird/52.4.0
> 
> Hi Bruce
> 
> 
> On 11/8/2017 8:15 PM, Bruce Richardson Wrote:
> > On Wed, Nov 08, 2017 at 09:54:37AM +0000, Jia He wrote:
> > > We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
> > > due to a possible race condition.
> > > 
> > > To fix this race, there are 2 options as suggested by Jerin: 1. use
> > > rte_smp_rmb 2. use load_acquire/store_release(refer to [2]).
> > > CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is
> > > "y" only on arm64 so far.
> > > 
> > > The reason why providing 2 options is due to the performance benchmark
> > > difference in different arm machines.
> > > 
> > > Already fuctionally tested on the machines as follows: - on X86 - on
> > > arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y - on arm64 with
> > > CONFIG_RTE_RING_USE_C11_MEM_MODEL=n
> > > 
> > > --- Changelog: V4: split into small patches V3: arch specific
> > > implementation for enqueue/dequeue barrier V2: let users choose
> > > whether using load_acquire/store_release V1: rte_smp_rmb() between 2
> > > loads
> > > 
> > > Jia He (4): eal/arm64: remove the braces {} for dmb() and dsb() ring:
> > > guarantee load/load order in enqueue and dequeue ring: introduce new
> > > header file to include common functions ring: introduce new header
> > > file to support C11 memory model
> > > 
> > I'm wondering what the merge plans are for this set, given we are now
> > past RC3 in 17.11? As the rings are broken on ARM machines we need to
> > merge in some fix, but I'm a little concerned about the scope of the
> > changes from the 3rd and 4th patches. Would it be acceptable to just
> > merge in patches 1 & 2 in 17.11 and leave the rework and C11 memory
> > model additions in patches 3 & 4 to 18.02 release?
> As far as I'm concerned, it is ok.

It is OK to me as well. May be Jia can send 0-1 and 2-3 as separate
series with exiting comments.


> 
> Cheers,
> Jia
> 

Re: [dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[Ananyev, Konstantin]

> -----Original Message-----
> From: Jia He [mailto:hejianet@gmail.com]
> Sent: Wednesday, November 8, 2017 3:12 PM
> To: Richardson, Bruce <bruce.richardson@intel.com>
> Cc: jerin.jacob@caviumnetworks.com; dev@dpdk.org; olivier.matz@6wind.com; Ananyev, Konstantin <konstantin.ananyev@intel.com>;
> jianbo.liu@arm.com; hemant.agrawal@nxp.com
> Subject: Re: [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder
> 
> Hi Bruce
> 
> 
> On 11/8/2017 8:15 PM, Bruce Richardson Wrote:
> > On Wed, Nov 08, 2017 at 09:54:37AM +0000, Jia He wrote:
> >> We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
> >> due to a possible race condition.
> >>
> >> To fix this race, there are 2 options as suggested by Jerin: 1. use
> >> rte_smp_rmb 2. use load_acquire/store_release(refer to [2]).
> >> CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is
> >> "y" only on arm64 so far.
> >>
> >> The reason why providing 2 options is due to the performance benchmark
> >> difference in different arm machines.
> >>
> >> Already fuctionally tested on the machines as follows: - on X86 - on
> >> arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y - on arm64 with
> >> CONFIG_RTE_RING_USE_C11_MEM_MODEL=n
> >>
> >> --- Changelog: V4: split into small patches V3: arch specific
> >> implementation for enqueue/dequeue barrier V2: let users choose
> >> whether using load_acquire/store_release V1: rte_smp_rmb() between 2
> >> loads
> >>
> >> Jia He (4): eal/arm64: remove the braces {} for dmb() and dsb() ring:
> >> guarantee load/load order in enqueue and dequeue ring: introduce new
> >> header file to include common functions ring: introduce new header
> >> file to support C11 memory model
> >>
> > I'm wondering what the merge plans are for this set, given we are now
> > past RC3 in 17.11? As the rings are broken on ARM machines we need to
> > merge in some fix, but I'm a little concerned about the scope of the
> > changes from the 3rd and 4th patches. Would it be acceptable to just
> > merge in patches 1 & 2 in 17.11 and leave the rework and C11 memory
> > model additions in patches 3 & 4 to 18.02 release?
> As far as I'm concerned, it is ok.

Sounds good to me too.
Konstantin 

[dpdk-dev] [PATCH v5 0/1] fix race condition in enqueue/dequeue because of cpu reorder

[Jia He]

We watched a rte panic of mbuf_autotest in our qualcomm arm64 server due
to a possible race condition. To fix this race condition, rmb() is needed to add
between the 2 loads.

Already fuctionally tested on the machines as follows:
- on X86
- on arm64 

---
Changelog
V5: split it into 2 patchset due to the milestone concerns, this is the 1st one
V4: split into small patches
V3: arch specific implementation for enqueue/dequeue barrier
V2: let users choose whether using load_acquire/store_release
V1: rte_smp_rmb() between 2 loads

Jia He (4):
  eal/arm64: remove the braces {} for dmb() and dsb()
  ring: guarantee load/load order in enqueue and dequeue
  ring: introduce new header file to include common functions
  ring: introduce new header file to support C11 memory model

 config/common_armv8a_linuxapp                      |   2 +
 .../common/include/arch/arm/rte_atomic_64.h        |   4 +-
 lib/librte_eventdev/rte_event_ring.h               |   6 +-
 lib/librte_ring/Makefile                           |   4 +-
 lib/librte_ring/rte_ring.h                         | 161 ++---------------
 lib/librte_ring/rte_ring_c11_mem.h                 | 185 ++++++++++++++++++++
 lib/librte_ring/rte_ring_generic.h                 | 194 +++++++++++++++++++++
 7 files changed, 404 insertions(+), 152 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h
 create mode 100644 lib/librte_ring/rte_ring_generic.h

-- 
2.7.4

[dpdk-dev] [PATCH v5 1/1] ring: guarantee load/load order in enqueue and dequeue

[Jia He]

We watched a rte panic of mbuf_autotest in our qualcomm arm64 server.
In __rte_ring_move_cons_head()
...
        do {
                /* Restore n as it may change every loop */
                n = max;

                *old_head = r->cons.head;                //1st load
                const uint32_t prod_tail = r->prod.tail; //2nd load

cpu1(producer)          cpu2(consumer)          cpu3(consumer)
                        load r->prod.tail
in enqueue:
load r->cons.tail
load r->prod.head

store r->prod.tail

                                                load r->cons.head
                                                load r->prod.tail
                                                ...
                                                store r->cons.{head,tail}
                        load r->cons.head

In weak memory order architectures(powerpc,arm), the 2nd load might be
reodered before the 1st load, that makes *entries is bigger than we
wanted. This nasty reording messed enque/deque up. Then, r->cons.head
will be bigger than prod_tail, then make *entries very big and the
consumer will go forward incorrectly.

After this patch, even with above context switches, the old cons.head
will be recaculated after failure of rte_atomic32_cmpset. So no race
conditions left.

There is no such issue on X86, because X86 is strong memory order model.
But rte_smp_rmb() doesn't have impact on runtime performance on X86, so
keep the same code without architectures specific concerns.

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Signed-off-by: jie2.liu@hxt-semitech.com
Signed-off-by: bing.zhao@hxt-semitech.com
---
 lib/librte_ring/rte_ring.h | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 5e9b3b7..3e8085a 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -409,6 +409,11 @@ __rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
 		n = max;
 
 		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86 */
+		rte_smp_rmb();
+
 		const uint32_t cons_tail = r->cons.tail;
 		/*
 		 *  The subtraction is done between two unsigned 32bits value
@@ -517,6 +522,11 @@ __rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
 		n = max;
 
 		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86 */
+		rte_smp_rmb();
+
 		const uint32_t prod_tail = r->prod.tail;
 		/* The subtraction is done between two unsigned 32bits value
 		 * (the result is always modulo 32 bits even if we have
-- 
2.7.4

Re: [dpdk-dev] [PATCH v5 1/1] ring: guarantee load/load order in enqueue and dequeue

[Jerin Jacob]

-----Original Message-----
> Date: Fri, 10 Nov 2017 01:51:09 +0000
> From: Jia He <hejianet@gmail.com>
> To: jerin.jacob@caviumnetworks.com, dev@dpdk.org, olivier.matz@6wind.com
> Cc: konstantin.ananyev@intel.com, bruce.richardson@intel.com,
>  jianbo.liu@arm.com, hemant.agrawal@nxp.com, Jia He <hejianet@gmail.com>,
>  Jia He <jia.he@hxt-semitech.com>, jie2.liu@hxt-semitech.com,
>  bing.zhao@hxt-semitech.com
> Subject: [PATCH v5 1/1] ring: guarantee load/load order in enqueue and
>  dequeue
> X-Mailer: git-send-email 2.7.4
> 
> We watched a rte panic of mbuf_autotest in our qualcomm arm64 server.

I think, the above text can be improved. Something like below.


Fix for the rte_panic() in mbuf_autotest on qualcomm arm64 server(...SoC
name...)

Root cause:
> In __rte_ring_move_cons_head()
> ...
>         do {
>                 /* Restore n as it may change every loop */
>                 n = max;
> 
>                 *old_head = r->cons.head;                //1st load
>                 const uint32_t prod_tail = r->prod.tail; //2nd load
> 
> cpu1(producer)          cpu2(consumer)          cpu3(consumer)
>                         load r->prod.tail
> in enqueue:
> load r->cons.tail
> load r->prod.head
> 
> store r->prod.tail
> 
>                                                 load r->cons.head
>                                                 load r->prod.tail
>                                                 ...
>                                                 store r->cons.{head,tail}
>                         load r->cons.head
> 
> In weak memory order architectures(powerpc,arm), the 2nd load might be
> reodered before the 1st load, that makes *entries is bigger than we
> wanted. This nasty reording messed enque/deque up. Then, r->cons.head
> will be bigger than prod_tail, then make *entries very big and the
> consumer will go forward incorrectly.
> 
> After this patch, even with above context switches, the old cons.head
> will be recaculated after failure of rte_atomic32_cmpset. So no race
> conditions left.
> 
> There is no such issue on X86, because X86 is strong memory order model.
> But rte_smp_rmb() doesn't have impact on runtime performance on X86, so
> keep the same code without architectures specific concerns.
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Signed-off-by: jie2.liu@hxt-semitech.com
> Signed-off-by: bing.zhao@hxt-semitech.com

➜ [master][dpdk.org] $ ./devtools/checkpatches.sh 

### ring: guarantee load/load order in enqueue and dequeue

WARNING:BLOCK_COMMENT_STYLE: Block comments use a trailing */ on a
separate line
#58: FILE: lib/librte_ring/rte_ring.h:414:
+		 * memory model. It is noop on x86 */

WARNING:BLOCK_COMMENT_STYLE: Block comments use a trailing */ on a
separate line
#70: FILE: lib/librte_ring/rte_ring.h:527:
+		 * memory model. It is noop on x86 */

total: 0 errors, 2 warnings, 22 lines checked

0/1 valid patch
➜ [master][dpdk.org] $ ./devtools/check-git-log.sh 
Wrong tag:
	Signed-off-by: jie2.liu@hxt-semitech.com
	Signed-off-by: bing.zhao@hxt-semitech.com


With above fixes:
Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Re: [dpdk-dev] [PATCH v5 1/1] ring: guarantee load/load order in enqueue and dequeue

[Jianbo Liu]

The 11/10/2017 08:16, Jerin Jacob wrote:
> -----Original Message-----
> > Date: Fri, 10 Nov 2017 01:51:09 +0000
> > From: Jia He <hejianet@gmail.com>
> > To: jerin.jacob@caviumnetworks.com, dev@dpdk.org, olivier.matz@6wind.com
> > Cc: konstantin.ananyev@intel.com, bruce.richardson@intel.com,
> >  jianbo.liu@arm.com, hemant.agrawal@nxp.com, Jia He <hejianet@gmail.com>,
> >  Jia He <jia.he@hxt-semitech.com>, jie2.liu@hxt-semitech.com,
> >  bing.zhao@hxt-semitech.com
> > Subject: [PATCH v5 1/1] ring: guarantee load/load order in enqueue and
> >  dequeue
> > X-Mailer: git-send-email 2.7.4
> >
> > We watched a rte panic of mbuf_autotest in our qualcomm arm64 server.
>
> I think, the above text can be improved. Something like below.
>
>
> Fix for the rte_panic() in mbuf_autotest on qualcomm arm64 server(...SoC
> name...)
>
> Root cause:
> > In __rte_ring_move_cons_head()
> > ...
> >         do {
> >                 /* Restore n as it may change every loop */
> >                 n = max;
> >
> >                 *old_head = r->cons.head;                //1st load
> >                 const uint32_t prod_tail = r->prod.tail; //2nd load
> >
> > cpu1(producer)          cpu2(consumer)          cpu3(consumer)
> >                         load r->prod.tail
> > in enqueue:
> > load r->cons.tail
> > load r->prod.head
> >
> > store r->prod.tail
> >
> >                                                 load r->cons.head
> >                                                 load r->prod.tail
> >                                                 ...
> >                                                 store r->cons.{head,tail}
> >                         load r->cons.head
> >
> > In weak memory order architectures(powerpc,arm), the 2nd load might be
> > reodered before the 1st load, that makes *entries is bigger than we
> > wanted. This nasty reording messed enque/deque up. Then, r->cons.head
> > will be bigger than prod_tail, then make *entries very big and the
> > consumer will go forward incorrectly.
> >
> > After this patch, even with above context switches, the old cons.head
> > will be recaculated after failure of rte_atomic32_cmpset. So no race
> > conditions left.
> >
> > There is no such issue on X86, because X86 is strong memory order model.
> > But rte_smp_rmb() doesn't have impact on runtime performance on X86, so
> > keep the same code without architectures specific concerns.
> >
> > Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> > Signed-off-by: jie2.liu@hxt-semitech.com
> > Signed-off-by: bing.zhao@hxt-semitech.com
>
> ➜ [master][dpdk.org] $ ./devtools/checkpatches.sh
>
> ### ring: guarantee load/load order in enqueue and dequeue
>
> WARNING:BLOCK_COMMENT_STYLE: Block comments use a trailing */ on a
> separate line
> #58: FILE: lib/librte_ring/rte_ring.h:414:
> +              * memory model. It is noop on x86 */
>
> WARNING:BLOCK_COMMENT_STYLE: Block comments use a trailing */ on a
> separate line
> #70: FILE: lib/librte_ring/rte_ring.h:527:
> +              * memory model. It is noop on x86 */
>
> total: 0 errors, 2 warnings, 22 lines checked
>
> 0/1 valid patch
> ➜ [master][dpdk.org] $ ./devtools/check-git-log.sh
> Wrong tag:
>       Signed-off-by: jie2.liu@hxt-semitech.com
>       Signed-off-by: bing.zhao@hxt-semitech.com
>
>
> With above fixes:
> Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
>
>
>

Acked-by: Jianbo Liu <jianbo.liu@arm.com>

And add it to stable.
 Cc: stable@dpdk.org

--
IMPORTANT NOTICE: The contents of this email and any attachments are confidential and may also be privileged. If you are not the intended recipient, please notify the sender immediately and do not disclose the contents to any other person, use it for any purpose, or store or copy the information in any medium. Thank you.

[dpdk-dev] [PATCH v6] guarantee load/load order in enqueue and dequeue

[Jia He]

From: Jia He <jia.he@hxt-semitech.com>

We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
(Amberwing) due to a possible race condition. To fix this race condition,
rmb() is needed to add between the 2 loads.

Already fuctionally tested on the machines as follows:
- on X86
- on arm64 

---
Changelog
V6: improve the text description and fix the checkpatch warnings
V5: split it into 2 patchset due to the milestone concerns, this is the 1st one
V4: split into small patches
V3: arch specific implementation for enqueue/dequeue barrier
V2: let users choose whether using load_acquire/store_release
V1: rte_smp_rmb() between 2 loads

Jia He (1):
  ring: guarantee load/load order in enqueue and dequeue

 lib/librte_ring/rte_ring.h | 12 ++++++++++++
 1 file changed, 12 insertions(+)

-- 
2.7.4

[dpdk-dev] [PATCH v6] ring: guarantee load/load order in enqueue and dequeue

[Jia He]

We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
(Amberwing).

Root cause:
In __rte_ring_move_cons_head()
...
        do {
                /* Restore n as it may change every loop */
                n = max;

                *old_head = r->cons.head;                //1st load
                const uint32_t prod_tail = r->prod.tail; //2nd load

In weak memory order architectures(powerpc,arm), the 2nd load might be
reodered before the 1st load, that makes *entries is bigger than we wanted.
This nasty reording messed enque/deque up.

cpu1(producer)          cpu2(consumer)          cpu3(consumer)
                        load r->prod.tail
in enqueue:
load r->cons.tail
load r->prod.head

store r->prod.tail

                                                load r->cons.head
                                                load r->prod.tail
                                                ...
                                                store r->cons.{head,tail}
                        load r->cons.head

Then, r->cons.head will be bigger than prod_tail, then make *entries very
big and the consumer will go forward incorrectly.

After this patch, the old cons.head will be recaculated after failure of
rte_atomic32_cmpset

There is no such issue on X86, because X86 is strong memory order model.
But rte_smp_rmb() doesn't have impact on runtime performance on X86, so
keep the same code without architectures specific concerns.

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Signed-off-by: jie2.liu@hxt-semitech.com
Signed-off-by: bing.zhao@hxt-semitech.com
Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
Acked-by: Jianbo Liu <jianbo.liu@arm.com>
Cc: stable@dpdk.org

---
 lib/librte_ring/rte_ring.h | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 5e9b3b7..e924438 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -409,6 +409,12 @@ __rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
 		n = max;
 
 		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
 		const uint32_t cons_tail = r->cons.tail;
 		/*
 		 *  The subtraction is done between two unsigned 32bits value
@@ -517,6 +523,12 @@ __rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
 		n = max;
 
 		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
 		const uint32_t prod_tail = r->prod.tail;
 		/* The subtraction is done between two unsigned 32bits value
 		 * (the result is always modulo 32 bits even if we have
-- 
2.7.4

[dpdk-dev] [PATCH v5 0/3] support c11 memory model barrier in librte_ring

[Jia He]

From: Jia He <jia.he@hxt-semitech.com>

To support C11 memory model barrier, 2 options are suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines.

Already fuctionally tested on the machines as follows:
- on X86
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=n

---
Changelog:
V5: split it into 2 patchset due to the milestone concerns, this is the 2st 
    one. Also fix checkpatch.pl warnings
V4: split into small patches
V3: arch specific implementation for enqueue/dequeue barrier
V2: let users choose whether using load_acquire/store_release
V1: rte_smp_rmb() between 2 loads

*** BLURB HERE ***

Jia He (3):
  eal/arm64: remove the braces {} for dmb() and dsb()
  ring: introduce new header file to include common functions
  ring: introduce new header file to support C11 memory model

 config/common_armv8a_linuxapp                      |   2 +
 .../common/include/arch/arm/rte_atomic_64.h        |   4 +-
 lib/librte_eventdev/rte_event_ring.h               |   6 +-
 lib/librte_ring/Makefile                           |   4 +-
 lib/librte_ring/rte_ring.h                         | 173 ++----------------
 lib/librte_ring/rte_ring_c11_mem.h                 | 186 ++++++++++++++++++++
 lib/librte_ring/rte_ring_generic.h                 | 195 +++++++++++++++++++++
 7 files changed, 406 insertions(+), 164 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h
 create mode 100644 lib/librte_ring/rte_ring_generic.h

-- 
2.7.4

[dpdk-dev] [PATCH v5 1/3] eal/arm64: remove the braces {} for dmb() and dsb()

[Jia He]

for the code as follows:
if (condition)
	rte_smp_rmb();
else
	rte_smp_wmb();
Without this patch, compiler will report this error:
error: 'else' without a previous 'if'

Fixes: 84733fd0d75e ("eal/arm64: fix memory barrier definition")
Signed-off-by: Jia He <jia.he@hxt-semitech.com>
---
 lib/librte_eal/common/include/arch/arm/rte_atomic_64.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
index 0b70d62..71da29c 100644
--- a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
+++ b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
@@ -43,8 +43,8 @@ extern "C" {
 
 #include "generic/rte_atomic.h"
 
-#define dsb(opt)  { asm volatile("dsb " #opt : : : "memory"); }
-#define dmb(opt)  { asm volatile("dmb " #opt : : : "memory"); }
+#define dsb(opt) asm volatile("dsb " #opt : : : "memory")
+#define dmb(opt) asm volatile("dmb " #opt : : : "memory")
 
 #define rte_mb() dsb(sy)
 
-- 
2.7.4

[dpdk-dev] [PATCH v5 2/3] ring: introduce new header file to include common functions

[Jia He]

From: Jia He <jia.he@hxt-semitech.com>

move the common part of rte_ring.h into rte_ring_generic.h
move the memory barrier part into update_tail()
no functional changes here

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
Suggested-by: Ananyev, Konstantin <konstantin.ananyev@intel.com>
---
 lib/librte_eventdev/rte_event_ring.h |   6 +-
 lib/librte_ring/Makefile             |   3 +-
 lib/librte_ring/rte_ring.h           | 161 +----------------------------
 lib/librte_ring/rte_ring_generic.h   | 195 +++++++++++++++++++++++++++++++++++
 4 files changed, 203 insertions(+), 162 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_generic.h

diff --git a/lib/librte_eventdev/rte_event_ring.h b/lib/librte_eventdev/rte_event_ring.h
index ea9b688..3e49458 100644
--- a/lib/librte_eventdev/rte_event_ring.h
+++ b/lib/librte_eventdev/rte_event_ring.h
@@ -126,9 +126,8 @@ rte_event_ring_enqueue_burst(struct rte_event_ring *r,
 		goto end;
 
 	ENQUEUE_PTRS(&r->r, &r[1], prod_head, events, n, struct rte_event);
-	rte_smp_wmb();
 
-	update_tail(&r->r.prod, prod_head, prod_next, 1);
+	update_tail(&r->r.prod, prod_head, prod_next, 1, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -168,9 +167,8 @@ rte_event_ring_dequeue_burst(struct rte_event_ring *r,
 		goto end;
 
 	DEQUEUE_PTRS(&r->r, &r[1], cons_head, events, n, struct rte_event);
-	rte_smp_rmb();
 
-	update_tail(&r->r.cons, cons_head, cons_next, 1);
+	update_tail(&r->r.cons, cons_head, cons_next, 1, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index e34d9d9..c959945 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -45,6 +45,7 @@ LIBABIVER := 1
 SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
-SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
+					rte_ring_generic.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index e924438..519614c 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,91 +356,8 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-static __rte_always_inline void
-update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
-		uint32_t single)
-{
-	/*
-	 * If there are other enqueues/dequeues in progress that preceded us,
-	 * we need to wait for them to complete
-	 */
-	if (!single)
-		while (unlikely(ht->tail != old_val))
-			rte_pause();
-
-	ht->tail = new_val;
-}
-
-/**
- * @internal This function updates the producer head for enqueue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sp
- *   Indicates whether multi-producer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where enqueue starts
- * @param new_head
- *   Returns the current/new head value i.e. where enqueue finishes
- * @param free_entries
- *   Returns the amount of free space in the ring BEFORE head was moved
- * @return
- *   Actual number of objects enqueued.
- *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *free_entries)
-{
-	const uint32_t capacity = r->capacity;
-	unsigned int max = n;
-	int success;
-
-	do {
-		/* Reset n to the initial burst count */
-		n = max;
-
-		*old_head = r->prod.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86
-		 */
-		rte_smp_rmb();
-
-		const uint32_t cons_tail = r->cons.tail;
-		/*
-		 *  The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * *old_head > cons_tail). So 'free_entries' is always between 0
-		 * and capacity (which is < size).
-		 */
-		*free_entries = (capacity + cons_tail - *old_head);
-
-		/* check that we have enough room in ring */
-		if (unlikely(n > *free_entries))
-			n = (behavior == RTE_RING_QUEUE_FIXED) ?
-					0 : *free_entries;
-
-		if (n == 0)
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sp)
-			r->prod.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->prod.head,
-					*old_head, *new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
+/* Move common functions to generic file */
+#include "rte_ring_generic.h"
 
 /**
  * @internal Enqueue several objects on the ring
@@ -476,9 +393,8 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 		goto end;
 
 	ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);
-	rte_smp_wmb();
 
-	update_tail(&r->prod, prod_head, prod_next, is_sp);
+	update_tail(&r->prod, prod_head, prod_next, is_sp, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -486,74 +402,6 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 }
 
 /**
- * @internal This function updates the consumer head for dequeue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sc
- *   Indicates whether multi-consumer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where dequeue starts
- * @param new_head
- *   Returns the current/new head value i.e. where dequeue finishes
- * @param entries
- *   Returns the number of entries in the ring BEFORE head was moved
- * @return
- *   - Actual number of objects dequeued.
- *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *entries)
-{
-	unsigned int max = n;
-	int success;
-
-	/* move cons.head atomically */
-	do {
-		/* Restore n as it may change every loop */
-		n = max;
-
-		*old_head = r->cons.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86
-		 */
-		rte_smp_rmb();
-
-		const uint32_t prod_tail = r->prod.tail;
-		/* The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * cons_head > prod_tail). So 'entries' is always between 0
-		 * and size(ring)-1. */
-		*entries = (prod_tail - *old_head);
-
-		/* Set the actual entries for dequeue */
-		if (n > *entries)
-			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
-
-		if (unlikely(n == 0))
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sc)
-			r->cons.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
-					*new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
-
-/**
  * @internal Dequeue several objects from the ring
  *
  * @param r
@@ -587,9 +435,8 @@ __rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,
 		goto end;
 
 	DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *);
-	rte_smp_rmb();
 
-	update_tail(&r->cons, cons_head, cons_next, is_sc);
+	update_tail(&r->cons, cons_head, cons_next, is_sc, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/rte_ring_generic.h b/lib/librte_ring/rte_ring_generic.h
new file mode 100644
index 0000000..eaef94f
--- /dev/null
+++ b/lib/librte_ring/rte_ring_generic.h
@@ -0,0 +1,195 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_GENERIC_H_
+#define _RTE_RING_GENERIC_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	if (enqueue)
+		rte_smp_wmb();
+	else
+		rte_smp_rmb();
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	ht->tail = new_val;
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->prod.head,
+					*old_head, *new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+
+		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1.
+		 */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
+					*new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_GENERIC_H_ */
-- 
2.7.4

[dpdk-dev] [PATCH v6 3/3] ring: introduce new header file to support C11 memory model

[Jia He]

To support C11 memory model barrier, 2 options are suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release(refer to [1]).
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines, refer to [2].

We haven't tested on ppc64. If anyone verifies it, he can add
CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.

[1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
[2] http://dpdk.org/ml/archives/dev/2017-October/080861.html

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
---
 config/common_armv8a_linuxapp      |   2 +
 lib/librte_ring/Makefile           |   3 +-
 lib/librte_ring/rte_ring.h         |  14 ++-
 lib/librte_ring/rte_ring_c11_mem.h | 186 +++++++++++++++++++++++++++++++++++++
 4 files changed, 203 insertions(+), 2 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h

diff --git a/config/common_armv8a_linuxapp b/config/common_armv8a_linuxapp
index 6732d1e..5b7b2eb 100644
--- a/config/common_armv8a_linuxapp
+++ b/config/common_armv8a_linuxapp
@@ -49,3 +49,5 @@ CONFIG_RTE_LIBRTE_SFC_EFX_PMD=n
 CONFIG_RTE_LIBRTE_AVP_PMD=n
 
 CONFIG_RTE_SCHED_VECTOR=n
+
+CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index c959945..a2682e7 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -46,6 +46,7 @@ SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
 SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
-					rte_ring_generic.h
+					rte_ring_generic.h \
+					rte_ring_c11_mem.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 519614c..3343eba 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,8 +356,20 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-/* Move common functions to generic file */
+/* Between load and load. there might be cpu reorder in weak model
+ * (powerpc/arm).
+ * There are 2 choices for the users
+ * 1.use rmb() memory barrier
+ * 2.use one-direcion load_acquire/store_release barrier,defined by
+ * CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
+ * It depends on performance test results.
+ * By default, move common functions to rte_ring_generic.h
+ */
+#ifdef RTE_RING_USE_C11_MEM_MODEL
+#include "rte_ring_c11_mem.h"
+#else
 #include "rte_ring_generic.h"
+#endif
 
 /**
  * @internal Enqueue several objects on the ring
diff --git a/lib/librte_ring/rte_ring_c11_mem.h b/lib/librte_ring/rte_ring_c11_mem.h
new file mode 100644
index 0000000..ca21e95
--- /dev/null
+++ b/lib/librte_ring/rte_ring_c11_mem.h
@@ -0,0 +1,186 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_C11_MEM_H_
+#define _RTE_RING_C11_MEM_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	RTE_SET_USED(enqueue);
+
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	__atomic_store_n(&ht->tail, new_val, __ATOMIC_RELEASE);
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = __atomic_load_n(&r->prod.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->prod.head,
+					old_head, *new_head,
+					0, __ATOMIC_ACQUIRE,
+					__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+		*old_head = __atomic_load_n(&r->cons.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1.
+		 */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->cons.head,
+							old_head, *new_head,
+							0, __ATOMIC_ACQUIRE,
+							__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_C11_MEM_H_ */
-- 
2.7.4

Re: [dpdk-dev] [PATCH v5 1/1] ring: guarantee load/load order in enqueue and dequeue

[Ananyev, Konstantin]

> -----Original Message-----
> From: Jia He [mailto:hejianet@gmail.com]
> Sent: Friday, November 10, 2017 1:51 AM
> To: jerin.jacob@caviumnetworks.com; dev@dpdk.org; olivier.matz@6wind.com
> Cc: Ananyev, Konstantin <konstantin.ananyev@intel.com>; Richardson, Bruce <bruce.richardson@intel.com>; jianbo.liu@arm.com;
> hemant.agrawal@nxp.com; Jia He <hejianet@gmail.com>; Jia He <jia.he@hxt-semitech.com>; jie2.liu@hxt-semitech.com; bing.zhao@hxt-
> semitech.com
> Subject: [PATCH v5 1/1] ring: guarantee load/load order in enqueue and dequeue
> 
> We watched a rte panic of mbuf_autotest in our qualcomm arm64 server.
> In __rte_ring_move_cons_head()
> ...
>         do {
>                 /* Restore n as it may change every loop */
>                 n = max;
> 
>                 *old_head = r->cons.head;                //1st load
>                 const uint32_t prod_tail = r->prod.tail; //2nd load
> 
> cpu1(producer)          cpu2(consumer)          cpu3(consumer)
>                         load r->prod.tail
> in enqueue:
> load r->cons.tail
> load r->prod.head
> 
> store r->prod.tail
> 
>                                                 load r->cons.head
>                                                 load r->prod.tail
>                                                 ...
>                                                 store r->cons.{head,tail}
>                         load r->cons.head
> 
> In weak memory order architectures(powerpc,arm), the 2nd load might be
> reodered before the 1st load, that makes *entries is bigger than we
> wanted. This nasty reording messed enque/deque up. Then, r->cons.head
> will be bigger than prod_tail, then make *entries very big and the
> consumer will go forward incorrectly.
> 
> After this patch, even with above context switches, the old cons.head
> will be recaculated after failure of rte_atomic32_cmpset. So no race
> conditions left.
> 
> There is no such issue on X86, because X86 is strong memory order model.
> But rte_smp_rmb() doesn't have impact on runtime performance on X86, so
> keep the same code without architectures specific concerns.
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Signed-off-by: jie2.liu@hxt-semitech.com
> Signed-off-by: bing.zhao@hxt-semitech.com
> ---
>  lib/librte_ring/rte_ring.h | 10 ++++++++++
>  1 file changed, 10 insertions(+)
> 
> diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
> index 5e9b3b7..3e8085a 100644
> --- a/lib/librte_ring/rte_ring.h
> +++ b/lib/librte_ring/rte_ring.h
> @@ -409,6 +409,11 @@ __rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
>  		n = max;
> 
>  		*old_head = r->prod.head;
> +
> +		/* add rmb barrier to avoid load/load reorder in weak
> +		 * memory model. It is noop on x86 */
> +		rte_smp_rmb();
> +
>  		const uint32_t cons_tail = r->cons.tail;
>  		/*
>  		 *  The subtraction is done between two unsigned 32bits value
> @@ -517,6 +522,11 @@ __rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
>  		n = max;
> 
>  		*old_head = r->cons.head;
> +
> +		/* add rmb barrier to avoid load/load reorder in weak
> +		 * memory model. It is noop on x86 */
> +		rte_smp_rmb();
> +
>  		const uint32_t prod_tail = r->prod.tail;
>  		/* The subtraction is done between two unsigned 32bits value
>  		 * (the result is always modulo 32 bits even if we have
> --

Acked-by: Konstantin Ananyev <konstantin.ananyev@intel.com>

> 2.7.4

Re: [dpdk-dev] [dpdk-stable] [PATCH v6] ring: guarantee load/load order in enqueue and dequeue

[Thomas Monjalon]

10/11/2017 04:30, Jia He:
> We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
> (Amberwing).
> 
> Root cause:
> In __rte_ring_move_cons_head()
> ...
>         do {
>                 /* Restore n as it may change every loop */
>                 n = max;
> 
>                 *old_head = r->cons.head;                //1st load
>                 const uint32_t prod_tail = r->prod.tail; //2nd load
> 
> In weak memory order architectures(powerpc,arm), the 2nd load might be
> reodered before the 1st load, that makes *entries is bigger than we wanted.
> This nasty reording messed enque/deque up.
> 
> cpu1(producer)          cpu2(consumer)          cpu3(consumer)
>                         load r->prod.tail
> in enqueue:
> load r->cons.tail
> load r->prod.head
> 
> store r->prod.tail
> 
>                                                 load r->cons.head
>                                                 load r->prod.tail
>                                                 ...
>                                                 store r->cons.{head,tail}
>                         load r->cons.head
> 
> Then, r->cons.head will be bigger than prod_tail, then make *entries very
> big and the consumer will go forward incorrectly.
> 
> After this patch, the old cons.head will be recaculated after failure of
> rte_atomic32_cmpset
> 
> There is no such issue on X86, because X86 is strong memory order model.
> But rte_smp_rmb() doesn't have impact on runtime performance on X86, so
> keep the same code without architectures specific concerns.
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Signed-off-by: jie2.liu@hxt-semitech.com
> Signed-off-by: bing.zhao@hxt-semitech.com
> Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
> Acked-by: Jianbo Liu <jianbo.liu@arm.com>
> Cc: stable@dpdk.org

Applied, thanks

Re: [dpdk-dev] [PATCH v4 0/4] fix race condition in enqueue/dequeue because of cpu reorder

[Bruce Richardson]

On Fri, Nov 24, 2017 at 10:08:18AM +0800, Jia He wrote:
> Hi Bruce
> 
> I knew little about DPDK's milestone
> 
> I read some hints from http://dpdk.org/dev/roadmap
> 
> 18.02
> 
>  * Proposal deadline: November 24, 2017
>  * Integration deadline: January 5, 2018
> 
> I wonder whether I need to resend the patchset(the 2nd part) again after
> November 24?
> Thanks for any suggestions
> 
> Cheers,
> Jia

It depends upon the status in patchwork. Right now I see three patches
listed from you there, but strangely, despite being a set of 3, 2 are at
v5, and the last is at v6. If this set is not correct, please send a new
revision of the patches and mark the old ones in patchwork as
superceded. The main objective is to ensure that you have one copy of
the patches listed there that the committers can apply.

/Bruce

> On 11/8/2017 11:11 PM, Jia He Wrote:
> > Hi Bruce
> > 
> > 
> > On 11/8/2017 8:15 PM, Bruce Richardson Wrote:
> > > On Wed, Nov 08, 2017 at 09:54:37AM +0000, Jia He wrote:
> > > > We watched a rte panic of mbuf_autotest in our qualcomm arm64 server
> > > > due to a possible race condition.
> > > > 
> > > > To fix this race, there are 2 options as suggested by Jerin: 1. use
> > > > rte_smp_rmb 2. use load_acquire/store_release(refer to [2]).
> > > > CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is
> > > > "y" only on arm64 so far.
> > > > 
> > > > The reason why providing 2 options is due to the performance benchmark
> > > > difference in different arm machines.
> > > > 
> > > > Already fuctionally tested on the machines as follows: - on X86 - on
> > > > arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y - on arm64 with
> > > > CONFIG_RTE_RING_USE_C11_MEM_MODEL=n
> > > > 
> > > > --- Changelog: V4: split into small patches V3: arch specific
> > > > implementation for enqueue/dequeue barrier V2: let users choose
> > > > whether using load_acquire/store_release V1: rte_smp_rmb() between 2
> > > > loads
> > > > 
> > > > Jia He (4): eal/arm64: remove the braces {} for dmb() and dsb() ring:
> > > > guarantee load/load order in enqueue and dequeue ring: introduce new
> > > > header file to include common functions ring: introduce new header
> > > > file to support C11 memory model
> > > > 
> > > I'm wondering what the merge plans are for this set, given we are now
> > > past RC3 in 17.11? As the rings are broken on ARM machines we need to
> > > merge in some fix, but I'm a little concerned about the scope of the
> > > changes from the 3rd and 4th patches. Would it be acceptable to just
> > > merge in patches 1 & 2 in 17.11 and leave the rework and C11 memory
> > > model additions in patches 3 & 4 to 18.02 release?
> > As far as I'm concerned, it is ok.
> > 
> > Cheers,
> > Jia
> 

[dpdk-dev] [PATCH V6 0/3] support c11 memory model barrier in librte_ring

[Jia He]

To support C11 memory model barrier, 2 options are suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines.

Already fuctionally tested on the machines as follows:
- on X86
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=n

---
Changelog:
V6: minor change in subject and log
V5: split it into 2 patchset due to the milestone concerns, this is the 2st 
    one. Also fix checkpatch.pl warnings
V4: split into small patches
V3: arch specific implementation for enqueue/dequeue barrier
V2: let users choose whether using load_acquire/store_release
V1: rte_smp_rmb() between 2 loads

Jia He (3):
  eal/arm64: remove the braces {} for dmb(),dsb()
  ring: introduce new header file to include common functions
  ring: introduce new header file to support C11 memory model

 config/common_armv8a_linuxapp                      |   2 +
 .../common/include/arch/arm/rte_atomic_64.h        |   4 +-
 lib/librte_eventdev/rte_event_ring.h               |   6 +-
 lib/librte_ring/Makefile                           |   4 +-
 lib/librte_ring/rte_ring.h                         | 173 ++----------------
 lib/librte_ring/rte_ring_generic.h                 | 195 +++++++++++++++++++++
 6 files changed, 220 insertions(+), 164 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_generic.h

-- 
2.7.4

[dpdk-dev] [PATCH V6 1/3] eal/arm64: remove the braces {} for dmb() and dsb()

[Jia He]

for the code as follows:
if (condition)
	rte_smp_rmb();
else
	rte_smp_wmb();
Without this patch, compiler will report this error:
error: 'else' without a previous 'if'

Fixes: 84733fd0d75e ("eal/arm: fix memory barrier definition")
Cc: stable@dpdk.org
Signed-off-by: Jia He <jia.he@hxt-semitech.com>
---
 lib/librte_eal/common/include/arch/arm/rte_atomic_64.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
index 0b70d62..71da29c 100644
--- a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
+++ b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
@@ -43,8 +43,8 @@ extern "C" {
 
 #include "generic/rte_atomic.h"
 
-#define dsb(opt)  { asm volatile("dsb " #opt : : : "memory"); }
-#define dmb(opt)  { asm volatile("dmb " #opt : : : "memory"); }
+#define dsb(opt) asm volatile("dsb " #opt : : : "memory")
+#define dmb(opt) asm volatile("dmb " #opt : : : "memory")
 
 #define rte_mb() dsb(sy)
 
-- 
2.7.4

[dpdk-dev] [PATCH V6 2/3] ring: introduce new header file to include common functions

[Jia He]

move the common part of rte_ring.h into rte_ring_generic.h.
move the memory barrier part into update_tail().

no functional changes here.

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
Suggested-by: Ananyev, Konstantin <konstantin.ananyev@intel.com>
---
 lib/librte_eventdev/rte_event_ring.h |   6 +-
 lib/librte_ring/Makefile             |   3 +-
 lib/librte_ring/rte_ring.h           | 161 +----------------------------
 lib/librte_ring/rte_ring_generic.h   | 195 +++++++++++++++++++++++++++++++++++
 4 files changed, 203 insertions(+), 162 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_generic.h

diff --git a/lib/librte_eventdev/rte_event_ring.h b/lib/librte_eventdev/rte_event_ring.h
index ea9b688..3e49458 100644
--- a/lib/librte_eventdev/rte_event_ring.h
+++ b/lib/librte_eventdev/rte_event_ring.h
@@ -126,9 +126,8 @@ rte_event_ring_enqueue_burst(struct rte_event_ring *r,
 		goto end;
 
 	ENQUEUE_PTRS(&r->r, &r[1], prod_head, events, n, struct rte_event);
-	rte_smp_wmb();
 
-	update_tail(&r->r.prod, prod_head, prod_next, 1);
+	update_tail(&r->r.prod, prod_head, prod_next, 1, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -168,9 +167,8 @@ rte_event_ring_dequeue_burst(struct rte_event_ring *r,
 		goto end;
 
 	DEQUEUE_PTRS(&r->r, &r[1], cons_head, events, n, struct rte_event);
-	rte_smp_rmb();
 
-	update_tail(&r->r.cons, cons_head, cons_next, 1);
+	update_tail(&r->r.cons, cons_head, cons_next, 1, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index e34d9d9..c959945 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -45,6 +45,7 @@ LIBABIVER := 1
 SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
-SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
+					rte_ring_generic.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index e924438..519614c 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,91 +356,8 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-static __rte_always_inline void
-update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
-		uint32_t single)
-{
-	/*
-	 * If there are other enqueues/dequeues in progress that preceded us,
-	 * we need to wait for them to complete
-	 */
-	if (!single)
-		while (unlikely(ht->tail != old_val))
-			rte_pause();
-
-	ht->tail = new_val;
-}
-
-/**
- * @internal This function updates the producer head for enqueue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sp
- *   Indicates whether multi-producer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where enqueue starts
- * @param new_head
- *   Returns the current/new head value i.e. where enqueue finishes
- * @param free_entries
- *   Returns the amount of free space in the ring BEFORE head was moved
- * @return
- *   Actual number of objects enqueued.
- *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *free_entries)
-{
-	const uint32_t capacity = r->capacity;
-	unsigned int max = n;
-	int success;
-
-	do {
-		/* Reset n to the initial burst count */
-		n = max;
-
-		*old_head = r->prod.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86
-		 */
-		rte_smp_rmb();
-
-		const uint32_t cons_tail = r->cons.tail;
-		/*
-		 *  The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * *old_head > cons_tail). So 'free_entries' is always between 0
-		 * and capacity (which is < size).
-		 */
-		*free_entries = (capacity + cons_tail - *old_head);
-
-		/* check that we have enough room in ring */
-		if (unlikely(n > *free_entries))
-			n = (behavior == RTE_RING_QUEUE_FIXED) ?
-					0 : *free_entries;
-
-		if (n == 0)
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sp)
-			r->prod.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->prod.head,
-					*old_head, *new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
+/* Move common functions to generic file */
+#include "rte_ring_generic.h"
 
 /**
  * @internal Enqueue several objects on the ring
@@ -476,9 +393,8 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 		goto end;
 
 	ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);
-	rte_smp_wmb();
 
-	update_tail(&r->prod, prod_head, prod_next, is_sp);
+	update_tail(&r->prod, prod_head, prod_next, is_sp, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -486,74 +402,6 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 }
 
 /**
- * @internal This function updates the consumer head for dequeue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sc
- *   Indicates whether multi-consumer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where dequeue starts
- * @param new_head
- *   Returns the current/new head value i.e. where dequeue finishes
- * @param entries
- *   Returns the number of entries in the ring BEFORE head was moved
- * @return
- *   - Actual number of objects dequeued.
- *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *entries)
-{
-	unsigned int max = n;
-	int success;
-
-	/* move cons.head atomically */
-	do {
-		/* Restore n as it may change every loop */
-		n = max;
-
-		*old_head = r->cons.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86
-		 */
-		rte_smp_rmb();
-
-		const uint32_t prod_tail = r->prod.tail;
-		/* The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * cons_head > prod_tail). So 'entries' is always between 0
-		 * and size(ring)-1. */
-		*entries = (prod_tail - *old_head);
-
-		/* Set the actual entries for dequeue */
-		if (n > *entries)
-			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
-
-		if (unlikely(n == 0))
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sc)
-			r->cons.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
-					*new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
-
-/**
  * @internal Dequeue several objects from the ring
  *
  * @param r
@@ -587,9 +435,8 @@ __rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,
 		goto end;
 
 	DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *);
-	rte_smp_rmb();
 
-	update_tail(&r->cons, cons_head, cons_next, is_sc);
+	update_tail(&r->cons, cons_head, cons_next, is_sc, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/rte_ring_generic.h b/lib/librte_ring/rte_ring_generic.h
new file mode 100644
index 0000000..eaef94f
--- /dev/null
+++ b/lib/librte_ring/rte_ring_generic.h
@@ -0,0 +1,195 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_GENERIC_H_
+#define _RTE_RING_GENERIC_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	if (enqueue)
+		rte_smp_wmb();
+	else
+		rte_smp_rmb();
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	ht->tail = new_val;
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->prod.head,
+					*old_head, *new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+
+		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1.
+		 */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
+					*new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_GENERIC_H_ */
-- 
2.7.4

[dpdk-dev] [PATCH V6 3/3] ring: introduce new header file to support C11 memory model

[Jia He]

To support C11 memory model barrier, 2 options are suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release(refer to [1]).
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines, refer to [2].

We haven't tested on ppc64. If anyone verifies it, he can add
CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.

[1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
[2] http://dpdk.org/ml/archives/dev/2017-October/080861.html

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
---
 config/common_armv8a_linuxapp      |   2 +
 lib/librte_ring/Makefile           |   3 +-
 lib/librte_ring/rte_ring.h         |  14 ++-
 lib/librte_ring/rte_ring_c11_mem.h | 186 +++++++++++++++++++++++++++++++++++++
 4 files changed, 203 insertions(+), 2 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h

diff --git a/config/common_armv8a_linuxapp b/config/common_armv8a_linuxapp
index 6732d1e..5b7b2eb 100644
--- a/config/common_armv8a_linuxapp
+++ b/config/common_armv8a_linuxapp
@@ -49,3 +49,5 @@ CONFIG_RTE_LIBRTE_SFC_EFX_PMD=n
 CONFIG_RTE_LIBRTE_AVP_PMD=n
 
 CONFIG_RTE_SCHED_VECTOR=n
+
+CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index c959945..a2682e7 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -46,6 +46,7 @@ SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
 SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
-					rte_ring_generic.h
+					rte_ring_generic.h \
+					rte_ring_c11_mem.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 519614c..3343eba 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,8 +356,20 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-/* Move common functions to generic file */
+/* Between load and load. there might be cpu reorder in weak model
+ * (powerpc/arm).
+ * There are 2 choices for the users
+ * 1.use rmb() memory barrier
+ * 2.use one-direcion load_acquire/store_release barrier,defined by
+ * CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
+ * It depends on performance test results.
+ * By default, move common functions to rte_ring_generic.h
+ */
+#ifdef RTE_RING_USE_C11_MEM_MODEL
+#include "rte_ring_c11_mem.h"
+#else
 #include "rte_ring_generic.h"
+#endif
 
 /**
  * @internal Enqueue several objects on the ring
diff --git a/lib/librte_ring/rte_ring_c11_mem.h b/lib/librte_ring/rte_ring_c11_mem.h
new file mode 100644
index 0000000..ca21e95
--- /dev/null
+++ b/lib/librte_ring/rte_ring_c11_mem.h
@@ -0,0 +1,186 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_C11_MEM_H_
+#define _RTE_RING_C11_MEM_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	RTE_SET_USED(enqueue);
+
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	__atomic_store_n(&ht->tail, new_val, __ATOMIC_RELEASE);
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = __atomic_load_n(&r->prod.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->prod.head,
+					old_head, *new_head,
+					0, __ATOMIC_ACQUIRE,
+					__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+		*old_head = __atomic_load_n(&r->cons.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1.
+		 */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->cons.head,
+							old_head, *new_head,
+							0, __ATOMIC_ACQUIRE,
+							__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_C11_MEM_H_ */
-- 
2.7.4

Re: [dpdk-dev] [PATCH V6 1/3] eal/arm64: remove the braces {} for dmb() and dsb()

[Jerin Jacob]

-----Original Message-----
> Date: Sun, 26 Nov 2017 18:00:22 -0800
> From: Jia He <hejianet@gmail.com>
> To: jerin.jacob@caviumnetworks.com, dev@dpdk.org,
>  bruce.richardson@intel.com, konstantin.ananyev@intel.com
> Cc: olivier.matz@6wind.com, jianbo.liu@arm.com, hemant.agrawal@nxp.com, Jia
>  He <hejianet@gmail.com>, stable@dpdk.org, Jia He <jia.he@hxt-semitech.com>
> Subject: [PATCH V6 1/3] eal/arm64: remove the braces {} for dmb() and dsb()
> X-Mailer: git-send-email 2.7.4
> 
> for the code as follows:
> if (condition)
> 	rte_smp_rmb();
> else
> 	rte_smp_wmb();
> Without this patch, compiler will report this error:
> error: 'else' without a previous 'if'
> 
> Fixes: 84733fd0d75e ("eal/arm: fix memory barrier definition")
> Cc: stable@dpdk.org
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>

Please fix the below checkpatch errors. 

Wrong tag:
        Suggested-by: Ananyev, Konstantin <konstantin.ananyev@intel.com>
Wrong 'Fixes' reference:
        Fixes: 84733fd0d75e ("eal/arm: fix memory barrier definition")

With above fix:
Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Re: [dpdk-dev] [PATCH V6 2/3] ring: introduce new header file to include common functions

[Jerin Jacob]

-----Original Message-----
> Date: Sun, 26 Nov 2017 18:00:23 -0800
> From: Jia He <hejianet@gmail.com>
> To: jerin.jacob@caviumnetworks.com, dev@dpdk.org,
>  bruce.richardson@intel.com, konstantin.ananyev@intel.com
> Cc: olivier.matz@6wind.com, jianbo.liu@arm.com, hemant.agrawal@nxp.com, Jia
>  He <hejianet@gmail.com>, Jia He <jia.he@hxt-semitech.com>
> Subject: [PATCH V6 2/3] ring: introduce new header file to include common
>  functions
> X-Mailer: git-send-email 2.7.4
> 
> move the common part of rte_ring.h into rte_ring_generic.h.
> move the memory barrier part into update_tail().
> 
> no functional changes here.
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
> Suggested-by: Ananyev, Konstantin <konstantin.ananyev@intel.com>

Wrong tag: complaint from checkpatch.
	Suggested-by: Ananyev, Konstantin <konstantin.ananyev@intel.com>

> ---
> + */
> +
> +#ifndef _RTE_RING_GENERIC_H_
> +#define _RTE_RING_GENERIC_H_
> +
> +static __rte_always_inline void
> +update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
> +		uint32_t single, uint32_t enqueue)
> +{

How about making enqueue as const. ie. const uint32_t enqueue ?

> +	if (enqueue)
> +		rte_smp_wmb();
> +	else
> +		rte_smp_rmb();

Other than that, it looks good to me.

Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Re: [dpdk-dev] [PATCH V6 3/3] ring: introduce new header file to support C11 memory model

[Jerin Jacob]

-----Original Message-----
> Date: Sun, 26 Nov 2017 18:00:24 -0800
> From: Jia He <hejianet@gmail.com>
> To: jerin.jacob@caviumnetworks.com, dev@dpdk.org,
>  bruce.richardson@intel.com, konstantin.ananyev@intel.com
> Cc: olivier.matz@6wind.com, jianbo.liu@arm.com, hemant.agrawal@nxp.com, Jia
>  He <hejianet@gmail.com>, Jia He <jia.he@hxt-semitech.com>
> Subject: [PATCH V6 3/3] ring: introduce new header file to support C11
>  memory model
> X-Mailer: git-send-email 2.7.4
> 
> To support C11 memory model barrier, 2 options are suggested by Jerin:
> 1. use rte_smp_rmb
> 2. use load_acquire/store_release(refer to [1]).
> CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
> only on arm64 so far.
> 
> The reason why providing 2 options is due to the performance benchmark
> difference in different arm machines, refer to [2].
> 
> We haven't tested on ppc64. If anyone verifies it, he can add
> CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.
> 
> [1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
> [2] http://dpdk.org/ml/archives/dev/2017-October/080861.html
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

[dpdk-dev] [PATCH V7 0/3] support c11 memory model barrier in librte_ring

[Jia He]

To support C11 memory model barrier, 2 options are suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines.

Already fuctionally tested on the machines as follows:
- on X86
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
- on arm64 with CONFIG_RTE_RING_USE_C11_MEM_MODEL=n

---
Changelog:
V7: fix check-git-log warnings which is suggested by Jerin
V6: minor change in subject and log
V5: split it into 2 patchset due to the milestone concerns, this is the 2st 
    one. Also fix checkpatch.pl warnings
V4: split into small patches
V3: arch specific implementation for enqueue/dequeue barrier
V2: let users choose whether using load_acquire/store_release
V1: rte_smp_rmb() between 2 loads

Jia He (3):
  eal/arm64: remove the braces {} for dmb() and dsb()
  ring: introduce new header file to include common functions
  ring: introduce new header file to support C11 memory model

 config/common_armv8a_linuxapp                      |   2 +
 .../common/include/arch/arm/rte_atomic_64.h        |   4 +-
 lib/librte_eventdev/rte_event_ring.h               |   6 +-
 lib/librte_ring/Makefile                           |   4 +-
 lib/librte_ring/rte_ring.h                         | 173 ++----------------
 lib/librte_ring/rte_ring_c11_mem.h                 | 186 ++++++++++++++++++++
 lib/librte_ring/rte_ring_generic.h                 | 195 +++++++++++++++++++++
 7 files changed, 406 insertions(+), 164 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h
 create mode 100644 lib/librte_ring/rte_ring_generic.h

-- 
2.7.4

[dpdk-dev] [PATCH V7 1/3] eal/arm64: remove the braces {} for dmb() and dsb()

[Jia He]

for the code as follows:
if (condition)
	rte_smp_rmb();
else
	rte_smp_wmb();
Without this patch, compiler will report this error:
error: 'else' without a previous 'if'

Fixes: 84733fd0d75e ("eal/arm64: fix memory barrier definition")
Cc: stable@dpdk.org
Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
---
 lib/librte_eal/common/include/arch/arm/rte_atomic_64.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
index 0b70d62..71da29c 100644
--- a/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
+++ b/lib/librte_eal/common/include/arch/arm/rte_atomic_64.h
@@ -43,8 +43,8 @@ extern "C" {
 
 #include "generic/rte_atomic.h"
 
-#define dsb(opt)  { asm volatile("dsb " #opt : : : "memory"); }
-#define dmb(opt)  { asm volatile("dmb " #opt : : : "memory"); }
+#define dsb(opt) asm volatile("dsb " #opt : : : "memory")
+#define dmb(opt) asm volatile("dmb " #opt : : : "memory")
 
 #define rte_mb() dsb(sy)
 
-- 
2.7.4

[dpdk-dev] [PATCH V7 2/3] ring: introduce new header file to include common functions

[Jia He]

move the common part of rte_ring.h into rte_ring_generic.h.
move the memory barrier part into update_tail().

no functional changes here.

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
Suggested-by: Ananyev Konstantin <konstantin.ananyev@intel.com>
Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
---
 lib/librte_eventdev/rte_event_ring.h |   6 +-
 lib/librte_ring/Makefile             |   3 +-
 lib/librte_ring/rte_ring.h           | 161 +----------------------------
 lib/librte_ring/rte_ring_generic.h   | 195 +++++++++++++++++++++++++++++++++++
 4 files changed, 203 insertions(+), 162 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_generic.h

diff --git a/lib/librte_eventdev/rte_event_ring.h b/lib/librte_eventdev/rte_event_ring.h
index ea9b688..3e49458 100644
--- a/lib/librte_eventdev/rte_event_ring.h
+++ b/lib/librte_eventdev/rte_event_ring.h
@@ -126,9 +126,8 @@ rte_event_ring_enqueue_burst(struct rte_event_ring *r,
 		goto end;
 
 	ENQUEUE_PTRS(&r->r, &r[1], prod_head, events, n, struct rte_event);
-	rte_smp_wmb();
 
-	update_tail(&r->r.prod, prod_head, prod_next, 1);
+	update_tail(&r->r.prod, prod_head, prod_next, 1, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -168,9 +167,8 @@ rte_event_ring_dequeue_burst(struct rte_event_ring *r,
 		goto end;
 
 	DEQUEUE_PTRS(&r->r, &r[1], cons_head, events, n, struct rte_event);
-	rte_smp_rmb();
 
-	update_tail(&r->r.cons, cons_head, cons_next, 1);
+	update_tail(&r->r.cons, cons_head, cons_next, 1, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index e34d9d9..c959945 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -45,6 +45,7 @@ LIBABIVER := 1
 SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
-SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
+					rte_ring_generic.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index e924438..519614c 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,91 +356,8 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-static __rte_always_inline void
-update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
-		uint32_t single)
-{
-	/*
-	 * If there are other enqueues/dequeues in progress that preceded us,
-	 * we need to wait for them to complete
-	 */
-	if (!single)
-		while (unlikely(ht->tail != old_val))
-			rte_pause();
-
-	ht->tail = new_val;
-}
-
-/**
- * @internal This function updates the producer head for enqueue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sp
- *   Indicates whether multi-producer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where enqueue starts
- * @param new_head
- *   Returns the current/new head value i.e. where enqueue finishes
- * @param free_entries
- *   Returns the amount of free space in the ring BEFORE head was moved
- * @return
- *   Actual number of objects enqueued.
- *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *free_entries)
-{
-	const uint32_t capacity = r->capacity;
-	unsigned int max = n;
-	int success;
-
-	do {
-		/* Reset n to the initial burst count */
-		n = max;
-
-		*old_head = r->prod.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86
-		 */
-		rte_smp_rmb();
-
-		const uint32_t cons_tail = r->cons.tail;
-		/*
-		 *  The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * *old_head > cons_tail). So 'free_entries' is always between 0
-		 * and capacity (which is < size).
-		 */
-		*free_entries = (capacity + cons_tail - *old_head);
-
-		/* check that we have enough room in ring */
-		if (unlikely(n > *free_entries))
-			n = (behavior == RTE_RING_QUEUE_FIXED) ?
-					0 : *free_entries;
-
-		if (n == 0)
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sp)
-			r->prod.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->prod.head,
-					*old_head, *new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
+/* Move common functions to generic file */
+#include "rte_ring_generic.h"
 
 /**
  * @internal Enqueue several objects on the ring
@@ -476,9 +393,8 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 		goto end;
 
 	ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);
-	rte_smp_wmb();
 
-	update_tail(&r->prod, prod_head, prod_next, is_sp);
+	update_tail(&r->prod, prod_head, prod_next, is_sp, 1);
 end:
 	if (free_space != NULL)
 		*free_space = free_entries - n;
@@ -486,74 +402,6 @@ __rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
 }
 
 /**
- * @internal This function updates the consumer head for dequeue
- *
- * @param r
- *   A pointer to the ring structure
- * @param is_sc
- *   Indicates whether multi-consumer path is needed or not
- * @param n
- *   The number of elements we will want to enqueue, i.e. how far should the
- *   head be moved
- * @param behavior
- *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
- *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
- * @param old_head
- *   Returns head value as it was before the move, i.e. where dequeue starts
- * @param new_head
- *   Returns the current/new head value i.e. where dequeue finishes
- * @param entries
- *   Returns the number of entries in the ring BEFORE head was moved
- * @return
- *   - Actual number of objects dequeued.
- *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
- */
-static __rte_always_inline unsigned int
-__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
-		unsigned int n, enum rte_ring_queue_behavior behavior,
-		uint32_t *old_head, uint32_t *new_head,
-		uint32_t *entries)
-{
-	unsigned int max = n;
-	int success;
-
-	/* move cons.head atomically */
-	do {
-		/* Restore n as it may change every loop */
-		n = max;
-
-		*old_head = r->cons.head;
-
-		/* add rmb barrier to avoid load/load reorder in weak
-		 * memory model. It is noop on x86
-		 */
-		rte_smp_rmb();
-
-		const uint32_t prod_tail = r->prod.tail;
-		/* The subtraction is done between two unsigned 32bits value
-		 * (the result is always modulo 32 bits even if we have
-		 * cons_head > prod_tail). So 'entries' is always between 0
-		 * and size(ring)-1. */
-		*entries = (prod_tail - *old_head);
-
-		/* Set the actual entries for dequeue */
-		if (n > *entries)
-			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
-
-		if (unlikely(n == 0))
-			return 0;
-
-		*new_head = *old_head + n;
-		if (is_sc)
-			r->cons.head = *new_head, success = 1;
-		else
-			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
-					*new_head);
-	} while (unlikely(success == 0));
-	return n;
-}
-
-/**
  * @internal Dequeue several objects from the ring
  *
  * @param r
@@ -587,9 +435,8 @@ __rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,
 		goto end;
 
 	DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *);
-	rte_smp_rmb();
 
-	update_tail(&r->cons, cons_head, cons_next, is_sc);
+	update_tail(&r->cons, cons_head, cons_next, is_sc, 0);
 
 end:
 	if (available != NULL)
diff --git a/lib/librte_ring/rte_ring_generic.h b/lib/librte_ring/rte_ring_generic.h
new file mode 100644
index 0000000..eaef94f
--- /dev/null
+++ b/lib/librte_ring/rte_ring_generic.h
@@ -0,0 +1,195 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_GENERIC_H_
+#define _RTE_RING_GENERIC_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	if (enqueue)
+		rte_smp_wmb();
+	else
+		rte_smp_rmb();
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	ht->tail = new_val;
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = r->prod.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->prod.head,
+					*old_head, *new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+
+		*old_head = r->cons.head;
+
+		/* add rmb barrier to avoid load/load reorder in weak
+		 * memory model. It is noop on x86
+		 */
+		rte_smp_rmb();
+
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1.
+		 */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = rte_atomic32_cmpset(&r->cons.head, *old_head,
+					*new_head);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_GENERIC_H_ */
-- 
2.7.4

[dpdk-dev] [PATCH V7 3/3] ring: introduce new header file to support C11 memory model

[Jia He]

To support C11 memory model barrier, 2 options are suggested by Jerin:
1. use rte_smp_rmb
2. use load_acquire/store_release(refer to [1]).
CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
only on arm64 so far.

The reason why providing 2 options is due to the performance benchmark
difference in different arm machines, refer to [2].

We haven't tested on ppc64. If anyone verifies it, he can add
CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.

[1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
[2] http://dpdk.org/ml/archives/dev/2017-October/080861.html

Signed-off-by: Jia He <jia.he@hxt-semitech.com>
Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
---
 config/common_armv8a_linuxapp      |   2 +
 lib/librte_ring/Makefile           |   3 +-
 lib/librte_ring/rte_ring.h         |  14 ++-
 lib/librte_ring/rte_ring_c11_mem.h | 186 +++++++++++++++++++++++++++++++++++++
 4 files changed, 203 insertions(+), 2 deletions(-)
 create mode 100644 lib/librte_ring/rte_ring_c11_mem.h

diff --git a/config/common_armv8a_linuxapp b/config/common_armv8a_linuxapp
index 6732d1e..5b7b2eb 100644
--- a/config/common_armv8a_linuxapp
+++ b/config/common_armv8a_linuxapp
@@ -49,3 +49,5 @@ CONFIG_RTE_LIBRTE_SFC_EFX_PMD=n
 CONFIG_RTE_LIBRTE_AVP_PMD=n
 
 CONFIG_RTE_SCHED_VECTOR=n
+
+CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index c959945..a2682e7 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -46,6 +46,7 @@ SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
 
 # install includes
 SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \
-					rte_ring_generic.h
+					rte_ring_generic.h \
+					rte_ring_c11_mem.h
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 519614c..3343eba 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -356,8 +356,20 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
 	} \
 } while (0)
 
-/* Move common functions to generic file */
+/* Between load and load. there might be cpu reorder in weak model
+ * (powerpc/arm).
+ * There are 2 choices for the users
+ * 1.use rmb() memory barrier
+ * 2.use one-direcion load_acquire/store_release barrier,defined by
+ * CONFIG_RTE_RING_USE_C11_MEM_MODEL=y
+ * It depends on performance test results.
+ * By default, move common functions to rte_ring_generic.h
+ */
+#ifdef RTE_RING_USE_C11_MEM_MODEL
+#include "rte_ring_c11_mem.h"
+#else
 #include "rte_ring_generic.h"
+#endif
 
 /**
  * @internal Enqueue several objects on the ring
diff --git a/lib/librte_ring/rte_ring_c11_mem.h b/lib/librte_ring/rte_ring_c11_mem.h
new file mode 100644
index 0000000..ca21e95
--- /dev/null
+++ b/lib/librte_ring/rte_ring_c11_mem.h
@@ -0,0 +1,186 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2017 hxt-semitech. All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of hxt-semitech nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _RTE_RING_C11_MEM_H_
+#define _RTE_RING_C11_MEM_H_
+
+static __rte_always_inline void
+update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,
+		uint32_t single, uint32_t enqueue)
+{
+	RTE_SET_USED(enqueue);
+
+	/*
+	 * If there are other enqueues/dequeues in progress that preceded us,
+	 * we need to wait for them to complete
+	 */
+	if (!single)
+		while (unlikely(ht->tail != old_val))
+			rte_pause();
+
+	__atomic_store_n(&ht->tail, new_val, __ATOMIC_RELEASE);
+}
+
+/**
+ * @internal This function updates the producer head for enqueue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sp
+ *   Indicates whether multi-producer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where enqueue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where enqueue finishes
+ * @param free_entries
+ *   Returns the amount of free space in the ring BEFORE head was moved
+ * @return
+ *   Actual number of objects enqueued.
+ *   If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *free_entries)
+{
+	const uint32_t capacity = r->capacity;
+	unsigned int max = n;
+	int success;
+
+	do {
+		/* Reset n to the initial burst count */
+		n = max;
+
+		*old_head = __atomic_load_n(&r->prod.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t cons_tail = r->cons.tail;
+		/*
+		 *  The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * *old_head > cons_tail). So 'free_entries' is always between 0
+		 * and capacity (which is < size).
+		 */
+		*free_entries = (capacity + cons_tail - *old_head);
+
+		/* check that we have enough room in ring */
+		if (unlikely(n > *free_entries))
+			n = (behavior == RTE_RING_QUEUE_FIXED) ?
+					0 : *free_entries;
+
+		if (n == 0)
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sp)
+			r->prod.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->prod.head,
+					old_head, *new_head,
+					0, __ATOMIC_ACQUIRE,
+					__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+/**
+ * @internal This function updates the consumer head for dequeue
+ *
+ * @param r
+ *   A pointer to the ring structure
+ * @param is_sc
+ *   Indicates whether multi-consumer path is needed or not
+ * @param n
+ *   The number of elements we will want to enqueue, i.e. how far should the
+ *   head be moved
+ * @param behavior
+ *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
+ *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring
+ * @param old_head
+ *   Returns head value as it was before the move, i.e. where dequeue starts
+ * @param new_head
+ *   Returns the current/new head value i.e. where dequeue finishes
+ * @param entries
+ *   Returns the number of entries in the ring BEFORE head was moved
+ * @return
+ *   - Actual number of objects dequeued.
+ *     If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.
+ */
+static __rte_always_inline unsigned int
+__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,
+		unsigned int n, enum rte_ring_queue_behavior behavior,
+		uint32_t *old_head, uint32_t *new_head,
+		uint32_t *entries)
+{
+	unsigned int max = n;
+	int success;
+
+	/* move cons.head atomically */
+	do {
+		/* Restore n as it may change every loop */
+		n = max;
+		*old_head = __atomic_load_n(&r->cons.head,
+					__ATOMIC_ACQUIRE);
+		const uint32_t prod_tail = r->prod.tail;
+		/* The subtraction is done between two unsigned 32bits value
+		 * (the result is always modulo 32 bits even if we have
+		 * cons_head > prod_tail). So 'entries' is always between 0
+		 * and size(ring)-1.
+		 */
+		*entries = (prod_tail - *old_head);
+
+		/* Set the actual entries for dequeue */
+		if (n > *entries)
+			n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
+
+		if (unlikely(n == 0))
+			return 0;
+
+		*new_head = *old_head + n;
+		if (is_sc)
+			r->cons.head = *new_head, success = 1;
+		else
+			success = __atomic_compare_exchange_n(&r->cons.head,
+							old_head, *new_head,
+							0, __ATOMIC_ACQUIRE,
+							__ATOMIC_RELAXED);
+	} while (unlikely(success == 0));
+	return n;
+}
+
+#endif /* _RTE_RING_C11_MEM_H_ */
-- 
2.7.4

Re: [dpdk-dev] [PATCH V7 3/3] ring: introduce new header file to support C11 memory model

[Jianbo Liu]

The 12/03/2017 17:50, Jia He wrote:
> To support C11 memory model barrier, 2 options are suggested by Jerin:
> 1. use rte_smp_rmb
> 2. use load_acquire/store_release(refer to [1]).
> CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
> only on arm64 so far.
>
> The reason why providing 2 options is due to the performance benchmark
> difference in different arm machines, refer to [2].
>
> We haven't tested on ppc64. If anyone verifies it, he can add
> CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.
>
> [1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
> [2] http://dpdk.org/ml/archives/dev/2017-October/080861.html
>
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
> Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Acked-by: Jianbo Liu <jianbo.liu@arm.com>

> ---
>  config/common_armv8a_linuxapp      |   2 +
>  lib/librte_ring/Makefile           |   3 +-
>  lib/librte_ring/rte_ring.h         |  14 ++-
>  lib/librte_ring/rte_ring_c11_mem.h | 186 +++++++++++++++++++++++++++++++++++++
>  4 files changed, 203 insertions(+), 2 deletions(-)
>  create mode 100644 lib/librte_ring/rte_ring_c11_mem.h
>
IMPORTANT NOTICE: The contents of this email and any attachments are confidential and may also be privileged. If you are not the intended recipient, please notify the sender immediately and do not disclose the contents to any other person, use it for any purpose, or store or copy the information in any medium. Thank you.

Re: [dpdk-dev] [PATCH V7 2/3] ring: introduce new header file to include common functions

[Thomas Monjalon]

04/12/2017 02:50, Jia He:
> move the common part of rte_ring.h into rte_ring_generic.h.
> move the memory barrier part into update_tail().
> 
> no functional changes here.
[...]
> --- /dev/null
> +++ b/lib/librte_ring/rte_ring_generic.h
> @@ -0,0 +1,195 @@
> +/*-
> + *   BSD LICENSE
> + *
> + *   Copyright(c) 2017 hxt-semitech. All rights reserved.

As you are moving existing code, I think you should keep
the original copyright.

Re: [dpdk-dev] [PATCH V7 3/3] ring: introduce new header file to support C11 memory model

[Thomas Monjalon]

Hi,
Please find few comments.
Sorry for the late review.
We need also to get the acknowledgement from Olivier.

04/12/2017 02:50, Jia He:
> --- a/config/common_armv8a_linuxapp
> +++ b/config/common_armv8a_linuxapp
> @@ -49,3 +49,5 @@ CONFIG_RTE_LIBRTE_SFC_EFX_PMD=n
>  CONFIG_RTE_LIBRTE_AVP_PMD=n
>  
>  CONFIG_RTE_SCHED_VECTOR=n
> +
> +CONFIG_RTE_RING_USE_C11_MEM_MODEL=y

This config option should be added in the common file (as disabled).

> --- /dev/null
> +++ b/lib/librte_ring/rte_ring_c11_mem.h
> @@ -0,0 +1,186 @@
> +/*-
> + *   BSD LICENSE
> + *
> + *   Copyright(c) 2017 hxt-semitech. All rights reserved.
> + *
> + *   Redistribution and use in source and binary forms, with or without
> + *   modification, are permitted provided that the following conditions
> + *   are met:
> + *
> + *     * Redistributions of source code must retain the above copyright
> + *       notice, this list of conditions and the following disclaimer.
> + *     * Redistributions in binary form must reproduce the above copyright
> + *       notice, this list of conditions and the following disclaimer in
> + *       the documentation and/or other materials provided with the
> + *       distribution.
> + *     * Neither the name of hxt-semitech nor the names of its
> + *       contributors may be used to endorse or promote products derived
> + *       from this software without specific prior written permission.
> + *
> + *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
> + *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
> + *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
> + *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
> + *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
> + *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
> + *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
> + *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
> + *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
> + *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
> + *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
> + */

If you have to spin a v8, please use SPDX tag for the license.

Re: [dpdk-dev] [PATCH V7 2/3] ring: introduce new header file to include common functions

[Jia He]

On 1/13/2018 1:09 AM, Thomas Monjalon Wrote:
> 04/12/2017 02:50, Jia He:
>> move the common part of rte_ring.h into rte_ring_generic.h.
>> move the memory barrier part into update_tail().
>>
>> no functional changes here.
> [...]
>> --- /dev/null
>> +++ b/lib/librte_ring/rte_ring_generic.h
>> @@ -0,0 +1,195 @@
>> +/*-
>> + *   BSD LICENSE
>> + *
>> + *   Copyright(c) 2017 hxt-semitech. All rights reserved.
> As you are moving existing code, I think you should keep
> the original copyright.
Ok, thanks for the comments

-- 
Cheers,
Jia

Re: [dpdk-dev] [PATCH V7 3/3] ring: introduce new header file to support C11 memory model

[Olivier Matz]

On Sun, Dec 03, 2017 at 05:50:12PM -0800, Jia He wrote:
> To support C11 memory model barrier, 2 options are suggested by Jerin:
> 1. use rte_smp_rmb
> 2. use load_acquire/store_release(refer to [1]).
> CONFIG_RTE_RING_USE_C11_MEM_MODEL is provided, and by default it is "y"
> only on arm64 so far.
> 
> The reason why providing 2 options is due to the performance benchmark
> difference in different arm machines, refer to [2].
> 
> We haven't tested on ppc64. If anyone verifies it, he can add
> CONFIG_RTE_RING_USE_C11_MEM_MODEL=y to ppc64 config files.
> 
> [1] https://github.com/freebsd/freebsd/blob/master/sys/sys/buf_ring.h#L170
> [2] http://dpdk.org/ml/archives/dev/2017-October/080861.html
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
> Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Acked-by: Olivier Matz <olivier.matz@6wind.com>

Re: [dpdk-dev] [PATCH V7 2/3] ring: introduce new header file to include common functions

[Olivier Matz]

On Sun, Dec 03, 2017 at 05:50:11PM -0800, Jia He wrote:
> move the common part of rte_ring.h into rte_ring_generic.h.
> move the memory barrier part into update_tail().
> 
> no functional changes here.
> 
> Signed-off-by: Jia He <jia.he@hxt-semitech.com>
> Suggested-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>
> Suggested-by: Ananyev Konstantin <konstantin.ananyev@intel.com>
> Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

Acked-by: Olivier Matz <olivier.matz@6wind.com>