Minor mistake in ring (en|de)queueing

PATCH In review Andrzej Ostruszka 5 participants 11 messages Jan 3, 2022 – Feb 5, 2022

lore

Share & tools

lore

Andrzej Ostruszka

author

2022-01-03 14:23 UTC #0 lore raw

cover Minor mistake in ring (en|de)queueing

Hi all,

Recently I was going through the ring implementation and I believe I've found a small mistake. Not a functional error, just a slightly suboptimal behaviour for the specific case when we want to enqueue exactly the number of elements that we can before wrapping to the ring beginning (the same goes for dequeueing).

Imagine we have a ring of size 16 depicted below with indexes underneath and consumer/producer pointing as shown.

| ****** |
0123456789ABCDEF
^ ^
c p

Now if we try to enqueue 8 elements we will end up at the check:

	if (likely(idx + n < size)) {

where idx (=8) is a producer head and n=8. We will fail this check even though we have the case of 8 elements available without wrapping to the beginning of the ring.

I hope I'm not completely off the base here :), if I'm not then the subsequent patch attempts to fix this.

Wit regards
Andrzej Ostruszka

Andrzej Ostruszka (1):
ring: fix off by 1 mistake

 lib/ring/rte_ring_elem_pvt.h | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

Signature · 1 line

2.34.1.448.ga2b2bfdf31-goog

Andrzej Ostruszka

author

2022-01-03 14:23 UTC #1 lore raw

PATCH 1/1 ring: fix off by 1 mistake

When enqueueing/dequeueing to/from the ring we try to optimize by manual loop unrolling. The check for this optimization looks like:

	if (likely(idx + n < size)) {

where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.

This is not a functional error since we fall back to the loop with correct checks on indexes. Just a minor suboptimal behaviour for the case when we want to enqueue/dequeue exactly the number of elements that we have in the ring before wrapping to its beginning.

Signed-off-by Andrzej Ostruszka <[email protected]>

---
 lib/ring/rte_ring_elem_pvt.h | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

lib/ring/rte_ring_elem_pvt.h +6 -6

index 275ec55393..83788c56e6 100644
@@ -17,7 +17,7 @@ __rte_ring_enqueue_elems_32(struct rte_ring *r, const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	const uint32_t *obj = (const uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -62,7 +62,7 @@ __rte_ring_enqueue_elems_64(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	const unaligned_uint64_t *obj = (const unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -95,7 +95,7 @@ __rte_ring_enqueue_elems_128(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	const rte_int128_t *obj = (const rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(ring + idx),
 				(const void *)(obj + i), 32);

@@ -151,7 +151,7 @@ __rte_ring_dequeue_elems_32(struct rte_ring *r, const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	uint32_t *obj = (uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			obj[i] = ring[idx];
 			obj[i + 1] = ring[idx + 1];

@@ -196,7 +196,7 @@ __rte_ring_dequeue_elems_64(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	unaligned_uint64_t *obj = (unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			obj[i] = ring[idx];
 			obj[i + 1] = ring[idx + 1];

@@ -229,7 +229,7 @@ __rte_ring_dequeue_elems_128(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	rte_int128_t *obj = (rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(obj + i), (void *)(ring + idx), 32);
 		switch (n & 0x1) {

Signature · 1 line

2.34.1.448.ga2b2bfdf31-goog

Morten Brørup

Reviewed-by on v1 1/1

2022-01-03 14:56 UTC #2 lore raw

RE: [PATCH 1/1] ring: fix off by 1 mistake

+Ring queue maintainers: Honnappa Nagarahalli <[email protected]>, Konstantin Ananyev <[email protected]>

Quoted 89 lines

From: Andrzej Ostruszka [mailto:[email protected]]
Sent: Monday, 3 January 2022 15.22

When enqueueing/dequeueing to/from the ring we try to optimize by manual
loop unrolling. The check for this optimization looks like:

	if (likely(idx + n < size)) {

where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.

This is not a functional error since we fall back to the loop with correct checks on indexes. Just a minor suboptimal behaviour for the case when we want to enqueue/dequeue exactly the number of elements that
we have in the ring before wrapping to its beginning.

Signed-off-by Andrzej Ostruszka <[email protected]>

---
 lib/ring/rte_ring_elem_pvt.h | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

lib/ring/rte_ring_elem_pvt.h +1 -1

index 275ec55393..83788c56e6 100644
@@ -17,7 +17,7 @@ __rte_ring_enqueue_elems_32(struct rte_ring *r, const
 uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	const uint32_t *obj = (const uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			ring[idx] = obj[i];

lib/ring/rte_ring_elem_pvt.h +1 -1

 			ring[idx + 1] = obj[i + 1];

@@ -62,7 +62,7 @@ __rte_ring_enqueue_elems_64(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	const unaligned_uint64_t *obj = (const unaligned_uint64_t
 *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {

lib/ring/rte_ring_elem_pvt.h +1 -1

 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -95,7 +95,7 @@ __rte_ring_enqueue_elems_128(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	const rte_int128_t *obj = (const rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(ring + idx),

lib/ring/rte_ring_elem_pvt.h +1 -1

 				(const void *)(obj + i), 32);

@@ -151,7 +151,7 @@ __rte_ring_dequeue_elems_32(struct rte_ring *r,
 const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	uint32_t *obj = (uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			obj[i] = ring[idx];

lib/ring/rte_ring_elem_pvt.h +1 -1

 			obj[i + 1] = ring[idx + 1];

@@ -196,7 +196,7 @@ __rte_ring_dequeue_elems_64(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	unaligned_uint64_t *obj = (unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			obj[i] = ring[idx];

lib/ring/rte_ring_elem_pvt.h +1 -1

 			obj[i + 1] = ring[idx + 1];

@@ -229,7 +229,7 @@ __rte_ring_dequeue_elems_128(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	rte_int128_t *obj = (rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(obj + i), (void *)(ring + idx), 32);

Quoted 4 lines

switch (n & 0x1) {
--
2.34.1.448.ga2b2bfdf31-goog

Well spotted! I took a very good look at it, and came to the same conclusion: It not a functional bug; the only consequence is that the optimized code path may not be taken in a situation where it could be taken. But it should be fixed as suggested in your patch.

Reviewed-by Morten Brørup <[email protected]>

Olivier Matz

Reviewed-by on v1 1/1

2022-01-06 10:45 UTC #3 lore raw

Re: [PATCH 1/1] ring: fix off by 1 mistake

In reply to Andrzej Ostruszka

Hi Andrzej,

Quoted 1 line · Andrzej Ostruszka

ring: fix off by 1 mistake

I suggest something less scary for the title:

ring: optimize corner case for enqueue/dequeue

Quoted 9 lines

When enqueueing/dequeueing to/from the ring we try to optimize by manual loop unrolling. The check for this optimization looks like:
	if (likely(idx + n < size)) {
where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.

Quoted 6 lines

This is not a functional error since we fall back to the loop with correct checks on indexes. Just a minor suboptimal behaviour for the case when we want to enqueue/dequeue exactly the number of elements that we have in the ring before wrapping to its beginning.
Signed-off-by Andrzej Ostruszka <[email protected]>

Reviewed-by Olivier Matz <[email protected]>

I'll tend to add:

Fixes cc4b218790f6 ("ring: support configurable element size")

But the same error was in the ENQUEUE_PTRS()/DEQUEUE_PTRS() macros since the beginning, so we may also add:

Fixes 286bd05bf70d ("ring: optimisations")

This macro was removed in commit 2d6ed071a8b9 ("ring: use custom element for fixed size API")

Thanks!

Ananyev, Konstantin

Acked-by on v1 1/1

2022-01-10 15:09 UTC #4 lore raw

RE: [PATCH 1/1] ring: fix off by 1 mistake

In reply to Andrzej Ostruszka

Quoted 69 lines

When enqueueing/dequeueing to/from the ring we try to optimize by manual loop unrolling. The check for this optimization looks like:

	if (likely(idx + n < size)) {

where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.

Signed-off-by Andrzej Ostruszka <[email protected]>

---
 lib/ring/rte_ring_elem_pvt.h | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

lib/ring/rte_ring_elem_pvt.h +5 -5

index 275ec55393..83788c56e6 100644
@@ -17,7 +17,7 @@ __rte_ring_enqueue_elems_32(struct rte_ring *r, const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	const uint32_t *obj = (const uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -62,7 +62,7 @@ __rte_ring_enqueue_elems_64(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	const unaligned_uint64_t *obj = (const unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -95,7 +95,7 @@ __rte_ring_enqueue_elems_128(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	const rte_int128_t *obj = (const rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(ring + idx),
 				(const void *)(obj + i), 32);

@@ -151,7 +151,7 @@ __rte_ring_dequeue_elems_32(struct rte_ring *r, const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	uint32_t *obj = (uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			obj[i] = ring[idx];
 			obj[i + 1] = ring[idx + 1];

@@ -196,7 +196,7 @@ __rte_ring_dequeue_elems_64(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	unaligned_uint64_t *obj = (unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			obj[i] = ring[idx];
 			obj[i + 1] = ring[idx + 1];

lib/ring/rte_ring_elem_pvt.h __rte_ring_dequeue_elems_128(struct rte_ring *r, uint32_t prod_head,

Quoted 9 lines

lib/ring/rte_ring_elem_pvt.h +1 -2

@@ -229,7 +229,7 @@ __rte_ring_dequeue_elems_128(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	rte_int128_t *obj = (rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(obj + i), (void *)(ring + idx), 32);
 		switch (n & 0x1) {
--

Acked-by Konstantin Ananyev <[email protected]>

Quoted 1 line

2.34.1.448.ga2b2bfdf31-goog

Andrzej Ostruszka

author

2022-01-11 11:50 UTC #5 lore raw

Re: [PATCH 1/1] ring: fix off by 1 mistake

Thank you Morten, Olivier and Konstantin for taking look at it. I've just sent another version, with updates in commit message suggested by Olivier.

With regards
Andrzej Ostruszka

Andrzej Ostruszka

author

2022-01-11 11:38 UTC #6 lore raw

PATCH 1/1 ring: optimize corner case for enqueue/dequeue

In reply to Andrzej Ostruszka

When enqueueing/dequeueing to/from the ring we try to optimize by manual loop unrolling. The check for this optimization looks like:

	if (likely(idx + n < size)) {

where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.

Fixes cc4b218790f6 ("ring: support configurable element size")

Fixes 286bd05bf70d ("ring: optimisations")

Signed-off-by Andrzej Ostruszka <[email protected]>

Reviewed-by Olivier Matz <[email protected]>

Acked-by Konstantin Ananyev <[email protected]>

---
 lib/ring/rte_ring_elem_pvt.h | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

lib/ring/rte_ring_elem_pvt.h +6 -6

index 275ec55393..83788c56e6 100644
@@ -17,7 +17,7 @@ __rte_ring_enqueue_elems_32(struct rte_ring *r, const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	const uint32_t *obj = (const uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -62,7 +62,7 @@ __rte_ring_enqueue_elems_64(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	const unaligned_uint64_t *obj = (const unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -95,7 +95,7 @@ __rte_ring_enqueue_elems_128(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	const rte_int128_t *obj = (const rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(ring + idx),
 				(const void *)(obj + i), 32);

@@ -151,7 +151,7 @@ __rte_ring_dequeue_elems_32(struct rte_ring *r, const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	uint32_t *obj = (uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			obj[i] = ring[idx];
 			obj[i + 1] = ring[idx + 1];

@@ -196,7 +196,7 @@ __rte_ring_dequeue_elems_64(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	unaligned_uint64_t *obj = (unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			obj[i] = ring[idx];
 			obj[i + 1] = ring[idx + 1];

@@ -229,7 +229,7 @@ __rte_ring_dequeue_elems_128(struct rte_ring *r, uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	rte_int128_t *obj = (rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(obj + i), (void *)(ring + idx), 32);
 		switch (n & 0x1) {

Signature · 1 line

2.34.1.575.g55b058a8bb-goog

Morten Brørup

Reviewed-by on v1 1/1

2022-01-11 12:00 UTC #7 lore raw

RE: [PATCH] ring: optimize corner case for enqueue/dequeue

Quoted 94 lines

From: Andrzej Ostruszka [mailto:[email protected]]
Sent: Tuesday, 11 January 2022 12.38

When enqueueing/dequeueing to/from the ring we try to optimize by manual
loop unrolling. The check for this optimization looks like:

	if (likely(idx + n < size)) {

where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.

This is not a functional error since we fall back to the loop with correct checks on indexes. Just a minor suboptimal behaviour for the case when we want to enqueue/dequeue exactly the number of elements that
we have in the ring before wrapping to its beginning.

Fixes cc4b218790f6 ("ring: support configurable element size")

Fixes 286bd05bf70d ("ring: optimisations")

Signed-off-by Andrzej Ostruszka <[email protected]>

Reviewed-by Olivier Matz <[email protected]>

Acked-by Konstantin Ananyev <[email protected]>

---
 lib/ring/rte_ring_elem_pvt.h | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

lib/ring/rte_ring_elem_pvt.h +1 -1

index 275ec55393..83788c56e6 100644
@@ -17,7 +17,7 @@ __rte_ring_enqueue_elems_32(struct rte_ring *r, const
 uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	const uint32_t *obj = (const uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			ring[idx] = obj[i];

lib/ring/rte_ring_elem_pvt.h +1 -1

 			ring[idx + 1] = obj[i + 1];

@@ -62,7 +62,7 @@ __rte_ring_enqueue_elems_64(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	const unaligned_uint64_t *obj = (const unaligned_uint64_t
 *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {

lib/ring/rte_ring_elem_pvt.h +1 -1

 			ring[idx] = obj[i];
 			ring[idx + 1] = obj[i + 1];

@@ -95,7 +95,7 @@ __rte_ring_enqueue_elems_128(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	const rte_int128_t *obj = (const rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(ring + idx),

lib/ring/rte_ring_elem_pvt.h +1 -1

 				(const void *)(obj + i), 32);

@@ -151,7 +151,7 @@ __rte_ring_dequeue_elems_32(struct rte_ring *r,
 const uint32_t size,
 	unsigned int i;
 	uint32_t *ring = (uint32_t *)&r[1];
 	uint32_t *obj = (uint32_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x7); i += 8, idx += 8) {
 			obj[i] = ring[idx];

lib/ring/rte_ring_elem_pvt.h +1 -1

 			obj[i + 1] = ring[idx + 1];

@@ -196,7 +196,7 @@ __rte_ring_dequeue_elems_64(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	uint64_t *ring = (uint64_t *)&r[1];
 	unaligned_uint64_t *obj = (unaligned_uint64_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x3); i += 4, idx += 4) {
 			obj[i] = ring[idx];

lib/ring/rte_ring_elem_pvt.h +1 -1

 			obj[i + 1] = ring[idx + 1];

@@ -229,7 +229,7 @@ __rte_ring_dequeue_elems_128(struct rte_ring *r,
 uint32_t prod_head,
 	uint32_t idx = prod_head & r->mask;
 	rte_int128_t *ring = (rte_int128_t *)&r[1];
 	rte_int128_t *obj = (rte_int128_t *)obj_table;
-	if (likely(idx + n < size)) {
+	if (likely(idx + n <= size)) {
 		for (i = 0; i < (n & ~0x1); i += 2, idx += 2)
 			memcpy((void *)(obj + i), (void *)(ring + idx), 32);

Quoted 4 lines

switch (n & 0x1) {
--
2.34.1.575.g55b058a8bb-goog

Also this version of the patch:

Reviewed-by Morten Brørup <[email protected]>

Andrzej Ostruszka

author

2022-01-11 13:46 UTC #8 lore raw

Re: [PATCH] ring: optimize corner case for enqueue/dequeue

On Tue, Jan 11, 2022 at 01:00:25PM +0100, Morten Brørup wrote:
[...]

Quoted 3 lines

Also this version of the patch:
Reviewed-by Morten Brørup <[email protected]>

My apologies Morten, don't know how I missed your tag.

With regards
Andrzej Ostruszka

Morten Brørup

2022-01-11 13:53 UTC #9 lore raw

RE: [PATCH] ring: optimize corner case for enqueue/dequeue

Quoted 5 lines

From: Andrzej Ostruszka [mailto:[email protected]]
Sent: Tuesday, 11 January 2022 14.46
On Tue, Jan 11, 2022 at 01:00:25PM +0100, Morten Brørup wrote:
[...]

Quoted 8 lines

Quoted 3 lines
Also this version of the patch:
Reviewed-by Morten Brørup <[email protected]>
My apologies Morten, don't know how I missed your tag.
With regards
Andrzej Ostruszka

No worries. :-)

I added the tag again, for Patchwork to show a sufficient level of acceptance when the maintainer considers the patch for merging.

-Morten

David Marchand

Reviewed-by Morten Brørup on v1 1/1 Applied

2022-02-05 16:49 UTC #10 lore raw

Re: [PATCH] ring: optimize corner case for enqueue/dequeue

In reply to Andrzej Ostruszka

Quoted 14 lines · Andrzej Ostruszka

When enqueueing/dequeueing to/from the ring we try to optimize by manual loop unrolling. The check for this optimization looks like:
        if (likely(idx + n < size)) {
where 'idx' points to the first usable element (empty slot for enqueue, data for dequeue). The correct comparison here should be '<=' instead of '<'.
This is not a functional error since we fall back to the loop with correct checks on indexes. Just a minor suboptimal behaviour for the case when we want to enqueue/dequeue exactly the number of elements that we have in the ring before wrapping to its beginning.

Quoted 6 lines

Fixes cc4b218790f6 ("ring: support configurable element size")
Fixes 286bd05bf70d ("ring: optimisations")
Signed-off-by Andrzej Ostruszka <[email protected]>
Reviewed-by Olivier Matz <[email protected]>
Acked-by Konstantin Ananyev <[email protected]>

Reviewed-by Morten Brørup <[email protected]>

Applied, thanks.

Signature · 1 line

David Marchand