All of lore.kernel.org
 help / color / mirror / Atom feed
From: "Morten Brørup" <mb@smartsharesystems.com>
To: dev@dpdk.org
Cc: "Morten Brørup" <mb@smartsharesystems.com>
Subject: [RFC PATCH v7] pile stack and mempool driver
Date: Mon,  3 Aug 2026 08:14:46 +0000	[thread overview]
Message-ID: <20260803081446.1292377-1-mb@smartsharesystems.com> (raw)
In-Reply-To: <98CBD80474FA8B44BF855DF32C47DC35F659AB@smartserver.smartshare.dk>

Early submission of:
- A new "pile" stack-like implementation using the Stack API, and
- an accompanying "pile" mempool driver.
And:
- Some mempool optimizations.
- Deprecating "__rte_restrict" in favor of keyword "__restrict",
  supported by all relevant C/C++ compilers.
- An x86 rte_memcpy() optimization for some compile time known sizes
  (64-byte blocks up to 512 or 256 bytes).
- Decorating the object tables in the Stack API with "restrict".

For CI test and community feedback.

Needless to say, this must be separated into a series of patches,
or multiple independent series of patches.
For now, I'm submitting a snapshot of work in progress.

The "pile" somewhat resembles the lock-free stack, but operates on
bulks (arrays) of objects, to significantly reduce linked list
traversal.
With the pile's default bulk size of 32 objects, a mempool cache
flush/refill traverses a linked list of only 16 elements, whereas
the lock-free stack would traverse a linked list of 512 elements.

Some performance numbers from mempool_perf_autotest_2cores, all
with cache=1024 cores=2 n_keep=32768:

start performance test (using ring_mp_mc, with cache)
n_get_bulk= 64 n_put_bulk= 64 constant_n=0 rate_persec= 753985338
n_get_bulk=256 n_put_bulk=256 constant_n=0 rate_persec= 755805913

start performance test for lf_stack (with cache)
n_get_bulk= 64 n_put_bulk= 64 constant_n=0 rate_persec=  29132352
n_get_bulk=256 n_put_bulk=256 constant_n=0 rate_persec=  29276708

start performance test for pile (with cache)
n_get_bulk= 64 n_put_bulk= 64 constant_n=0 rate_persec= 560159479
n_get_bulk=256 n_put_bulk=256 constant_n=0 rate_persec= 557910933

Hat tip to Bruce for bringing attention to the ring not being the
optimal mempool driver!

Signed-off-by: Morten Brørup <mb@smartsharesystems.com>
---
v2:
* Fix indentation, long lines, etc. (checkpatch)
* Fix label followed by a declaration is a C23 extension. (CI)
* Added __rte_internal to pile init and get_memsize. (AI)
* Fix roll back bulk elements in wrong order with fragmentation. (AI)
* Minor changes suggested by AI.
v3:
* Revert rename unused field in rte_stack_pile_bulk_elem structure.
v4:
* Revert add __rte_internal.
  Compilation fails, and existing stack implementations don't have it.
v5:
* Remove compiler diagnostic pragmas in stack overflow test case.
  (Stephen)
* Temporarily remove stack overflow test case until a sufficiently
  obfuscated variant doesn't trigger a compiler warning about array
  overrun.
* Revert most changes in rte_stack_std.h, and only change what is
  necessary.
* Add FIXME in the pile implementation, noting that support for the
  generic memory model should be removed here too, if removed in the
  lock-free stack.
  (Inspired by Stephen)
v6:
* Add note about roll back of objects in fragmentation element. (AI)
* Move declaration of temporary variable up, to please compilers. (CI)
* Revert mempool cache size adjustments in some drivers;
  let the mempool creation function adjust at runtime instead.
v7:
* Use the "__restrict" keyword, supported by all relevant C/C++ compilers.
  Degrade "__rte_restrict" to a backwards compatibility macro,
  document it as deprecated, and check for it in checkpatches.sh.
* Eliminate risk of namespace pollution by prefixing ALIGNMENT_MASK
  macro in x86 rte_memcpy.h header file.
* Replace conditional code by defining RTE_MEMCPY_BLOCK_64_MAX. (AI)
* Fix off-by-one in assumption when fetching free elements for the
  excess objects in the fragmentation element. (AI, advanced model)
* Fix memcmp() size in test_stack_push_pop(). (AI, advanced model)
* Do not try fragmentation when we know (from fetching bulk elements
  above) that no bulk element is available.
  (Inspired by AI, advanced model)
* Fix function descriptions mentioning wrong parameter name.
  (AI, advanced model)
* Fix typo in documentation. (AI, advanced model)
* Mention the pile in the mempool stack documentation.
  (AI, advanced model)
* Set the mbuf default pool ops to "pile", for CI test purposes only.
* Select the C11 memory model for x86, for CI test purposes only.
---
 app/test/test_mempool.c                   |   3 +-
 app/test/test_stack.c                     |  64 ++++-
 app/test/test_stack_perf.c                |  15 +-
 config/rte_config.h                       |   7 +-
 config/x86/meson.build                    |   1 +
 devtools/checkpatches.sh                  |   8 +
 doc/guides/mempool/stack.rst              |  12 +-
 doc/guides/prog_guide/stack_lib.rst       |  67 ++++-
 drivers/mempool/stack/rte_mempool_stack.c |  40 +++
 drivers/net/sxe2/sxe2_txrx_vec_avx512.c   |   2 +-
 drivers/net/tap/rte_eth_tap.c             |   2 +-
 lib/eal/include/rte_bitset.h              |   2 +-
 lib/eal/include/rte_common.h              |  21 +-
 lib/eal/x86/include/rte_memcpy.h          |  69 +++--
 lib/mempool/mempool_trace.h               |   1 -
 lib/mempool/rte_mempool.c                 |  75 +++--
 lib/mempool/rte_mempool.h                 |  76 +++--
 lib/stack/meson.build                     |   3 +-
 lib/stack/rte_stack.c                     |  18 +-
 lib/stack/rte_stack.h                     |  83 +++++-
 lib/stack/rte_stack_lf.h                  |   6 +-
 lib/stack/rte_stack_lf_c11.h              |   2 +-
 lib/stack/rte_stack_lf_generic.h          |   2 +-
 lib/stack/rte_stack_lf_stubs.h            |   2 +-
 lib/stack/rte_stack_pile.c                |  35 +++
 lib/stack/rte_stack_pile.h                | 330 ++++++++++++++++++++++
 lib/stack/rte_stack_std.h                 |  24 +-
 27 files changed, 836 insertions(+), 134 deletions(-)
 create mode 100644 lib/stack/rte_stack_pile.c
 create mode 100644 lib/stack/rte_stack_pile.h

diff --git a/app/test/test_mempool.c b/app/test/test_mempool.c
index e54249ce61..76d45cea2a 100644
--- a/app/test/test_mempool.c
+++ b/app/test/test_mempool.c
@@ -112,8 +112,7 @@ test_mempool_basic(struct rte_mempool *mp, int use_external_cache)
 		GOTO_ERR(ret, out);
 
 	printf("get private data\n");
-	if (rte_mempool_get_priv(mp) != (char *)mp +
-			RTE_MEMPOOL_HEADER_SIZE(mp, mp->cache_size))
+	if (rte_mempool_get_priv(mp) != (char *)mp + sizeof(struct rte_mempool))
 		GOTO_ERR(ret, out);
 
 #ifndef RTE_EXEC_ENV_FREEBSD /* rte_mem_virt2iova() not supported on bsd */
diff --git a/app/test/test_stack.c b/app/test/test_stack.c
index 5517982774..a40c73d63d 100644
--- a/app/test/test_stack.c
+++ b/app/test/test_stack.c
@@ -81,13 +81,30 @@ test_stack_push_pop(struct rte_stack *s, void **obj_table, unsigned int bulk_sz)
 		}
 	}
 
