[dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Lazaros Koromilas]

The mempool cache is only available to EAL threads as a per-lcore
resource. Change this so that the user can create and provide their own
cache on mempool get and put operations. This works with non-EAL threads
too. This commit introduces the new API calls:

    rte_mempool_cache_create(size, socket_id)
    rte_mempool_cache_flush(cache, mp)
    rte_mempool_cache_free(cache)
    rte_mempool_default_cache(mp, lcore_id)
    rte_mempool_generic_put(mp, obj_table, n, cache, is_mp)
    rte_mempool_generic_get(mp, obj_table, n, cache, is_mc)

Removes the API calls:

    rte_mempool_sp_put_bulk(mp, obj_table, n)
    rte_mempool_sc_get_bulk(mp, obj_table, n)
    rte_mempool_sp_put(mp, obj)
    rte_mempool_sc_get(mp, obj)

And the remaining API calls use the per-lcore default local cache:

    rte_mempool_put_bulk(mp, obj_table, n)
    rte_mempool_get_bulk(mp, obj_table, n)
    rte_mempool_put(mp, obj)
    rte_mempool_get(mp, obj)

Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
---
 app/test/test_mempool.c                |  58 +++++--
 app/test/test_mempool_perf.c           |  46 +++++-
 lib/librte_eal/common/eal_common_log.c |   8 +-
 lib/librte_mempool/rte_mempool.c       |  76 ++++++++-
 lib/librte_mempool/rte_mempool.h       | 291 +++++++++++++--------------------
 5 files changed, 275 insertions(+), 204 deletions(-)

diff --git a/app/test/test_mempool.c b/app/test/test_mempool.c
index 10e1fa4..2dc0cf2 100644
--- a/app/test/test_mempool.c
+++ b/app/test/test_mempool.c
@@ -79,6 +79,7 @@
 
 static struct rte_mempool *mp;
 static struct rte_mempool *mp_cache, *mp_nocache;
+static int use_external_cache;
 
 static rte_atomic32_t synchro;
 
@@ -107,19 +108,33 @@ test_mempool_basic(void)
 	char *obj_data;
 	int ret = 0;
 	unsigned i, j;
+	struct rte_mempool_cache *cache;
+
+	if (use_external_cache)
+		/* Create a user-owned mempool cache. */
+		cache = rte_mempool_cache_create(RTE_MEMPOOL_CACHE_MAX_SIZE,
+						 SOCKET_ID_ANY);
+	else
+		cache = rte_mempool_default_cache(mp, rte_lcore_id());
 
 	/* dump the mempool status */
 	rte_mempool_dump(stdout, mp);
 
 	printf("get an object\n");
-	if (rte_mempool_get(mp, &obj) < 0)
+	if (rte_mempool_generic_get(mp, &obj, 1, cache, 1) < 0)
 		return -1;
 	rte_mempool_dump(stdout, mp);
 
 	/* tests that improve coverage */
 	printf("get object count\n");
-	if (rte_mempool_count(mp) != MEMPOOL_SIZE - 1)
-		return -1;
+	if (use_external_cache) {
+		/* We have to count the extra caches, one in this case. */
+		if (rte_mempool_count(mp) + cache->len != MEMPOOL_SIZE - 1)
+			return -1;
+	} else {
+		if (rte_mempool_count(mp) != MEMPOOL_SIZE - 1)
+			return -1;
+	}
 
 	printf("get private data\n");
 	if (rte_mempool_get_priv(mp) != (char *)mp +
@@ -134,21 +149,21 @@ test_mempool_basic(void)
 		return -1;
 
 	printf("put the object back\n");
-	rte_mempool_put(mp, obj);
+	rte_mempool_generic_put(mp, &obj, 1, cache, 1);
 	rte_mempool_dump(stdout, mp);
 
 	printf("get 2 objects\n");
-	if (rte_mempool_get(mp, &obj) < 0)
+	if (rte_mempool_generic_get(mp, &obj, 1, cache, 1) < 0)
 		return -1;
-	if (rte_mempool_get(mp, &obj2) < 0) {
-		rte_mempool_put(mp, obj);
+	if (rte_mempool_generic_get(mp, &obj2, 1, cache, 1) < 0) {
+		rte_mempool_generic_put(mp, &obj, 1, cache, 1);
 		return -1;
 	}
 	rte_mempool_dump(stdout, mp);
 
 	printf("put the objects back\n");
-	rte_mempool_put(mp, obj);
-	rte_mempool_put(mp, obj2);
+	rte_mempool_generic_put(mp, &obj, 1, cache, 1);
+	rte_mempool_generic_put(mp, &obj2, 1, cache, 1);
 	rte_mempool_dump(stdout, mp);
 
 	/*
@@ -161,7 +176,7 @@ test_mempool_basic(void)
 	}
 
 	for (i=0; i<MEMPOOL_SIZE; i++) {
-		if (rte_mempool_get(mp, &objtable[i]) < 0)
+		if (rte_mempool_generic_get(mp, &objtable[i], 1, cache, 1) < 0)
 			break;
 	}
 
@@ -183,13 +198,18 @@ test_mempool_basic(void)
 				ret = -1;
 		}
 
-		rte_mempool_put(mp, objtable[i]);
+		rte_mempool_generic_put(mp, &objtable[i], 1, cache, 1);
 	}
 
 	free(objtable);
 	if (ret == -1)
 		printf("objects were modified!\n");
 
+	if (use_external_cache) {
+		rte_mempool_cache_flush(cache, mp);
+		rte_mempool_cache_free(cache);
+	}
+
 	return ret;
 }
 
@@ -245,7 +265,7 @@ static int test_mempool_single_producer(void)
 			printf("test_mempool_single_producer there is an obj not owned by this mempool\n");
 			return -1;
 		}
-		rte_mempool_sp_put(mp_spsc, obj);
+		rte_mempool_put(mp_spsc, obj);
 		rte_spinlock_lock(&scsp_spinlock);
 		scsp_obj_table[i] = NULL;
 		rte_spinlock_unlock(&scsp_spinlock);
@@ -278,7 +298,7 @@ static int test_mempool_single_consumer(void)
 		rte_spinlock_unlock(&scsp_spinlock);
 		if (i >= MAX_KEEP)
 			continue;
-		if (rte_mempool_sc_get(mp_spsc, &obj) < 0)
+		if (rte_mempool_get(mp_spsc, &obj) < 0)
 			break;
 		rte_spinlock_lock(&scsp_spinlock);
 		scsp_obj_table[i] = obj;
@@ -371,12 +391,12 @@ test_mempool_basic_ex(struct rte_mempool * mp)
 	}
 
 	for (i = 0; i < MEMPOOL_SIZE; i ++) {
-		if (rte_mempool_mc_get(mp, &obj[i]) < 0) {
+		if (rte_mempool_get(mp, &obj[i]) < 0) {
 			printf("fail_mp_basic_ex fail to get mempool object for [%u]\n", i);
 			goto fail_mp_basic_ex;
 		}
 	}
-	if (rte_mempool_mc_get(mp, &err_obj) == 0) {
+	if (rte_mempool_get(mp, &err_obj) == 0) {
 		printf("test_mempool_basic_ex get an impossible obj from mempool\n");
 		goto fail_mp_basic_ex;
 	}
@@ -387,7 +407,7 @@ test_mempool_basic_ex(struct rte_mempool * mp)
 	}
 
 	for (i = 0; i < MEMPOOL_SIZE; i ++) {
-		rte_mempool_mp_put(mp, obj[i]);
+		rte_mempool_put(mp, obj[i]);
 	}
 	if (rte_mempool_full(mp) != 1) {
 		printf("test_mempool_basic_ex the mempool is not full but it should be\n");
@@ -499,6 +519,12 @@ test_mempool(void)
 	if (test_mempool_basic() < 0)
 		return -1;
 
+	/* basic tests with user-owned cache */
+	mp = mp_nocache;
+	use_external_cache = 1;
+	if (test_mempool_basic() < 0)
+		return -1;
+
 	/* more basic tests without cache */
 	if (test_mempool_basic_ex(mp_nocache) < 0)
 		return -1;
diff --git a/app/test/test_mempool_perf.c b/app/test/test_mempool_perf.c
index cdc02a0..e917f4d 100644
--- a/app/test/test_mempool_perf.c
+++ b/app/test/test_mempool_perf.c
@@ -78,6 +78,9 @@
  *      - One core without cache
  *      - Two cores without cache
  *      - Max. cores without cache
+ *      - One core with user-owned cache
+ *      - Two cores with user-owned cache
+ *      - Max. cores with user-owned cache
  *
  *    - Bulk size (*n_get_bulk*, *n_put_bulk*)
  *
@@ -98,6 +101,8 @@
 
 static struct rte_mempool *mp;
 static struct rte_mempool *mp_cache, *mp_nocache;
+static int use_external_cache;
+static unsigned external_cache_size = RTE_MEMPOOL_CACHE_MAX_SIZE;
 
 static rte_atomic32_t synchro;
 
@@ -137,6 +142,14 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
 	int ret;
 	uint64_t start_cycles, end_cycles;
 	uint64_t time_diff = 0, hz = rte_get_timer_hz();
+	struct rte_mempool_cache *cache;
+
+	if (use_external_cache)
+		/* Create a user-owned mempool cache. */
+		cache = rte_mempool_cache_create(external_cache_size,
+						 SOCKET_ID_ANY);
+	else
+		cache = rte_mempool_default_cache(mp, lcore_id);
 
 	/* n_get_bulk and n_put_bulk must be divisors of n_keep */
 	if (((n_keep / n_get_bulk) * n_get_bulk) != n_keep)
@@ -157,8 +170,9 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
 			/* get n_keep objects by bulk of n_bulk */
 			idx = 0;
 			while (idx < n_keep) {
-				ret = rte_mempool_get_bulk(mp, &obj_table[idx],
-							   n_get_bulk);
+				ret = rte_mempool_generic_get(mp, &obj_table[idx],
+							      n_get_bulk,
+							      cache, 1);
 				if (unlikely(ret < 0)) {
 					rte_mempool_dump(stdout, mp);
 					rte_ring_dump(stdout, mp->ring);
@@ -171,8 +185,9 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
 			/* put the objects back */
 			idx = 0;
 			while (idx < n_keep) {
-				rte_mempool_put_bulk(mp, &obj_table[idx],
-						     n_put_bulk);
+				rte_mempool_generic_put(mp, &obj_table[idx],
+							n_put_bulk,
+							cache, 1);
 				idx += n_put_bulk;
 			}
 		}
@@ -181,6 +196,11 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
 		stats[lcore_id].enq_count += N;
 	}
 
+	if (use_external_cache) {
+		rte_mempool_cache_flush(cache, mp);
+		rte_mempool_cache_free(cache);
+	}
+
 	return 0;
 }
 
@@ -200,7 +220,9 @@ launch_cores(unsigned cores)
 
 	printf("mempool_autotest cache=%u cores=%u n_get_bulk=%u "
 	       "n_put_bulk=%u n_keep=%u ",
-	       (unsigned) mp->cache_size, cores, n_get_bulk, n_put_bulk, n_keep);
+	       use_external_cache ?
+	           external_cache_size : (unsigned) mp->cache_size,
+	       cores, n_get_bulk, n_put_bulk, n_keep);
 
 	if (rte_mempool_count(mp) != MEMPOOL_SIZE) {
 		printf("mempool is not full\n");
@@ -324,6 +346,20 @@ test_mempool_perf(void)
 	if (do_one_mempool_test(rte_lcore_count()) < 0)
 		return -1;
 
+	/* performance test with 1, 2 and max cores */
+	printf("start performance test (with user-owned cache)\n");
+	mp = mp_nocache;
+	use_external_cache = 1;
+
+	if (do_one_mempool_test(1) < 0)
+		return -1;
+
+	if (do_one_mempool_test(2) < 0)
+		return -1;
+
+	if (do_one_mempool_test(rte_lcore_count()) < 0)
+		return -1;
+
 	rte_mempool_list_dump(stdout);
 
 	return 0;
diff --git a/lib/librte_eal/common/eal_common_log.c b/lib/librte_eal/common/eal_common_log.c
index 1ae8de7..07aee22 100644
--- a/lib/librte_eal/common/eal_common_log.c
+++ b/lib/librte_eal/common/eal_common_log.c
@@ -125,7 +125,7 @@ rte_log_add_in_history(const char *buf, size_t size)
 		}
 	}
 	else {
-		if (rte_mempool_mc_get(log_history_mp, &obj) < 0)
+		if (rte_mempool_get(log_history_mp, &obj) < 0)
 			obj = NULL;
 		hist_buf = obj;
 	}
@@ -138,7 +138,7 @@ rte_log_add_in_history(const char *buf, size_t size)
 
 	/* not enough room for msg, buffer go back in mempool */
 	if (size >= hist_buf_size) {
-		rte_mempool_mp_put(log_history_mp, hist_buf);
+		rte_mempool_put(log_history_mp, hist_buf);
 		rte_spinlock_unlock(&log_list_lock);
 		return -ENOBUFS;
 	}
@@ -252,12 +252,12 @@ rte_log_dump_history(FILE *out)
 
 		/* write on stdout */
 		if (fwrite(hist_buf->buf, hist_buf->size, 1, out) == 0) {
-			rte_mempool_mp_put(log_history_mp, hist_buf);
+			rte_mempool_put(log_history_mp, hist_buf);
 			break;
 		}
 
 		/* put back message structure in pool */
-		rte_mempool_mp_put(log_history_mp, hist_buf);
+		rte_mempool_put(log_history_mp, hist_buf);
 	}
 	fflush(out);
 
diff --git a/lib/librte_mempool/rte_mempool.c b/lib/librte_mempool/rte_mempool.c
index 73ca770..4d977c1 100644
--- a/lib/librte_mempool/rte_mempool.c
+++ b/lib/librte_mempool/rte_mempool.c
@@ -375,6 +375,63 @@ rte_mempool_xmem_usage(void *vaddr, uint32_t elt_num, size_t elt_sz,
 	return usz;
 }
 
