Re: [dpdk-dev] [PATCH v7 3/6] ring: add a lock-free implementation

[Stephen Hemminger <stephen@networkplumber.org>]

On Mon, 18 Mar 2019 16:35:52 -0500
Gage Eads <gage.eads@intel.com> wrote:

>  
> +/* The actual enqueue of pointers on the lock-free ring, used by the
> + * single-producer lock-free enqueue function.
> + */
> +#define ENQUEUE_PTRS_LF(r, base, prod_head, obj_table, n) do { \
> +	unsigned int i; \
> +	const uint32_t size = (r)->size; \
> +	size_t idx = prod_head & (r)->mask; \
> +	size_t new_cnt = prod_head + size; \
> +	struct rte_ring_lf_entry *ring = (struct rte_ring_lf_entry *)base; \
> +	unsigned int mask = ~0x3; \
> +	if (likely(idx + n < size)) { \
> +		for (i = 0; i < (n & mask); i += 4, idx += 4) { \
> +			ring[idx].ptr = obj_table[i]; \
> +			ring[idx].cnt = (new_cnt + i) >> r->log2_size; \
> +			ring[idx + 1].ptr = obj_table[i + 1]; \
> +			ring[idx + 1].cnt = (new_cnt + i + 1) >> r->log2_size; \
> +			ring[idx + 2].ptr = obj_table[i + 2]; \
> +			ring[idx + 2].cnt = (new_cnt + i + 2) >> r->log2_size; \
> +			ring[idx + 3].ptr = obj_table[i + 3]; \
> +			ring[idx + 3].cnt = (new_cnt + i + 3) >> r->log2_size; \
> +		} \
> +		switch (n & 0x3) { \
> +		case 3: \
> +			ring[idx].cnt = (new_cnt + i) >> r->log2_size; \
> +			ring[idx++].ptr = obj_table[i++]; /* fallthrough */ \
> +		case 2: \
> +			ring[idx].cnt = (new_cnt + i) >> r->log2_size; \
> +			ring[idx++].ptr = obj_table[i++]; /* fallthrough */ \
> +		case 1: \
> +			ring[idx].cnt = (new_cnt + i) >> r->log2_size; \
> +			ring[idx++].ptr = obj_table[i++]; \
> +		} \
> +	} else { \
> +		for (i = 0; idx < size; i++, idx++) { \
> +			ring[idx].cnt = (new_cnt + i) >> r->log2_size; \
> +			ring[idx].ptr = obj_table[i]; \
> +		} \
> +		for (idx = 0; i < n; i++, idx++) {    \
> +			ring[idx].cnt = (new_cnt + i) >> r->log2_size; \
> +			ring[idx].ptr = obj_table[i]; \
> +		} \
> +	} \
> +} while (0)
> +
>  /* the actual copy of pointers on the ring to obj_table.
>   * Placed here since identical code needed in both
>   * single and multi consumer dequeue functions */
> @@ -314,6 +384,43 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
>  	} \
>  } while (0)
>  
> +/* The actual copy of pointers on the lock-free ring to obj_table. */
> +#define DEQUEUE_PTRS_LF(r, base, cons_head, obj_table, n) do { \
> +	unsigned int i; \
> +	size_t idx = cons_head & (r)->mask; \
> +	const uint32_t size = (r)->size; \
> +	struct rte_ring_lf_entry *ring = (struct rte_ring_lf_entry *)base; \
> +	unsigned int mask = ~0x3; \
> +	if (likely(idx + n < size)) { \
> +		for (i = 0; i < (n & mask); i += 4, idx += 4) {\
> +			obj_table[i] = ring[idx].ptr; \
> +			obj_table[i + 1] = ring[idx + 1].ptr; \
> +			obj_table[i + 2] = ring[idx + 2].ptr; \
> +			obj_table[i + 3] = ring[idx + 3].ptr; \
> +		} \
> +		switch (n & 0x3) { \
> +		case 3: \
> +			obj_table[i++] = ring[idx++].ptr; /* fallthrough */ \
> +		case 2: \
> +			obj_table[i++] = ring[idx++].ptr; /* fallthrough */ \
> +		case 1: \
> +			obj_table[i++] = ring[idx++].ptr; \
> +		} \
> +	} else { \
> +		for (i = 0; idx < size; i++, idx++) \
> +			obj_table[i] = ring[idx].ptr; \
> +		for (idx = 0; i < n; i++, idx++) \
> +			obj_table[i] = ring[idx].ptr; \
> +	} \
> +} while (0)
> +

Generic programming like this in C is hard to maintain. Please don't
use large macros. It is better to have two copies than a single large macro.
Better yet use a single inline function than can be expanded in different ways
based on the pointer size.