Re: [dpdk-dev] [PATCH v4 1/4] mempool: add event callbacks

[Andrew Rybchenko <andrew.rybchenko@oktetlabs.ru>]

On 10/13/21 2:01 PM, Dmitry Kozlyuk wrote:
> Data path performance can benefit if the PMD knows which memory it will
> need to handle in advance, before the first mbuf is sent to the PMD.
> It is impractical, however, to consider all allocated memory for this
> purpose. Most often mbuf memory comes from mempools that can come and
> go. PMD can enumerate existing mempools on device start, but it also
> needs to track creation and destruction of mempools after the forwarding
> starts but before an mbuf from the new mempool is sent to the device.
> 
> Add an API to register callback for mempool life cycle events:
> * rte_mempool_event_callback_register()
> * rte_mempool_event_callback_unregister()
> Currently tracked events are:
> * RTE_MEMPOOL_EVENT_READY (after populating a mempool)
> * RTE_MEMPOOL_EVENT_DESTROY (before freeing a mempool)
> Provide a unit test for the new API.
> The new API is internal, because it is primarily demanded by PMDs that
> may need to deal with any mempools and do not control their creation,
> while an application, on the other hand, knows which mempools it creates
> and doesn't care about internal mempools PMDs might create.
> 
> Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> Acked-by: Matan Azrad <matan@nvidia.com>

With below review notes processed

Reviewed-by: Andrew Rybchenko <andrew.rybchenko@oktetlabs.ru>

[snip]

> diff --git a/lib/mempool/rte_mempool.c b/lib/mempool/rte_mempool.c
> index c5f859ae71..51c0ba2931 100644
> --- a/lib/mempool/rte_mempool.c
> +++ b/lib/mempool/rte_mempool.c

[snip]