+static void
+mempool_cache_init(struct rte_mempool_cache *cache, uint32_t size)
+{
+	cache->size = size;
+	cache->flushthresh = CALC_CACHE_FLUSHTHRESH(size);
+	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.
+ */
+struct rte_mempool_cache *
+rte_mempool_cache_create(uint32_t size, int socket_id)
+{
+	struct rte_mempool_cache *cache;
+
+	if (size > RTE_MEMPOOL_CACHE_MAX_SIZE) {
+		rte_errno = EINVAL;
+		return NULL;
+	}
+
+	cache = rte_zmalloc_socket("MEMPOOL_CACHE", sizeof(*cache),
+				   RTE_CACHE_LINE_SIZE, socket_id);
+	if (cache == NULL) {
+		RTE_LOG(ERR, MEMPOOL, "Cannot allocate mempool cache!\n");
+		rte_errno = ENOMEM;
+		return NULL;
+	}
+
+	mempool_cache_init(cache, size);
+
+	return cache;
+}
+
+/*
+ * Free a cache. It's the responsibility of the user to make sure that any
+ * remaining objects in the cache are flushed to the corresponding
+ * mempool.
+ */
+void
+rte_mempool_cache_free(struct rte_mempool_cache *cache)
+{
+	rte_free(cache);
+}
+
+/*
+ * Put all objects in the cache to the specified mempool's ring.
+ */
+void
+rte_mempool_cache_flush(struct rte_mempool_cache *cache,
+			struct rte_mempool *mp)
+{
+	rte_ring_enqueue_bulk(mp->ring, cache->objs, cache->len);
+}
+
 #ifndef RTE_LIBRTE_XEN_DOM0
 /* stub if DOM0 support not configured */
 struct rte_mempool *
@@ -448,6 +505,7 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
 	struct rte_mempool_objsz objsz;
 	void *startaddr;
 	int page_size = getpagesize();
+	unsigned lcore_id;
 
 	/* compilation-time checks */
 	RTE_BUILD_BUG_ON((sizeof(struct rte_mempool) &
@@ -583,8 +641,8 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
 	mp->elt_size = objsz.elt_size;
 	mp->header_size = objsz.header_size;
 	mp->trailer_size = objsz.trailer_size;
+	/* Size of default caches, zero means disabled. */
 	mp->cache_size = cache_size;
-	mp->cache_flushthresh = CALC_CACHE_FLUSHTHRESH(cache_size);
 	mp->private_data_size = private_data_size;
 
 	/*
@@ -594,6 +652,13 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
 	mp->local_cache = (struct rte_mempool_cache *)
 			((char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, 0));
 
+	/* Init all default caches. */
+	if (cache_size != 0) {
+		for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++)
+			mempool_cache_init(&mp->local_cache[lcore_id],
+					   cache_size);
+	}
+
 	/* calculate address of the first element for continuous mempool. */
 	obj = (char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, cache_size) +
 		private_data_size;
@@ -672,7 +737,7 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
 	unsigned count = 0;
 	unsigned cache_count;
 
-	fprintf(f, "  cache infos:\n");
+	fprintf(f, "  internal cache infos:\n");
 	fprintf(f, "    cache_size=%"PRIu32"\n", mp->cache_size);
 
 	if (mp->cache_size == 0)
@@ -680,7 +745,8 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
 
 	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
 		cache_count = mp->local_cache[lcore_id].len;
-		fprintf(f, "    cache_count[%u]=%u\n", lcore_id, cache_count);
+		fprintf(f, "    cache_count[%u]=%"PRIu32"\n",
+			lcore_id, cache_count);
 		count += cache_count;
 	}
 	fprintf(f, "    total_cache_count=%u\n", count);
@@ -760,7 +826,9 @@ mempool_audit_cache(const struct rte_mempool *mp)
 		return;
 
 	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
-		if (mp->local_cache[lcore_id].len > mp->cache_flushthresh) {
+		const struct rte_mempool_cache *cache;
+		cache = &mp->local_cache[lcore_id];
+		if (cache->len > cache->flushthresh) {
 			RTE_LOG(CRIT, MEMPOOL, "badness on cache[%u]\n",
 				lcore_id);
 			rte_panic("MEMPOOL: invalid cache len\n");
diff --git a/lib/librte_mempool/rte_mempool.h b/lib/librte_mempool/rte_mempool.h
index 8595e77..21d43e2 100644
--- a/lib/librte_mempool/rte_mempool.h
+++ b/lib/librte_mempool/rte_mempool.h
@@ -99,7 +99,9 @@ struct rte_mempool_debug_stats {
  * A structure that stores a per-core object cache.
  */
 struct rte_mempool_cache {
-	unsigned len; /**< Cache len */
+	uint32_t size;        /**< Size of the cache */
+	uint32_t flushthresh; /**< Threshold before we flush excess elements */
+	uint32_t len;         /**< Current cache count */
 	/*
 	 * Cache is allocated to this size to allow it to overflow in certain
 	 * cases to avoid needless emptying of cache.
@@ -182,9 +184,7 @@ struct rte_mempool {
 	phys_addr_t phys_addr;           /**< Phys. addr. of mempool struct. */
 	int flags;                       /**< Flags of the mempool. */
 	uint32_t size;                   /**< Size of the mempool. */
-	uint32_t cache_size;             /**< Size of per-lcore local cache. */
-	uint32_t cache_flushthresh;
-	/**< Threshold before we flush excess elements. */
+	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). */
@@ -742,6 +742,66 @@ rte_dom0_mempool_create(const char *name, unsigned n, unsigned elt_size,
 void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
 
 /**
+ * Create a user-owned mempool cache.
+ *
+ * This can be used by non-EAL threads to enable caching when they
+ * interact with a mempool.
+ *
+ * @param size
+ *   The size of the mempool cache. See rte_mempool_create()'s cache_size
+ *   parameter description for more information. The same limits and
+ *   considerations apply here too.
+ * @param socket_id
+ *   The socket identifier in the case of NUMA. The value can be
+ *   SOCKET_ID_ANY if there is no NUMA constraint for the reserved zone.
+ */
+struct rte_mempool_cache *
+rte_mempool_cache_create(uint32_t size, int socket_id);
+
+/**
+ * Free a user-owned mempool cache.
+ *
+ * @param cache
+ *   A pointer to the mempool cache.
+ */
+void
+rte_mempool_cache_free(struct rte_mempool_cache *cache);
+
+/**
+ * Flush a user-owned mempool cache to the specified mempool.
+ *
+ * @param cache
+ *   A pointer to the mempool cache.
+ * @param mp
+ *   A pointer to the mempool.
+ */
+void
+rte_mempool_cache_flush(struct rte_mempool_cache *cache,
+			struct rte_mempool *mp);
+
+/**
+ * Get a pointer to the per-lcore default mempool cache.
+ *
+ * @param mp
+ *   A pointer to the mempool structure.
+ * @param lcore_id
+ *   The logical core id.
+ * @return
+ *   A pointer to the mempool cache or NULL if disabled or non-EAL thread.
+ */
+static inline struct rte_mempool_cache *
+rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
+{
+	if (mp->cache_size == 0)
+		return NULL;
+
+	if (lcore_id >= RTE_MAX_LCORE)
+		return NULL;
+
+	return &mp->local_cache[lcore_id];
+}
+
+/**
  * @internal Put several objects back in the mempool; used internally.
  * @param mp
  *   A pointer to the mempool structure.
@@ -750,33 +810,29 @@ void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
  * @param n
  *   The number of objects to store back in the mempool, must be strictly
  *   positive.
+ * @param cache
+ *   A pointer to a mempool cache structure. May be NULL if not needed.
  * @param is_mp
  *   Mono-producer (0) or multi-producers (1).
  */
 static inline void __attribute__((always_inline))
-__mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
-		    unsigned n, int is_mp)
+__mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
+		      unsigned n, struct rte_mempool_cache *cache, int is_mp)
 {
-	struct rte_mempool_cache *cache;
 	uint32_t index;
 	void **cache_objs;
-	unsigned lcore_id = rte_lcore_id();
-	uint32_t cache_size = mp->cache_size;
-	uint32_t flushthresh = mp->cache_flushthresh;
 
 	/* increment stat now, adding in mempool always success */
 	__MEMPOOL_STAT_ADD(mp, put, n);
 
-	/* cache is not enabled or single producer or non-EAL thread */
-	if (unlikely(cache_size == 0 || is_mp == 0 ||
-		     lcore_id >= RTE_MAX_LCORE))
+	/* No cache provided or cache is not enabled or single producer */
+	if (unlikely(cache == NULL || cache->size == 0 || is_mp == 0))
 		goto ring_enqueue;
 
 	/* Go straight to ring if put would overflow mem allocated for cache */
 	if (unlikely(n > RTE_MEMPOOL_CACHE_MAX_SIZE))
 		goto ring_enqueue;
 
-	cache = &mp->local_cache[lcore_id];
 	cache_objs = &cache->objs[cache->len];
 
 	/*
@@ -792,10 +848,10 @@ __mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
 
 	cache->len += n;
 
-	if (cache->len >= flushthresh) {
-		rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache_size],
-				cache->len - cache_size);
-		cache->len = cache_size;
+	if (cache->len >= cache->flushthresh) {
+		rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache->size],
+				cache->len - cache->size);
+		cache->len = cache->size;
 	}
 
 	return;
@@ -820,41 +876,27 @@ ring_enqueue:
 #endif
 }
 
-
 /**
- * Put several objects back in the mempool (multi-producers safe).
+ * Put several objects back in the mempool.
  *
  * @param mp
  *   A pointer to the mempool structure.
  * @param obj_table
  *   A pointer to a table of void * pointers (objects).
  * @param n
- *   The number of objects to add in the mempool from the obj_table.
+ *   The number of objects to store back in the mempool, must be strictly
+ *   positive.
+ * @param cache
+ *   A pointer to a mempool cache structure. May be NULL if not needed.
+ * @param is_mp
+ *   Mono-producer (0) or multi-producers (1).
  */
 static inline void __attribute__((always_inline))
-rte_mempool_mp_put_bulk(struct rte_mempool *mp, void * const *obj_table,
-			unsigned n)
-{
-	__mempool_check_cookies(mp, obj_table, n, 0);
-	__mempool_put_bulk(mp, obj_table, n, 1);
-}
-
-/**
- * Put several objects back in the mempool (NOT multi-producers safe).
- *
- * @param mp
- *   A pointer to the mempool structure.
- * @param obj_table
- *   A pointer to a table of void * pointers (objects).
- * @param n
- *   The number of objects to add in the mempool from obj_table.
- */
-static inline void
-rte_mempool_sp_put_bulk(struct rte_mempool *mp, void * const *obj_table,
-			unsigned n)
+rte_mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
+			unsigned n, struct rte_mempool_cache *cache, int is_mp)
 {
 	__mempool_check_cookies(mp, obj_table, n, 0);
-	__mempool_put_bulk(mp, obj_table, n, 0);
+	__mempool_generic_put(mp, obj_table, n, cache, is_mp);
 }
 
 /**
@@ -875,36 +917,11 @@ static inline void __attribute__((always_inline))
 rte_mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
 		     unsigned n)
 {
-	__mempool_check_cookies(mp, obj_table, n, 0);
-	__mempool_put_bulk(mp, obj_table, n, !(mp->flags & MEMPOOL_F_SP_PUT));
-}
-
-/**
- * Put one object in the mempool (multi-producers safe).
- *
- * @param mp
- *   A pointer to the mempool structure.
- * @param obj
- *   A pointer to the object to be added.
- */
-static inline void __attribute__((always_inline))
-rte_mempool_mp_put(struct rte_mempool *mp, void *obj)
-{
-	rte_mempool_mp_put_bulk(mp, &obj, 1);
-}
+	struct rte_mempool_cache *cache;
 
-/**
- * Put one object back in the mempool (NOT multi-producers safe).
- *
- * @param mp
- *   A pointer to the mempool structure.
- * @param obj
- *   A pointer to the object to be added.
- */
-static inline void __attribute__((always_inline))
-rte_mempool_sp_put(struct rte_mempool *mp, void *obj)
-{
-	rte_mempool_sp_put_bulk(mp, &obj, 1);
+	cache = rte_mempool_default_cache(mp, rte_lcore_id());
+	rte_mempool_generic_put(mp, obj_table, n, cache,
+				!(mp->flags & MEMPOOL_F_SP_PUT));
 }
 
 /**
@@ -933,6 +950,8 @@ rte_mempool_put(struct rte_mempool *mp, void *obj)
  *   A pointer to a table of void * pointers (objects).
  * @param n
  *   The number of objects to get, must be strictly positive.
+ * @param cache
+ *   A pointer to a mempool cache structure. May be NULL if not needed.
  * @param is_mc
  *   Mono-consumer (0) or multi-consumers (1).
  * @return
@@ -940,28 +959,24 @@ rte_mempool_put(struct rte_mempool *mp, void *obj)
  *   - <0: Error; code of ring dequeue function.
  */
 static inline int __attribute__((always_inline))
-__mempool_get_bulk(struct rte_mempool *mp, void **obj_table,
-		   unsigned n, int is_mc)
+__mempool_generic_get(struct rte_mempool *mp, void **obj_table,
+		      unsigned n, struct rte_mempool_cache *cache, int is_mc)
 {
 	int ret;
-	struct rte_mempool_cache *cache;
 	uint32_t index, len;
 	void **cache_objs;
-	unsigned lcore_id = rte_lcore_id();
-	uint32_t cache_size = mp->cache_size;
 
-	/* cache is not enabled or single consumer */
-	if (unlikely(cache_size == 0 || is_mc == 0 ||
-		     n >= cache_size || lcore_id >= RTE_MAX_LCORE))
+	/* No cache provided or cache is not enabled or single consumer */
+	if (unlikely(cache == NULL || cache->size == 0 || is_mc == 0 ||
+		     n >= cache->size))
 		goto ring_dequeue;
 
-	cache = &mp->local_cache[lcore_id];
 	cache_objs = cache->objs;
 
 	/* Can this be satisfied from the cache? */
 	if (cache->len < n) {
 		/* No. Backfill the cache first, and then fill from it */
-		uint32_t req = n + (cache_size - cache->len);
+		uint32_t req = n + (cache->size - cache->len);
 
 		/* How many do we require i.e. number to fill the cache + the request */
 		ret = rte_ring_mc_dequeue_bulk(mp->ring, &cache->objs[cache->len], req);
@@ -1005,57 +1020,28 @@ ring_dequeue:
 }
 
 /**
- * Get several objects from the mempool (multi-consumers safe).
- *
- * If cache is enabled, objects will be retrieved first from cache,
- * subsequently from the common pool. Note that it can return -ENOENT when
- * the local cache and common pool are empty, even if cache from other
- * lcores are full.
- *
- * @param mp
- *   A pointer to the mempool structure.
- * @param obj_table
- *   A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- *   The number of objects to get from mempool to obj_table.
- * @return
- *   - 0: Success; objects taken.
- *   - -ENOENT: Not enough entries in the mempool; no object is retrieved.
- */
-static inline int __attribute__((always_inline))
-rte_mempool_mc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
-{
-	int ret;
-	ret = __mempool_get_bulk(mp, obj_table, n, 1);
-	if (ret == 0)
-		__mempool_check_cookies(mp, obj_table, n, 1);
-	return ret;
-}
-
-/**
- * Get several objects from the mempool (NOT multi-consumers safe).
- *
- * If cache is enabled, objects will be retrieved first from cache,
- * subsequently from the common pool. Note that it can return -ENOENT when
- * the local cache and common pool are empty, even if cache from other
- * lcores are full.
+ * Get several objects from the mempool.
  *
  * @param mp
  *   A pointer to the mempool structure.
  * @param obj_table
- *   A pointer to a table of void * pointers (objects) that will be filled.
+ *   A pointer to a table of void * pointers (objects).
  * @param n
- *   The number of objects to get from the mempool to obj_table.
+ *   The number of objects to get, must be strictly positive.
+ * @param cache
+ *   A pointer to a mempool cache structure. May be NULL if not needed.
+ * @param is_mc
+ *   Mono-consumer (0) or multi-consumers (1).
  * @return
- *   - 0: Success; objects taken.
- *   - -ENOENT: Not enough entries in the mempool; no object is
- *     retrieved.
+ *   - >=0: Success; number of objects supplied.
+ *   - <0: Error; code of ring dequeue function.
  */
 static inline int __attribute__((always_inline))
-rte_mempool_sc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
+rte_mempool_generic_get(struct rte_mempool *mp, void **obj_table,
+			unsigned n, struct rte_mempool_cache *cache, int is_mc)
 {
 	int ret;
-	ret = __mempool_get_bulk(mp, obj_table, n, 0);
+	ret = __mempool_generic_get(mp, obj_table, n, cache, is_mc);
 	if (ret == 0)
 		__mempool_check_cookies(mp, obj_table, n, 1);
 	return ret;
@@ -1086,56 +1072,11 @@ rte_mempool_sc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
 static inline int __attribute__((always_inline))
 rte_mempool_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
 {
-	int ret;
-	ret = __mempool_get_bulk(mp, obj_table, n,
-				 !(mp->flags & MEMPOOL_F_SC_GET));
-	if (ret == 0)
-		__mempool_check_cookies(mp, obj_table, n, 1);
-	return ret;
-}
-
-/**
- * Get one object from the mempool (multi-consumers safe).
- *
- * If cache is enabled, objects will be retrieved first from cache,
- * subsequently from the common pool. Note that it can return -ENOENT when
- * the local cache and common pool are empty, even if cache from other
- * lcores are full.
- *
- * @param mp
- *   A pointer to the mempool structure.
- * @param obj_p
- *   A pointer to a void * pointer (object) that will be filled.
- * @return
- *   - 0: Success; objects taken.
- *   - -ENOENT: Not enough entries in the mempool; no object is retrieved.
- */
-static inline int __attribute__((always_inline))
-rte_mempool_mc_get(struct rte_mempool *mp, void **obj_p)
-{
-	return rte_mempool_mc_get_bulk(mp, obj_p, 1);
-}
+	struct rte_mempool_cache *cache;
 
-/**
- * Get one object from the mempool (NOT multi-consumers safe).
- *
- * If cache is enabled, objects will be retrieved first from cache,
- * subsequently from the common pool. Note that it can return -ENOENT when
- * the local cache and common pool are empty, even if cache from other
- * lcores are full.
- *
- * @param mp
- *   A pointer to the mempool structure.
- * @param obj_p
- *   A pointer to a void * pointer (object) that will be filled.
- * @return
- *   - 0: Success; objects taken.
- *   - -ENOENT: Not enough entries in the mempool; no object is retrieved.
- */
-static inline int __attribute__((always_inline))
-rte_mempool_sc_get(struct rte_mempool *mp, void **obj_p)
-{
-	return rte_mempool_sc_get_bulk(mp, obj_p, 1);
+	cache = rte_mempool_default_cache(mp, rte_lcore_id());
+	return rte_mempool_generic_get(mp, obj_table, n, cache,
+				       !(mp->flags & MEMPOOL_F_SC_GET));
 }
 
 /**
@@ -1169,7 +1110,7 @@ rte_mempool_get(struct rte_mempool *mp, void **obj_p)
  *
  * When cache is enabled, this function has to browse the length of
  * all lcores, so it should not be used in a data path, but only for
- * debug purposes.
+ * debug purposes. User-owned mempool caches are not accounted for.
  *
  * @param mp
  *   A pointer to the mempool structure.
@@ -1188,7 +1129,7 @@ unsigned rte_mempool_count(const struct rte_mempool *mp);
  *
  * When cache is enabled, this function has to browse the length of
  * all lcores, so it should not be used in a data path, but only for
- * debug purposes.
+ * debug purposes. User-owned mempool caches are not accounted for.
  *
  * @param mp
  *   A pointer to the mempool structure.
@@ -1206,7 +1147,7 @@ rte_mempool_free_count(const struct rte_mempool *mp)
  *
  * When cache is enabled, this function has to browse the length of all
  * lcores, so it should not be used in a data path, but only for debug
- * purposes.
+ * purposes. User-owned mempool caches are not accounted for.
  *
  * @param mp
  *   A pointer to the mempool structure.
@@ -1225,7 +1166,7 @@ rte_mempool_full(const struct rte_mempool *mp)
  *
  * When cache is enabled, this function has to browse the length of all
  * lcores, so it should not be used in a data path, but only for debug
- * purposes.
+ * purposes. User-owned mempool caches are not accounted for.
  *
  * @param mp
  *   A pointer to the mempool structure.
-- 
1.9.1

[dpdk-dev] [PATCH v2 2/2] mempool: use bit flags instead of is_mp and is_mc

[Lazaros Koromilas]

Pass the same flags as in rte_mempool_create().

Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
---
 app/test/test_mempool.c          | 18 +++++------
 app/test/test_mempool_perf.c     |  4 +--
 lib/librte_mempool/rte_mempool.h | 66 +++++++++++++++++++++-------------------
 3 files changed, 45 insertions(+), 43 deletions(-)

diff --git a/app/test/test_mempool.c b/app/test/test_mempool.c
index 2dc0cf2..445f450 100644
--- a/app/test/test_mempool.c
+++ b/app/test/test_mempool.c
@@ -121,7 +121,7 @@ test_mempool_basic(void)
 	rte_mempool_dump(stdout, mp);
 
 	printf("get an object\n");
-	if (rte_mempool_generic_get(mp, &obj, 1, cache, 1) < 0)
+	if (rte_mempool_generic_get(mp, &obj, 1, cache, 0) < 0)
 		return -1;
 	rte_mempool_dump(stdout, mp);
 
@@ -149,21 +149,21 @@ test_mempool_basic(void)
 		return -1;
 
 	printf("put the object back\n");
-	rte_mempool_generic_put(mp, &obj, 1, cache, 1);
+	rte_mempool_generic_put(mp, &obj, 1, cache, 0);
 	rte_mempool_dump(stdout, mp);
 
 	printf("get 2 objects\n");
-	if (rte_mempool_generic_get(mp, &obj, 1, cache, 1) < 0)
+	if (rte_mempool_generic_get(mp, &obj, 1, cache, 0) < 0)
 		return -1;
-	if (rte_mempool_generic_get(mp, &obj2, 1, cache, 1) < 0) {
-		rte_mempool_generic_put(mp, &obj, 1, cache, 1);
+	if (rte_mempool_generic_get(mp, &obj2, 1, cache, 0) < 0) {
+		rte_mempool_generic_put(mp, &obj, 1, cache, 0);
 		return -1;
 	}
 	rte_mempool_dump(stdout, mp);
 
 	printf("put the objects back\n");
-	rte_mempool_generic_put(mp, &obj, 1, cache, 1);
-	rte_mempool_generic_put(mp, &obj2, 1, cache, 1);
+	rte_mempool_generic_put(mp, &obj, 1, cache, 0);
+	rte_mempool_generic_put(mp, &obj2, 1, cache, 0);
 	rte_mempool_dump(stdout, mp);
 
 	/*
@@ -176,7 +176,7 @@ test_mempool_basic(void)
 	}
 
 	for (i=0; i<MEMPOOL_SIZE; i++) {
-		if (rte_mempool_generic_get(mp, &objtable[i], 1, cache, 1) < 0)
+		if (rte_mempool_generic_get(mp, &objtable[i], 1, cache, 0) < 0)
 			break;
 	}
 
@@ -198,7 +198,7 @@ test_mempool_basic(void)
 				ret = -1;
 		}
 
-		rte_mempool_generic_put(mp, &objtable[i], 1, cache, 1);
+		rte_mempool_generic_put(mp, &objtable[i], 1, cache, 0);
 	}
 
 	free(objtable);
diff --git a/app/test/test_mempool_perf.c b/app/test/test_mempool_perf.c
index e917f4d..fa7e817 100644
--- a/app/test/test_mempool_perf.c
+++ b/app/test/test_mempool_perf.c
@@ -172,7 +172,7 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
 			while (idx < n_keep) {
 				ret = rte_mempool_generic_get(mp, &obj_table[idx],
 							      n_get_bulk,
-							      cache, 1);
+							      cache, 0);
 				if (unlikely(ret < 0)) {
 					rte_mempool_dump(stdout, mp);
 					rte_ring_dump(stdout, mp->ring);
@@ -187,7 +187,7 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
 			while (idx < n_keep) {
 				rte_mempool_generic_put(mp, &obj_table[idx],
 							n_put_bulk,
-							cache, 1);
+							cache, 0);
 				idx += n_put_bulk;
 			}
 		}
diff --git a/lib/librte_mempool/rte_mempool.h b/lib/librte_mempool/rte_mempool.h
index 21d43e2..fe4fed9 100644
--- a/lib/librte_mempool/rte_mempool.h
+++ b/lib/librte_mempool/rte_mempool.h
@@ -812,12 +812,13 @@ rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
  *   positive.
  * @param cache
  *   A pointer to a mempool cache structure. May be NULL if not needed.
- * @param is_mp
- *   Mono-producer (0) or multi-producers (1).
+ * @param flags
+ *   The flags used for the mempool creation.
+ *   Single-producer (MEMPOOL_F_SP_PUT flag) or multi-producers.
  */
 static inline void __attribute__((always_inline))
 __mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
-		      unsigned n, struct rte_mempool_cache *cache, int is_mp)
+		      unsigned n, struct rte_mempool_cache *cache, int flags)
 {
 	uint32_t index;
 	void **cache_objs;
@@ -826,7 +827,8 @@ __mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
 	__MEMPOOL_STAT_ADD(mp, put, n);
 
 	/* No cache provided or cache is not enabled or single producer */
-	if (unlikely(cache == NULL || cache->size == 0 || is_mp == 0))
+	if (unlikely(cache == NULL || cache->size == 0 ||
+		     flags & MEMPOOL_F_SP_PUT))
 		goto ring_enqueue;
 
 	/* Go straight to ring if put would overflow mem allocated for cache */
@@ -860,19 +862,18 @@ ring_enqueue:
 
 	/* push remaining objects in ring */
 #ifdef RTE_LIBRTE_MEMPOOL_DEBUG
-	if (is_mp) {
-		if (rte_ring_mp_enqueue_bulk(mp->ring, obj_table, n) < 0)
-			rte_panic("cannot put objects in mempool\n");
-	}
-	else {
+	if (flags & MEMPOOL_F_SP_PUT) {
 		if (rte_ring_sp_enqueue_bulk(mp->ring, obj_table, n) < 0)
 			rte_panic("cannot put objects in mempool\n");
+	} else {
+		if (rte_ring_mp_enqueue_bulk(mp->ring, obj_table, n) < 0)
+			rte_panic("cannot put objects in mempool\n");
 	}
 #else
-	if (is_mp)
-		rte_ring_mp_enqueue_bulk(mp->ring, obj_table, n);
-	else
+	if (flags & MEMPOOL_F_SP_PUT)
 		rte_ring_sp_enqueue_bulk(mp->ring, obj_table, n);
+	else
+		rte_ring_mp_enqueue_bulk(mp->ring, obj_table, n);
 #endif
 }
 
@@ -888,15 +889,16 @@ ring_enqueue:
  *   positive.
  * @param cache
  *   A pointer to a mempool cache structure. May be NULL if not needed.
- * @param is_mp
- *   Mono-producer (0) or multi-producers (1).
+ * @param flags
+ *   The flags used for the mempool creation.
+ *   Single-producer (MEMPOOL_F_SP_PUT flag) or multi-producers.
  */
 static inline void __attribute__((always_inline))
 rte_mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
-			unsigned n, struct rte_mempool_cache *cache, int is_mp)
+			unsigned n, struct rte_mempool_cache *cache, int flags)
 {
 	__mempool_check_cookies(mp, obj_table, n, 0);
-	__mempool_generic_put(mp, obj_table, n, cache, is_mp);
+	__mempool_generic_put(mp, obj_table, n, cache, flags);
 }
 
 /**
@@ -920,8 +922,7 @@ rte_mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
 	struct rte_mempool_cache *cache;
 
 	cache = rte_mempool_default_cache(mp, rte_lcore_id());
-	rte_mempool_generic_put(mp, obj_table, n, cache,
-				!(mp->flags & MEMPOOL_F_SP_PUT));
+	rte_mempool_generic_put(mp, obj_table, n, cache, mp->flags);
 }
 
 /**
@@ -952,23 +953,24 @@ rte_mempool_put(struct rte_mempool *mp, void *obj)
  *   The number of objects to get, must be strictly positive.
  * @param cache
  *   A pointer to a mempool cache structure. May be NULL if not needed.
- * @param is_mc
- *   Mono-consumer (0) or multi-consumers (1).
+ * @param flags
+ *   The flags used for the mempool creation.
+ *   Single-consumer (MEMPOOL_F_SC_GET flag) or multi-consumers.
  * @return
  *   - >=0: Success; number of objects supplied.
  *   - <0: Error; code of ring dequeue function.
  */
 static inline int __attribute__((always_inline))
 __mempool_generic_get(struct rte_mempool *mp, void **obj_table,
-		      unsigned n, struct rte_mempool_cache *cache, int is_mc)
+		      unsigned n, struct rte_mempool_cache *cache, int flags)
 {
 	int ret;
 	uint32_t index, len;
 	void **cache_objs;
 
 	/* No cache provided or cache is not enabled or single consumer */
-	if (unlikely(cache == NULL || cache->size == 0 || is_mc == 0 ||
-		     n >= cache->size))
+	if (unlikely(cache == NULL || cache->size == 0 ||
+		     flags & MEMPOOL_F_SC_GET || n >= cache->size))
 		goto ring_dequeue;
 
 	cache_objs = cache->objs;
@@ -1006,10 +1008,10 @@ __mempool_generic_get(struct rte_mempool *mp, void **obj_table,
 ring_dequeue:
 
 	/* get remaining objects from ring */
-	if (is_mc)
-		ret = rte_ring_mc_dequeue_bulk(mp->ring, obj_table, n);
-	else
+	if (flags & MEMPOOL_F_SC_GET)
 		ret = rte_ring_sc_dequeue_bulk(mp->ring, obj_table, n);
+	else
+		ret = rte_ring_mc_dequeue_bulk(mp->ring, obj_table, n);
 
 	if (ret < 0)
 		__MEMPOOL_STAT_ADD(mp, get_fail, n);
@@ -1030,18 +1032,19 @@ ring_dequeue:
  *   The number of objects to get, must be strictly positive.
  * @param cache
  *   A pointer to a mempool cache structure. May be NULL if not needed.
- * @param is_mc
- *   Mono-consumer (0) or multi-consumers (1).
+ * @param flags
+ *   The flags used for the mempool creation.
+ *   Single-consumer (MEMPOOL_F_SC_GET flag) or multi-consumers.
  * @return
  *   - >=0: Success; number of objects supplied.
  *   - <0: Error; code of ring dequeue function.
  */
 static inline int __attribute__((always_inline))
 rte_mempool_generic_get(struct rte_mempool *mp, void **obj_table,
-			unsigned n, struct rte_mempool_cache *cache, int is_mc)
+			unsigned n, struct rte_mempool_cache *cache, int flags)
 {
 	int ret;
-	ret = __mempool_generic_get(mp, obj_table, n, cache, is_mc);
+	ret = __mempool_generic_get(mp, obj_table, n, cache, flags);
 	if (ret == 0)
 		__mempool_check_cookies(mp, obj_table, n, 1);
 	return ret;
@@ -1075,8 +1078,7 @@ rte_mempool_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
 	struct rte_mempool_cache *cache;
 
 	cache = rte_mempool_default_cache(mp, rte_lcore_id());
-	return rte_mempool_generic_get(mp, obj_table, n, cache,
-				       !(mp->flags & MEMPOOL_F_SC_GET));
+	return rte_mempool_generic_get(mp, obj_table, n, cache, mp->flags);
 }
 
 /**
-- 
1.9.1

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Lazaros Koromilas]

Hi all,

I forgot to mention that this series applies on top of:

http://www.dpdk.org/dev/patchwork/patch/10492/

Thanks,
Lazaros.

On Mon, Apr 4, 2016 at 6:43 PM, Lazaros Koromilas <l@nofutznetworks.com> wrote:
> The mempool cache is only available to EAL threads as a per-lcore
> resource. Change this so that the user can create and provide their own
> cache on mempool get and put operations. This works with non-EAL threads
> too. This commit introduces the new API calls:
>
>     rte_mempool_cache_create(size, socket_id)
>     rte_mempool_cache_flush(cache, mp)
>     rte_mempool_cache_free(cache)
>     rte_mempool_default_cache(mp, lcore_id)
>     rte_mempool_generic_put(mp, obj_table, n, cache, is_mp)
>     rte_mempool_generic_get(mp, obj_table, n, cache, is_mc)
>
> Removes the API calls:
>
>     rte_mempool_sp_put_bulk(mp, obj_table, n)
>     rte_mempool_sc_get_bulk(mp, obj_table, n)
>     rte_mempool_sp_put(mp, obj)
>     rte_mempool_sc_get(mp, obj)
>
> And the remaining API calls use the per-lcore default local cache:
>
>     rte_mempool_put_bulk(mp, obj_table, n)
>     rte_mempool_get_bulk(mp, obj_table, n)
>     rte_mempool_put(mp, obj)
>     rte_mempool_get(mp, obj)
>
> Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
> ---
>  app/test/test_mempool.c                |  58 +++++--
>  app/test/test_mempool_perf.c           |  46 +++++-
>  lib/librte_eal/common/eal_common_log.c |   8 +-
>  lib/librte_mempool/rte_mempool.c       |  76 ++++++++-
>  lib/librte_mempool/rte_mempool.h       | 291 +++++++++++++--------------------
>  5 files changed, 275 insertions(+), 204 deletions(-)
>
> diff --git a/app/test/test_mempool.c b/app/test/test_mempool.c
> index 10e1fa4..2dc0cf2 100644
> --- a/app/test/test_mempool.c
> +++ b/app/test/test_mempool.c
> @@ -79,6 +79,7 @@
>
>  static struct rte_mempool *mp;
>  static struct rte_mempool *mp_cache, *mp_nocache;
> +static int use_external_cache;
>
>  static rte_atomic32_t synchro;
>
> @@ -107,19 +108,33 @@ test_mempool_basic(void)
>         char *obj_data;
>         int ret = 0;
>         unsigned i, j;
> +       struct rte_mempool_cache *cache;
> +
> +       if (use_external_cache)
> +               /* Create a user-owned mempool cache. */
> +               cache = rte_mempool_cache_create(RTE_MEMPOOL_CACHE_MAX_SIZE,
> +                                                SOCKET_ID_ANY);
> +       else
> +               cache = rte_mempool_default_cache(mp, rte_lcore_id());
>
>         /* dump the mempool status */
>         rte_mempool_dump(stdout, mp);
>
>         printf("get an object\n");
> -       if (rte_mempool_get(mp, &obj) < 0)
> +       if (rte_mempool_generic_get(mp, &obj, 1, cache, 1) < 0)
>                 return -1;
>         rte_mempool_dump(stdout, mp);
>
>         /* tests that improve coverage */
>         printf("get object count\n");
> -       if (rte_mempool_count(mp) != MEMPOOL_SIZE - 1)
> -               return -1;
> +       if (use_external_cache) {
> +               /* We have to count the extra caches, one in this case. */
> +               if (rte_mempool_count(mp) + cache->len != MEMPOOL_SIZE - 1)
> +                       return -1;
> +       } else {
> +               if (rte_mempool_count(mp) != MEMPOOL_SIZE - 1)
> +                       return -1;
> +       }
>
>         printf("get private data\n");
>         if (rte_mempool_get_priv(mp) != (char *)mp +
> @@ -134,21 +149,21 @@ test_mempool_basic(void)
>                 return -1;
>
>         printf("put the object back\n");
> -       rte_mempool_put(mp, obj);
> +       rte_mempool_generic_put(mp, &obj, 1, cache, 1);
>         rte_mempool_dump(stdout, mp);
>
>         printf("get 2 objects\n");
> -       if (rte_mempool_get(mp, &obj) < 0)
> +       if (rte_mempool_generic_get(mp, &obj, 1, cache, 1) < 0)
>                 return -1;
> -       if (rte_mempool_get(mp, &obj2) < 0) {
> -               rte_mempool_put(mp, obj);
> +       if (rte_mempool_generic_get(mp, &obj2, 1, cache, 1) < 0) {
> +               rte_mempool_generic_put(mp, &obj, 1, cache, 1);
>                 return -1;
>         }
>         rte_mempool_dump(stdout, mp);
>
>         printf("put the objects back\n");
> -       rte_mempool_put(mp, obj);
> -       rte_mempool_put(mp, obj2);
> +       rte_mempool_generic_put(mp, &obj, 1, cache, 1);
> +       rte_mempool_generic_put(mp, &obj2, 1, cache, 1);
>         rte_mempool_dump(stdout, mp);
>
>         /*
> @@ -161,7 +176,7 @@ test_mempool_basic(void)
>         }
>
>         for (i=0; i<MEMPOOL_SIZE; i++) {
> -               if (rte_mempool_get(mp, &objtable[i]) < 0)
> +               if (rte_mempool_generic_get(mp, &objtable[i], 1, cache, 1) < 0)
>                         break;
>         }
>
> @@ -183,13 +198,18 @@ test_mempool_basic(void)
>                                 ret = -1;
>                 }
>
> -               rte_mempool_put(mp, objtable[i]);
> +               rte_mempool_generic_put(mp, &objtable[i], 1, cache, 1);
>         }
>
>         free(objtable);
>         if (ret == -1)
>                 printf("objects were modified!\n");
>
> +       if (use_external_cache) {
> +               rte_mempool_cache_flush(cache, mp);
> +               rte_mempool_cache_free(cache);
> +       }
> +
>         return ret;
>  }
>
> @@ -245,7 +265,7 @@ static int test_mempool_single_producer(void)
>                         printf("test_mempool_single_producer there is an obj not owned by this mempool\n");
>                         return -1;
>                 }
> -               rte_mempool_sp_put(mp_spsc, obj);
> +               rte_mempool_put(mp_spsc, obj);
>                 rte_spinlock_lock(&scsp_spinlock);
>                 scsp_obj_table[i] = NULL;
>                 rte_spinlock_unlock(&scsp_spinlock);
> @@ -278,7 +298,7 @@ static int test_mempool_single_consumer(void)
>                 rte_spinlock_unlock(&scsp_spinlock);
>                 if (i >= MAX_KEEP)
>                         continue;
> -               if (rte_mempool_sc_get(mp_spsc, &obj) < 0)
> +               if (rte_mempool_get(mp_spsc, &obj) < 0)
>                         break;
>                 rte_spinlock_lock(&scsp_spinlock);
>                 scsp_obj_table[i] = obj;
> @@ -371,12 +391,12 @@ test_mempool_basic_ex(struct rte_mempool * mp)
>         }
>
>         for (i = 0; i < MEMPOOL_SIZE; i ++) {
> -               if (rte_mempool_mc_get(mp, &obj[i]) < 0) {
> +               if (rte_mempool_get(mp, &obj[i]) < 0) {
>                         printf("fail_mp_basic_ex fail to get mempool object for [%u]\n", i);
>                         goto fail_mp_basic_ex;
>                 }
>         }
> -       if (rte_mempool_mc_get(mp, &err_obj) == 0) {
> +       if (rte_mempool_get(mp, &err_obj) == 0) {
>                 printf("test_mempool_basic_ex get an impossible obj from mempool\n");
>                 goto fail_mp_basic_ex;
>         }
> @@ -387,7 +407,7 @@ test_mempool_basic_ex(struct rte_mempool * mp)
>         }
>
>         for (i = 0; i < MEMPOOL_SIZE; i ++) {
> -               rte_mempool_mp_put(mp, obj[i]);
> +               rte_mempool_put(mp, obj[i]);
>         }
>         if (rte_mempool_full(mp) != 1) {
>                 printf("test_mempool_basic_ex the mempool is not full but it should be\n");
> @@ -499,6 +519,12 @@ test_mempool(void)
>         if (test_mempool_basic() < 0)
>                 return -1;
>
> +       /* basic tests with user-owned cache */
> +       mp = mp_nocache;
> +       use_external_cache = 1;
> +       if (test_mempool_basic() < 0)
> +               return -1;
> +
>         /* more basic tests without cache */
>         if (test_mempool_basic_ex(mp_nocache) < 0)
>                 return -1;
> diff --git a/app/test/test_mempool_perf.c b/app/test/test_mempool_perf.c
> index cdc02a0..e917f4d 100644
> --- a/app/test/test_mempool_perf.c
> +++ b/app/test/test_mempool_perf.c
> @@ -78,6 +78,9 @@
>   *      - One core without cache
>   *      - Two cores without cache
>   *      - Max. cores without cache
> + *      - One core with user-owned cache
> + *      - Two cores with user-owned cache
> + *      - Max. cores with user-owned cache
>   *
>   *    - Bulk size (*n_get_bulk*, *n_put_bulk*)
>   *
> @@ -98,6 +101,8 @@
>
>  static struct rte_mempool *mp;
>  static struct rte_mempool *mp_cache, *mp_nocache;
> +static int use_external_cache;
> +static unsigned external_cache_size = RTE_MEMPOOL_CACHE_MAX_SIZE;
>
>  static rte_atomic32_t synchro;
>
> @@ -137,6 +142,14 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
>         int ret;
>         uint64_t start_cycles, end_cycles;
>         uint64_t time_diff = 0, hz = rte_get_timer_hz();
> +       struct rte_mempool_cache *cache;
> +
> +       if (use_external_cache)
> +               /* Create a user-owned mempool cache. */
> +               cache = rte_mempool_cache_create(external_cache_size,
> +                                                SOCKET_ID_ANY);
> +       else
> +               cache = rte_mempool_default_cache(mp, lcore_id);
>
>         /* n_get_bulk and n_put_bulk must be divisors of n_keep */
>         if (((n_keep / n_get_bulk) * n_get_bulk) != n_keep)
> @@ -157,8 +170,9 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
>                         /* get n_keep objects by bulk of n_bulk */
>                         idx = 0;
>                         while (idx < n_keep) {
> -                               ret = rte_mempool_get_bulk(mp, &obj_table[idx],
> -                                                          n_get_bulk);
> +                               ret = rte_mempool_generic_get(mp, &obj_table[idx],
> +                                                             n_get_bulk,
> +                                                             cache, 1);
>                                 if (unlikely(ret < 0)) {
>                                         rte_mempool_dump(stdout, mp);
>                                         rte_ring_dump(stdout, mp->ring);
> @@ -171,8 +185,9 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
>                         /* put the objects back */
>                         idx = 0;
>                         while (idx < n_keep) {
> -                               rte_mempool_put_bulk(mp, &obj_table[idx],
> -                                                    n_put_bulk);
> +                               rte_mempool_generic_put(mp, &obj_table[idx],
> +                                                       n_put_bulk,
> +                                                       cache, 1);
>                                 idx += n_put_bulk;
>                         }
>                 }
> @@ -181,6 +196,11 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
>                 stats[lcore_id].enq_count += N;
>         }
>
> +       if (use_external_cache) {
> +               rte_mempool_cache_flush(cache, mp);
> +               rte_mempool_cache_free(cache);
> +       }
> +
>         return 0;
>  }
>
> @@ -200,7 +220,9 @@ launch_cores(unsigned cores)
>
>         printf("mempool_autotest cache=%u cores=%u n_get_bulk=%u "
>                "n_put_bulk=%u n_keep=%u ",
> -              (unsigned) mp->cache_size, cores, n_get_bulk, n_put_bulk, n_keep);
> +              use_external_cache ?
> +                  external_cache_size : (unsigned) mp->cache_size,
> +              cores, n_get_bulk, n_put_bulk, n_keep);
>
>         if (rte_mempool_count(mp) != MEMPOOL_SIZE) {
>                 printf("mempool is not full\n");
> @@ -324,6 +346,20 @@ test_mempool_perf(void)
>         if (do_one_mempool_test(rte_lcore_count()) < 0)
>                 return -1;
>
> +       /* performance test with 1, 2 and max cores */
> +       printf("start performance test (with user-owned cache)\n");
> +       mp = mp_nocache;
> +       use_external_cache = 1;
> +
> +       if (do_one_mempool_test(1) < 0)
> +               return -1;
> +
> +       if (do_one_mempool_test(2) < 0)
> +               return -1;
> +
> +       if (do_one_mempool_test(rte_lcore_count()) < 0)
> +               return -1;
> +
>         rte_mempool_list_dump(stdout);
>
>         return 0;
> diff --git a/lib/librte_eal/common/eal_common_log.c b/lib/librte_eal/common/eal_common_log.c
> index 1ae8de7..07aee22 100644
> --- a/lib/librte_eal/common/eal_common_log.c
> +++ b/lib/librte_eal/common/eal_common_log.c
> @@ -125,7 +125,7 @@ rte_log_add_in_history(const char *buf, size_t size)
>                 }
>         }
>         else {
> -               if (rte_mempool_mc_get(log_history_mp, &obj) < 0)
> +               if (rte_mempool_get(log_history_mp, &obj) < 0)
>                         obj = NULL;
>                 hist_buf = obj;
>         }
> @@ -138,7 +138,7 @@ rte_log_add_in_history(const char *buf, size_t size)
>
>         /* not enough room for msg, buffer go back in mempool */
>         if (size >= hist_buf_size) {
> -               rte_mempool_mp_put(log_history_mp, hist_buf);
> +               rte_mempool_put(log_history_mp, hist_buf);
>                 rte_spinlock_unlock(&log_list_lock);
>                 return -ENOBUFS;
>         }
> @@ -252,12 +252,12 @@ rte_log_dump_history(FILE *out)
>
>                 /* write on stdout */
>                 if (fwrite(hist_buf->buf, hist_buf->size, 1, out) == 0) {
> -                       rte_mempool_mp_put(log_history_mp, hist_buf);
> +                       rte_mempool_put(log_history_mp, hist_buf);
>                         break;
>                 }
>
>                 /* put back message structure in pool */
> -               rte_mempool_mp_put(log_history_mp, hist_buf);
> +               rte_mempool_put(log_history_mp, hist_buf);
>         }
>         fflush(out);
>
> diff --git a/lib/librte_mempool/rte_mempool.c b/lib/librte_mempool/rte_mempool.c
> index 73ca770..4d977c1 100644
> --- a/lib/librte_mempool/rte_mempool.c
> +++ b/lib/librte_mempool/rte_mempool.c
> @@ -375,6 +375,63 @@ rte_mempool_xmem_usage(void *vaddr, uint32_t elt_num, size_t elt_sz,
>         return usz;
>  }
>
> +static void
> +mempool_cache_init(struct rte_mempool_cache *cache, uint32_t size)
> +{
> +       cache->size = size;
> +       cache->flushthresh = CALC_CACHE_FLUSHTHRESH(size);
> +       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.
> + */
> +struct rte_mempool_cache *
> +rte_mempool_cache_create(uint32_t size, int socket_id)
> +{
> +       struct rte_mempool_cache *cache;
> +
> +       if (size > RTE_MEMPOOL_CACHE_MAX_SIZE) {
> +               rte_errno = EINVAL;
> +               return NULL;
> +       }
> +
> +       cache = rte_zmalloc_socket("MEMPOOL_CACHE", sizeof(*cache),
> +                                  RTE_CACHE_LINE_SIZE, socket_id);
> +       if (cache == NULL) {
> +               RTE_LOG(ERR, MEMPOOL, "Cannot allocate mempool cache!\n");
> +               rte_errno = ENOMEM;
> +               return NULL;
> +       }
> +
> +       mempool_cache_init(cache, size);
> +
> +       return cache;
> +}
> +
> +/*
> + * Free a cache. It's the responsibility of the user to make sure that any
> + * remaining objects in the cache are flushed to the corresponding
> + * mempool.
> + */
> +void
> +rte_mempool_cache_free(struct rte_mempool_cache *cache)
> +{
> +       rte_free(cache);
> +}
> +
> +/*
> + * Put all objects in the cache to the specified mempool's ring.
> + */
> +void
> +rte_mempool_cache_flush(struct rte_mempool_cache *cache,
> +                       struct rte_mempool *mp)
> +{
> +       rte_ring_enqueue_bulk(mp->ring, cache->objs, cache->len);
> +}
> +
>  #ifndef RTE_LIBRTE_XEN_DOM0
>  /* stub if DOM0 support not configured */
>  struct rte_mempool *
> @@ -448,6 +505,7 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>         struct rte_mempool_objsz objsz;
>         void *startaddr;
>         int page_size = getpagesize();
> +       unsigned lcore_id;
>
>         /* compilation-time checks */
>         RTE_BUILD_BUG_ON((sizeof(struct rte_mempool) &
> @@ -583,8 +641,8 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>         mp->elt_size = objsz.elt_size;
>         mp->header_size = objsz.header_size;
>         mp->trailer_size = objsz.trailer_size;
> +       /* Size of default caches, zero means disabled. */
>         mp->cache_size = cache_size;
> -       mp->cache_flushthresh = CALC_CACHE_FLUSHTHRESH(cache_size);
>         mp->private_data_size = private_data_size;
>
>         /*
> @@ -594,6 +652,13 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>         mp->local_cache = (struct rte_mempool_cache *)
>                         ((char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, 0));
>
> +       /* Init all default caches. */
> +       if (cache_size != 0) {
> +               for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++)
> +                       mempool_cache_init(&mp->local_cache[lcore_id],
> +                                          cache_size);
> +       }
> +
>         /* calculate address of the first element for continuous mempool. */
>         obj = (char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, cache_size) +
>                 private_data_size;
> @@ -672,7 +737,7 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
>         unsigned count = 0;
>         unsigned cache_count;
>
> -       fprintf(f, "  cache infos:\n");
> +       fprintf(f, "  internal cache infos:\n");
>         fprintf(f, "    cache_size=%"PRIu32"\n", mp->cache_size);
>
>         if (mp->cache_size == 0)
> @@ -680,7 +745,8 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
>
>         for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
>                 cache_count = mp->local_cache[lcore_id].len;
> -               fprintf(f, "    cache_count[%u]=%u\n", lcore_id, cache_count);
> +               fprintf(f, "    cache_count[%u]=%"PRIu32"\n",
> +                       lcore_id, cache_count);
>                 count += cache_count;
>         }
>         fprintf(f, "    total_cache_count=%u\n", count);
> @@ -760,7 +826,9 @@ mempool_audit_cache(const struct rte_mempool *mp)
>                 return;
>
>         for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
> -               if (mp->local_cache[lcore_id].len > mp->cache_flushthresh) {
> +               const struct rte_mempool_cache *cache;
> +               cache = &mp->local_cache[lcore_id];
> +               if (cache->len > cache->flushthresh) {
>                         RTE_LOG(CRIT, MEMPOOL, "badness on cache[%u]\n",
>                                 lcore_id);
>                         rte_panic("MEMPOOL: invalid cache len\n");
> diff --git a/lib/librte_mempool/rte_mempool.h b/lib/librte_mempool/rte_mempool.h
> index 8595e77..21d43e2 100644
> --- a/lib/librte_mempool/rte_mempool.h
> +++ b/lib/librte_mempool/rte_mempool.h
> @@ -99,7 +99,9 @@ struct rte_mempool_debug_stats {
>   * A structure that stores a per-core object cache.
>   */
>  struct rte_mempool_cache {
> -       unsigned len; /**< Cache len */
> +       uint32_t size;        /**< Size of the cache */
> +       uint32_t flushthresh; /**< Threshold before we flush excess elements */
> +       uint32_t len;         /**< Current cache count */
>         /*
>          * Cache is allocated to this size to allow it to overflow in certain
>          * cases to avoid needless emptying of cache.
> @@ -182,9 +184,7 @@ struct rte_mempool {
>         phys_addr_t phys_addr;           /**< Phys. addr. of mempool struct. */
>         int flags;                       /**< Flags of the mempool. */
>         uint32_t size;                   /**< Size of the mempool. */
> -       uint32_t cache_size;             /**< Size of per-lcore local cache. */
> -       uint32_t cache_flushthresh;
> -       /**< Threshold before we flush excess elements. */
> +       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). */
> @@ -742,6 +742,66 @@ rte_dom0_mempool_create(const char *name, unsigned n, unsigned elt_size,
>  void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
>
>  /**
> + * Create a user-owned mempool cache.
> + *
> + * This can be used by non-EAL threads to enable caching when they
> + * interact with a mempool.
> + *
> + * @param size
> + *   The size of the mempool cache. See rte_mempool_create()'s cache_size
> + *   parameter description for more information. The same limits and
> + *   considerations apply here too.
> + * @param socket_id
> + *   The socket identifier in the case of NUMA. The value can be
> + *   SOCKET_ID_ANY if there is no NUMA constraint for the reserved zone.
> + */
> +struct rte_mempool_cache *
> +rte_mempool_cache_create(uint32_t size, int socket_id);
> +
> +/**
> + * Free a user-owned mempool cache.
> + *
> + * @param cache
> + *   A pointer to the mempool cache.
> + */
> +void
> +rte_mempool_cache_free(struct rte_mempool_cache *cache);
> +
> +/**
> + * Flush a user-owned mempool cache to the specified mempool.
> + *
> + * @param cache
> + *   A pointer to the mempool cache.
> + * @param mp
> + *   A pointer to the mempool.
> + */
> +void
> +rte_mempool_cache_flush(struct rte_mempool_cache *cache,
> +                       struct rte_mempool *mp);
> +
> +/**
> + * Get a pointer to the per-lcore default mempool cache.
> + *
> + * @param mp
> + *   A pointer to the mempool structure.
> + * @param lcore_id
> + *   The logical core id.
> + * @return
> + *   A pointer to the mempool cache or NULL if disabled or non-EAL thread.
> + */
> +static inline struct rte_mempool_cache *
> +rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
> +{
> +       if (mp->cache_size == 0)
> +               return NULL;
> +
> +       if (lcore_id >= RTE_MAX_LCORE)
> +               return NULL;
> +
> +       return &mp->local_cache[lcore_id];
> +}
> +
> +/**
>   * @internal Put several objects back in the mempool; used internally.
>   * @param mp
>   *   A pointer to the mempool structure.
> @@ -750,33 +810,29 @@ void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
>   * @param n
>   *   The number of objects to store back in the mempool, must be strictly
>   *   positive.
> + * @param cache
> + *   A pointer to a mempool cache structure. May be NULL if not needed.
>   * @param is_mp
>   *   Mono-producer (0) or multi-producers (1).
>   */
>  static inline void __attribute__((always_inline))
> -__mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -                   unsigned n, int is_mp)
> +__mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
> +                     unsigned n, struct rte_mempool_cache *cache, int is_mp)
>  {
> -       struct rte_mempool_cache *cache;
>         uint32_t index;
>         void **cache_objs;
> -       unsigned lcore_id = rte_lcore_id();
> -       uint32_t cache_size = mp->cache_size;
> -       uint32_t flushthresh = mp->cache_flushthresh;
>
>         /* increment stat now, adding in mempool always success */
>         __MEMPOOL_STAT_ADD(mp, put, n);
>
> -       /* cache is not enabled or single producer or non-EAL thread */
> -       if (unlikely(cache_size == 0 || is_mp == 0 ||
> -                    lcore_id >= RTE_MAX_LCORE))
> +       /* No cache provided or cache is not enabled or single producer */
> +       if (unlikely(cache == NULL || cache->size == 0 || is_mp == 0))
>                 goto ring_enqueue;
>
>         /* Go straight to ring if put would overflow mem allocated for cache */
>         if (unlikely(n > RTE_MEMPOOL_CACHE_MAX_SIZE))
>                 goto ring_enqueue;
>
> -       cache = &mp->local_cache[lcore_id];
>         cache_objs = &cache->objs[cache->len];
>
>         /*
> @@ -792,10 +848,10 @@ __mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
>
>         cache->len += n;
>
> -       if (cache->len >= flushthresh) {
> -               rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache_size],
> -                               cache->len - cache_size);
> -               cache->len = cache_size;
> +       if (cache->len >= cache->flushthresh) {
> +               rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache->size],
> +                               cache->len - cache->size);
> +               cache->len = cache->size;
>         }
>
>         return;
> @@ -820,41 +876,27 @@ ring_enqueue:
>  #endif
>  }
>
> -
>  /**
> - * Put several objects back in the mempool (multi-producers safe).
> + * Put several objects back in the mempool.
>   *
>   * @param mp
>   *   A pointer to the mempool structure.
>   * @param obj_table
>   *   A pointer to a table of void * pointers (objects).
>   * @param n
> - *   The number of objects to add in the mempool from the obj_table.
> + *   The number of objects to store back in the mempool, must be strictly
> + *   positive.
> + * @param cache
> + *   A pointer to a mempool cache structure. May be NULL if not needed.
> + * @param is_mp
> + *   Mono-producer (0) or multi-producers (1).
>   */
>  static inline void __attribute__((always_inline))
> -rte_mempool_mp_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -                       unsigned n)
> -{
> -       __mempool_check_cookies(mp, obj_table, n, 0);
> -       __mempool_put_bulk(mp, obj_table, n, 1);
> -}
> -
> -/**
> - * Put several objects back in the mempool (NOT multi-producers safe).
> - *
> - * @param mp
> - *   A pointer to the mempool structure.
> - * @param obj_table
> - *   A pointer to a table of void * pointers (objects).
> - * @param n
> - *   The number of objects to add in the mempool from obj_table.
> - */
> -static inline void
> -rte_mempool_sp_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -                       unsigned n)
> +rte_mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
> +                       unsigned n, struct rte_mempool_cache *cache, int is_mp)
>  {
>         __mempool_check_cookies(mp, obj_table, n, 0);
> -       __mempool_put_bulk(mp, obj_table, n, 0);
> +       __mempool_generic_put(mp, obj_table, n, cache, is_mp);
>  }
>
>  /**
> @@ -875,36 +917,11 @@ static inline void __attribute__((always_inline))
>  rte_mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
>                      unsigned n)
>  {
> -       __mempool_check_cookies(mp, obj_table, n, 0);
> -       __mempool_put_bulk(mp, obj_table, n, !(mp->flags & MEMPOOL_F_SP_PUT));
> -}
> -
> -/**
> - * Put one object in the mempool (multi-producers safe).
> - *
> - * @param mp
> - *   A pointer to the mempool structure.
> - * @param obj
> - *   A pointer to the object to be added.
> - */
> -static inline void __attribute__((always_inline))
> -rte_mempool_mp_put(struct rte_mempool *mp, void *obj)
> -{
> -       rte_mempool_mp_put_bulk(mp, &obj, 1);
> -}
> +       struct rte_mempool_cache *cache;
>
> -/**
> - * Put one object back in the mempool (NOT multi-producers safe).
> - *
> - * @param mp
> - *   A pointer to the mempool structure.
> - * @param obj
> - *   A pointer to the object to be added.
> - */
> -static inline void __attribute__((always_inline))
> -rte_mempool_sp_put(struct rte_mempool *mp, void *obj)
> -{
> -       rte_mempool_sp_put_bulk(mp, &obj, 1);
> +       cache = rte_mempool_default_cache(mp, rte_lcore_id());
> +       rte_mempool_generic_put(mp, obj_table, n, cache,
> +                               !(mp->flags & MEMPOOL_F_SP_PUT));
>  }
>
>  /**
> @@ -933,6 +950,8 @@ rte_mempool_put(struct rte_mempool *mp, void *obj)
>   *   A pointer to a table of void * pointers (objects).
>   * @param n
>   *   The number of objects to get, must be strictly positive.
> + * @param cache
> + *   A pointer to a mempool cache structure. May be NULL if not needed.
>   * @param is_mc
>   *   Mono-consumer (0) or multi-consumers (1).
>   * @return
> @@ -940,28 +959,24 @@ rte_mempool_put(struct rte_mempool *mp, void *obj)
>   *   - <0: Error; code of ring dequeue function.
>   */
>  static inline int __attribute__((always_inline))
> -__mempool_get_bulk(struct rte_mempool *mp, void **obj_table,
> -                  unsigned n, int is_mc)
> +__mempool_generic_get(struct rte_mempool *mp, void **obj_table,
> +                     unsigned n, struct rte_mempool_cache *cache, int is_mc)
>  {
>         int ret;
> -       struct rte_mempool_cache *cache;
>         uint32_t index, len;
>         void **cache_objs;
> -       unsigned lcore_id = rte_lcore_id();
> -       uint32_t cache_size = mp->cache_size;
>
> -       /* cache is not enabled or single consumer */
> -       if (unlikely(cache_size == 0 || is_mc == 0 ||
> -                    n >= cache_size || lcore_id >= RTE_MAX_LCORE))
> +       /* No cache provided or cache is not enabled or single consumer */
> +       if (unlikely(cache == NULL || cache->size == 0 || is_mc == 0 ||
> +                    n >= cache->size))
>                 goto ring_dequeue;
>
> -       cache = &mp->local_cache[lcore_id];
>         cache_objs = cache->objs;
>
>         /* Can this be satisfied from the cache? */
>         if (cache->len < n) {
>                 /* No. Backfill the cache first, and then fill from it */
> -               uint32_t req = n + (cache_size - cache->len);
> +               uint32_t req = n + (cache->size - cache->len);
>
>                 /* How many do we require i.e. number to fill the cache + the request */
>                 ret = rte_ring_mc_dequeue_bulk(mp->ring, &cache->objs[cache->len], req);
> @@ -1005,57 +1020,28 @@ ring_dequeue:
>  }
>
>  /**
> - * Get several objects from the mempool (multi-consumers safe).
> - *
> - * If cache is enabled, objects will be retrieved first from cache,
> - * subsequently from the common pool. Note that it can return -ENOENT when
> - * the local cache and common pool are empty, even if cache from other
> - * lcores are full.
> - *
> - * @param mp
> - *   A pointer to the mempool structure.
> - * @param obj_table
> - *   A pointer to a table of void * pointers (objects) that will be filled.
> - * @param n
> - *   The number of objects to get from mempool to obj_table.
> - * @return
> - *   - 0: Success; objects taken.
> - *   - -ENOENT: Not enough entries in the mempool; no object is retrieved.
> - */
> -static inline int __attribute__((always_inline))
> -rte_mempool_mc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
> -{
> -       int ret;
> -       ret = __mempool_get_bulk(mp, obj_table, n, 1);
> -       if (ret == 0)
> -               __mempool_check_cookies(mp, obj_table, n, 1);
> -       return ret;
> -}
> -
> -/**
> - * Get several objects from the mempool (NOT multi-consumers safe).
> - *
> - * If cache is enabled, objects will be retrieved first from cache,
> - * subsequently from the common pool. Note that it can return -ENOENT when
> - * the local cache and common pool are empty, even if cache from other
> - * lcores are full.
> + * Get several objects from the mempool.
>   *
>   * @param mp
>   *   A pointer to the mempool structure.
>   * @param obj_table
> - *   A pointer to a table of void * pointers (objects) that will be filled.
> + *   A pointer to a table of void * pointers (objects).
>   * @param n
> - *   The number of objects to get from the mempool to obj_table.
> + *   The number of objects to get, must be strictly positive.
> + * @param cache
> + *   A pointer to a mempool cache structure. May be NULL if not needed.
> + * @param is_mc
> + *   Mono-consumer (0) or multi-consumers (1).
>   * @return
> - *   - 0: Success; objects taken.
> - *   - -ENOENT: Not enough entries in the mempool; no object is
> - *     retrieved.
> + *   - >=0: Success; number of objects supplied.
> + *   - <0: Error; code of ring dequeue function.
>   */
>  static inline int __attribute__((always_inline))
> -rte_mempool_sc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
> +rte_mempool_generic_get(struct rte_mempool *mp, void **obj_table,
> +                       unsigned n, struct rte_mempool_cache *cache, int is_mc)
>  {
>         int ret;
> -       ret = __mempool_get_bulk(mp, obj_table, n, 0);
> +       ret = __mempool_generic_get(mp, obj_table, n, cache, is_mc);
>         if (ret == 0)
>                 __mempool_check_cookies(mp, obj_table, n, 1);
>         return ret;
> @@ -1086,56 +1072,11 @@ rte_mempool_sc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
>  static inline int __attribute__((always_inline))
>  rte_mempool_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)
>  {
> -       int ret;
> -       ret = __mempool_get_bulk(mp, obj_table, n,
> -                                !(mp->flags & MEMPOOL_F_SC_GET));
> -       if (ret == 0)
> -               __mempool_check_cookies(mp, obj_table, n, 1);
> -       return ret;
> -}
> -
> -/**
> - * Get one object from the mempool (multi-consumers safe).
> - *
> - * If cache is enabled, objects will be retrieved first from cache,
> - * subsequently from the common pool. Note that it can return -ENOENT when
> - * the local cache and common pool are empty, even if cache from other
> - * lcores are full.
> - *
> - * @param mp
> - *   A pointer to the mempool structure.
> - * @param obj_p
> - *   A pointer to a void * pointer (object) that will be filled.
> - * @return
> - *   - 0: Success; objects taken.
> - *   - -ENOENT: Not enough entries in the mempool; no object is retrieved.
> - */
> -static inline int __attribute__((always_inline))
> -rte_mempool_mc_get(struct rte_mempool *mp, void **obj_p)
> -{
> -       return rte_mempool_mc_get_bulk(mp, obj_p, 1);
> -}
> +       struct rte_mempool_cache *cache;
>
> -/**
> - * Get one object from the mempool (NOT multi-consumers safe).
> - *
> - * If cache is enabled, objects will be retrieved first from cache,
> - * subsequently from the common pool. Note that it can return -ENOENT when
> - * the local cache and common pool are empty, even if cache from other
> - * lcores are full.
> - *
> - * @param mp
> - *   A pointer to the mempool structure.
> - * @param obj_p
> - *   A pointer to a void * pointer (object) that will be filled.
> - * @return
> - *   - 0: Success; objects taken.
> - *   - -ENOENT: Not enough entries in the mempool; no object is retrieved.
> - */
> -static inline int __attribute__((always_inline))
> -rte_mempool_sc_get(struct rte_mempool *mp, void **obj_p)
> -{
> -       return rte_mempool_sc_get_bulk(mp, obj_p, 1);
> +       cache = rte_mempool_default_cache(mp, rte_lcore_id());
> +       return rte_mempool_generic_get(mp, obj_table, n, cache,
> +                                      !(mp->flags & MEMPOOL_F_SC_GET));
>  }
>
>  /**
> @@ -1169,7 +1110,7 @@ rte_mempool_get(struct rte_mempool *mp, void **obj_p)
>   *
>   * When cache is enabled, this function has to browse the length of
>   * all lcores, so it should not be used in a data path, but only for
> - * debug purposes.
> + * debug purposes. User-owned mempool caches are not accounted for.
>   *
>   * @param mp
>   *   A pointer to the mempool structure.
> @@ -1188,7 +1129,7 @@ unsigned rte_mempool_count(const struct rte_mempool *mp);
>   *
>   * When cache is enabled, this function has to browse the length of
>   * all lcores, so it should not be used in a data path, but only for
> - * debug purposes.
> + * debug purposes. User-owned mempool caches are not accounted for.
>   *
>   * @param mp
>   *   A pointer to the mempool structure.
> @@ -1206,7 +1147,7 @@ rte_mempool_free_count(const struct rte_mempool *mp)
>   *
>   * When cache is enabled, this function has to browse the length of all
>   * lcores, so it should not be used in a data path, but only for debug
> - * purposes.
> + * purposes. User-owned mempool caches are not accounted for.
>   *
>   * @param mp
>   *   A pointer to the mempool structure.
> @@ -1225,7 +1166,7 @@ rte_mempool_full(const struct rte_mempool *mp)
>   *
>   * When cache is enabled, this function has to browse the length of all
>   * lcores, so it should not be used in a data path, but only for debug
> - * purposes.
> + * purposes. User-owned mempool caches are not accounted for.
>   *
>   * @param mp
>   *   A pointer to the mempool structure.
> --
> 1.9.1
>

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Ananyev, Konstantin]

Hi Lazaros,

Looks ok to me in general, few comments below.
One more generic question - did you observe any performance impact 
caused by these changes?
Konstantin

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Lazaros Koromilas
> Sent: Monday, April 04, 2016 4:43 PM
> To: dev@dpdk.org
> Subject: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches
> 
> The mempool cache is only available to EAL threads as a per-lcore
> resource. Change this so that the user can create and provide their own
> cache on mempool get and put operations. This works with non-EAL threads
> too. This commit introduces the new API calls:
> 
>     rte_mempool_cache_create(size, socket_id)
>     rte_mempool_cache_flush(cache, mp)
>     rte_mempool_cache_free(cache)
>     rte_mempool_default_cache(mp, lcore_id)
>     rte_mempool_generic_put(mp, obj_table, n, cache, is_mp)
>     rte_mempool_generic_get(mp, obj_table, n, cache, is_mc)
> 
> Removes the API calls:
> 
>     rte_mempool_sp_put_bulk(mp, obj_table, n)
>     rte_mempool_sc_get_bulk(mp, obj_table, n)
>     rte_mempool_sp_put(mp, obj)
>     rte_mempool_sc_get(mp, obj)


Hmm, shouldn't we deprecate it first for a release before removing completely?
Let say for now you can just make them macros that calls the remaining functions or so.

> 
> And the remaining API calls use the per-lcore default local cache:
> 
>     rte_mempool_put_bulk(mp, obj_table, n)
>     rte_mempool_get_bulk(mp, obj_table, n)
>     rte_mempool_put(mp, obj)
>     rte_mempool_get(mp, obj)
> 
> Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
> ---
>  app/test/test_mempool.c                |  58 +++++--
>  app/test/test_mempool_perf.c           |  46 +++++-
>  lib/librte_eal/common/eal_common_log.c |   8 +-
>  lib/librte_mempool/rte_mempool.c       |  76 ++++++++-
>  lib/librte_mempool/rte_mempool.h       | 291 +++++++++++++--------------------
>  5 files changed, 275 insertions(+), 204 deletions(-)
> 
> 
> diff --git a/lib/librte_mempool/rte_mempool.c b/lib/librte_mempool/rte_mempool.c
> index 73ca770..4d977c1 100644
> --- a/lib/librte_mempool/rte_mempool.c
> +++ b/lib/librte_mempool/rte_mempool.c
> @@ -375,6 +375,63 @@ rte_mempool_xmem_usage(void *vaddr, uint32_t elt_num, size_t elt_sz,
>  	return usz;
>  }
> 
> +static void
> +mempool_cache_init(struct rte_mempool_cache *cache, uint32_t size)
> +{
> +	cache->size = size;
> +	cache->flushthresh = CALC_CACHE_FLUSHTHRESH(size);
> +	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.
> + */
> +struct rte_mempool_cache *
> +rte_mempool_cache_create(uint32_t size, int socket_id)
> +{
> +	struct rte_mempool_cache *cache;
> +
> +	if (size > RTE_MEMPOOL_CACHE_MAX_SIZE) {
> +		rte_errno = EINVAL;
> +		return NULL;
> +	}
> +
> +	cache = rte_zmalloc_socket("MEMPOOL_CACHE", sizeof(*cache),
> +				   RTE_CACHE_LINE_SIZE, socket_id);
> +	if (cache == NULL) {
> +		RTE_LOG(ERR, MEMPOOL, "Cannot allocate mempool cache!\n");
> +		rte_errno = ENOMEM;
> +		return NULL;
> +	}
> +
> +	mempool_cache_init(cache, size);
> +
> +	return cache;
> +}
> +
> +/*
> + * Free a cache. It's the responsibility of the user to make sure that any
> + * remaining objects in the cache are flushed to the corresponding
> + * mempool.
> + */
> +void
> +rte_mempool_cache_free(struct rte_mempool_cache *cache)
> +{
> +	rte_free(cache);
> +}
> +
> +/*
> + * Put all objects in the cache to the specified mempool's ring.
> + */
> +void
> +rte_mempool_cache_flush(struct rte_mempool_cache *cache,
> +			struct rte_mempool *mp)
> +{
> +	rte_ring_enqueue_bulk(mp->ring, cache->objs, cache->len);

Shouldn't you also reset cache->len too here?
cache->len = 0;
Another thought - might be that function deserved to be inline one.

> +}
> +
>  #ifndef RTE_LIBRTE_XEN_DOM0
>  /* stub if DOM0 support not configured */
>  struct rte_mempool *
> @@ -448,6 +505,7 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>  	struct rte_mempool_objsz objsz;
>  	void *startaddr;
>  	int page_size = getpagesize();
> +	unsigned lcore_id;
> 
>  	/* compilation-time checks */
>  	RTE_BUILD_BUG_ON((sizeof(struct rte_mempool) &
> @@ -583,8 +641,8 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>  	mp->elt_size = objsz.elt_size;
>  	mp->header_size = objsz.header_size;
>  	mp->trailer_size = objsz.trailer_size;
> +	/* Size of default caches, zero means disabled. */
>  	mp->cache_size = cache_size;
> -	mp->cache_flushthresh = CALC_CACHE_FLUSHTHRESH(cache_size);
>  	mp->private_data_size = private_data_size;
> 
>  	/*
> @@ -594,6 +652,13 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>  	mp->local_cache = (struct rte_mempool_cache *)
>  			((char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, 0));
> 
> +	/* Init all default caches. */
> +	if (cache_size != 0) {
> +		for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++)
> +			mempool_cache_init(&mp->local_cache[lcore_id],
> +					   cache_size);
> +	}
> +
>  	/* calculate address of the first element for continuous mempool. */
>  	obj = (char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, cache_size) +
>  		private_data_size;
> @@ -672,7 +737,7 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
>  	unsigned count = 0;
>  	unsigned cache_count;
> 
> -	fprintf(f, "  cache infos:\n");
> +	fprintf(f, "  internal cache infos:\n");
>  	fprintf(f, "    cache_size=%"PRIu32"\n", mp->cache_size);
> 
>  	if (mp->cache_size == 0)
> @@ -680,7 +745,8 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
> 
>  	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
>  		cache_count = mp->local_cache[lcore_id].len;
> -		fprintf(f, "    cache_count[%u]=%u\n", lcore_id, cache_count);
> +		fprintf(f, "    cache_count[%u]=%"PRIu32"\n",
> +			lcore_id, cache_count);
>  		count += cache_count;
>  	}
>  	fprintf(f, "    total_cache_count=%u\n", count);
> @@ -760,7 +826,9 @@ mempool_audit_cache(const struct rte_mempool *mp)
>  		return;
> 
>  	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
> -		if (mp->local_cache[lcore_id].len > mp->cache_flushthresh) {
> +		const struct rte_mempool_cache *cache;
> +		cache = &mp->local_cache[lcore_id];
> +		if (cache->len > cache->flushthresh) {
>  			RTE_LOG(CRIT, MEMPOOL, "badness on cache[%u]\n",
>  				lcore_id);
>  			rte_panic("MEMPOOL: invalid cache len\n");
> diff --git a/lib/librte_mempool/rte_mempool.h b/lib/librte_mempool/rte_mempool.h
> index 8595e77..21d43e2 100644
> --- a/lib/librte_mempool/rte_mempool.h
> +++ b/lib/librte_mempool/rte_mempool.h
> @@ -99,7 +99,9 @@ struct rte_mempool_debug_stats {
>   * A structure that stores a per-core object cache.
>   */
>  struct rte_mempool_cache {
> -	unsigned len; /**< Cache len */
> +	uint32_t size;        /**< Size of the cache */
> +	uint32_t flushthresh; /**< Threshold before we flush excess elements */
> +	uint32_t len;         /**< Current cache count */
>  	/*
>  	 * Cache is allocated to this size to allow it to overflow in certain
>  	 * cases to avoid needless emptying of cache.
> @@ -182,9 +184,7 @@ struct rte_mempool {
>  	phys_addr_t phys_addr;           /**< Phys. addr. of mempool struct. */
>  	int flags;                       /**< Flags of the mempool. */
>  	uint32_t size;                   /**< Size of the mempool. */
> -	uint32_t cache_size;             /**< Size of per-lcore local cache. */
> -	uint32_t cache_flushthresh;
> -	/**< Threshold before we flush excess elements. */
> +	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). */
> @@ -742,6 +742,66 @@ rte_dom0_mempool_create(const char *name, unsigned n, unsigned elt_size,
>  void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
> 
>  /**
> + * Create a user-owned mempool cache.
> + *
> + * This can be used by non-EAL threads to enable caching when they
> + * interact with a mempool.
> + *
> + * @param size
> + *   The size of the mempool cache. See rte_mempool_create()'s cache_size
> + *   parameter description for more information. The same limits and
> + *   considerations apply here too.
> + * @param socket_id
> + *   The socket identifier in the case of NUMA. The value can be
> + *   SOCKET_ID_ANY if there is no NUMA constraint for the reserved zone.
> + */
> +struct rte_mempool_cache *
> +rte_mempool_cache_create(uint32_t size, int socket_id);
> +
> +/**
> + * Free a user-owned mempool cache.
> + *
> + * @param cache
> + *   A pointer to the mempool cache.
> + */
> +void
> +rte_mempool_cache_free(struct rte_mempool_cache *cache);
> +
> +/**
> + * Flush a user-owned mempool cache to the specified mempool.
> + *
> + * @param cache
> + *   A pointer to the mempool cache.
> + * @param mp
> + *   A pointer to the mempool.
> + */
> +void
> +rte_mempool_cache_flush(struct rte_mempool_cache *cache,
> +			struct rte_mempool *mp);
> +
> +/**
> + * Get a pointer to the per-lcore default mempool cache.
> + *
> + * @param mp
> + *   A pointer to the mempool structure.
> + * @param lcore_id
> + *   The logical core id.
> + * @return
> + *   A pointer to the mempool cache or NULL if disabled or non-EAL thread.
> + */
> +static inline struct rte_mempool_cache *
> +rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
> +{
> +	if (mp->cache_size == 0)
> +		return NULL;
> +
> +	if (lcore_id >= RTE_MAX_LCORE)
> +		return NULL;
> +
> +	return &mp->local_cache[lcore_id];
> +}
> +
> +/**
>   * @internal Put several objects back in the mempool; used internally.
>   * @param mp
>   *   A pointer to the mempool structure.
> @@ -750,33 +810,29 @@ void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
>   * @param n
>   *   The number of objects to store back in the mempool, must be strictly
>   *   positive.
> + * @param cache
> + *   A pointer to a mempool cache structure. May be NULL if not needed.
>   * @param is_mp
>   *   Mono-producer (0) or multi-producers (1).
>   */
>  static inline void __attribute__((always_inline))
> -__mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -		    unsigned n, int is_mp)
> +__mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
> +		      unsigned n, struct rte_mempool_cache *cache, int is_mp)
>  {
> -	struct rte_mempool_cache *cache;
>  	uint32_t index;
>  	void **cache_objs;
> -	unsigned lcore_id = rte_lcore_id();
> -	uint32_t cache_size = mp->cache_size;
> -	uint32_t flushthresh = mp->cache_flushthresh;
> 
>  	/* increment stat now, adding in mempool always success */
>  	__MEMPOOL_STAT_ADD(mp, put, n);
> 
> -	/* cache is not enabled or single producer or non-EAL thread */
> -	if (unlikely(cache_size == 0 || is_mp == 0 ||
> -		     lcore_id >= RTE_MAX_LCORE))
> +	/* No cache provided or cache is not enabled or single producer */
> +	if (unlikely(cache == NULL || cache->size == 0 || is_mp == 0))

Here and in other places do we really need that check: 'cache->size == 0'?
I mean at mempool_create() if mp->cache_size would be zero, then we wouldn't allocate
any local_cache[] ant all and cache would be 0 anyway.
Also in rte_mempool_cache_create() we can check that size > 0 and return -EINVAL otherwise.

>  		goto ring_enqueue;
> 
>  	/* Go straight to ring if put would overflow mem allocated for cache */
>  	if (unlikely(n > RTE_MEMPOOL_CACHE_MAX_SIZE))
>  		goto ring_enqueue;
> 
> -	cache = &mp->local_cache[lcore_id];
>  	cache_objs = &cache->objs[cache->len];
> 
>  	/*
> @@ -792,10 +848,10 @@ __mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> 
>  	cache->len += n;
> 
> -	if (cache->len >= flushthresh) {
> -		rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache_size],
> -				cache->len - cache_size);
> -		cache->len = cache_size;
> +	if (cache->len >= cache->flushthresh) {
> +		rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache->size],
> +				cache->len - cache->size);
> +		cache->len = cache->size;
>  	}
> 
>  	return;
> @@ -820,41 +876,27 @@ ring_enqueue:
>  #endif
>  }
> 

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Lazaros Koromilas]

Hi Konstantin,
Thanks for the review.

Regards,
Lazaros.

On Mon, Apr 18, 2016 at 4:17 PM, Ananyev, Konstantin
<konstantin.ananyev@intel.com> wrote:
> Hi Lazaros,
>
> Looks ok to me in general, few comments below.
> One more generic question - did you observe any performance impact
> caused by these changes?
> Konstantin

I didn't observe any notable difference to the default per-lcore cache
case. Here is an excerpt from the mempool_perf test:

$ egrep '(^start|n_get_bulk=32 n_put_bulk=32 n_keep=128)'
x86_64-native-linuxapp-gcc.log
start performance test (without cache)
mempool_autotest cache=0 cores=1 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=714958438
mempool_autotest cache=0 cores=2 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=795738111
mempool_autotest cache=0 cores=4 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=313655295
start performance test (with cache)
mempool_autotest cache=512 cores=1 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=780455116
mempool_autotest cache=512 cores=2 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=1046937599
mempool_autotest cache=512 cores=4 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=1988362238
start performance test (with user-owned cache)
mempool_autotest cache=512 cores=1 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=787519897
mempool_autotest cache=512 cores=2 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=1047029350
mempool_autotest cache=512 cores=4 n_get_bulk=32 n_put_bulk=32
n_keep=128 rate_persec=1965896498

>
>> -----Original Message-----
>> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Lazaros Koromilas
>> Sent: Monday, April 04, 2016 4:43 PM
>> To: dev@dpdk.org
>> Subject: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches
>>
>> The mempool cache is only available to EAL threads as a per-lcore
>> resource. Change this so that the user can create and provide their own
>> cache on mempool get and put operations. This works with non-EAL threads
>> too. This commit introduces the new API calls:
>>
>>     rte_mempool_cache_create(size, socket_id)
>>     rte_mempool_cache_flush(cache, mp)
>>     rte_mempool_cache_free(cache)
>>     rte_mempool_default_cache(mp, lcore_id)
>>     rte_mempool_generic_put(mp, obj_table, n, cache, is_mp)
>>     rte_mempool_generic_get(mp, obj_table, n, cache, is_mc)
>>
>> Removes the API calls:
>>
>>     rte_mempool_sp_put_bulk(mp, obj_table, n)
>>     rte_mempool_sc_get_bulk(mp, obj_table, n)
>>     rte_mempool_sp_put(mp, obj)
>>     rte_mempool_sc_get(mp, obj)
>
>
> Hmm, shouldn't we deprecate it first for a release before removing completely?
> Let say for now you can just make them macros that calls the remaining functions or so.

How do we mark the calls as deprecated? The librte_compat stuff don't
apply here as we don't have a different version of the same symbol or
something. Do I need to put them as a notice?

>
>>
>> And the remaining API calls use the per-lcore default local cache:
>>
>>     rte_mempool_put_bulk(mp, obj_table, n)
>>     rte_mempool_get_bulk(mp, obj_table, n)
>>     rte_mempool_put(mp, obj)
>>     rte_mempool_get(mp, obj)
>>
>> Signed-off-by: Lazaros Koromilas <l@nofutznetworks.com>
>> ---
>>  app/test/test_mempool.c                |  58 +++++--
>>  app/test/test_mempool_perf.c           |  46 +++++-
>>  lib/librte_eal/common/eal_common_log.c |   8 +-
>>  lib/librte_mempool/rte_mempool.c       |  76 ++++++++-
>>  lib/librte_mempool/rte_mempool.h       | 291 +++++++++++++--------------------
>>  5 files changed, 275 insertions(+), 204 deletions(-)
>>
>>
>> diff --git a/lib/librte_mempool/rte_mempool.c b/lib/librte_mempool/rte_mempool.c
>> index 73ca770..4d977c1 100644
>> --- a/lib/librte_mempool/rte_mempool.c
>> +++ b/lib/librte_mempool/rte_mempool.c
>> @@ -375,6 +375,63 @@ rte_mempool_xmem_usage(void *vaddr, uint32_t elt_num, size_t elt_sz,
>>       return usz;
>>  }
>>
>> +static void
>> +mempool_cache_init(struct rte_mempool_cache *cache, uint32_t size)
>> +{
>> +     cache->size = size;
>> +     cache->flushthresh = CALC_CACHE_FLUSHTHRESH(size);
>> +     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.
>> + */
>> +struct rte_mempool_cache *
>> +rte_mempool_cache_create(uint32_t size, int socket_id)
>> +{
>> +     struct rte_mempool_cache *cache;
>> +
>> +     if (size > RTE_MEMPOOL_CACHE_MAX_SIZE) {
>> +             rte_errno = EINVAL;
>> +             return NULL;
>> +     }
>> +
>> +     cache = rte_zmalloc_socket("MEMPOOL_CACHE", sizeof(*cache),
>> +                                RTE_CACHE_LINE_SIZE, socket_id);
>> +     if (cache == NULL) {
>> +             RTE_LOG(ERR, MEMPOOL, "Cannot allocate mempool cache!\n");
>> +             rte_errno = ENOMEM;
>> +             return NULL;
>> +     }
>> +
>> +     mempool_cache_init(cache, size);
>> +
>> +     return cache;
>> +}
>> +
>> +/*
>> + * Free a cache. It's the responsibility of the user to make sure that any
>> + * remaining objects in the cache are flushed to the corresponding
>> + * mempool.
>> + */
>> +void
>> +rte_mempool_cache_free(struct rte_mempool_cache *cache)
>> +{
>> +     rte_free(cache);
>> +}
>> +
>> +/*
>> + * Put all objects in the cache to the specified mempool's ring.
>> + */
>> +void
>> +rte_mempool_cache_flush(struct rte_mempool_cache *cache,
>> +                     struct rte_mempool *mp)
>> +{
>> +     rte_ring_enqueue_bulk(mp->ring, cache->objs, cache->len);
>
> Shouldn't you also reset cache->len too here?
> cache->len = 0;
> Another thought - might be that function deserved to be inline one.

Yes, thanks! I'll make it inline too.

>> +}
>> +
>>  #ifndef RTE_LIBRTE_XEN_DOM0
>>  /* stub if DOM0 support not configured */
>>  struct rte_mempool *
>> @@ -448,6 +505,7 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>>       struct rte_mempool_objsz objsz;
>>       void *startaddr;
>>       int page_size = getpagesize();
>> +     unsigned lcore_id;
>>
>>       /* compilation-time checks */
>>       RTE_BUILD_BUG_ON((sizeof(struct rte_mempool) &
>> @@ -583,8 +641,8 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>>       mp->elt_size = objsz.elt_size;
>>       mp->header_size = objsz.header_size;
>>       mp->trailer_size = objsz.trailer_size;
>> +     /* Size of default caches, zero means disabled. */
>>       mp->cache_size = cache_size;
>> -     mp->cache_flushthresh = CALC_CACHE_FLUSHTHRESH(cache_size);
>>       mp->private_data_size = private_data_size;
>>
>>       /*
>> @@ -594,6 +652,13 @@ rte_mempool_xmem_create(const char *name, unsigned n, unsigned elt_size,
>>       mp->local_cache = (struct rte_mempool_cache *)
>>                       ((char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, 0));
>>
>> +     /* Init all default caches. */
>> +     if (cache_size != 0) {
>> +             for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++)
>> +                     mempool_cache_init(&mp->local_cache[lcore_id],
>> +                                        cache_size);
>> +     }
>> +
>>       /* calculate address of the first element for continuous mempool. */
>>       obj = (char *)mp + MEMPOOL_HEADER_SIZE(mp, pg_num, cache_size) +
>>               private_data_size;
>> @@ -672,7 +737,7 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
>>       unsigned count = 0;
>>       unsigned cache_count;
>>
>> -     fprintf(f, "  cache infos:\n");
>> +     fprintf(f, "  internal cache infos:\n");
>>       fprintf(f, "    cache_size=%"PRIu32"\n", mp->cache_size);
>>
>>       if (mp->cache_size == 0)
>> @@ -680,7 +745,8 @@ rte_mempool_dump_cache(FILE *f, const struct rte_mempool *mp)
>>
>>       for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
>>               cache_count = mp->local_cache[lcore_id].len;
>> -             fprintf(f, "    cache_count[%u]=%u\n", lcore_id, cache_count);
>> +             fprintf(f, "    cache_count[%u]=%"PRIu32"\n",
>> +                     lcore_id, cache_count);
>>               count += cache_count;
>>       }
>>       fprintf(f, "    total_cache_count=%u\n", count);
>> @@ -760,7 +826,9 @@ mempool_audit_cache(const struct rte_mempool *mp)
>>               return;
>>
>>       for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
>> -             if (mp->local_cache[lcore_id].len > mp->cache_flushthresh) {
>> +             const struct rte_mempool_cache *cache;
>> +             cache = &mp->local_cache[lcore_id];
>> +             if (cache->len > cache->flushthresh) {
>>                       RTE_LOG(CRIT, MEMPOOL, "badness on cache[%u]\n",
>>                               lcore_id);
>>                       rte_panic("MEMPOOL: invalid cache len\n");
>> diff --git a/lib/librte_mempool/rte_mempool.h b/lib/librte_mempool/rte_mempool.h
>> index 8595e77..21d43e2 100644
>> --- a/lib/librte_mempool/rte_mempool.h
>> +++ b/lib/librte_mempool/rte_mempool.h
>> @@ -99,7 +99,9 @@ struct rte_mempool_debug_stats {
>>   * A structure that stores a per-core object cache.
>>   */
>>  struct rte_mempool_cache {
>> -     unsigned len; /**< Cache len */
>> +     uint32_t size;        /**< Size of the cache */
>> +     uint32_t flushthresh; /**< Threshold before we flush excess elements */
>> +     uint32_t len;         /**< Current cache count */
>>       /*
>>        * Cache is allocated to this size to allow it to overflow in certain
>>        * cases to avoid needless emptying of cache.
>> @@ -182,9 +184,7 @@ struct rte_mempool {
>>       phys_addr_t phys_addr;           /**< Phys. addr. of mempool struct. */
>>       int flags;                       /**< Flags of the mempool. */
>>       uint32_t size;                   /**< Size of the mempool. */
>> -     uint32_t cache_size;             /**< Size of per-lcore local cache. */
>> -     uint32_t cache_flushthresh;
>> -     /**< Threshold before we flush excess elements. */
>> +     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). */
>> @@ -742,6 +742,66 @@ rte_dom0_mempool_create(const char *name, unsigned n, unsigned elt_size,
>>  void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
>>
>>  /**
>> + * Create a user-owned mempool cache.
>> + *
>> + * This can be used by non-EAL threads to enable caching when they
>> + * interact with a mempool.
>> + *
>> + * @param size
>> + *   The size of the mempool cache. See rte_mempool_create()'s cache_size
>> + *   parameter description for more information. The same limits and
>> + *   considerations apply here too.
>> + * @param socket_id
>> + *   The socket identifier in the case of NUMA. The value can be
>> + *   SOCKET_ID_ANY if there is no NUMA constraint for the reserved zone.
>> + */
>> +struct rte_mempool_cache *
>> +rte_mempool_cache_create(uint32_t size, int socket_id);
>> +
>> +/**
>> + * Free a user-owned mempool cache.
>> + *
>> + * @param cache
>> + *   A pointer to the mempool cache.
>> + */
>> +void
>> +rte_mempool_cache_free(struct rte_mempool_cache *cache);
>> +
>> +/**
>> + * Flush a user-owned mempool cache to the specified mempool.
>> + *
>> + * @param cache
>> + *   A pointer to the mempool cache.
>> + * @param mp
>> + *   A pointer to the mempool.
>> + */
>> +void
>> +rte_mempool_cache_flush(struct rte_mempool_cache *cache,
>> +                     struct rte_mempool *mp);
>> +
>> +/**
>> + * Get a pointer to the per-lcore default mempool cache.
>> + *
>> + * @param mp
>> + *   A pointer to the mempool structure.
>> + * @param lcore_id
>> + *   The logical core id.
>> + * @return
>> + *   A pointer to the mempool cache or NULL if disabled or non-EAL thread.
>> + */
>> +static inline struct rte_mempool_cache *
>> +rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
>> +{
>> +     if (mp->cache_size == 0)
>> +             return NULL;
>> +
>> +     if (lcore_id >= RTE_MAX_LCORE)
>> +             return NULL;
>> +
>> +     return &mp->local_cache[lcore_id];
>> +}
>> +
>> +/**
>>   * @internal Put several objects back in the mempool; used internally.
>>   * @param mp
>>   *   A pointer to the mempool structure.
>> @@ -750,33 +810,29 @@ void rte_mempool_dump(FILE *f, const struct rte_mempool *mp);
>>   * @param n
>>   *   The number of objects to store back in the mempool, must be strictly
>>   *   positive.
>> + * @param cache
>> + *   A pointer to a mempool cache structure. May be NULL if not needed.
>>   * @param is_mp
>>   *   Mono-producer (0) or multi-producers (1).
>>   */
>>  static inline void __attribute__((always_inline))
>> -__mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
>> -                 unsigned n, int is_mp)
>> +__mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
>> +                   unsigned n, struct rte_mempool_cache *cache, int is_mp)
>>  {
>> -     struct rte_mempool_cache *cache;
>>       uint32_t index;
>>       void **cache_objs;
>> -     unsigned lcore_id = rte_lcore_id();
>> -     uint32_t cache_size = mp->cache_size;
>> -     uint32_t flushthresh = mp->cache_flushthresh;
>>
>>       /* increment stat now, adding in mempool always success */
>>       __MEMPOOL_STAT_ADD(mp, put, n);
>>
>> -     /* cache is not enabled or single producer or non-EAL thread */
>> -     if (unlikely(cache_size == 0 || is_mp == 0 ||
>> -                  lcore_id >= RTE_MAX_LCORE))
>> +     /* No cache provided or cache is not enabled or single producer */
>> +     if (unlikely(cache == NULL || cache->size == 0 || is_mp == 0))
>
> Here and in other places do we really need that check: 'cache->size == 0'?
> I mean at mempool_create() if mp->cache_size would be zero, then we wouldn't allocate
> any local_cache[] ant all and cache would be 0 anyway.
> Also in rte_mempool_cache_create() we can check that size > 0 and return -EINVAL otherwise.

Makes sense, old calls now use rte_mempool_default_cache() which
returns NULL on that case. Will fix.

>>               goto ring_enqueue;
>>
>>       /* Go straight to ring if put would overflow mem allocated for cache */
>>       if (unlikely(n > RTE_MEMPOOL_CACHE_MAX_SIZE))
>>               goto ring_enqueue;
>>
>> -     cache = &mp->local_cache[lcore_id];
>>       cache_objs = &cache->objs[cache->len];
>>
>>       /*
>> @@ -792,10 +848,10 @@ __mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
>>
>>       cache->len += n;
>>
>> -     if (cache->len >= flushthresh) {
>> -             rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache_size],
>> -                             cache->len - cache_size);
>> -             cache->len = cache_size;
>> +     if (cache->len >= cache->flushthresh) {
>> +             rte_ring_mp_enqueue_bulk(mp->ring, &cache->objs[cache->size],
>> +                             cache->len - cache->size);
>> +             cache->len = cache->size;
>>       }
>>
>>       return;
>> @@ -820,41 +876,27 @@ ring_enqueue:
>>  #endif
>>  }
>>

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Thomas Monjalon]

2016-04-19 18:39, Lazaros Koromilas:
> >> Removes the API calls:
> >>
> >>     rte_mempool_sp_put_bulk(mp, obj_table, n)
> >>     rte_mempool_sc_get_bulk(mp, obj_table, n)
> >>     rte_mempool_sp_put(mp, obj)
> >>     rte_mempool_sc_get(mp, obj)
> >
> > Hmm, shouldn't we deprecate it first for a release before removing completely?
> > Let say for now you can just make them macros that calls the remaining functions or so.
> 
> How do we mark the calls as deprecated? The librte_compat stuff don't
> apply here as we don't have a different version of the same symbol or
> something. Do I need to put them as a notice?

Yes a notice and add __rte_deprecated.

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Olivier MATZ]

Hi Lazaros,

Sorry for the late review. Please find some comments,
in addition to what Konstantin already said.

On 04/04/2016 05:43 PM, Lazaros Koromilas wrote:

> --- a/app/test/test_mempool.c
> +++ b/app/test/test_mempool.c
> @@ -79,6 +79,7 @@
>
>   static struct rte_mempool *mp;
>   static struct rte_mempool *mp_cache, *mp_nocache;
> +static int use_external_cache;
>
>   static rte_atomic32_t synchro;
>
> @@ -107,19 +108,33 @@ test_mempool_basic(void)
>   	char *obj_data;
>   	int ret = 0;
>   	unsigned i, j;
> +	struct rte_mempool_cache *cache;
> +
> +	if (use_external_cache)
> +		/* Create a user-owned mempool cache. */
> +		cache = rte_mempool_cache_create(RTE_MEMPOOL_CACHE_MAX_SIZE,
> +						 SOCKET_ID_ANY);
> +	else
> +		cache = rte_mempool_default_cache(mp, rte_lcore_id());

Shouldn't we return an error if rte_mempool_default_cache()
failed? Even if the cache can be NULL for get/put, it would
crash on the flush() operation, so it's better to return an
error if the cache cannot be allocated.

I also think the resource should be freed on error, maybe
by doing "goto fail" instead of "return -1" in the subsequent
checks. Note that I also reworked this test in my patchset, see:
http://dpdk.org/dev/patchwork/patch/12069/

I think the "use_external_cache" parameter should be a parameter
instead of a global variable, like I've done for the mempool pointer.


> --- a/app/test/test_mempool_perf.c
> +++ b/app/test/test_mempool_perf.c
> @@ -98,6 +101,8 @@
>
>   static struct rte_mempool *mp;
>   static struct rte_mempool *mp_cache, *mp_nocache;
> +static int use_external_cache;
> +static unsigned external_cache_size = RTE_MEMPOOL_CACHE_MAX_SIZE;
>
>   static rte_atomic32_t synchro;
>

The same comment (global vs parameter) could apply here, but it would
require to rework the full test file... so maybe it's off topic.


> @@ -137,6 +142,14 @@ per_lcore_mempool_test(__attribute__((unused)) void *arg)
>   	int ret;
>   	uint64_t start_cycles, end_cycles;
>   	uint64_t time_diff = 0, hz = rte_get_timer_hz();
> +	struct rte_mempool_cache *cache;
> +
> +	if (use_external_cache)
> +		/* Create a user-owned mempool cache. */
> +		cache = rte_mempool_cache_create(external_cache_size,
> +						 SOCKET_ID_ANY);
> +	else
> +		cache = rte_mempool_default_cache(mp, lcore_id);
>
>   	/* n_get_bulk and n_put_bulk must be divisors of n_keep */
>   	if (((n_keep / n_get_bulk) * n_get_bulk) != n_keep)

Same comments than above (check return value != NULL).

The cache creation could be moved some lines below to avoid
to free the resource on error.


> --- a/lib/librte_eal/common/eal_common_log.c
> +++ b/lib/librte_eal/common/eal_common_log.c
> @@ -125,7 +125,7 @@ rte_log_add_in_history(const char *buf, size_t size)
>   		}
>   	}
>   	else {
> -		if (rte_mempool_mc_get(log_history_mp, &obj) < 0)
> +		if (rte_mempool_get(log_history_mp, &obj) < 0)
>   			obj = NULL;
>   		hist_buf = obj;
>   	}

After seeing many changes like this, I wonder if it's possible
to move these in a separate commit:
"mempool: deprecate specific get/put functions"

It would remove some noise in the "interesting" part. I suggest
the following order:
   mempool: deprecate specific get/put functions
   mempool: use bit flags instead of is_mp and is_mc
   mempool: allow for user-owned mempool	caches

What do you think?


>   static inline void __attribute__((always_inline))
> -__mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -		    unsigned n, int is_mp)
> +__mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
> +		      unsigned n, struct rte_mempool_cache *cache, int is_mp)
>   {
> -	struct rte_mempool_cache *cache;
>   	uint32_t index;
>   	void **cache_objs;
> -	unsigned lcore_id = rte_lcore_id();
> -	uint32_t cache_size = mp->cache_size;
> -	uint32_t flushthresh = mp->cache_flushthresh;
>
>   	/* increment stat now, adding in mempool always success */
>   	__MEMPOOL_STAT_ADD(mp, put, n);
>
> -	/* cache is not enabled or single producer or non-EAL thread */
> -	if (unlikely(cache_size == 0 || is_mp == 0 ||
> -		     lcore_id >= RTE_MAX_LCORE))
> +	/* No cache provided or cache is not enabled or single producer */
> +	if (unlikely(cache == NULL || cache->size == 0 || is_mp == 0))
>   		goto ring_enqueue;

Is it possible that cache->size == 0?
I suggest to remove that test and ensure that size != 0 at cache
creation.

> -__mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -		    unsigned n, int is_mp)

> +__mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
> +		      unsigned n, struct rte_mempool_cache *cache, int is_mp)

> -rte_mempool_mp_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -			unsigned n)

> -rte_mempool_sp_put_bulk(struct rte_mempool *mp, void * const *obj_table,
> -			unsigned n)

> +rte_mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
> +			unsigned n, struct rte_mempool_cache *cache, int is_mp)

> -rte_mempool_mp_put(struct rte_mempool *mp, void *obj)

> -rte_mempool_sp_put(struct rte_mempool *mp, void *obj)

> -__mempool_get_bulk(struct rte_mempool *mp, void **obj_table,
> -		   unsigned n, int is_mc)

> +__mempool_generic_get(struct rte_mempool *mp, void **obj_table,
> +		      unsigned n, struct rte_mempool_cache *cache, int is_mc)

> -rte_mempool_mc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)

