[PATCH] stack: introduce pile

[Konstantin Ananyev]

> 
> Added a new high-performance lock-free "pile", using the Stack API.
> The pile behaves roughly like a stack, but is not strictly LIFO.
> 
> The pile is optimized for pushing/popping bulks of objects, which
> it does significantly faster than the lock-free stack.
> 
> Pushing/popping a number of objects not divisible by the compile time
> configurable bulk size is handled gracefully, but not as fast as
> complete bulks.
> 
> Performance examples, stack_pile_perf_autotest vs. stack_lf_autotest:
> 
> On a single core, pushing/popping 1 or 8 objects is similar speed.
> On a single core, pushing/popping 32 objects is 2x faster.
> On a single core, pushing/popping 512 objects is 10x faster.
> 
> On four cores, pushing/popping 1, 8 or 32 objects is slightly faster.
> On four cores, pushing/popping 512 objects is 4x faster.
> 
> Signed-off-by: Morten Brørup <mb at smartsharesystems.com>
> ---

...
> +
> +/**
> + * Pop several objects from the pile (lock-free, MT-safe).
> + *
> + * @param s
> + *   A pointer to the pile structure.
> + * @param obj_table
> + *   A pointer to a table of void * pointers (objects).
> + * @param n
> + *   The number of objects to pull from the pile.
> + * @return
> + *   Actual number of objects popped (either 0 or *n*).
> + */
> +static __rte_always_inline unsigned int
> +__rte_stack_pile_pop(struct rte_stack *s,
> +		void **obj_table,
> +		unsigned int n)
> +{
> +	RTE_ASSERT(s != NULL);
> +	RTE_ASSERT(obj_table != NULL);
> +
> +	struct rte_stack_pile *pile = &s->stack_pile;
> +	struct rte_stack_pile_bulk_elem *bulk_first = NULL, *bulk_last = NULL;
> +	struct rte_stack_lf_elem *solo_first = NULL, *solo_last = NULL,
> *tmp_solo;
> +	alignas(RTE_CACHE_LINE_SIZE) void
> *obj_frag[RTE_STACK_PILE_BULK_SIZE];
> +	struct rte_stack_pile_bulk_elem *frag = NULL;
> +	unsigned int n_bulk = n / RTE_STACK_PILE_BULK_SIZE;
> +	unsigned int n_solo = n & (RTE_STACK_PILE_BULK_SIZE - 1);
> +	unsigned int i;
> +
> +	if (unlikely(n_bulk == 0)) {
> +		if (unlikely(n_solo == 0))
> +			return 0;
> +		goto solo;
> +	}
> +
> +bulk:
> +	/* Fetch n_bulk * RTE_STACK_PILE_BULK_SIZE objects as bulk elements.
> */
> +	bulk_first = __rte_stack_pile_bulk_pop_elems(&pile->bulk, n_bulk,
> obj_table, &bulk_last);
> +	if (unlikely(bulk_first == NULL)) {
> +		/*
> +		 * Not available.
> +		 * Retry with fewer bulk elements; objects to be fetched as solo
> elements instead.
> +		 */
> +		n_solo += RTE_STACK_PILE_BULK_SIZE;
> +		n_bulk--;
> +		if (n_bulk > 0)
> +			goto bulk;
> +		else
> +			goto solo;
> +	}
> +
> +	if (likely(n_solo == 0))
> +		goto done;
> +
> +solo:
> +	/* Fetch n_solo objects as solo elements. */
> +	solo_first = __rte_stack_lf_pop_elems(&pile->solo, n_solo,
> +			&obj_table[n_bulk * RTE_STACK_PILE_BULK_SIZE],
> &solo_last);
> +	if (solo_first != NULL)
> +		goto done;

A question: does it mean that if user asked to pop  just one elem, it can fail
If solo list is empty, while there are plenty of elements in bulk section?

> +
> +	/* Solo elements not available. Try fragmentation. */
> +	if (unlikely(n_solo >= RTE_STACK_PILE_BULK_SIZE))
> +		goto fail; /* Ran out of bulk elements above. Don't try to fetch
> one more. */
> +
> +	/* Fetch a fragmentation element as a bulk element. */
> +	frag = __rte_stack_pile_bulk_pop_elems(&pile->bulk, 1, obj_frag, NULL);
> +	if (unlikely(frag == NULL))
> +		goto fail;
> +
> +	/* Get n_solo objects from the fragmentation element. */
> +	__rte_assume(n_solo > 0);
> +	__rte_assume(n_solo < RTE_STACK_PILE_BULK_SIZE);
> +	for (i = 0; i < n_solo; i++)
> +		obj_table[n_bulk * RTE_STACK_PILE_BULK_SIZE + i] = obj_frag[i];
> +
> +	/* Fetch free elements for the excess objects. */
> +	__rte_assume(RTE_STACK_PILE_BULK_SIZE - n_solo > 0);
> +	__rte_assume(RTE_STACK_PILE_BULK_SIZE - n_solo <
> RTE_STACK_PILE_BULK_SIZE);
> +	solo_first = __rte_stack_lf_pop_elems(&pile->free_solo,
> +			RTE_STACK_PILE_BULK_SIZE - n_solo, NULL, &solo_last);
> +	if (unlikely(solo_first == NULL))
> +		goto fail;
> +
> +	/* Construct the solo elements from the excess objects. */
> +	tmp_solo = solo_first;
> +	__rte_assume(n_solo > 0);
> +	__rte_assume(n_solo < RTE_STACK_PILE_BULK_SIZE);
> +	for (i = n_solo; i < RTE_STACK_PILE_BULK_SIZE; i++, tmp_solo = tmp_solo-
> >next)
> +		tmp_solo->data = obj_frag[i];
> +
> +	/* Push the excess objects as solo elements. */
> +	__rte_stack_lf_push_elems(&pile->solo, solo_first, solo_last,
> +			RTE_STACK_PILE_BULK_SIZE - n_solo);
> +	n_solo = 0;
> +
> +	/* Add the fragmentation element to the bulk elements, so it can be
> freed with them. */
> +	if (n_bulk > 0)
> +		bulk_last->next = frag;
> +	else
> +		bulk_first = frag;
> +	bulk_last = frag;
> +	n_bulk++;
> +
> +done:
> +	/* Success. Free the elements. */
> +	if (n_bulk > 0)
> +		__rte_stack_pile_bulk_push_elems(&pile->free_bulk, bulk_first,
> bulk_last, n_bulk);
> +	if (n_solo > 0)
> +		__rte_stack_lf_push_elems(&pile->free_solo, solo_first,
> solo_last, n_solo);
> +
> +	return n;
> +
> +fail:
> +	/* Failed. Roll back. */
> +	if (frag != NULL) {
> +		/*
> +		 * No further action than this is required to roll the
> fragmentation
> +		 * element back into the pile of bulk elements, as the objects in
> +		 * the fragmentation element are intact.
> +		 */
> +		if (n_bulk > 0)
> +			bulk_last->next = frag;
> +		else
> +			bulk_first = frag;
> +		bulk_last = frag;
> +		n_bulk += 1;
> +	}
> +	if (n_bulk > 0)
> +		__rte_stack_pile_bulk_push_elems(&pile->bulk, bulk_first,
> bulk_last, n_bulk);
> +
> +	return 0;
> +}
> +
> +/**
> + * @internal Initialize a pile stack.
> + *
> + * @param s
> + *   A pointer to the stack structure.
> + * @param count
> + *   The size of the stack.
> + */
> +void
> +rte_stack_pile_init(struct rte_stack *s, unsigned int count);
> +
> +/**
> + * @internal Return the memory required for a pile stack.
> + *
> + * @param count
> + *   The size of the stack.
> + * @return
> + *   The bytes to allocate for a pile stack.
> + */
> +ssize_t
> +rte_stack_pile_get_memsize(unsigned int count);
> +
> +#endif /* _RTE_STACK_PILE_H_ */
> --
> 2.43.0