-	for (i = 0; i < STACK_SIZE; i++) {
-		if (obj_table[i] != popped_objs[STACK_SIZE - i - 1]) {
-			printf("[%s():%u] Incorrect value %p at index 0x%x\n",
-			       __func__, __LINE__,
-			       popped_objs[STACK_SIZE - i - 1], i);
-			rte_free(popped_objs);
-			return -1;
+	if (!(s->flags & RTE_STACK_F_PILE)) {
+		for (i = 0; i < STACK_SIZE; i++) {
+			if (obj_table[i] != popped_objs[STACK_SIZE - i - 1]) {
+				printf("[%s():%u] Incorrect value %p at index 0x%x\n",
+				       __func__, __LINE__,
+				       popped_objs[STACK_SIZE - i - 1], i);
+				rte_free(popped_objs);
+				return -1;
+			}
+		}
+	}
+
+	if ((s->flags & RTE_STACK_F_PILE) && (bulk_sz & (RTE_STACK_PILE_BULK_SIZE - 1)) == 0) {
+		for (i = 0; i < STACK_SIZE; i += RTE_STACK_PILE_BULK_SIZE) {
+			if (memcmp(&obj_table[i],
+					&popped_objs[STACK_SIZE - RTE_STACK_PILE_BULK_SIZE - i],
+					sizeof(void *) * RTE_STACK_PILE_BULK_SIZE) != 0) {
+				printf("[%s():%u] Incorrect values %p at index 0x%x with bulk size %u\n",
+				       __func__, __LINE__,
+				       popped_objs[STACK_SIZE - RTE_STACK_PILE_BULK_SIZE - i],
+				       i, bulk_sz);
+				rte_free(popped_objs);
+				return -1;
+			}
 		}
 	}
 
@@ -152,12 +169,24 @@ test_stack_basic(uint32_t flags)
 		goto fail_test;
 	}
 
-	ret = rte_stack_push(s, obj_table, 2 * STACK_SIZE);
-	if (ret != 0) {
-		printf("[%s():%u] Excess objects push succeeded\n",
-		       __func__, __LINE__);
-		goto fail_test;
+#if 0 /* FIXME: Omitted. Doesn't compile [-Warray-bounds=]. Write an obfuscated method. */
+	if (!(s->flags & RTE_STACK_F_PILE)) {
+		ret = rte_stack_push(s, obj_table, 2 * STACK_SIZE);
+		if (ret != 0) {
+			printf("[%s():%u] Excess objects push succeeded\n",
+			       __func__, __LINE__);
+			goto fail_test;
+		}
 	}
+	if (s->flags & RTE_STACK_F_PILE) {
+		ret = rte_stack_push(s, obj_table, STACK_SIZE * RTE_STACK_PILE_BULK_SIZE + 1);
+		if (ret != 0) {
+			printf("[%s():%u] Excess objects push succeeded\n",
+			       __func__, __LINE__);
+			goto fail_test;
+		}
+	}
+#endif
 
 	ret = rte_stack_pop(s, obj_table, 1);
 	if (ret != 0) {
@@ -384,5 +413,16 @@ test_lf_stack(void)
 #endif
 }
 
+static int
+test_pile(void)
+{
+#if defined(RTE_STACK_PILE_SUPPORTED)
+	return __test_stack(RTE_STACK_F_PILE);
+#else
+	return TEST_SKIPPED;
+#endif
+}
+
 REGISTER_FAST_TEST(stack_autotest, NOHUGE_SKIP, ASAN_OK, test_stack);
 REGISTER_FAST_TEST(stack_lf_autotest, NOHUGE_SKIP, ASAN_OK, test_lf_stack);
+REGISTER_FAST_TEST(stack_pile_autotest, NOHUGE_SKIP, ASAN_OK, test_pile);
diff --git a/app/test/test_stack_perf.c b/app/test/test_stack_perf.c
index 3f17a2606c..a15f3719c2 100644
--- a/app/test/test_stack_perf.c
+++ b/app/test/test_stack_perf.c
@@ -14,14 +14,14 @@
 #include "test.h"
 
 #define STACK_NAME "STACK_PERF"
-#define MAX_BURST 32
+#define MAX_BURST (RTE_MEMPOOL_CACHE_MAX_SIZE / 2)
 #define STACK_SIZE (RTE_MAX_LCORE * MAX_BURST)
 
 /*
  * Push/pop bulk sizes, marked volatile so they aren't treated as compile-time
  * constants.
  */
-static volatile unsigned int bulk_sizes[] = {8, MAX_BURST};
+static volatile unsigned int bulk_sizes[] = {1, 8, 32, MAX_BURST};
 
 static RTE_ATOMIC(uint32_t) lcore_barrier;
 
@@ -354,5 +354,16 @@ test_lf_stack_perf(void)
 #endif
 }
 
+static int
+test_pile_perf(void)
+{
+#if defined(RTE_STACK_PILE_SUPPORTED)
+	return __test_stack_perf(RTE_STACK_F_PILE);
+#else
+	return TEST_SKIPPED;
+#endif
+}
+
 REGISTER_PERF_TEST(stack_perf_autotest, test_stack_perf);
 REGISTER_PERF_TEST(stack_lf_perf_autotest, test_lf_stack_perf);
+REGISTER_PERF_TEST(stack_pile_perf_autotest, test_pile_perf);
diff --git a/config/rte_config.h b/config/rte_config.h
index 0447cdf2ad..6085d2e1a0 100644
--- a/config/rte_config.h
+++ b/config/rte_config.h
@@ -56,14 +56,17 @@
 #define RTE_CONTIGMEM_DEFAULT_BUF_SIZE (512*1024*1024)
 
 /* mempool defines */
-#define RTE_MEMPOOL_CACHE_MAX_SIZE 512
+#define RTE_MEMPOOL_CACHE_MAX_SIZE 1024
 /* RTE_LIBRTE_MEMPOOL_STATS is not set */
 /* RTE_LIBRTE_MEMPOOL_DEBUG is not set */
 
 /* mbuf defines */
-#define RTE_MBUF_DEFAULT_MEMPOOL_OPS "ring_mp_mc"
+#define RTE_MBUF_DEFAULT_MEMPOOL_OPS "pile" /* FIXME: Test only. Default: "ring_mp_mc" */
 /* RTE_MBUF_HISTORY_DEBUG is not set */
 
+/* stack defines */
+#define RTE_STACK_PILE_BULK_SIZE 32
+
 /* ether defines */
 #define RTE_MAX_QUEUES_PER_PORT 1024
 #define RTE_ETHDEV_RXTX_CALLBACKS 1
diff --git a/config/x86/meson.build b/config/x86/meson.build
index 124b204847..28be579556 100644
--- a/config/x86/meson.build
+++ b/config/x86/meson.build
@@ -49,6 +49,7 @@ else
 endif
 
 dpdk_conf.set('RTE_MAX_NUMA_NODES', 32)
+dpdk_conf.set('RTE_USE_C11_MEM_MODEL', true) # FIXME: Test only.
 
 if is_ms_compiler
     subdir_done()
diff --git a/devtools/checkpatches.sh b/devtools/checkpatches.sh
index 18bd825372..63c1ece0b6 100755
--- a/devtools/checkpatches.sh
+++ b/devtools/checkpatches.sh
@@ -154,6 +154,14 @@ check_forbidden_additions() { # <patch>
 		-f $(dirname $(readlink -f $0))/check-forbidden-tokens.awk \
 		"$1" || res=1
 
+	# refrain from using __rte_restrict
+	awk -v FOLDERS="lib drivers app examples" \
+		-v EXPRESSIONS="__rte_restrict" \
+		-v RET_ON_FAIL=1 \
+		-v MESSAGE='Using __rte_restrict, prefer restrict' \
+		-f $(dirname $(readlink -f $0))/check-forbidden-tokens.awk \
+		"$1" || res=1
+
 	# refrain from using compiler __atomic_xxx builtins
 	awk -v FOLDERS="lib drivers app examples" \
 		-v SKIP_FILES='drivers/common/cnxk/' \
diff --git a/doc/guides/mempool/stack.rst b/doc/guides/mempool/stack.rst
index 80ea07e65d..708f48f952 100644
--- a/doc/guides/mempool/stack.rst
+++ b/doc/guides/mempool/stack.rst
@@ -1,5 +1,6 @@
 ..  SPDX-License-Identifier: BSD-3-Clause
     Copyright(c) 2020 Intel Corporation.
+    Copyright(c) 2026 SmartShare Systems.
 
 Stack Mempool Driver
 ====================
@@ -28,6 +29,12 @@ can be selected as described in :ref:`Mempool_Handlers`:
   The underlying **rte_stack** operates in lock-free mode. For more
   information please refer to :ref:`Stack_Library_LF_Stack`.
 
+- ``pile``
+
+  The underlying **rte_stack** operates in lock-free mode,
+  and is optimized for bulks of objects.
+  For more information please refer to :ref:`_Stack_Library_Pile`.
+
 The standard stack outperforms the lock-free stack on average, however the
 standard stack is non-preemptive: if a mempool user is preempted while holding
 the stack lock, that thread will block all other mempool accesses until it
@@ -35,9 +42,12 @@ returns and releases the lock. As a result, an application using the standard
 stack whose threads can be preempted can suffer from brief, infrequent
 performance hiccups.
 
-The lock-free stack, by design, is not susceptible to this problem; one thread can
+The lock-free stack and the pile, by design, are not susceptible to this problem; one thread can
 be preempted at any point during a push or pop operation and will not impede
 the progress of any other thread.
 
+The pile is not LIFO per object, but per bulk of objects.
+Although the pile is optimized for bulks of objects, it can handle any request size.
+
 For a more detailed description of the stack implementations, please refer to
 :doc:`/prog_guide/stack_lib`.
diff --git a/doc/guides/prog_guide/stack_lib.rst b/doc/guides/prog_guide/stack_lib.rst
index fdf056730c..d5a498e778 100644
--- a/doc/guides/prog_guide/stack_lib.rst
+++ b/doc/guides/prog_guide/stack_lib.rst
@@ -1,5 +1,6 @@
 ..  SPDX-License-Identifier: BSD-3-Clause
     Copyright(c) 2019 Intel Corporation.
+    Copyright(c) 2026 SmartShare Systems.
 
 Stack Library
 =============
@@ -9,9 +10,10 @@ stack of pointers.
 
 The stack library provides the following basic operations:
 
-*  Create a uniquely named stack of a user-specified size and using a
+*  Create a uniquely named stack (or pile) of a user-specified size and using a
    user-specified socket, with either standard (lock-based) or lock-free
    behavior.
+   The pile resembles a lock-free stack, but is not strictly LIFO.
 
 *  Push and pop a burst of one or more stack objects (pointers).
    These functions are multi-thread safe.
@@ -25,8 +27,9 @@ The stack library provides the following basic operations:
 Implementation
 --------------
 
-The library supports two types of stacks: standard (lock-based) and lock-free.
-Both types use the same set of interfaces, but their implementations differ.
+The library supports three types of stacks: standard (lock-based), lock-free,
+and pile (lock-free, not strictly LIFO, optimized for bulk operations).
+All types use the same set of interfaces, but their implementations differ.
 
 .. _Stack_Library_Std_Stack:
 
@@ -64,7 +67,7 @@ The linked list elements themselves are maintained in a lock-free LIFO, and are
 allocated before stack pushes and freed after stack pops. Since the stack has a
 fixed maximum depth, these elements do not need to be dynamically created.
 
-The lock-free behavior is selected by passing the *RTE_STACK_F_LF* flag to
+The lock-free behavior is selected by passing the ``RTE_STACK_F_LF`` flag to
 ``rte_stack_create()``.
 
 Preventing the ABA problem
@@ -86,3 +89,59 @@ both pop stale data and incorrectly change the head pointer. By adding a
 modification counter that is updated on every push and pop as part of the
 compare-and-swap, the algorithm can detect when the list changes even if the
 head pointer remains the same.
+
+.. _Stack_Library_Pile:
+
+Pile
+~~~~
+
+The pile is a stack-like implementation, optimized for bulk operations.
+It is only LIFO on bulk level, not on object level; i.e. arrays of bulks are
+pushed and popped in LIFO manner, but objects within each bulk are not ordered
+as expected by a stack.
+
+The pile implementation generally resembles that of the lock-free stack.
+In addition to the lock-free stack's linked list of solo (single-object) elements,
+it also contains a linked list of bulk (multi-object) elements.
+And similar to the linked list of free elements, it contains two linked lists of
+free elements, one for each element type (bulk and solo).
+The lock-free property means that multiple threads can push and pop simultaneously.
+One thread being preempted/delayed in a push or pop operation will not
+impede the forward progress of any other thread.
+
+Push operations are performed by splitting the burst in two: objects fitting into
+bulk elements, and any remaining objects (after filling bulk elements) into
+solo elements, and then performing two lock-free push operations,
+one for each element type (solo and bulk).
+
+Pop operations are performed by splitting the burst in two: objects fitting into
+bulk elements, and any remaining objects (not filling a bulk element) into
+solo elements. Two lock-free pop operations are performed,
+first for bulk elements, and then for solo elements.
+If the pop operation for bulk elements fails, it keeps retrying, requesting one
+less bulk element. The number of solo elements in the following request is
+correspondingly increased.
+
+The pile's lock-free list push and pop operations use the lock-free stack's
+implementations (and uses type casting to mimic C++ class inheritance).
+
+The linked list elements themselves are maintained in two lock-free LIFOs,
+one for bulk elements and one for solo elements, and are
+allocated before pushes and freed after pops. Since the pile has a
+fixed maximum depth, these elements do not need to be dynamically created.
+
+The pile behavior is selected by passing the ``RTE_STACK_F_PILE`` flag to
+``rte_stack_create()``.
+
+The pile bulk size can be changed by modifying ``RTE_STACK_PILE_BULK_SIZE`` in
+``config/rte_config.h``.
+For optimal performance when using the pile mempool driver, the
+mempool cache size / 2 should be divisible by the pile bulk size.
+
+.. note::
+    The pile is designed and optimized for use with bulks of objects.
+    Bursts not a multiple of the bulk size are still handled in a lock-free,
+    forward-progress-guaranteed manner. However, pop operations may exhibit
+    significantly lower performance in instances where the optimal number of
+    bulk elements is unavailable, and it is necessary to retry (fetching
+    increasingly fewer bulk elements and correspondingly more solo elements).
diff --git a/drivers/mempool/stack/rte_mempool_stack.c b/drivers/mempool/stack/rte_mempool_stack.c
index 1476905227..7467b8b39e 100644
--- a/drivers/mempool/stack/rte_mempool_stack.c
+++ b/drivers/mempool/stack/rte_mempool_stack.c
@@ -41,6 +41,36 @@ lf_stack_alloc(struct rte_mempool *mp)
 	return __stack_alloc(mp, RTE_STACK_F_LF);
 }
 
+static int
+pile_alloc(struct rte_mempool *mp)
+{
+	return __stack_alloc(mp, RTE_STACK_F_PILE);
+}
+
+static int
+pile_enqueue(struct rte_mempool *mp, void * const *obj_table,
+	      unsigned int n)
+{
+	struct rte_stack *s = mp->pool_data;
+
+	RTE_ASSERT(s != NULL);
+	RTE_ASSERT(obj_table != NULL);
+
+	return __rte_stack_pile_push(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+}
+
+static int
+pile_dequeue(struct rte_mempool *mp, void **obj_table,
+	      unsigned int n)
+{
+	struct rte_stack *s = mp->pool_data;
+
+	RTE_ASSERT(s != NULL);
+	RTE_ASSERT(obj_table != NULL);
+
+	return __rte_stack_pile_pop(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+}
+
 static int
 stack_enqueue(struct rte_mempool *mp, void * const *obj_table,
 	      unsigned int n)
@@ -93,5 +123,15 @@ static struct rte_mempool_ops ops_lf_stack = {
 	.get_count = stack_get_count
 };
 
+static struct rte_mempool_ops ops_pile = {
+	.name = "pile",
+	.alloc = pile_alloc,
+	.free = stack_free,
+	.enqueue = pile_enqueue,
+	.dequeue = pile_dequeue,
+	.get_count = stack_get_count
+};
+
 RTE_MEMPOOL_REGISTER_OPS(ops_stack);
 RTE_MEMPOOL_REGISTER_OPS(ops_lf_stack);
+RTE_MEMPOOL_REGISTER_OPS(ops_pile);
diff --git a/drivers/net/sxe2/sxe2_txrx_vec_avx512.c b/drivers/net/sxe2/sxe2_txrx_vec_avx512.c
index a830c7a33b..2e680f4027 100644
--- a/drivers/net/sxe2/sxe2_txrx_vec_avx512.c
+++ b/drivers/net/sxe2/sxe2_txrx_vec_avx512.c
@@ -67,7 +67,7 @@ static __rte_always_inline int32_t sxe2_tx_bufs_free_vec_avx512(struct sxe2_tx_q
 		}
 		cache->len += rs_thresh;
 
-		if (cache->len >= cache->flushthresh) {
+		if (cache->len >= cache->size) {
 			(void)rte_mempool_ops_enqueue_bulk(mp,
 					&cache->objs[cache->size], cache->len - cache->size);
 			cache->len = cache->size;
diff --git a/drivers/net/tap/rte_eth_tap.c b/drivers/net/tap/rte_eth_tap.c
index b93452f168..b3142561c2 100644
--- a/drivers/net/tap/rte_eth_tap.c
+++ b/drivers/net/tap/rte_eth_tap.c
@@ -61,7 +61,7 @@
 #define TAP_MAX_MAC_ADDRS	16
 #define TAP_GSO_MBUFS_PER_CORE	128
 #define TAP_GSO_MBUF_SEG_SIZE	128
-#define TAP_GSO_MBUF_CACHE_SIZE	4
+#define TAP_GSO_MBUF_CACHE_SIZE	32
 #define TAP_GSO_MBUFS_NUM \
 	(TAP_GSO_MBUFS_PER_CORE * TAP_GSO_MBUF_CACHE_SIZE)
 
diff --git a/lib/eal/include/rte_bitset.h b/lib/eal/include/rte_bitset.h
index 4e6d44874a..8fc4a76da7 100644
--- a/lib/eal/include/rte_bitset.h
+++ b/lib/eal/include/rte_bitset.h
@@ -978,7 +978,7 @@ rte_bitset_find_clear_wrap(const uint64_t *bitset, size_t size, size_t start_bit
  */
 __rte_experimental
 static inline void
-rte_bitset_copy(uint64_t *__rte_restrict dst_bitset, const uint64_t *__rte_restrict src_bitset,
+rte_bitset_copy(uint64_t * __restrict dst_bitset, const uint64_t * __restrict src_bitset,
 		size_t size)
 {
 	rte_memcpy(dst_bitset, src_bitset, RTE_BITSET_SIZE(size));
diff --git a/lib/eal/include/rte_common.h b/lib/eal/include/rte_common.h
index 79d2a0ab93..212f43c494 100644
--- a/lib/eal/include/rte_common.h
+++ b/lib/eal/include/rte_common.h
@@ -250,12 +250,11 @@ typedef uint16_t unaligned_uint16_t;
 
 /**
  * Mark pointer as restricted with regard to pointer aliasing.
+ * For backwards compatibility only.
+ * @deprecated
+ * Use the ``__restrict`` keyword (recognized by supported C and C++ compilers) instead.
  */
-#if !defined(__STDC_VERSION__) || __STDC_VERSION__ < 199901L
 #define __rte_restrict __restrict
-#else
-#define __rte_restrict restrict
-#endif
 
 /**
  * definition to mark a variable or function parameter as used so
@@ -567,6 +566,15 @@ static void __attribute__((destructor(RTE_PRIO(prio)), used)) func(void)
 #define __rte_assume(condition) __assume(condition)
 #endif
 
+/**
+ * Alignment hint precondition
+ */
+#ifdef RTE_TOOLCHAIN_MSVC
+#define __rte_assume_aligned(ptr, alignment) (ptr)
+#else
+#define __rte_assume_aligned(ptr, alignment) __builtin_assume_aligned(ptr, alignment)
+#endif
+
 /**
  * Disable AddressSanitizer on some code
  */
@@ -729,7 +737,7 @@ static void __attribute__((destructor(RTE_PRIO(prio)), used)) func(void)
  *   True(1) where the pointer is correctly aligned, false(0) otherwise
  */
 static inline int
-rte_is_aligned(const void * const __rte_restrict ptr, const unsigned int align)
+rte_is_aligned(const void * const __restrict ptr, const unsigned int align)
 {
 	return ((uintptr_t)ptr & (align - 1)) == 0;
 }
@@ -775,6 +783,9 @@ rte_is_aligned(const void * const __rte_restrict ptr, const unsigned int align)
 /** Force minimum cache line alignment. */
 #define __rte_cache_min_aligned __rte_aligned(RTE_CACHE_LINE_MIN_SIZE)
 
+/** Cache alignment hint precondition */
+#define __rte_assume_cache_aligned(ptr) __rte_assume_aligned(ptr, RTE_CACHE_LINE_SIZE)
+
 #define _RTE_CACHE_GUARD_HELPER2(unique) \
 	alignas(RTE_CACHE_LINE_SIZE) \
 	char cache_guard_ ## unique[RTE_CACHE_LINE_SIZE * RTE_CACHE_GUARD_LINES]
diff --git a/lib/eal/x86/include/rte_memcpy.h b/lib/eal/x86/include/rte_memcpy.h
index 8ed8c55010..5e3e55737c 100644
--- a/lib/eal/x86/include/rte_memcpy.h
+++ b/lib/eal/x86/include/rte_memcpy.h
@@ -45,7 +45,7 @@ extern "C" {
  *   Pointer to the destination data.
  */
 static __rte_always_inline void *
-rte_memcpy(void *__rte_restrict dst, const void *__rte_restrict src, size_t n);
+rte_memcpy(void * __restrict dst, const void * __restrict src, size_t n);
 
 /**
  * Copy bytes from one location to another,
@@ -53,7 +53,7 @@ rte_memcpy(void *__rte_restrict dst, const void *__rte_restrict src, size_t n);
  * Use with n <= 15.
  */
 static __rte_always_inline void *
-rte_mov15_or_less(void *__rte_restrict dst, const void *__rte_restrict src, size_t n)
+rte_mov15_or_less(void * __restrict dst, const void * __restrict src, size_t n)
 {
 	/**
 	 * Use the following structs to avoid violating C standard
@@ -98,7 +98,7 @@ rte_mov15_or_less(void *__rte_restrict dst, const void *__rte_restrict src, size
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov16(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
+rte_mov16(uint8_t * __restrict dst, const uint8_t * __restrict src)
 {
 	__m128i xmm0;
 
@@ -111,7 +111,7 @@ rte_mov16(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov32(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
+rte_mov32(uint8_t * __restrict dst, const uint8_t * __restrict src)
 {
 #if defined RTE_MEMCPY_AVX
 	__m256i ymm0;
@@ -129,7 +129,7 @@ rte_mov32(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov48(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
+rte_mov48(uint8_t * __restrict dst, const uint8_t * __restrict src)
 {
 #if defined RTE_MEMCPY_AVX
 	rte_mov32((uint8_t *)dst, (const uint8_t *)src);
@@ -146,7 +146,7 @@ rte_mov48(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov64(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
+rte_mov64(uint8_t * __restrict dst, const uint8_t * __restrict src)
 {
 #if defined __AVX512F__ && defined RTE_MEMCPY_AVX512
 	__m512i zmm0;
@@ -164,7 +164,7 @@ rte_mov64(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov128(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
+rte_mov128(uint8_t * __restrict dst, const uint8_t * __restrict src)
 {
 	rte_mov64(dst + 0 * 64, src + 0 * 64);
 	rte_mov64(dst + 1 * 64, src + 1 * 64);
@@ -175,7 +175,7 @@ rte_mov128(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov256(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
+rte_mov256(uint8_t * __restrict dst, const uint8_t * __restrict src)
 {
 	rte_mov128(dst + 0 * 128, src + 0 * 128);
 	rte_mov128(dst + 1 * 128, src + 1 * 128);
@@ -187,14 +187,15 @@ rte_mov256(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src)
  * AVX512 implementation below
  */
 
-#define ALIGNMENT_MASK 0x3F
+#define RTE_MEMCPY_ALIGNMENT_MASK 0x3F
+#define RTE_MEMCPY_BLOCK_64_MAX 512
 
 /**
  * Copy 128-byte blocks from one location to another,
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov128blocks(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src, size_t n)
+rte_mov128blocks(uint8_t * __restrict dst, const uint8_t * __restrict src, size_t n)
 {
 	__m512i zmm0, zmm1;
 
@@ -214,7 +215,7 @@ rte_mov128blocks(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src,
  * locations must not overlap.
  */
 static inline void
-rte_mov512blocks(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src, size_t n)
+rte_mov512blocks(uint8_t * __restrict dst, const uint8_t * __restrict src, size_t n)
 {
 	__m512i zmm0, zmm1, zmm2, zmm3, zmm4, zmm5, zmm6, zmm7;
 
@@ -247,7 +248,7 @@ rte_mov512blocks(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src,
  * Use with n > 64.
  */
 static __rte_always_inline void *
-rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_restrict src,
+rte_memcpy_generic_more_than_64(void * __restrict dst, const void * __restrict src,
 		size_t n)
 {
 	void *ret = dst;
@@ -333,14 +334,15 @@ rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_rest
  * AVX implementation below
  */
 
-#define ALIGNMENT_MASK 0x1F
+#define RTE_MEMCPY_ALIGNMENT_MASK 0x1F
+#define RTE_MEMCPY_BLOCK_64_MAX 256
 
 /**
  * Copy 128-byte blocks from one location to another,
  * locations must not overlap.
  */
 static __rte_always_inline void
-rte_mov128blocks(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src, size_t n)
+rte_mov128blocks(uint8_t * __restrict dst, const uint8_t * __restrict src, size_t n)
 {
 	__m256i ymm0, ymm1, ymm2, ymm3;
 
@@ -373,7 +375,7 @@ rte_mov128blocks(uint8_t *__rte_restrict dst, const uint8_t *__rte_restrict src,
  * Use with n > 64.
  */
 static __rte_always_inline void *
-rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_restrict src,
+rte_memcpy_generic_more_than_64(void * __restrict dst, const void * __restrict src,
 		size_t n)
 {
 	void *ret = dst;
@@ -444,7 +446,8 @@ rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_rest
  * SSE implementation below
  */
 
-#define ALIGNMENT_MASK 0x0F
+#define RTE_MEMCPY_ALIGNMENT_MASK 0x0F
+#define RTE_MEMCPY_BLOCK_64_MAX 512
 
 /**
  * Macro for copying unaligned block from one location to another with constant load offset,
@@ -546,7 +549,7 @@ rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_rest
  * Use with n > 64.
  */
 static __rte_always_inline void *
-rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_restrict src,
+rte_memcpy_generic_more_than_64(void * __restrict dst, const void * __restrict src,
 		size_t n)
 {
 	__m128i xmm0, xmm1, xmm2, xmm3, xmm4, xmm5, xmm6, xmm7, xmm8;
@@ -654,7 +657,7 @@ rte_memcpy_generic_more_than_64(void *__rte_restrict dst, const void *__rte_rest
  * Use with n > 64.
  */
 static __rte_always_inline void *
-rte_memcpy_aligned_more_than_64(void *__rte_restrict dst, const void *__rte_restrict src,
+rte_memcpy_aligned_more_than_64(void * __restrict dst, const void * __restrict src,
 		size_t n)
 {
 	void *ret = dst;
@@ -674,7 +677,7 @@ rte_memcpy_aligned_more_than_64(void *__rte_restrict dst, const void *__rte_rest
 }
 
 static __rte_always_inline void *
-rte_memcpy(void *__rte_restrict dst, const void *__rte_restrict src, size_t n)
+rte_memcpy(void * __restrict dst, const void * __restrict src, size_t n)
 {
 	/* Fast way when copy size doesn't exceed 64 bytes. */
 	if (n < 16)
@@ -707,15 +710,39 @@ rte_memcpy(void *__rte_restrict dst, const void *__rte_restrict src, size_t n)
 #endif
 		return dst;
 	}
+	/* Common way for small copy size of 64-byte blocks */
+	if (__rte_constant(n) && (n & 63) == 0 && n <= RTE_MEMCPY_BLOCK_64_MAX) {
+		void *ret = dst;
+
+		if (n & 512) {
+			rte_mov256((uint8_t *)dst + 0 * 256, (const uint8_t *)src + 0 * 256);
+			rte_mov256((uint8_t *)dst + 1 * 256, (const uint8_t *)src + 1 * 256);
+		}
+		if (n & 256) {
+			rte_mov256((uint8_t *)dst, (const uint8_t *)src);
+			src = (const uint8_t *)src + 256;
+			dst = (uint8_t *)dst + 256;
+		}
+		if (n & 128) {
+			rte_mov128((uint8_t *)dst, (const uint8_t *)src);
+			src = (const uint8_t *)src + 128;
+			dst = (uint8_t *)dst + 128;
+		}
+		if (n & 64)
+			rte_mov64((uint8_t *)dst, (const uint8_t *)src);
+
+		return ret;
+	}
 
 	/* Implementation for size > 64 bytes depends on alignment with vector register size. */
-	if (!(((uintptr_t)dst | (uintptr_t)src) & ALIGNMENT_MASK))
+	if (!(((uintptr_t)dst | (uintptr_t)src) & RTE_MEMCPY_ALIGNMENT_MASK))
 		return rte_memcpy_aligned_more_than_64(dst, src, n);
 	else
 		return rte_memcpy_generic_more_than_64(dst, src, n);
 }
 
-#undef ALIGNMENT_MASK
+#undef RTE_MEMCPY_ALIGNMENT_MASK
+#undef RTE_MEMCPY_BLOCK_64_MAX
 
 #ifdef __cplusplus
 }
diff --git a/lib/mempool/mempool_trace.h b/lib/mempool/mempool_trace.h
index 23cda1473c..60e47cf67b 100644
--- a/lib/mempool/mempool_trace.h
+++ b/lib/mempool/mempool_trace.h
@@ -119,7 +119,6 @@ RTE_TRACE_POINT(
 	rte_trace_point_emit_i32(socket_id);
 	rte_trace_point_emit_ptr(cache);
 	rte_trace_point_emit_u32(cache->len);
-	rte_trace_point_emit_u32(cache->flushthresh);
 )
 
 RTE_TRACE_POINT(
diff --git a/lib/mempool/rte_mempool.c b/lib/mempool/rte_mempool.c
index 817e2b8dc1..04b959c61f 100644
--- a/lib/mempool/rte_mempool.c
+++ b/lib/mempool/rte_mempool.c
@@ -753,14 +753,13 @@ static void
 mempool_cache_init(struct rte_mempool_cache *cache, uint32_t size)
 {
 	cache->size = size;
-	cache->flushthresh = size; /* Obsolete; for API/ABI compatibility purposes only */
 	cache->len = 0;
 }
 
 /*
  * Create and initialize a cache for objects that are retrieved from and
  * returned to an underlying mempool. This structure is identical to the
- * local_cache[lcore_id] pointed to by the mempool structure.
+ * local_cache[lcore_id] entry in the mempool structure.
  */
 RTE_EXPORT_SYMBOL(rte_mempool_cache_create)
 struct rte_mempool_cache *
@@ -768,6 +767,23 @@ rte_mempool_cache_create(uint32_t size, int socket_id)
 {
 	struct rte_mempool_cache *cache;
 
+	/*
+	 * Alignment requirement for performance optimized move within the mempool cache.
+	 * @ref rte_mempool_do_generic_put() implementation.
+	 */
+	if (size & 31) {
+		uint32_t rounded = RTE_ALIGN_MUL_FLOOR(size, 32);
+		if (rounded == 0) {
+			RTE_MEMPOOL_LOG(ERR,
+					"Tiny cache size not divisible by 32.");
+			rte_errno = EINVAL;
+			return NULL;
+		}
+		RTE_MEMPOOL_LOG(DEBUG,
+				"Rounding down cache size to nearest multiple of 32.");
+		size = rounded;
+	}
+
 	if (size == 0 || size > RTE_MEMPOOL_CACHE_MAX_SIZE) {
 		rte_errno = EINVAL;
 		return NULL;
@@ -838,9 +854,28 @@ rte_mempool_create_empty(const char *name, unsigned n, unsigned elt_size,
 		return NULL;
 	}
 
+	/*
+	 * Alignment requirement for performance optimized move within the mempool cache.
+	 * @ref rte_mempool_do_generic_put() implementation.
+	 */
+	RTE_BUILD_BUG_ON(((sizeof(void *) * RTE_MEMPOOL_CACHE_MAX_SIZE / 2) &
+			RTE_CACHE_LINE_MASK) != 0);
+	RTE_BUILD_BUG_ON((RTE_MEMPOOL_CACHE_MAX_SIZE & 31) != 0);
+	if (cache_size & 31) {
+		unsigned int rounded = RTE_ALIGN_MUL_FLOOR(cache_size, 32);
+		if (rounded > 0)
+			RTE_MEMPOOL_LOG(DEBUG,
+					"Rounding down cache size to nearest multiple of 32.");
+		else
+			RTE_MEMPOOL_LOG(WARNING,
+					"Tiny cache size not divisible by 32. Disabling cache.");
+		cache_size = rounded;
+	}
+
 	/* asked cache too big */
 	if (cache_size > RTE_MEMPOOL_CACHE_MAX_SIZE ||
 	    cache_size > n) {
+		RTE_MEMPOOL_LOG(ERR, "Cache size too big.");
 		rte_errno = EINVAL;
 		return NULL;
 	}
@@ -884,7 +919,7 @@ rte_mempool_create_empty(const char *name, unsigned n, unsigned elt_size,
 		goto exit_unlock;
 	}
 
-	mempool_size = RTE_MEMPOOL_HEADER_SIZE(mp, cache_size);
+	mempool_size = sizeof(struct rte_mempool);
 	mempool_size += private_data_size;
 	mempool_size = RTE_ALIGN_CEIL(mempool_size, RTE_MEMPOOL_ALIGN);
 
@@ -900,7 +935,7 @@ rte_mempool_create_empty(const char *name, unsigned n, unsigned elt_size,
 
 	/* init the mempool structure */
 	mp = mz->addr;
-	memset(mp, 0, RTE_MEMPOOL_HEADER_SIZE(mp, cache_size));
+	memset(mp, 0, mempool_size);
 	ret = strlcpy(mp->name, name, sizeof(mp->name));
 	if (ret < 0 || ret >= (int)sizeof(mp->name)) {
 		rte_errno = ENAMETOOLONG;
@@ -937,13 +972,6 @@ rte_mempool_create_empty(const char *name, unsigned n, unsigned elt_size,
 		goto exit_unlock;
 	}
 
-	/*
-	 * local_cache pointer is set even if cache_size is zero.
-	 * The local_cache points to just past the elt_pa[] array.
-	 */
-	mp->local_cache = (struct rte_mempool_cache *)
-		RTE_PTR_ADD(mp, RTE_MEMPOOL_HEADER_SIZE(mp, 0));
-
 	/* Init all default caches. */
 	if (cache_size != 0) {
 		for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++)
@@ -1197,6 +1225,7 @@ mempool_obj_audit(struct rte_mempool *mp, __rte_unused void *opaque,
 	RTE_MEMPOOL_CHECK_COOKIES(mp, &obj, 1, 2);
 }
 
+/* check cookies before and after objects */
 static void
 mempool_audit_cookies(struct rte_mempool *mp)
 {
@@ -1213,23 +1242,28 @@ mempool_audit_cookies(struct rte_mempool *mp)
 #define mempool_audit_cookies(mp) do {} while(0)
 #endif
 
-/* check cookies before and after objects */
+/* check cache size consistency */
 static void
 mempool_audit_cache(const struct rte_mempool *mp)
 {
-	/* check cache size consistency */
 	unsigned lcore_id;
+	const uint32_t cache_size = mp->cache_size;
 
-	if (mp->cache_size == 0)
-		return;
+	if (cache_size > RTE_MEMPOOL_CACHE_MAX_SIZE) {
+		RTE_MEMPOOL_LOG(CRIT, "badness on cache size");
+		rte_panic("MEMPOOL: invalid cache size\n");
+	}
 
 	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
 		const struct rte_mempool_cache *cache;
 		cache = &mp->local_cache[lcore_id];
-		if (cache->len > RTE_DIM(cache->objs)) {
-			RTE_MEMPOOL_LOG(CRIT, "badness on cache[%u]",
-				lcore_id);
-			rte_panic("MEMPOOL: invalid cache len\n");
+		if (cache->size != cache_size) {
+			RTE_MEMPOOL_LOG(CRIT, "badness on cache[%u] size", lcore_id);
+			rte_panic("MEMPOOL: invalid cache[%u] size\n", lcore_id);
+		}
+		if (cache->len > cache_size) {
+			RTE_MEMPOOL_LOG(CRIT, "badness on cache[%u] len", lcore_id);
+			rte_panic("MEMPOOL: invalid cache[%u] len\n", lcore_id);
 		}
 	}
 }
@@ -1241,9 +1275,6 @@ rte_mempool_audit(struct rte_mempool *mp)
 {
 	mempool_audit_cache(mp);
 	mempool_audit_cookies(mp);
-
-	/* For case where mempool DEBUG is not set, and cache size is 0 */
-	RTE_SET_USED(mp);
 }
 
 /* dump the status of the mempool on the console */
diff --git a/lib/mempool/rte_mempool.h b/lib/mempool/rte_mempool.h
index 50d958c7c6..4a061de1ef 100644
--- a/lib/mempool/rte_mempool.h
+++ b/lib/mempool/rte_mempool.h
@@ -89,14 +89,14 @@ struct __rte_cache_aligned rte_mempool_debug_stats {
  */
 struct __rte_cache_aligned rte_mempool_cache {
 	uint32_t size;	      /**< Size of the cache */
-	uint32_t flushthresh; /**< Obsolete; for API/ABI compatibility purposes only */
 	uint32_t len;	      /**< Current cache count */
 #ifdef RTE_LIBRTE_MEMPOOL_STATS
-	uint32_t unused;
 	/*
 	 * Alternative location for the most frequently updated mempool statistics (per-lcore),
 	 * providing faster update access when using a mempool cache.
+	 * Note: 16-byte aligned for optimal SIMD access, when updating pairs of counters.
 	 */
+	alignas(16)
 	struct {
 		uint64_t put_bulk;          /**< Number of puts. */
 		uint64_t put_objs;          /**< Number of objects successfully put. */
@@ -104,15 +104,9 @@ struct __rte_cache_aligned rte_mempool_cache {
 		uint64_t get_success_objs;  /**< Objects successfully allocated. */
 	} stats;                        /**< Statistics */
 #endif
-	/**
-	 * Cache objects
-	 *
-	 * Note:
-	 * Cache is allocated at double size for API/ABI compatibility purposes only.
-	 * When reducing its size at an API/ABI breaking release,
-	 * remember to add a cache guard after it.
-	 */
-	alignas(RTE_CACHE_LINE_SIZE) void *objs[RTE_MEMPOOL_CACHE_MAX_SIZE * 2];
+	/** Cache objects */
+	alignas(RTE_CACHE_LINE_SIZE) void *objs[RTE_MEMPOOL_CACHE_MAX_SIZE];
+	RTE_CACHE_GUARD;
 };
 
 /**
@@ -240,8 +234,7 @@ struct __rte_cache_aligned rte_mempool {
 	unsigned int flags;              /**< Flags of the mempool. */
 	int socket_id;                   /**< Socket id passed at create. */
 	uint32_t size;                   /**< Max size of the mempool. */
-	uint32_t cache_size;
-	/**< Size of per-lcore default local cache. */
+	uint32_t cache_size;             /**< Size of per-lcore default local cache. */
 
 	uint32_t elt_size;               /**< Size of an element. */
 	uint32_t header_size;            /**< Size of header (before elt). */
@@ -257,13 +250,13 @@ struct __rte_cache_aligned rte_mempool {
 	 */
 	int32_t ops_index;
 
-	struct rte_mempool_cache *local_cache; /**< Per-lcore local cache */
-
 	uint32_t populated_size;         /**< Number of populated objects. */
 	struct rte_mempool_objhdr_list elt_list; /**< List of objects in pool */
 	uint32_t nb_mem_chunks;          /**< Number of memory chunks */
 	struct rte_mempool_memhdr_list mem_list; /**< List of memory chunks */
 
+	struct rte_mempool_cache local_cache[RTE_MAX_LCORE]; /**< Per-lcore local cache */
+
 #ifdef RTE_LIBRTE_MEMPOOL_STATS
 	/** Per-lcore statistics.
 	 *
@@ -271,6 +264,8 @@ struct __rte_cache_aligned rte_mempool {
 	 */
 	struct rte_mempool_debug_stats stats[RTE_MAX_LCORE + 1];
 #endif
+
+	/* Private data are located immediately after the mempool structure. */
 };
 
 /** Spreading among memory channels not required. */
@@ -362,18 +357,6 @@ struct __rte_cache_aligned rte_mempool {
 #define RTE_MEMPOOL_CACHE_STAT_ADD(cache, name, n) do {} while (0)
 #endif
 
-/**
- * @internal Calculate the size of the mempool header.
- *
- * @param mp
- *   Pointer to the memory pool.
- * @param cs
- *   Size of the per-lcore cache.
- */
-#define RTE_MEMPOOL_HEADER_SIZE(mp, cs) \
-	(sizeof(*(mp)) + (((cs) == 0) ? 0 : \
-	(sizeof(struct rte_mempool_cache) * RTE_MAX_LCORE)))
-
 /* return the header of a mempool object (internal) */
 static inline struct rte_mempool_objhdr *
 rte_mempool_get_header(void *obj)
@@ -718,7 +701,7 @@ struct __rte_cache_aligned rte_mempool_ops {
 	rte_mempool_dequeue_contig_blocks_t dequeue_contig_blocks;
 };
 
-#define RTE_MEMPOOL_MAX_OPS_IDX 16  /**< Max registered ops structs */
+#define RTE_MEMPOOL_MAX_OPS_IDX 32  /**< Max registered ops structs */
 
 /**
  * Structure storing the table of registered ops structs, each of which contain
@@ -1049,7 +1032,7 @@ rte_mempool_free(struct rte_mempool *mp);
  *   If cache_size is non-zero, the rte_mempool library will try to
  *   limit the accesses to the common lockless pool, by maintaining a
  *   per-lcore object cache. This argument must be lower or equal to
- *   RTE_MEMPOOL_CACHE_MAX_SIZE and n.
+ *   RTE_MEMPOOL_CACHE_MAX_SIZE and n, and it must be divisible by 32.
  *   The access to the per-lcore table is of course
  *   faster than the multi-producer/consumer pool. The cache can be
  *   disabled if the cache_size argument is set to 0; it can be useful to
@@ -1368,15 +1351,16 @@ rte_mempool_cache_free(struct rte_mempool_cache *cache);
 static __rte_always_inline struct rte_mempool_cache *
 rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
 {
-	if (unlikely(mp->cache_size == 0))
+	if (unlikely(lcore_id == LCORE_ID_ANY))
 		return NULL;
 
-	if (unlikely(lcore_id == LCORE_ID_ANY))
+	struct rte_mempool_cache *cache = &mp->local_cache[lcore_id];
+
+	if (unlikely(cache->size == 0))
 		return NULL;
 
-	rte_mempool_trace_default_cache(mp, lcore_id,
-		&mp->local_cache[lcore_id]);
-	return &mp->local_cache[lcore_id];
+	rte_mempool_trace_default_cache(mp, lcore_id, cache);
+	return cache;
 }
 
 /**
@@ -1445,9 +1429,24 @@ rte_mempool_do_generic_put(struct rte_mempool *mp, void * const *obj_table,
 		 * are more hot, from the upper half of the cache.
 		 */
 		__rte_assume(cache->len > cache->size / 2);
-		rte_mempool_ops_enqueue_bulk(mp, &cache->objs[0], cache->size / 2);
-		rte_memcpy(&cache->objs[0], &cache->objs[cache->size / 2],
-				sizeof(void *) * (cache->len - cache->size / 2));
+		rte_mempool_ops_enqueue_bulk(mp, cache->objs, cache->size / 2);
+		/*
+		 * For improved rte_memcpy() performance, move down objects
+		 * from CPU cache line aligned address in chunks of 32 bytes.
+		 * Note: For cache->objs[cache->size / 2] to be cache line aligned, cache->size
+		 * must be divisible by 32 on 32-bit architecture with 64-byte cache line,
+		 * divisible by 32 on 64-bit architecture with 128-byte cache line, and
+		 * be divisible by 16 on 64-bit architecture with 64-byte cache line.
+		 * For API consistency, require mempool cache size is divisible by 32.
+		 * This requirement is enforced when creating the cache.
+		 * @ref rte_mempool_create_empty() implementation.
+		 */
+		const size_t move = RTE_ALIGN_MUL_CEIL(
+				sizeof(void *) * (cache->len - cache->size / 2), 32);
+		__rte_assume(move >= 32);
+		__rte_assume((move & 31) == 0);
+		rte_memcpy(cache->objs, __rte_assume_cache_aligned(&cache->objs[cache->size / 2]),
+				move);
 		cache_objs = &cache->objs[cache->len - cache->size / 2];
 		cache->len = cache->len - cache->size / 2 + n;
 	} else {
@@ -1892,8 +1891,7 @@ void rte_mempool_audit(struct rte_mempool *mp);
  */
 static inline void *rte_mempool_get_priv(struct rte_mempool *mp)
 {
-	return (char *)mp +
-		RTE_MEMPOOL_HEADER_SIZE(mp, mp->cache_size);
+	return (char *)mp + sizeof(struct rte_mempool);
 }
 
 /**
diff --git a/lib/stack/meson.build b/lib/stack/meson.build
index 18177a742f..50e688522e 100644
--- a/lib/stack/meson.build
+++ b/lib/stack/meson.build
@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: BSD-3-Clause
 # Copyright(c) 2019 Intel Corporation
 
-sources = files('rte_stack.c', 'rte_stack_std.c', 'rte_stack_lf.c')
+sources = files('rte_stack.c', 'rte_stack_std.c', 'rte_stack_lf.c', 'rte_stack_pile.c')
 headers = files('rte_stack.h')
 # subheaders, not for direct inclusion by apps
 indirect_headers += files(
@@ -10,4 +10,5 @@ indirect_headers += files(
         'rte_stack_lf_generic.h',
         'rte_stack_lf_c11.h',
         'rte_stack_lf_stubs.h',
+        'rte_stack_pile.h',
 )
diff --git a/lib/stack/rte_stack.c b/lib/stack/rte_stack.c
index 4c78fe4b4b..a4bbf8a4d7 100644
--- a/lib/stack/rte_stack.c
+++ b/lib/stack/rte_stack.c
@@ -1,5 +1,6 @@
 /* SPDX-License-Identifier: BSD-3-Clause
  * Copyright(c) 2019 Intel Corporation
+ * Copyright(c) 2026 SmartShare Systems
  */
 
 #include <stdalign.h>
@@ -32,6 +33,8 @@ rte_stack_init(struct rte_stack *s, unsigned int count, uint32_t flags)
 
 	if (flags & RTE_STACK_F_LF)
 		rte_stack_lf_init(s, count);
+	else if (flags & RTE_STACK_F_PILE)
+		rte_stack_pile_init(s, count);
 	else
 		rte_stack_std_init(s);
 }
@@ -41,6 +44,8 @@ rte_stack_get_memsize(unsigned int count, uint32_t flags)
 {
 	if (flags & RTE_STACK_F_LF)
 		return rte_stack_lf_get_memsize(count);
+	else if (flags & RTE_STACK_F_PILE)
+		return rte_stack_pile_get_memsize(count);
 	else
 		return rte_stack_std_get_memsize(count);
 }
@@ -58,7 +63,11 @@ rte_stack_create(const char *name, unsigned int count, int socket_id,
 	unsigned int sz;
 	int ret;
 
-	if (flags & ~(RTE_STACK_F_LF)) {
+	if (flags & ~(RTE_STACK_F_LF | RTE_STACK_F_PILE)) {
+		STACK_LOG_ERR("Unsupported stack flags %#x", flags);
+		return NULL;
+	}
+	if ((flags & RTE_STACK_F_LF) && (flags & RTE_STACK_F_PILE)) {
 		STACK_LOG_ERR("Unsupported stack flags %#x", flags);
 		return NULL;
 	}
@@ -73,6 +82,13 @@ rte_stack_create(const char *name, unsigned int count, int socket_id,
 		return NULL;
 	}
 #endif
+#if !defined(RTE_STACK_PILE_SUPPORTED)
+	if (flags & RTE_STACK_F_PILE) {
+		STACK_LOG_ERR("Pile is not supported on your platform");
+		rte_errno = ENOTSUP;
+		return NULL;
+	}
+#endif
 
 	sz = rte_stack_get_memsize(count, flags);
 
diff --git a/lib/stack/rte_stack.h b/lib/stack/rte_stack.h
index fd17ac791d..84bb620c5a 100644
--- a/lib/stack/rte_stack.h
+++ b/lib/stack/rte_stack.h
@@ -1,5 +1,6 @@
 /* SPDX-License-Identifier: BSD-3-Clause
  * Copyright(c) 2019 Intel Corporation
+ * Copyright(c) 2026 SmartShare Systems
  */
 
 /**
@@ -28,11 +29,47 @@
 #define RTE_STACK_NAMESIZE (RTE_MEMZONE_NAMESIZE - \
 			   sizeof(RTE_STACK_MZ_PREFIX) + 1)
 
+static_assert(((sizeof(void *) * RTE_STACK_PILE_BULK_SIZE) & RTE_CACHE_LINE_MASK) == 0,
+		"Pile bulk size must be divisible by CPU cache line size");
+static_assert(RTE_IS_POWER_OF_2(RTE_STACK_PILE_BULK_SIZE),
+		"RTE_STACK_PILE_BULK_SIZE must be power of 2");
+
+/* Note: Also used as solo (single-object) pile element. */
 struct rte_stack_lf_elem {
 	void *data;			/**< Data pointer */
 	struct rte_stack_lf_elem *next;	/**< Next pointer */
 };
 
+/*
+ * Bulk (multi-object) pile element.
+ * Inherited from the rte_stack_lf_elem (single-object) class,
+ * and extended with an array for holding a bulk of object pointers.
+ */
+struct rte_stack_pile_bulk_elem {
+	/* The first part must be ABI compatible with the rte_stack_lf_elem parent class. */
+	void *data;                             /**< Unused, for rte_stack_lf_elem compatibility */
+	struct rte_stack_pile_bulk_elem *next;  /**< Next pointer */
+	/* The second part differs. */
+	alignas(RTE_CACHE_LINE_SIZE)
+	void *objs[RTE_STACK_PILE_BULK_SIZE];   /**< Bulk (multi-object) pointers */
+};
+
+static_assert(sizeof(struct rte_stack_lf_elem) ==
+		sizeof(struct rte_stack_lf_elem *) + sizeof(void *),
+		"Parent type has changed");
+static_assert(RTE_SIZEOF_FIELD(struct rte_stack_lf_elem, next) ==
+		RTE_SIZEOF_FIELD(struct rte_stack_pile_bulk_elem, next),
+		"Inherited type mismatch");
+static_assert(offsetof(struct rte_stack_lf_elem, next) ==
+		offsetof(struct rte_stack_pile_bulk_elem, next),
+		"Inherited type mismatch");
+static_assert(RTE_SIZEOF_FIELD(struct rte_stack_lf_elem, data) ==
+		RTE_SIZEOF_FIELD(struct rte_stack_pile_bulk_elem, data),
+		"Inherited type mismatch");
+static_assert(offsetof(struct rte_stack_lf_elem, data) ==
+		offsetof(struct rte_stack_pile_bulk_elem, data),
+		"Inherited type mismatch");
+
 struct __rte_aligned(16) rte_stack_lf_head {
 	struct rte_stack_lf_elem *top; /**< Stack top */
 	uint64_t cnt; /**< Modification counter for avoiding ABA problem */
@@ -51,12 +88,36 @@ struct rte_stack_lf_list {
 struct rte_stack_lf {
 	/** LIFO list of elements */
 	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_list used;
+	RTE_CACHE_GUARD;
 	/** LIFO list of free elements */
 	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_list free;
+	RTE_CACHE_GUARD;
 	/** LIFO elements */
 	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_elem elems[];
 };
 
+/* Pile structure containing three lock-free LIFO-like lists:
+ *  - A list of elements, each element holding a bulk of pointers to objects.
+ *  - A list of elements, each element holding one pointer to an object.
+ *  - A list of free linked-list elements.
+ */
+struct rte_stack_pile {
+	/** LIFO list of bulk (multi-object) elements */
+	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_list bulk;
+	RTE_CACHE_GUARD;
+	/** LIFO list of solo (single-object) elements */
+	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_list solo;
+	RTE_CACHE_GUARD;
+	/** LIFO list of free bulk elements */
+	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_list free_bulk;
+	RTE_CACHE_GUARD;
+	/** LIFO list of free solo elements */
+	alignas(RTE_CACHE_LINE_SIZE) struct rte_stack_lf_list free_solo;
+	RTE_CACHE_GUARD;
+	/** LIFO elements follow, first bulk, then solo */
+	alignas(RTE_CACHE_LINE_SIZE) void *elems[];
+};
+
 /* Structure containing the LIFO, its current length, and a lock for mutual
  * exclusion.
  */
@@ -78,6 +139,7 @@ struct __rte_cache_aligned rte_stack {
 	uint32_t flags; /**< Flags supplied at creation. */
 	union {
 		struct rte_stack_lf stack_lf; /**< Lock-free LIFO structure. */
+		struct rte_stack_pile stack_pile; /**< Lock-free pile (LIFO-like) structure. */
 		struct rte_stack_std stack_std;	/**< LIFO structure. */
 	};
 };
@@ -88,8 +150,19 @@ struct __rte_cache_aligned rte_stack {
  */
 #define RTE_STACK_F_LF 0x0001
 
+/**
+ * The stack-like pile uses lock-free push and pop functions.
+ * It is optimized for bulks of objects, and is not strictly LIFO.
+ * This flag is only supported on x86_64 or arm64 platforms, currently.
+ *
+ * @warning
+ * @b EXPERIMENTAL: this API may change, or be removed, without prior notice.
+ */
+#define RTE_STACK_F_PILE 0x0002
+
 #include "rte_stack_std.h"
 #include "rte_stack_lf.h"
+#include "rte_stack_pile.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -108,13 +181,15 @@ extern "C" {
  *   Actual number of objects pushed (either 0 or *n*).
  */
 static __rte_always_inline unsigned int
-rte_stack_push(struct rte_stack *s, void * const *obj_table, unsigned int n)
+rte_stack_push(struct rte_stack *s, void * const * __restrict obj_table, unsigned int n)
 {
 	RTE_ASSERT(s != NULL);
 	RTE_ASSERT(obj_table != NULL);
 
 	if (s->flags & RTE_STACK_F_LF)
 		return __rte_stack_lf_push(s, obj_table, n);
+	else if (s->flags & RTE_STACK_F_PILE)
+		return __rte_stack_pile_push(s, obj_table, n);
 	else
 		return __rte_stack_std_push(s, obj_table, n);
 }
@@ -132,13 +207,15 @@ rte_stack_push(struct rte_stack *s, void * const *obj_table, unsigned int n)
  *   Actual number of objects popped (either 0 or *n*).
  */
 static __rte_always_inline unsigned int
-rte_stack_pop(struct rte_stack *s, void **obj_table, unsigned int n)
+rte_stack_pop(struct rte_stack *s, void ** __restrict obj_table, unsigned int n)
 {
 	RTE_ASSERT(s != NULL);
 	RTE_ASSERT(obj_table != NULL);
 
 	if (s->flags & RTE_STACK_F_LF)
 		return __rte_stack_lf_pop(s, obj_table, n);
+	else if (s->flags & RTE_STACK_F_PILE)
+		return __rte_stack_pile_pop(s, obj_table, n);
 	else
 		return __rte_stack_std_pop(s, obj_table, n);
 }
@@ -158,6 +235,8 @@ rte_stack_count(struct rte_stack *s)
 
 	if (s->flags & RTE_STACK_F_LF)
 		return __rte_stack_lf_count(s);
+	else if (s->flags & RTE_STACK_F_PILE)
+		return __rte_stack_pile_count(s);
 	else
 		return __rte_stack_std_count(s);
 }
diff --git a/lib/stack/rte_stack_lf.h b/lib/stack/rte_stack_lf.h
index f2b012cd0e..e13b2a60d7 100644
--- a/lib/stack/rte_stack_lf.h
+++ b/lib/stack/rte_stack_lf.h
@@ -34,7 +34,7 @@
  */
 static __rte_always_inline unsigned int
 __rte_stack_lf_push(struct rte_stack *s,
-		    void * const *obj_table,
+		    void * const * __restrict obj_table,
 		    unsigned int n)
 {
 	struct rte_stack_lf_elem *tmp, *first, *last = NULL;
@@ -71,7 +71,8 @@ __rte_stack_lf_push(struct rte_stack *s,
  *   - Actual number of objects popped.
  */
 static __rte_always_inline unsigned int
-__rte_stack_lf_pop(struct rte_stack *s, void **obj_table, unsigned int n)
+__rte_stack_lf_pop(struct rte_stack *s, void ** __restrict obj_table,
+		   unsigned int n)
 {
 	struct rte_stack_lf_elem *first, *last = NULL;
 
@@ -79,6 +80,7 @@ __rte_stack_lf_pop(struct rte_stack *s, void **obj_table, unsigned int n)
 		return 0;
 
 	/* Pop n used elements */
+	__rte_assume(obj_table != NULL);
 	first = __rte_stack_lf_pop_elems(&s->stack_lf.used,
 					 n, obj_table, &last);
 	if (unlikely(first == NULL))
diff --git a/lib/stack/rte_stack_lf_c11.h b/lib/stack/rte_stack_lf_c11.h
index b97e02d6a1..d02bd1aece 100644
--- a/lib/stack/rte_stack_lf_c11.h
+++ b/lib/stack/rte_stack_lf_c11.h
@@ -99,7 +99,7 @@ __rte_stack_lf_push_elems(struct rte_stack_lf_list *list,
 static __rte_always_inline struct rte_stack_lf_elem *
 __rte_stack_lf_pop_elems(struct rte_stack_lf_list *list,
 			 unsigned int num,
-			 void **obj_table,
+			 void ** __restrict obj_table,
 			 struct rte_stack_lf_elem **last)
 {
 	struct rte_stack_lf_head old_head;
diff --git a/lib/stack/rte_stack_lf_generic.h b/lib/stack/rte_stack_lf_generic.h
index cc69e4d168..5187caf586 100644
--- a/lib/stack/rte_stack_lf_generic.h
+++ b/lib/stack/rte_stack_lf_generic.h
@@ -74,7 +74,7 @@ __rte_stack_lf_push_elems(struct rte_stack_lf_list *list,
 static __rte_always_inline struct rte_stack_lf_elem *
 __rte_stack_lf_pop_elems(struct rte_stack_lf_list *list,
 			 unsigned int num,
-			 void **obj_table,
+			 void ** __restrict obj_table,
 			 struct rte_stack_lf_elem **last)
 {
 	struct rte_stack_lf_head old_head;
diff --git a/lib/stack/rte_stack_lf_stubs.h b/lib/stack/rte_stack_lf_stubs.h
index a05abf1f1c..5810744d67 100644
--- a/lib/stack/rte_stack_lf_stubs.h
+++ b/lib/stack/rte_stack_lf_stubs.h
@@ -30,7 +30,7 @@ __rte_stack_lf_push_elems(struct rte_stack_lf_list *list,
 static __rte_always_inline struct rte_stack_lf_elem *
 __rte_stack_lf_pop_elems(struct rte_stack_lf_list *list,
 			 unsigned int num,
-			 void **obj_table,
+			 void ** __restrict obj_table,
 			 struct rte_stack_lf_elem **last)
 {
 	RTE_SET_USED(obj_table);
diff --git a/lib/stack/rte_stack_pile.c b/lib/stack/rte_stack_pile.c
new file mode 100644
index 0000000000..ccad544b8c
--- /dev/null
+++ b/lib/stack/rte_stack_pile.c
@@ -0,0 +1,35 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 SmartShare Systems
+ */
+
+#include "rte_stack.h"
+
+void
+rte_stack_pile_init(struct rte_stack *s, unsigned int count)
+{
+	unsigned int bulk = (count + RTE_STACK_PILE_BULK_SIZE - 1) / RTE_STACK_PILE_BULK_SIZE;
+	struct rte_stack_pile_bulk_elem *bulk_elems =
+			(struct rte_stack_pile_bulk_elem *)(&s->stack_pile.elems);
+	struct rte_stack_lf_elem *solo_elems = (struct rte_stack_lf_elem *)&bulk_elems[bulk];
+	unsigned int i;
+
+	for (i = 0; i < bulk; i++)
+		__rte_stack_pile_bulk_push_elems(&s->stack_pile.free_bulk,
+					  &bulk_elems[i], &bulk_elems[i], 1);
+	for (i = 0; i < count; i++)
+		__rte_stack_lf_push_elems(&s->stack_pile.free_solo,
+					  &solo_elems[i], &solo_elems[i], 1);
+}
+
+ssize_t
+rte_stack_pile_get_memsize(unsigned int count)
+{
+	unsigned int bulk = (count + RTE_STACK_PILE_BULK_SIZE - 1) / RTE_STACK_PILE_BULK_SIZE;
+	ssize_t sz = offsetof(struct rte_stack, stack_pile.elems);
+	sz += bulk * sizeof(struct rte_stack_pile_bulk_elem);
+	sz += count * sizeof(struct rte_stack_lf_elem);
+	sz += RTE_CACHE_LINE_ROUNDUP(sz);
+	sz += RTE_CACHE_GUARD_LINES * RTE_CACHE_LINE_SIZE;
+
+	return sz;
+}
diff --git a/lib/stack/rte_stack_pile.h b/lib/stack/rte_stack_pile.h
new file mode 100644
index 0000000000..8aa93b7e86
--- /dev/null
+++ b/lib/stack/rte_stack_pile.h
@@ -0,0 +1,330 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 SmartShare Systems
+ */
+
+#ifndef _RTE_STACK_PILE_H_
+#define _RTE_STACK_PILE_H_
+
+#include <rte_memcpy.h>
+
+#include "rte_stack_lf.h"
+#ifdef RTE_STACK_LF_SUPPORTED
+/**
+ * Indicates that RTE_STACK_F_PILE is supported.
+ */
+#define RTE_STACK_PILE_SUPPORTED
+#endif
+
+static __rte_always_inline unsigned int
+__rte_stack_pile_count(struct rte_stack *s)
+{
+	/* stack_lf_push() and stack_lf_pop() do not update the list's contents
+	 * and stack_lf->len atomically, which can cause the list to appear
+	 * shorter than it actually is if this function is called while other
+	 * threads are modifying the list.
+	 *
+	 * However, given the inherently approximate nature of the get_count
+	 * callback -- even if the list and its size were updated atomically,
+	 * the size could change between when get_count executes and when the
+	 * value is returned to the caller -- this is acceptable.
+	 *
+	 * The stack_lf->len updates are placed such that the list may appear to
+	 * have fewer elements than it does, but will never appear to have more
+	 * elements. If the mempool is near-empty to the point that this is a
+	 * concern, the user should consider increasing the mempool size.
+	 */
+#ifdef RTE_USE_C11_MEM_MODEL
+	return RTE_MIN((unsigned int)s->capacity,
+			(unsigned int)rte_atomic_load_explicit(&s->stack_pile.bulk.len,
+			rte_memory_order_relaxed) * RTE_STACK_PILE_BULK_SIZE +
+			(unsigned int)rte_atomic_load_explicit(&s->stack_pile.solo.len,
+			rte_memory_order_relaxed));
+#else /* FIXME: Remove if removed from lock-free stack. */
+	/* NOTE: review for potential ordering optimization */
+	return RTE_MIN((unsigned int)s->capacity,
+			(unsigned int)rte_atomic_load_explicit(&s->stack_pile.bulk.len,
+			rte_memory_order_seq_cst) * RTE_STACK_PILE_BULK_SIZE +
+			(unsigned int)rte_atomic_load_explicit(&s->stack_pile.solo.len,
+			rte_memory_order_seq_cst));
+#endif
+}
+
+static __rte_always_inline void
+__rte_stack_pile_bulk_push_elems(struct rte_stack_lf_list *list,
+		struct rte_stack_pile_bulk_elem *first,
+		struct rte_stack_pile_bulk_elem *last,
+		unsigned int num)
+{
+	__rte_stack_lf_push_elems(list,
+		(struct rte_stack_lf_elem *)first,
+		(struct rte_stack_lf_elem *)last,
+		num);
+}
+
+static __rte_always_inline struct rte_stack_pile_bulk_elem *
+__rte_stack_pile_bulk_pop_elems(struct rte_stack_lf_list *list,
+		unsigned int num,
+		void ** __restrict obj_table,
+		struct rte_stack_pile_bulk_elem **last)
+{
+	struct rte_stack_pile_bulk_elem *first = (struct rte_stack_pile_bulk_elem *)
+			__rte_stack_lf_pop_elems(list, num, NULL,
+			(struct rte_stack_lf_elem **)last);
+	if (first == NULL)
+		return NULL;
+
+	if (obj_table != NULL) {
+		/* Traverse the list to copy the bulks. */
+		struct rte_stack_pile_bulk_elem *tmp = first;
+		for (unsigned int i = 0; i < num; i++, tmp = tmp->next)
+			rte_memcpy(&obj_table[i * RTE_STACK_PILE_BULK_SIZE], tmp->objs,
+					sizeof(void *) * RTE_STACK_PILE_BULK_SIZE);
+	}
+
+	return first;
+}
+
+/**
+ * Push several objects on 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 push on the pile from the obj_table.
+ * @return
+ *   Actual number of objects pushed (either 0 or *n*).
+ */
+static __rte_always_inline unsigned int
+__rte_stack_pile_push(struct rte_stack *s,
+		void * const * __restrict 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, *tmp_bulk;
+	struct rte_stack_lf_elem *solo_first = NULL, *solo_last = NULL, *tmp_solo;
+	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;
+	}
+
+	/* Allocate n_bulk elements from the free list. */
+	bulk_first = __rte_stack_pile_bulk_pop_elems(&pile->free_bulk, n_bulk, NULL, &bulk_last);
+	if (unlikely(bulk_first == NULL))
+		return 0; /* Failed. */
+
+	if (likely(n_solo == 0))
+		goto bulk;
+
+solo:
+	/* Allocate n_solo elements from the free list. */
+	solo_first = __rte_stack_lf_pop_elems(&pile->free_solo, n_solo, NULL, &solo_last);
+	if (unlikely(solo_first == NULL)) {
+		/* Failed. Roll back. */
+		if (n_bulk > 0)
+			__rte_stack_pile_bulk_push_elems(&pile->free_bulk,
+					bulk_first, bulk_last, n_bulk);
+		return 0;
+	}
+
+	/*
+	 * Construct the solo elements.
+	 * Copy the objects, but ignore the object order.
+	 */
+	tmp_solo = solo_first;
+	__rte_assume(n_solo > 0);
+	__rte_assume(n_solo < RTE_STACK_PILE_BULK_SIZE);
+	for (i = 0; i < n_solo; i++, tmp_solo = tmp_solo->next)
+		tmp_solo->data = obj_table[n_bulk * RTE_STACK_PILE_BULK_SIZE + i];
+
+	/* Push them to the solo list. */
+	__rte_stack_lf_push_elems(&pile->solo, solo_first, solo_last, n_solo);
+
+	if (unlikely(n_bulk == 0))
+		return n; /* Done. */
+
+bulk:
+	/*
+	 * Construct the bulk elements.
+	 * Copy bulks in reverse order, but ignore the object order within each bulk.
+	 */
+	tmp_bulk = bulk_first;
+	__rte_assume(n_bulk > 0);
+	for (i = 0; i < n_bulk; i++, tmp_bulk = tmp_bulk->next)
+		rte_memcpy(tmp_bulk->objs, &obj_table[(n_bulk - i - 1) * RTE_STACK_PILE_BULK_SIZE],
+				sizeof(void *) * RTE_STACK_PILE_BULK_SIZE);
+
+	/* Push them to the bulk list. */
+	__rte_stack_pile_bulk_push_elems(&pile->bulk, bulk_first, bulk_last, n_bulk);
+
+	return n;
+}
+
+/**
+ * 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 ** __restrict 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;
+
+	/* 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 in front of the bulk elements, so it can be freed. */
+	if (n_bulk > 0)
+		frag->next = bulk_first;
+	else
+		bulk_last = frag;
+	bulk_first = 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) {
+			/* Attach the fragmentation element after the bulk elements. */
+			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_ */
diff --git a/lib/stack/rte_stack_std.h b/lib/stack/rte_stack_std.h
index ae28add5c4..8a3ca9edf6 100644
--- a/lib/stack/rte_stack_std.h
+++ b/lib/stack/rte_stack_std.h
@@ -20,25 +20,25 @@
  *   Actual number of objects pushed (either 0 or *n*).
  */
 static __rte_always_inline unsigned int
-__rte_stack_std_push(struct rte_stack *s, void * const *obj_table,
+__rte_stack_std_push(struct rte_stack *s, void * const * __restrict obj_table,
 		     unsigned int n)
 {
 	struct rte_stack_std *stack = &s->stack_std;
 	unsigned int index;
-	void **cache_objs;
+	void ** __restrict stack_objs;
 
 	rte_spinlock_lock(&stack->lock);
-	cache_objs = &stack->objs[stack->len];
+	stack_objs = &stack->objs[stack->len];
 
-	/* Is there sufficient space in the stack? */
-	if ((stack->len + n) > s->capacity) {
+	if (unlikely((stack->len + n) > s->capacity)) {
+		/* Insufficient space in the stack. */
 		rte_spinlock_unlock(&stack->lock);
 		return 0;
 	}
 
-	/* Add elements back into the cache */
+	/* Push objects to the stack */
 	for (index = 0; index < n; ++index, obj_table++)
-		cache_objs[index] = *obj_table;
+		stack_objs[index] = *obj_table;
 
 	stack->len += n;
 
@@ -59,24 +59,26 @@ __rte_stack_std_push(struct rte_stack *s, void * const *obj_table,
  *   Actual number of objects popped (either 0 or *n*).
  */
 static __rte_always_inline unsigned int
-__rte_stack_std_pop(struct rte_stack *s, void **obj_table, unsigned int n)
+__rte_stack_std_pop(struct rte_stack *s, void ** __restrict obj_table, unsigned int n)
 {
 	struct rte_stack_std *stack = &s->stack_std;
 	unsigned int index, len;
-	void **cache_objs;
+	void ** __restrict stack_objs;
 
 	rte_spinlock_lock(&stack->lock);
 
 	if (unlikely(n > stack->len)) {
+		/* Insufficient objects in the stack. */
 		rte_spinlock_unlock(&stack->lock);
 		return 0;
 	}
 
-	cache_objs = stack->objs;
+	stack_objs = stack->objs;
 
+	/* Pop objects from the stack */
 	for (index = 0, len = stack->len - 1; index < n;
 			++index, len--, obj_table++)
-		*obj_table = cache_objs[len];
+		*obj_table = stack_objs[len];
 
 	stack->len -= n;
 	rte_spinlock_unlock(&stack->lock);
-- 
2.43.0


  parent reply	other threads:[~2026-08-03  8:14 UTC|newest]

Thread overview: 15+ messages / expand[flat|nested]  mbox.gz  Atom feed  top
2026-08-01  7:15 [RFC PATCH] NEW: pile stack and mempool driver Morten Brørup
2026-08-01 10:09 ` [RFC PATCH v2] " Morten Brørup
2026-08-01 10:17 ` [RFC PATCH v3] " Morten Brørup
2026-08-01 11:13 ` [RFC PATCH v4] " Morten Brørup
2026-08-02  7:24 ` [RFC PATCH v5] " Morten Brørup
2026-08-02  9:59 ` [RFC PATCH v6] " Morten Brørup
2026-08-02 15:22   ` Stephen Hemminger
2026-08-02 16:45     ` Morten Brørup
2026-08-03  8:14 ` Morten Brørup [this message]
2026-08-03 23:02   ` [RFC PATCH v7] " Stephen Hemminger
2026-08-10  9:02 ` [RFC PATCH v8] " Morten Brørup
2026-08-10 13:58 ` [RFC PATCH v9] " Morten Brørup
2026-08-10 15:12 ` [RFC PATCH v10] " Morten Brørup
2026-08-10 16:15   ` Stephen Hemminger
2026-08-10 18:36   ` Morten Brørup

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

  Avoid top-posting and favor interleaved quoting:
  https://en.wikipedia.org/wiki/Posting_style#Interleaved_style

* Reply using the --to, --cc, and --in-reply-to
  switches of git-send-email(1):

  git send-email \
    --in-reply-to=20260803081446.1292377-1-mb@smartsharesystems.com \
    --to=mb@smartsharesystems.com \
    --cc=dev@dpdk.org \
    /path/to/YOUR_REPLY

  https://kernel.org/pub/software/scm/git/docs/git-send-email.html

* If your mail client supports setting the In-Reply-To header
  via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line before the message body.
This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.