> -rte_mempool_sc_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned n)

> +rte_mempool_generic_get(struct rte_mempool *mp, void **obj_table,

> -rte_mempool_mc_get(struct rte_mempool *mp, void **obj_p)

> -rte_mempool_sc_get(struct rte_mempool *mp, void **obj_p)

I think removing the mp/mc/sp/sc functions (or deprecate them for now,
as suggested by Konstantin/Thomas) is a good thing for code readability
in rte_mempool.h. Thanks for doing this!


Regards,
Olivier

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Olivier Matz]

Hi Lazaros,

On 05/11/2016 11:56 AM, Olivier MATZ wrote:
> Hi Lazaros,
> 
> Sorry for the late review. Please find some comments,
> in addition to what Konstantin already said.
> 

Will you have the time to send a v3 before the end of the
integration deadline at the end of the week?

I think it should be rebased on top of latest mempool series
from David Hunt:
http://dpdk.org/ml/archives/dev/2016-June/040897.html

Regards,
Olivier

Re: [dpdk-dev] [PATCH v2 1/2] mempool: allow for user-owned mempool caches

[Lazaros Koromilas]

Hi Olivier,

I have it in my queue, I'll do my best to have it before the deadline.

Thanks!
Lazaros.

On Mon, Jun 13, 2016 at 1:21 PM, Olivier Matz <olivier.matz@6wind.com> wrote:
> Hi Lazaros,
>
> On 05/11/2016 11:56 AM, Olivier MATZ wrote:
>> Hi Lazaros,
>>
>> Sorry for the late review. Please find some comments,
>> in addition to what Konstantin already said.
>>
>
> Will you have the time to send a v3 before the end of the
> integration deadline at the end of the week?
>
> I think it should be rebased on top of latest mempool series
> from David Hunt:
> http://dpdk.org/ml/archives/dev/2016-June/040897.html
>
> Regards,
> Olivier