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

[Sivaprasad Tummala <sivaprasad.tummala@amd.com>]

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@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);
+	} 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);
+	}
+	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);
+			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],
+					__ATOMIC_RELAXED);
+	if (unlikely(cb != NULL))
+		while (cb != NULL) {
+			nb_events = cb->fn.dequeue(dev_id, port_id, ev,
+			nb_events, cb->param);
+			cb = cb->next;
+		}
+
+	return nb_events;
+}
+
 int
 rte_event_port_get_monitor_addr(uint8_t dev_id, uint8_t port_id,
 				struct rte_power_monitor_cond *pmc)
diff --git a/lib/eventdev/rte_eventdev.h b/lib/eventdev/rte_eventdev.h
index 38dbbc2617..c0097c0a23 100644
--- a/lib/eventdev/rte_eventdev.h
+++ b/lib/eventdev/rte_eventdev.h
@@ -954,6 +954,101 @@ void
 rte_event_port_quiesce(uint8_t dev_id, uint8_t port_id,
 		       rte_eventdev_port_flush_t release_cb, void *args);
 
+struct rte_event_dequeue_callback;
+
+/**
+ * Function type used for dequeue event processing callbacks.
+ *
+ * The callback function is called on dequeue with a burst of events that have
+ * been received on the given event port.
+ *
+ * @param dev_id
+ *   The identifier of the device.
+ * @param port_id
+ *   The identifier of the event port.
+ * @param[out] 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
+ *   The maximum number of event objects to dequeue, typically number of
+ *   rte_event_port_dequeue_depth() available for this port.
+ * @param opaque
+ *   Opaque pointer of event port callback related data.
+ *
+ * @return
+ *  The number of event objects returned to the user.
+ */
+typedef uint16_t (*rte_dequeue_callback_fn)(uint8_t dev_id, uint8_t port_id,
+		struct rte_event *ev, uint16_t nb_events, void *user_param);
+
+/**
+ * Add a callback to be called on event dequeue on a given event device port.
+ *
+ * This API configures a function to be called for each burst of
+ * events dequeued on a given event device port. The return value is a pointer
+ * that can be used to later remove the callback using
+ * rte_event_remove_dequeue_callback().
+ *
+ * Multiple functions are called in the order that they are added.
+ *
+ * @param dev_id
+ *   The identifier of the device.
+ * @param port_id
+ *   The identifier of the event port.
+ * @param fn
+ *   The callback function
+ * @param user_param
+ *   A generic pointer parameter which will be passed to each invocation of the
+ *   callback function on this event device port. Inter-thread synchronization
+ *   of any user data changes is the responsibility of the user.
+ *
+ * @return
+ *   NULL on error.
+ *   On success, a pointer value which can later be used to remove the callback.
+ */
+__rte_experimental
+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);
+
+/**
+ * Remove a dequeue event callback from a given event device port.
+ *
+ * This API is used to removed callbacks that were added to a event device port
+ * using rte_event_add_dequeue_callback().
+ *
+ * Note: the callback is removed from the callback list but it isn't freed
+ * since the it may still be in use. The memory for the callback can be
+ * subsequently freed back by the application by calling rte_free():
+ *
+ * - Immediately - if the device is stopped, or the user knows that no
+ *   callbacks are in flight e.g. if called from the thread doing dequeue
+ *   on that port.
+ *
+ * - After a short delay - where the delay is sufficient to allow any
+ *   in-flight callbacks to complete. Alternately, the RCU mechanism can be
+ *   used to detect when data plane threads have ceased referencing the
+ *   callback memory.
+ *
+ * @param dev_id
+ *   The identifier of the device.
+ * @param port_id
+ *   The identifier of the event port.
+ * @param user_cb
+ *   The callback function
+ *
+ * @return
+ *   - 0: Success. Callback was removed.
+ *   - -ENODEV:  If *dev_id* is invalid.
+ *   - -EINVAL:  The port_id is out of range, or the callback
+ *               is NULL.
+ */
+__rte_experimental
+int
+rte_event_remove_dequeue_callback(uint8_t dev_id, uint8_t port_id,
+		const struct rte_event_dequeue_callback *user_cb);
+
+
 /**
  * The queue depth of the port on the enqueue side
  */
diff --git a/lib/eventdev/rte_eventdev_core.h b/lib/eventdev/rte_eventdev_core.h
index 5b405518d1..5ce93c4b6f 100644
--- a/lib/eventdev/rte_eventdev_core.h
+++ b/lib/eventdev/rte_eventdev_core.h
@@ -49,6 +49,14 @@ typedef uint16_t (*event_dma_adapter_enqueue_t)(void *port, struct rte_event ev[
 typedef int (*event_profile_switch_t)(void *port, uint8_t profile);
 /**< @internal Switch active link profile on the event port. */
 
+struct rte_eventdev_port_data {
+	void **data;
+	/**< points to array of internal port data pointers */
+	void **clbk;
+	/**< points to array of port callback data pointers */
+};
+/**< @internal Structure used to hold opaque eventdev port data. */
+
 struct rte_event_fp_ops {
 	void **data;
 	/**< points to array of internal port data pointers */
@@ -76,7 +84,9 @@ struct rte_event_fp_ops {
 	/**< PMD DMA adapter enqueue function. */
 	event_profile_switch_t profile_switch;
 	/**< PMD Event switch profile function. */
-	uintptr_t reserved[4];
+	struct rte_eventdev_port_data ev_port;
+	/**< Eventdev port data. */
+	uintptr_t reserved[1];
 } __rte_cache_aligned;
 
 extern struct rte_event_fp_ops rte_event_fp_ops[RTE_EVENT_MAX_DEVS];
diff --git a/lib/eventdev/version.map b/lib/eventdev/version.map
index fa9eb069ff..a0c7aa5bbd 100644
--- a/lib/eventdev/version.map
+++ b/lib/eventdev/version.map
@@ -155,6 +155,8 @@ EXPERIMENTAL {
 	rte_event_port_profile_unlink;
 	rte_event_port_profile_links_get;
 	rte_event_port_get_monitor_addr;
+	rte_event_add_dequeue_callback;
+	rte_event_remove_dequeue_callback;
 	__rte_eventdev_trace_port_profile_switch;
 };
 
@@ -165,6 +167,7 @@ INTERNAL {
 	event_dev_fp_ops_set;
 	event_dev_probing_finish;
 	rte_event_pmd_allocate;
+	rte_eventdev_pmd_dequeue_callback_process;
 	rte_event_pmd_get_named_dev;
 	rte_event_pmd_is_valid_dev;
 	rte_event_pmd_pci_probe;
-- 
2.34.1