> > > 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?
>
> No.
> In that case, it proceeds to fragmentation:
> It fetches a bulk element, returns the one asked object to the caller,
> and puts the remaining objects of the bulk into the solo section.
Yep, now I do see fragmentation block below.
My bad, I didn't read till the end of the function.
> >
> > > +
> > > + /* 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