> @@ -360,6 +372,10 @@ rte_mempool_populate_iova(struct rte_mempool *mp, char *vaddr,
>  	STAILQ_INSERT_TAIL(&mp->mem_list, memhdr, next);
>  	mp->nb_mem_chunks++;
>  
> +	/* Report the mempool as ready only when fully populated. */
> +	if (mp->populated_size >= mp->size)
> +		mempool_event_callback_invoke(RTE_MEMPOOL_EVENT_READY, mp);
> +
>  	rte_mempool_trace_populate_iova(mp, vaddr, iova, len, free_cb, opaque);
>  	return i;
>  
> @@ -722,6 +738,7 @@ rte_mempool_free(struct rte_mempool *mp)
>  	}
>  	rte_mcfg_tailq_write_unlock();
>  
> +	mempool_event_callback_invoke(RTE_MEMPOOL_EVENT_DESTROY, mp);
>  	rte_mempool_trace_free(mp);
>  	rte_mempool_free_memchunks(mp);
>  	rte_mempool_ops_free(mp);
> @@ -1343,3 +1360,123 @@ void rte_mempool_walk(void (*func)(struct rte_mempool *, void *),
>  
>  	rte_mcfg_mempool_read_unlock();
>  }
> +
> +struct mempool_callback {

It sounds like it is a mempool callback itself.
Consider: mempool_event_callback_data.
I think this way it will be consistent.

> +	rte_mempool_event_callback *func;
> +	void *user_data;
> +};
> +
> +static void
> +mempool_event_callback_invoke(enum rte_mempool_event event,
> +			      struct rte_mempool *mp)
> +{
> +	struct mempool_callback_list *list;
> +	struct rte_tailq_entry *te;
> +	void *tmp_te;
> +
> +	rte_mcfg_tailq_read_lock();
> +	list = RTE_TAILQ_CAST(callback_tailq.head, mempool_callback_list);
> +	RTE_TAILQ_FOREACH_SAFE(te, list, next, tmp_te) {
> +		struct mempool_callback *cb = te->data;
> +		rte_mcfg_tailq_read_unlock();
> +		cb->func(event, mp, cb->user_data);
> +		rte_mcfg_tailq_read_lock();
> +	}
> +	rte_mcfg_tailq_read_unlock();
> +}
> +
> +int
> +rte_mempool_event_callback_register(rte_mempool_event_callback *func,
> +				    void *user_data)
> +{
> +	struct mempool_callback_list *list;
> +	struct rte_tailq_entry *te = NULL;
> +	struct mempool_callback *cb;
> +	void *tmp_te;
> +	int ret;
> +
> +	if (func == NULL) {
> +		rte_errno = EINVAL;
> +		return -rte_errno;
> +	}
> +
> +	rte_mcfg_mempool_read_lock();
> +	rte_mcfg_tailq_write_lock();
> +
> +	list = RTE_TAILQ_CAST(callback_tailq.head, mempool_callback_list);
> +	RTE_TAILQ_FOREACH_SAFE(te, list, next, tmp_te) {
> +		struct mempool_callback *cb =
> +					(struct mempool_callback *)te->data;
> +		if (cb->func == func && cb->user_data == user_data) {
> +			ret = -EEXIST;
> +			goto exit;
> +		}
> +	}
> +
> +	te = rte_zmalloc("MEMPOOL_TAILQ_ENTRY", sizeof(*te), 0);
> +	if (te == NULL) {
> +		RTE_LOG(ERR, MEMPOOL,
> +			"Cannot allocate event callback tailq entry!\n");
> +		ret = -ENOMEM;
> +		goto exit;
> +	}
> +
> +	cb = rte_malloc("MEMPOOL_EVENT_CALLBACK", sizeof(*cb), 0);
> +	if (cb == NULL) {
> +		RTE_LOG(ERR, MEMPOOL,
> +			"Cannot allocate event callback!\n");
> +		rte_free(te);
> +		ret = -ENOMEM;
> +		goto exit;
> +	}
> +
> +	cb->func = func;
> +	cb->user_data = user_data;
> +	te->data = cb;
> +	TAILQ_INSERT_TAIL(list, te, next);
> +	ret = 0;
> +
> +exit:
> +	rte_mcfg_tailq_write_unlock();
> +	rte_mcfg_mempool_read_unlock();
> +	rte_errno = -ret;
> +	return ret;
> +}
> +
> +int
> +rte_mempool_event_callback_unregister(rte_mempool_event_callback *func,
> +				      void *user_data)
> +{
> +	struct mempool_callback_list *list;
> +	struct rte_tailq_entry *te = NULL;
> +	struct mempool_callback *cb;
> +	int ret;
> +
> +	if (rte_eal_process_type() != RTE_PROC_PRIMARY) {
> +		rte_errno = EPERM;
> +		return -1;

The function should behave consistencly. Below you
return negative error. Here just -1. I think it
would be more constent to return -rte_errno here.

> +	}
> +
> +	rte_mcfg_mempool_read_lock();
> +	rte_mcfg_tailq_write_lock();
> +	ret = -ENOENT;
> +	list = RTE_TAILQ_CAST(callback_tailq.head, mempool_callback_list);
> +	TAILQ_FOREACH(te, list, next) {
> +		cb = (struct mempool_callback *)te->data;
> +		if (cb->func == func && cb->user_data == user_data)
> +			break;
> +	}
> +	if (te != NULL) {

Here we rely on the fact that TAILQ_FOREACH()
exists with te==NULL in the case of no such
entry. I'd suggest to avoid the assumption.
I.e. do below two lines above before break and
have not the if condition her at all.

> +		TAILQ_REMOVE(list, te, next);
> +		ret = 0;
> +	}
> +	rte_mcfg_tailq_write_unlock();
> +	rte_mcfg_mempool_read_unlock();
> +
> +	if (ret == 0) {
> +		rte_free(te);
> +		rte_free(cb);
> +	}
> +	rte_errno = -ret;
> +	return ret;
> +}
> diff --git a/lib/mempool/rte_mempool.h b/lib/mempool/rte_mempool.h
> index f57ecbd6fc..663123042f 100644
> --- a/lib/mempool/rte_mempool.h
> +++ b/lib/mempool/rte_mempool.h
> @@ -1774,6 +1774,67 @@ void rte_mempool_walk(void (*func)(struct rte_mempool *, void *arg),
>  int
>  rte_mempool_get_page_size(struct rte_mempool *mp, size_t *pg_sz);
>  
> +/**
> + * Mempool event type.
> + * @internal

Shouldn't internal go first?

> + */
> +enum rte_mempool_event {
> +	/** Occurs after a mempool is successfully populated. */

successfully -> fully ?

> +	RTE_MEMPOOL_EVENT_READY = 0,
> +	/** Occurs before destruction of a mempool begins. */
> +	RTE_MEMPOOL_EVENT_DESTROY = 1,
> +};

[snip]