[PATCH v4 2/3] mempool: fix dump of unpopulated pool

[Robert Sanford <rsanford2@gmail.com>]

From: Robert Sanford <rsanford@akamai.com>

rte_mempool_create_empty() adds a mempool to the mempool list before
the mempool has a store. The store is allocated, and
RTE_MEMPOOL_F_POOL_CREATED is set, by the first populate call. In
between, rte_mempool_dump() queries the mempool driver for the number
of objects in the store. With the ring driver this dereferences a NULL
ring, and the process crashes. rte_mempool_list_dump() calls
rte_mempool_dump() for every mempool on the list, so it crashes when
another thread has created a mempool but not populated it yet. The
telemetry /mempool/info handler has the same problem: its
rte_mempool_walk() callback queries the driver for the same count.

Add rte_mempool_is_created(), which tests RTE_MEMPOOL_F_POOL_CREATED.
In rte_mempool_dump() and in the telemetry callback, query the mempool
driver only when it returns true, and report a common
pool count of 0 otherwise, since the mempool has no objects yet. The
rest of the mempool information is still reported. In statistics
builds, rte_mempool_dump() also skips rte_mempool_ops_get_info() for
such a mempool. The info is now zeroed first; it was left uninitialized
for drivers without a get_info operation.

The populating thread sets the flag without holding a lock, so set it
with an atomic fetch-or with release ordering, and test it with an
atomic load with acquire ordering. A reader that sees the flag then
also sees the store.

rte_mempool_walk() still passes mempools that are not populated yet to
its callback. Document that the callback must check
rte_mempool_is_created() before querying the mempool driver.

Fixes: 449c49b93a6b ("mempool: support handler operations")
Fixes: ab8a1cc4e3d8 ("mempool: add object usage counts in telemetry")
Cc: stable@dpdk.org

Signed-off-by: Robert Sanford <rsanford@akamai.com>
Acked-by: Morten Brørup <mb@smartsharesystems.com>
---
 doc/guides/rel_notes/release_26_11.rst |  7 +++++
 lib/mempool/rte_mempool.c              | 37 +++++++++++++++++++++++---
 lib/mempool/rte_mempool.h              | 26 ++++++++++++++++++
 3 files changed, 66 insertions(+), 4 deletions(-)

diff --git a/doc/guides/rel_notes/release_26_11.rst b/doc/guides/rel_notes/release_26_11.rst
index e62f273aad..2f50a602e8 100644
--- a/doc/guides/rel_notes/release_26_11.rst
+++ b/doc/guides/rel_notes/release_26_11.rst
@@ -82,6 +82,13 @@ New Features
   but objects within each bulk are not ordered as expected by a stack.
   Furthermore, it is not strictly bounded by its size, but might hold more objects.
 
+* **Added function to check if a mempool is created.**
+
+  Added the ``rte_mempool_is_created()`` function
+  to check whether the store of a mempool has been created.
+  ``rte_mempool_walk()`` callbacks can use it to skip mempools
+  that are not populated yet before querying the mempool driver.
+
 * **Added unplug operation support to DPAA bus.**
 
   Implemented device unplug operation to allow runtime removal of DPAA devices.
diff --git a/lib/mempool/rte_mempool.c b/lib/mempool/rte_mempool.c
index d9f339a3c2..6dc0f146a1 100644
--- a/lib/mempool/rte_mempool.c
+++ b/lib/mempool/rte_mempool.c
@@ -294,6 +294,22 @@ rte_mempool_free_memchunks(struct rte_mempool *mp)
 	}
 }
 
+RTE_EXPORT_SYMBOL(rte_mempool_is_created)
+bool
+rte_mempool_is_created(const struct rte_mempool *mp)
+{
+	/*
+	 * Mempools are on the mempool list before the store is allocated, so
+	 * rte_mempool_dump() and rte_mempool_walk() callbacks can find them
+	 * while another thread populates them.
+	 * Acquire: pairs with the release in mempool_ops_alloc_once().
+	 */
+	return (rte_atomic_load_explicit(
+			(const RTE_ATOMIC(unsigned int) *)&mp->flags,
+			rte_memory_order_acquire) &
+		RTE_MEMPOOL_F_POOL_CREATED) != 0;
+}
+
 static int
 mempool_ops_alloc_once(struct rte_mempool *mp)
 {
@@ -304,7 +320,13 @@ mempool_ops_alloc_once(struct rte_mempool *mp)
 		ret = rte_mempool_ops_alloc(mp);
 		if (ret != 0)
 			return ret;
-		mp->flags |= RTE_MEMPOOL_F_POOL_CREATED;
+		/*
+		 * Release: make the store visible before the flag, see
+		 * rte_mempool_is_created().
+		 */
+		rte_atomic_fetch_or_explicit(
+			(RTE_ATOMIC(unsigned int) *)&mp->flags,
+			RTE_MEMPOOL_F_POOL_CREATED, rte_memory_order_release);
 	}
 	return 0;
 }
