[PATCH v1 3/6] eventdev: support optional dequeue callbacks

[Tyler Retzlaff]

On Mon, Oct 16, 2023 at 01:57:12PM -0700, Sivaprasad Tummala wrote:
> Add optional support for inline event processing within pmd dequeue
> call. For a dequeue callback, events dequeued from the event port
> were passed them to a callback function if configured, to allow
> additional processing. e.g. unpack batch of packets from each event
> on dequeue, before passing back to the application.
> 
> Signed-off-by: Sivaprasad Tummala <sivaprasad.tummala at amd.com>
> ---
>  lib/eventdev/eventdev_pmd.h      |  38 +++++++++++
>  lib/eventdev/eventdev_private.c  |   2 +
>  lib/eventdev/rte_eventdev.c      | 107 +++++++++++++++++++++++++++++++
>  lib/eventdev/rte_eventdev.h      |  95 +++++++++++++++++++++++++++
>  lib/eventdev/rte_eventdev_core.h |  12 +++-
>  lib/eventdev/version.map         |   3 +
>  6 files changed, 256 insertions(+), 1 deletion(-)
> 
> diff --git a/lib/eventdev/eventdev_pmd.h b/lib/eventdev/eventdev_pmd.h
> index a0ee768ce7..ce067b1d5d 100644
> --- a/lib/eventdev/eventdev_pmd.h
> +++ b/lib/eventdev/eventdev_pmd.h
> @@ -97,6 +97,19 @@ struct rte_eventdev_global {
>  	uint8_t nb_devs;	/**< Number of devices found */
>  };
>  
> +/**
> + * @internal
> + * Structure used to hold information about the callbacks to be called for a
> + * port on dequeue.
> + */
> +struct rte_event_dequeue_callback {
> +	struct rte_event_dequeue_callback *next;
> +	union{
> +		rte_dequeue_callback_fn dequeue;
> +	} fn;
> +	void *param;
> +};
> +
>  /**
>   * @internal
>   * The data part, with no function pointers, associated with each device.
> @@ -171,6 +184,10 @@ struct rte_eventdev {
>  	/**< Pointer to PMD dequeue burst function. */
>  	event_maintain_t maintain;
>  	/**< Pointer to PMD port maintenance function. */
> +	struct rte_event_dequeue_callback *post_dequeue_burst_cbs[RTE_EVENT_MAX_PORTS_PER_DEV];
> +	/**<  User-supplied functions called from dequeue_burst to post-process
> +	 * received packets before passing them to the user
> +	 */
>  	event_tx_adapter_enqueue_t txa_enqueue_same_dest;
>  	/**< Pointer to PMD eth Tx adapter burst enqueue function with
>  	 * events destined to same Eth port & Tx queue.
> @@ -245,6 +262,27 @@ rte_event_pmd_is_valid_dev(uint8_t dev_id)
>  		return 1;
>  }
>  
> +/**
> + * Executes all the user application registered callbacks for the specific
> + * event device.
> + *
> + * @param dev_id
> + *   Event device index.
> + * @param port_id
> + *   Event port index
> + * @param ev
> + *   Points to an array of *nb_events* objects of type *rte_event* structure
> + *   for output to be populated with the dequeued event objects.
> + * @param nb_events
> + *   number of event objects
> + *
> + * @return
> + * The number of event objects
> + */
> +__rte_internal
> +uint16_t rte_eventdev_pmd_dequeue_callback_process(uint8_t dev_id,
> +		uint8_t port_id, struct rte_event ev[], uint16_t nb_events);
> +
>  /**
>   * Definitions of all functions exported by a driver through the
>   * generic structure of type *event_dev_ops* supplied in the
> diff --git a/lib/eventdev/eventdev_private.c b/lib/eventdev/eventdev_private.c
> index 017f97ccab..052c526ce0 100644
> --- a/lib/eventdev/eventdev_private.c
> +++ b/lib/eventdev/eventdev_private.c
> @@ -137,4 +137,6 @@ event_dev_fp_ops_set(struct rte_event_fp_ops *fp_op,
>  	fp_op->dma_enqueue = dev->dma_enqueue;
>  	fp_op->profile_switch = dev->profile_switch;
>  	fp_op->data = dev->data->ports;
> +	fp_op->ev_port.clbk = (void **)(uintptr_t)dev->post_dequeue_burst_cbs;
> +	fp_op->ev_port.data = dev->data->ports;
>  }
> diff --git a/lib/eventdev/rte_eventdev.c b/lib/eventdev/rte_eventdev.c
> index 5feb4326a2..f2540a6aa8 100644
> --- a/lib/eventdev/rte_eventdev.c
> +++ b/lib/eventdev/rte_eventdev.c
> @@ -18,6 +18,7 @@
>  #include <rte_common.h>
>  #include <rte_malloc.h>
>  #include <rte_errno.h>
> +#include <rte_stdatomic.h>
>  #include <ethdev_driver.h>
>  #include <rte_cryptodev.h>
>  #include <rte_dmadev.h>
> @@ -39,6 +40,9 @@ static struct rte_eventdev_global eventdev_globals = {
>  /* Public fastpath APIs. */
>  struct rte_event_fp_ops rte_event_fp_ops[RTE_EVENT_MAX_DEVS];
>  
> +/* spinlock for add/remove dequeue callbacks */
> +static rte_spinlock_t event_dev_dequeue_cb_lock = RTE_SPINLOCK_INITIALIZER;
> +
>  /* Event dev north bound API implementation */
>  
>  uint8_t
> @@ -884,6 +888,109 @@ rte_event_port_attr_get(uint8_t dev_id, uint8_t port_id, uint32_t attr_id,
>  	return 0;
>  }
>  
> +const struct rte_event_dequeue_callback *
> +rte_event_add_dequeue_callback(uint8_t dev_id, uint8_t port_id,
> +		rte_dequeue_callback_fn fn, void *user_param)
> +{
> +	struct rte_eventdev *dev;
> +	struct rte_event_dequeue_callback *cb;
> +	struct rte_event_dequeue_callback *tail;
> +
> +	/* check input parameters */
> +	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, NULL);
> +	dev = &rte_eventdevs[dev_id];
> +	if (!is_valid_port(dev, port_id)) {
> +		RTE_EDEV_LOG_ERR("Invalid port_id=%" PRIu8, port_id);
> +		return NULL;
> +	}
> +
> +	cb = rte_zmalloc(NULL, sizeof(*cb), 0);
> +	if (cb == NULL) {
> +		rte_errno = ENOMEM;
> +		return NULL;
> +	}
> +	cb->fn.dequeue = fn;
> +	cb->param = user_param;
> +
> +	rte_spinlock_lock(&event_dev_dequeue_cb_lock);
> +	/* Add the callbacks in fifo order. */
> +	tail = rte_eventdevs[dev_id].post_dequeue_burst_cbs[port_id];
> +	if (!tail) {
> +		/* Stores to cb->fn and cb->param should complete before
> +		 * cb is visible to data plane.
> +		 */
> +		rte_atomic_store_explicit(
> +			&rte_eventdevs[dev_id].post_dequeue_burst_cbs[port_id],
> +			cb, __ATOMIC_RELEASE);
                            ^^^^^^^^^^^^^^^^ rte_memory_order_release (rte_stdatomic.h)

> +	} else {
> +		while (tail->next)
> +			tail = tail->next;
> +		/* Stores to cb->fn and cb->param should complete before
> +		 * cb is visible to data plane.
> +		 */
> +		rte_atomic_store_explicit(&tail->next, cb, __ATOMIC_RELEASE);
							   ^^^^ same here

> +	}
> +	rte_spinlock_unlock(&event_dev_dequeue_cb_lock);
> +
> +	return cb;
> +}
> +
> +int
> +rte_event_remove_dequeue_callback(uint8_t dev_id, uint8_t port_id,
> +		const struct rte_event_dequeue_callback *user_cb)
> +{
> +	struct rte_eventdev *dev;
> +	struct rte_event_dequeue_callback *cb;
> +	struct rte_event_dequeue_callback **prev_cb;
> +
> +	/* Check input parameters. */
> +	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, -EINVAL);
> +	dev = &rte_eventdevs[dev_id];
> +	if (user_cb == NULL || !is_valid_port(dev, port_id))
> +		return -EINVAL;
> +
> +	rte_spinlock_lock(&event_dev_dequeue_cb_lock);
> +	prev_cb = &dev->post_dequeue_burst_cbs[port_id];
> +	for (; *prev_cb != NULL; prev_cb = &cb->next) {
> +		cb = *prev_cb;
> +		if (cb == user_cb) {
> +			/* Remove the user cb from the callback list. */
> +			rte_atomic_store_explicit(prev_cb, cb->next,
> +						__ATOMIC_RELAXED);
						^^^ and here

> +			break;
> +		}
> +	}
> +	rte_spinlock_unlock(&event_dev_dequeue_cb_lock);
> +
> +	return 0;
> +}
> +
> +uint16_t rte_eventdev_pmd_dequeue_callback_process(uint8_t dev_id,
> +		uint8_t port_id, struct rte_event ev[], uint16_t nb_events)
> +{
> +	struct rte_event_dequeue_callback *cb;
> +	const struct rte_event_fp_ops *fp_ops;
> +
> +	fp_ops = &rte_event_fp_ops[dev_id];
> +
> +	/* __ATOMIC_RELEASE memory order was used when the
> +	 * call back was inserted into the list.
> +	 * Since there is a clear dependency between loading
> +	 * cb and cb->fn/cb->next, __ATOMIC_ACQUIRE memory order is
> +	 * not required.
> +	 */
> +	cb = rte_atomic_load_explicit((void **)&fp_ops->ev_port.clbk[port_id],
				       ^^^^^^^ needs to be __rte_atomic
qualified

> +					__ATOMIC_RELAXED);
					^^^^ rte_memory_order