[PATCH v3 2/2] mempool: introduce pile driver

["Morten Brørup" <mb@smartsharesystems.com>]

Added a new "pile" mempool driver, based on the high-performance lock-free
"pile" stack variant.

Changed other stack drivers to call their specific push/pop functions,
instead of calling the generic stack push/pop API.

Signed-off-by: Morten Brørup <mb@smartsharesystems.com>
---
v3:
* Call other stack types' specific push/pop functions.
v2:
* There is no v2.
---
 doc/guides/mempool/stack.rst              | 12 ++-
 drivers/mempool/stack/rte_mempool_stack.c | 90 ++++++++++++++++++++---
 lib/mempool/rte_mempool.h                 |  2 +
 lib/mempool/rte_mempool_ops.c             | 27 +++++--
 4 files changed, 114 insertions(+), 17 deletions(-)

diff --git a/doc/guides/mempool/stack.rst b/doc/guides/mempool/stack.rst
index 80ea07e65d..c06ab2dc56 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/drivers/mempool/stack/rte_mempool_stack.c b/drivers/mempool/stack/rte_mempool_stack.c
index 1476905227..11c1243bec 100644
--- a/drivers/mempool/stack/rte_mempool_stack.c
+++ b/drivers/mempool/stack/rte_mempool_stack.c
@@ -30,7 +30,7 @@ __stack_alloc(struct rte_mempool *mp, uint32_t flags)
 }
 
 static int
-stack_alloc(struct rte_mempool *mp)
+std_stack_alloc(struct rte_mempool *mp)
 {
 	return __stack_alloc(mp, 0);
 }
@@ -42,21 +42,81 @@ lf_stack_alloc(struct rte_mempool *mp)
 }
 
 static int
-stack_enqueue(struct rte_mempool *mp, void * const *obj_table,
+pile_alloc(struct rte_mempool *mp)
+{
+	return __stack_alloc(mp, RTE_STACK_F_PILE);
+}
+
+static int
+std_stack_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_std_push(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+}
+
+static int
+std_stack_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_std_pop(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+}
+
+static int
+lf_stack_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_lf_push(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+}
+
+static int
+lf_stack_dequeue(struct rte_mempool *mp, void **obj_table,
 	      unsigned int n)
 {
 	struct rte_stack *s = mp->pool_data;
 
-	return rte_stack_push(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+	RTE_ASSERT(s != NULL);
+	RTE_ASSERT(obj_table != NULL);
+
+	return __rte_stack_lf_pop(s, obj_table, n) == 0 ? -ENOBUFS : 0;
 }
 
 static int
-stack_dequeue(struct rte_mempool *mp, void **obj_table,
+pile_enqueue(struct rte_mempool *mp, void * const *obj_table,
 	      unsigned int n)
 {
 	struct rte_stack *s = mp->pool_data;
 
-	return rte_stack_pop(s, obj_table, n) == 0 ? -ENOBUFS : 0;
+	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 unsigned
@@ -77,10 +137,10 @@ stack_free(struct rte_mempool *mp)
 
 static struct rte_mempool_ops ops_stack = {
 	.name = "stack",
-	.alloc = stack_alloc,
+	.alloc = std_stack_alloc,
 	.free = stack_free,
-	.enqueue = stack_enqueue,
-	.dequeue = stack_dequeue,
+	.enqueue = std_stack_enqueue,
+	.dequeue = std_stack_dequeue,
 	.get_count = stack_get_count
 };
 
@@ -88,10 +148,20 @@ static struct rte_mempool_ops ops_lf_stack = {
 	.name = "lf_stack",
 	.alloc = lf_stack_alloc,
 	.free = stack_free,
-	.enqueue = stack_enqueue,
-	.dequeue = stack_dequeue,
+	.enqueue = lf_stack_enqueue,
+	.dequeue = lf_stack_dequeue,
+	.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/lib/mempool/rte_mempool.h b/lib/mempool/rte_mempool.h
index 3e161bfdb9..db7def9096 100644
--- a/lib/mempool/rte_mempool.h
+++ b/lib/mempool/rte_mempool.h
@@ -979,6 +979,8 @@ rte_mempool_set_ops_byname(struct rte_mempool *mp, const char *name,
  *   - >=0: Success; return the index of the ops struct in the table.
  *   - -EINVAL - some missing callbacks while registering ops struct.
  *   - -ENOSPC - the maximum number of ops structs has been reached.
+ *   - -ENAMETOOLONG - the name of the ops is too long.
+ *   - -EEXIST - the name of the ops is already registered.
  */
 int rte_mempool_register_ops(const struct rte_mempool_ops *ops);
 
diff --git a/lib/mempool/rte_mempool_ops.c b/lib/mempool/rte_mempool_ops.c
index 066bec36fc..261ad217ee 100644
--- a/lib/mempool/rte_mempool_ops.c
+++ b/lib/mempool/rte_mempool_ops.c
@@ -27,7 +27,7 @@ int
 rte_mempool_register_ops(const struct rte_mempool_ops *h)
 {
 	struct rte_mempool_ops *ops;
-	int16_t ops_index;
+	unsigned int ops_index;
 
 	rte_spinlock_lock(&rte_mempool_ops_table.sl);
 
@@ -47,12 +47,21 @@ rte_mempool_register_ops(const struct rte_mempool_ops *h)
 		return -EINVAL;
 	}
 
-	if (strlen(h->name) >= sizeof(ops->name) - 1) {
+	if (strlen(h->name) > sizeof(ops->name) - 1) {
 		rte_spinlock_unlock(&rte_mempool_ops_table.sl);
-		RTE_MEMPOOL_LOG(DEBUG, "%s(): mempool_ops <%s>: name too long",
+		RTE_MEMPOOL_LOG(ERR, "%s(): mempool_ops <%s>: name too long",
 				__func__, h->name);
-		rte_errno = EEXIST;
-		return -EEXIST;
+		return -ENAMETOOLONG;
+	}
+
+	for (ops_index = 0; ops_index < rte_mempool_ops_table.num_ops; ops_index++) {
+		if (!strcmp(h->name,
+				rte_mempool_ops_table.ops[ops_index].name)) {
+			rte_spinlock_unlock(&rte_mempool_ops_table.sl);
+			RTE_MEMPOOL_LOG(ERR, "%s(): mempool_ops <%s>: name exists",
+					__func__, h->name);
+			return -EEXIST;
+		}
 	}
 
 	ops_index = rte_mempool_ops_table.num_ops++;
@@ -68,6 +77,9 @@ rte_mempool_register_ops(const struct rte_mempool_ops *h)
 	ops->get_info = h->get_info;
 	ops->dequeue_contig_blocks = h->dequeue_contig_blocks;
 
+	RTE_MEMPOOL_LOG(DEBUG,
+		"Registered mempool_ops[%u] <%s>", ops_index, h->name);
+
 	rte_spinlock_unlock(&rte_mempool_ops_table.sl);
 
 	return ops_index;
@@ -185,8 +197,11 @@ rte_mempool_set_ops_byname(struct rte_mempool *mp, const char *name,
 		}
 	}
 
-	if (ops == NULL)
+	if (ops == NULL) {
+		RTE_MEMPOOL_LOG(ERR,
+			"Unknown mempool_ops <%s>, of %u ops registered", name, i);
 		return -EINVAL;
+	}
 
 	mp->ops_index = i;
 	mp->pool_config = pool_config;
-- 
2.43.0