>
> 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 <[email protected]>
> ---
...
> +
> +/**
> + * 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
