DPDK patches and discussions
 help / color / mirror / Atom feed
* [dpdk-dev] [PATCH] ring: assert on zero objects dequeue/enqueue
@ 2016-03-15 16:58 Lazaros Koromilas
  2016-03-16 11:21 ` Bruce Richardson
  0 siblings, 1 reply; 3+ messages in thread
From: Lazaros Koromilas @ 2016-03-15 16:58 UTC (permalink / raw)
  To: dev

Issuing a zero objects dequeue with a single consumer has no effect.
Doing so with multiple consumers, can get more than one thread to succeed
the compare-and-set operation and observe starvation or even deadlock in
the while loop that checks for preceding dequeues.  The problematic piece
of code when n = 0:

    cons_next = cons_head + n;
    success = rte_atomic32_cmpset(&r->cons.head, cons_head, cons_next);

The same is possible on the enqueue path.

Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
---
 lib/librte_ring/rte_ring.h | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 943c97c..2bf9ce3 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -100,6 +100,7 @@ extern "C" {
 #include <rte_lcore.h>
 #include <rte_atomic.h>
 #include <rte_branch_prediction.h>
+#include <rte_debug.h>
 
 #define RTE_TAILQ_RING_NAME "RTE_RING"
 
@@ -211,6 +212,19 @@ struct rte_ring {
 #endif
 
 /**
+ * @internal Assert macro.
+ * @param exp
+ *   The expression to evaluate.
+ */
+#define RTE_RING_ASSERT(exp) do {                                 \
+		if (!(exp)) {                                     \
+			rte_panic("line%d\t"                      \
+			          "assert \"" #exp "\" failed\n", \
+			          __LINE__);                      \
+		}                                                 \
+	} while (0)
+
+/**
  * Calculate the memory size needed for a ring
  *
  * This function returns the number of bytes needed for a ring, given
@@ -406,6 +420,7 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
  *   A pointer to a table of void * pointers (objects).
  * @param n
  *   The number of objects to add in the ring from the obj_table.
+ *   Must be greater than zero.
  * @param behavior
  *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
  *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
@@ -431,6 +446,8 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
 	uint32_t mask = r->prod.mask;
 	int ret;
 
+	RTE_RING_ASSERT(n > 0);
+
 	/* move prod.head atomically */
 	do {
 		/* Reset n to the initial burst count */
@@ -510,6 +527,7 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
  *   A pointer to a table of void * pointers (objects).
  * @param n
  *   The number of objects to add in the ring from the obj_table.
+ *   Must be greater than zero.
  * @param behavior
  *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
  *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
@@ -533,6 +551,8 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
 	uint32_t mask = r->prod.mask;
 	int ret;
 
+	RTE_RING_ASSERT(n > 0);
+
 	prod_head = r->prod.head;
 	cons_tail = r->cons.tail;
 	/* The subtraction is done between two unsigned 32bits value
@@ -594,6 +614,7 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
  *   A pointer to a table of void * pointers (objects) that will be filled.
  * @param n
  *   The number of objects to dequeue from the ring to the obj_table.
+ *   Must be greater than zero.
  * @param behavior
  *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
  *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
@@ -618,6 +639,8 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
 	unsigned i, rep = 0;
 	uint32_t mask = r->prod.mask;
 
+	RTE_RING_ASSERT(n > 0);
+
 	/* move cons.head atomically */
 	do {
 		/* Restore n as it may change every loop */
@@ -689,6 +712,7 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
  *   A pointer to a table of void * pointers (objects) that will be filled.
  * @param n
  *   The number of objects to dequeue from the ring to the obj_table.
+ *   Must be greater than zero.
  * @param behavior
  *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
  *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
@@ -710,6 +734,8 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
 	unsigned i;
 	uint32_t mask = r->prod.mask;
 
+	RTE_RING_ASSERT(n > 0);
+
 	cons_head = r->cons.head;
 	prod_tail = r->prod.tail;
 	/* The subtraction is done between two unsigned 32bits value
-- 
1.9.1

^ permalink raw reply	[flat|nested] 3+ messages in thread

* Re: [dpdk-dev] [PATCH] ring: assert on zero objects dequeue/enqueue
  2016-03-15 16:58 [dpdk-dev] [PATCH] ring: assert on zero objects dequeue/enqueue Lazaros Koromilas
@ 2016-03-16 11:21 ` Bruce Richardson
  2016-03-17 15:48   ` Lazaros Koromilas
  0 siblings, 1 reply; 3+ messages in thread
From: Bruce Richardson @ 2016-03-16 11:21 UTC (permalink / raw)
  To: Lazaros Koromilas; +Cc: dev

On Tue, Mar 15, 2016 at 06:58:45PM +0200, Lazaros Koromilas wrote:
> Issuing a zero objects dequeue with a single consumer has no effect.
> Doing so with multiple consumers, can get more than one thread to succeed
> the compare-and-set operation and observe starvation or even deadlock in
> the while loop that checks for preceding dequeues.  The problematic piece
> of code when n = 0:
> 
>     cons_next = cons_head + n;
>     success = rte_atomic32_cmpset(&r->cons.head, cons_head, cons_next);
> 
> The same is possible on the enqueue path.
> 
> Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>

I'm not sure how serious a problem this really is, and I really suspect that
just calling rte_panic is rather an overreaction here. At worst, this should
be a check only when RTE_RING_DEBUG is on.

However, probably my preferred solution to this issue would be to just add
   if (n == 0)
	   return 0

to the mp and mc enqueue/dequeue functions. That way there is no performance
penalty for the higher-performing sp/sc paths, and you avoid and unnecessary
cmpset operations for the mp/mc cases.

/Bruce

> ---
>  lib/librte_ring/rte_ring.h | 26 ++++++++++++++++++++++++++
>  1 file changed, 26 insertions(+)
> 
> diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
> index 943c97c..2bf9ce3 100644
> --- a/lib/librte_ring/rte_ring.h
> +++ b/lib/librte_ring/rte_ring.h
> @@ -100,6 +100,7 @@ extern "C" {
>  #include <rte_lcore.h>
>  #include <rte_atomic.h>
>  #include <rte_branch_prediction.h>
> +#include <rte_debug.h>
>  
>  #define RTE_TAILQ_RING_NAME "RTE_RING"
>  
> @@ -211,6 +212,19 @@ struct rte_ring {
>  #endif
>  
>  /**
> + * @internal Assert macro.
> + * @param exp
> + *   The expression to evaluate.
> + */
> +#define RTE_RING_ASSERT(exp) do {                                 \
> +		if (!(exp)) {                                     \
> +			rte_panic("line%d\t"                      \
> +			          "assert \"" #exp "\" failed\n", \
> +			          __LINE__);                      \
> +		}                                                 \
> +	} while (0)
> +
> +/**
>   * Calculate the memory size needed for a ring
>   *
>   * This function returns the number of bytes needed for a ring, given
> @@ -406,6 +420,7 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
>   *   A pointer to a table of void * pointers (objects).
>   * @param n
>   *   The number of objects to add in the ring from the obj_table.
> + *   Must be greater than zero.
>   * @param behavior
>   *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
>   *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
> @@ -431,6 +446,8 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>  	uint32_t mask = r->prod.mask;
>  	int ret;
>  
> +	RTE_RING_ASSERT(n > 0);
> +
>  	/* move prod.head atomically */
>  	do {
>  		/* Reset n to the initial burst count */
> @@ -510,6 +527,7 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>   *   A pointer to a table of void * pointers (objects).
>   * @param n
>   *   The number of objects to add in the ring from the obj_table.
> + *   Must be greater than zero.
>   * @param behavior
>   *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
>   *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
> @@ -533,6 +551,8 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>  	uint32_t mask = r->prod.mask;
>  	int ret;
>  
> +	RTE_RING_ASSERT(n > 0);
> +
>  	prod_head = r->prod.head;
>  	cons_tail = r->cons.tail;
>  	/* The subtraction is done between two unsigned 32bits value
> @@ -594,6 +614,7 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>   *   A pointer to a table of void * pointers (objects) that will be filled.
>   * @param n
>   *   The number of objects to dequeue from the ring to the obj_table.
> + *   Must be greater than zero.
>   * @param behavior
>   *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
>   *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
> @@ -618,6 +639,8 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
>  	unsigned i, rep = 0;
>  	uint32_t mask = r->prod.mask;
>  
> +	RTE_RING_ASSERT(n > 0);
> +
>  	/* move cons.head atomically */
>  	do {
>  		/* Restore n as it may change every loop */
> @@ -689,6 +712,7 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
>   *   A pointer to a table of void * pointers (objects) that will be filled.
>   * @param n
>   *   The number of objects to dequeue from the ring to the obj_table.
> + *   Must be greater than zero.
>   * @param behavior
>   *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
>   *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
> @@ -710,6 +734,8 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
>  	unsigned i;
>  	uint32_t mask = r->prod.mask;
>  
> +	RTE_RING_ASSERT(n > 0);
> +
>  	cons_head = r->cons.head;
>  	prod_tail = r->prod.tail;
>  	/* The subtraction is done between two unsigned 32bits value
> -- 
> 1.9.1
> 

^ permalink raw reply	[flat|nested] 3+ messages in thread

* Re: [dpdk-dev] [PATCH] ring: assert on zero objects dequeue/enqueue
  2016-03-16 11:21 ` Bruce Richardson