@@ -1283,10 +1305,14 @@ rte_mempool_dump(FILE *f, struct rte_mempool *mp)
 	unsigned common_count;
 	unsigned cache_count;
 	size_t mem_len = 0;
+	bool created;
 
 	RTE_ASSERT(f != NULL);
 	RTE_ASSERT(mp != NULL);
 
+	/* the mempool driver can only be queried once the store exists */
+	created = rte_mempool_is_created(mp);
+
 	fprintf(f, "mempool <%s>@%p\n", mp->name, mp);
 	fprintf(f, "  flags=%x\n", mp->flags);
 	fprintf(f, "  socket_id=%d\n", mp->socket_id);
@@ -1318,14 +1344,16 @@ rte_mempool_dump(FILE *f, struct rte_mempool *mp)
 	}
 
 	cache_count = rte_mempool_dump_cache(f, mp);
-	common_count = rte_mempool_ops_get_count(mp);
+	common_count = created ? rte_mempool_ops_get_count(mp) : 0;
 	if ((cache_count + common_count) > mp->size)
 		common_count = mp->size - cache_count;
 	fprintf(f, "  common_pool_count=%u\n", common_count);
 
 	/* sum and dump statistics */
 #ifdef RTE_LIBRTE_MEMPOOL_STATS
-	rte_mempool_ops_get_info(mp, &info);
+	memset(&info, 0, sizeof(info));
+	if (created)
+		rte_mempool_ops_get_info(mp, &info);
 	memset(&sum, 0, sizeof(sum));
 	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE + 1; lcore_id++) {
 		sum.put_bulk += mp->stats[lcore_id].put_bulk;
@@ -1653,7 +1681,8 @@ mempool_info_cb(struct rte_mempool *mp, void *arg)
 			cache_count += cache->len;
 	}
 	rte_tel_data_add_dict_uint(info->d, "total_cache_count", cache_count);
-	common_count = rte_mempool_ops_get_count(mp);
+	/* the mempool driver can only be queried once the store exists */
+	common_count = rte_mempool_is_created(mp) ? rte_mempool_ops_get_count(mp) : 0;
 	if ((cache_count + common_count) > mp->size)
 		common_count = mp->size - cache_count;
 	rte_tel_data_add_dict_uint(info->d, "common_pool_count", common_count);
diff --git a/lib/mempool/rte_mempool.h b/lib/mempool/rte_mempool.h
index bd58d8a9d5..9bf1220b71 100644
--- a/lib/mempool/rte_mempool.h
+++ b/lib/mempool/rte_mempool.h
@@ -1325,6 +1325,9 @@ void rte_mempool_stats_reset(struct rte_mempool *mp);
 /**
  * Dump the status of the mempool to a file.
  *
+ * For a mempool that is not populated yet, the mempool driver is not
+ * queried, and the common pool count is reported as 0.
+ *
  * @param f
  *   A pointer to a file for output
  * @param mp
@@ -1791,6 +1794,24 @@ rte_mempool_get_contig_blocks(struct rte_mempool *mp,
 	return ret;
 }
 
+/**
+ * Check if the mempool driver has created the store of a mempool.
+ *
+ * rte_mempool_create_empty() adds a mempool to the mempool list, and
+ * assigns the default mempool driver, before the store exists. The
+ * mempool driver can still be changed with rte_mempool_set_ops_byname()
+ * until the store is created by the first populate call. Functions that
+ * query the mempool driver, such as rte_mempool_avail_count(), must only
+ * be called for a mempool that is created. rte_mempool_walk() callbacks
+ * can see mempools that are not created yet.
+ *
+ * @param mp
+ *   A pointer to the mempool structure.
+ * @return
+ *   true if the store has been created, false otherwise.
+ */
+bool rte_mempool_is_created(const struct rte_mempool *mp);
+
 /**
  * Return the number of entries in the mempool.
  *
@@ -1950,6 +1971,11 @@ uint32_t rte_mempool_calc_obj_size(uint32_t elt_size, uint32_t flags,
  * The mempool list is locked while the iterator function runs, so the
  * iterator function must not create or free a mempool.
  *
+ * The iterator function is also called for mempools that are not
+ * populated yet. Such a mempool has no store yet, so the iterator function
+ * must check rte_mempool_is_created() before calling functions that query
+ * the mempool driver, such as rte_mempool_avail_count().
+ *
  * @param func
  *   Iterator function
  * @param arg
-- 
2.34.1