@ 2016-03-17 15:48   ` Lazaros Koromilas
  0 siblings, 0 replies; 3+ messages in thread
From: Lazaros Koromilas @ 2016-03-17 15:48 UTC (permalink / raw)
  To: Bruce Richardson; +Cc: dev

Sure, I'm sending it again with your suggestions.

Lazaros.

On Wed, Mar 16, 2016 at 1:21 PM, Bruce Richardson
<bruce.richardson@intel.com> wrote:
> On Tue, Mar 15, 2016 at 06:58:45PM +0200, Lazaros Koromilas wrote:
>> Issuing a zero objects dequeue with a single consumer has no effect.
>> Doing so with multiple consumers, can get more than one thread to succeed
>> the compare-and-set operation and observe starvation or even deadlock in
>> the while loop that checks for preceding dequeues.  The problematic piece
>> of code when n = 0:
>>
>>     cons_next = cons_head + n;
>>     success = rte_atomic32_cmpset(&r->cons.head, cons_head, cons_next);
>>
>> The same is possible on the enqueue path.
>>
>> Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
>
> I'm not sure how serious a problem this really is, and I really suspect that
> just calling rte_panic is rather an overreaction here. At worst, this should
> be a check only when RTE_RING_DEBUG is on.
>
> However, probably my preferred solution to this issue would be to just add
>    if (n == 0)
>            return 0
>
> to the mp and mc enqueue/dequeue functions. That way there is no performance
> penalty for the higher-performing sp/sc paths, and you avoid and unnecessary
> cmpset operations for the mp/mc cases.
>
> /Bruce
>
>> ---
>>  lib/librte_ring/rte_ring.h | 26 ++++++++++++++++++++++++++
>>  1 file changed, 26 insertions(+)
>>
>> diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
>> index 943c97c..2bf9ce3 100644
>> --- a/lib/librte_ring/rte_ring.h
>> +++ b/lib/librte_ring/rte_ring.h
>> @@ -100,6 +100,7 @@ extern "C" {
>>  #include <rte_lcore.h>
>>  #include <rte_atomic.h>
>>  #include <rte_branch_prediction.h>
>> +#include <rte_debug.h>
>>
>>  #define RTE_TAILQ_RING_NAME "RTE_RING"
>>
>> @@ -211,6 +212,19 @@ struct rte_ring {
>>  #endif
>>
>>  /**
>> + * @internal Assert macro.
>> + * @param exp
>> + *   The expression to evaluate.
>> + */
>> +#define RTE_RING_ASSERT(exp) do {                                 \
>> +             if (!(exp)) {                                     \
>> +                     rte_panic("line%d\t"                      \
>> +                               "assert \"" #exp "\" failed\n", \
>> +                               __LINE__);                      \
>> +             }                                                 \
>> +     } while (0)
>> +
>> +/**
>>   * Calculate the memory size needed for a ring
>>   *
>>   * This function returns the number of bytes needed for a ring, given
>> @@ -406,6 +420,7 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
>>   *   A pointer to a table of void * pointers (objects).
>>   * @param n
>>   *   The number of objects to add in the ring from the obj_table.
>> + *   Must be greater than zero.
>>   * @param behavior
>>   *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
>>   *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
>> @@ -431,6 +446,8 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>>       uint32_t mask = r->prod.mask;
>>       int ret;
>>
>> +     RTE_RING_ASSERT(n > 0);
>> +
>>       /* move prod.head atomically */
>>       do {
>>               /* Reset n to the initial burst count */
>> @@ -510,6 +527,7 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>>   *   A pointer to a table of void * pointers (objects).
>>   * @param n
>>   *   The number of objects to add in the ring from the obj_table.
>> + *   Must be greater than zero.
>>   * @param behavior
>>   *   RTE_RING_QUEUE_FIXED:    Enqueue a fixed number of items from a ring
>>   *   RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
>> @@ -533,6 +551,8 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>>       uint32_t mask = r->prod.mask;
>>       int ret;
>>
>> +     RTE_RING_ASSERT(n > 0);
>> +
>>       prod_head = r->prod.head;
>>       cons_tail = r->cons.tail;
>>       /* The subtraction is done between two unsigned 32bits value
>> @@ -594,6 +614,7 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
>>   *   A pointer to a table of void * pointers (objects) that will be filled.
>>   * @param n
>>   *   The number of objects to dequeue from the ring to the obj_table.
>> + *   Must be greater than zero.
>>   * @param behavior
>>   *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
>>   *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
>> @@ -618,6 +639,8 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
>>       unsigned i, rep = 0;
>>       uint32_t mask = r->prod.mask;
>>
>> +     RTE_RING_ASSERT(n > 0);
>> +
>>       /* move cons.head atomically */
>>       do {
>>               /* Restore n as it may change every loop */
>> @@ -689,6 +712,7 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
>>   *   A pointer to a table of void * pointers (objects) that will be filled.
>>   * @param n
>>   *   The number of objects to dequeue from the ring to the obj_table.
>> + *   Must be greater than zero.
>>   * @param behavior
>>   *   RTE_RING_QUEUE_FIXED:    Dequeue a fixed number of items from a ring
>>   *   RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
>> @@ -710,6 +734,8 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
>>       unsigned i;
>>       uint32_t mask = r->prod.mask;
>>
>> +     RTE_RING_ASSERT(n > 0);
>> +
>>       cons_head = r->cons.head;
>>       prod_tail = r->prod.tail;
>>       /* The subtraction is done between two unsigned 32bits value
>> --
>> 1.9.1
>>

^ permalink raw reply	[flat|nested] 3+ messages in thread

end of thread, other threads:[~2016-03-17 15:48 UTC | newest]

Thread overview: 3+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2016-03-15 16:58 [dpdk-dev] [PATCH] ring: assert on zero objects dequeue/enqueue Lazaros Koromilas
2016-03-16 11:21 ` Bruce Richardson
2016-03-17 15:48   ` Lazaros Koromilas

This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for NNTP newsgroup(s).