[dpdk-dev] [PATCH v1 0/2] event enqueue-dequeue callback handler

[dpdk-dev] [PATCH v1 0/2] event enqueue-dequeue callback handler

[Vipin Varghese]

The patch series is an attempt to add callback handlers to event device.

Motivation
==========

Allow user to debug and modify the event list in event device enqueue-dequeue
stages.

The enqueue stage is modified to allow user handler to be invoked prior to
actual device enqueue. The dequeue stage is modified to allow user handler
to be invoked post to actual device dequeue.

Status
======

The new APIs are added as experimental.

Change Log:
==========

Vipin Varghese (2):
  lib/event: add callback handlers for event
  examples/event: add callback handle

 config/common_base                           |   1 +
 examples/eventdev_pipeline/main.c            |  49 +++
 lib/librte_eventdev/rte_eventdev.c           | 361 +++++++++++++++++++
 lib/librte_eventdev/rte_eventdev.h           | 267 +++++++++++++-
 lib/librte_eventdev/rte_eventdev_version.map |   8 +
 5 files changed, 682 insertions(+), 4 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v1 1/2] lib/event: add callback handlers for event

[Vipin Varghese]

Add callback event handler for enqueue dequeue operation on event
device. The pre-enqueue and post-dequeue are the selected to invoke
user callback handler.

Signed-off-by: Vipin Varghese <vipin.varghese@intel.com>
---
 config/common_base                           |   1 +
 lib/librte_eventdev/rte_eventdev.c           | 361 +++++++++++++++++++
 lib/librte_eventdev/rte_eventdev.h           | 267 +++++++++++++-
 lib/librte_eventdev/rte_eventdev_version.map |   8 +
 4 files changed, 633 insertions(+), 4 deletions(-)

diff --git a/config/common_base b/config/common_base
index 6f19ad5d2..ec29455d2 100644
--- a/config/common_base
+++ b/config/common_base
@@ -681,6 +681,7 @@ CONFIG_RTE_EVENT_TIMER_ADAPTER_NUM_MAX=32
 CONFIG_RTE_EVENT_ETH_INTR_RING_SIZE=1024
 CONFIG_RTE_EVENT_CRYPTO_ADAPTER_MAX_INSTANCE=32
 CONFIG_RTE_EVENT_ETH_TX_ADAPTER_MAX_INSTANCE=32
+CONFIG_RTE_EVENTDEV_ENQDEQ_CALLBACKS=n
 
 #
 # Compile PMD for skeleton event device
diff --git a/lib/librte_eventdev/rte_eventdev.c b/lib/librte_eventdev/rte_eventdev.c
index cc3199fb6..de9c772ce 100644
--- a/lib/librte_eventdev/rte_eventdev.c
+++ b/lib/librte_eventdev/rte_eventdev.c
@@ -44,6 +44,11 @@ static struct rte_eventdev_global eventdev_globals = {
 	.nb_devs		= 0
 };
 
+/* spinlock for pre-enqueue callbacks */
+rte_spinlock_t rte_eventdev_preenq_cb_lock = RTE_SPINLOCK_INITIALIZER;
+/* spinlock for post-dequeue callbacks */
+rte_spinlock_t rte_eventdev_pstdeq_cb_lock = RTE_SPINLOCK_INITIALIZER;
+
 /* Event dev north bound API implementation */
 
 uint8_t
@@ -524,6 +529,10 @@ rte_event_dev_configure(uint8_t dev_id,
 	}
 
 	dev->data->event_dev_cap = info.event_dev_cap;
+
+	TAILQ_INIT(&(dev->enq_cbs));
+	TAILQ_INIT(&(dev->deq_cbs));
+
 	return diag;
 }
 
@@ -1278,6 +1287,358 @@ rte_event_dev_close(uint8_t dev_id)
 	return (*dev->dev_ops->dev_close)(dev);
 }
 
+int __rte_experimental
+rte_event_preenq_callback_register(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	rte_errno = ENOTSUP;
+	return -ENOTSUP;
+#endif
+
+	struct rte_eventdev *dev;
+	rte_spinlock_t *lock = &rte_eventdev_preenq_cb_lock;
+	struct rte_eventdev_callback *user_cb = NULL;
+
+	if (!cb_fn)
+		return -EINVAL;
+
+	dev = &rte_eventdevs[eventdev_id];
+	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->dev_stop, -EINVAL);
+
+	rte_spinlock_lock(lock);
+
+	TAILQ_FOREACH(user_cb, &(dev->enq_cbs), next)
+	{
+		if (user_cb->cb_fn == cb_fn &&
+		user_cb->cb_arg == cb_arg)
+			break;
+	}
+
+	/* create a new callback. */
+	if (user_cb == NULL) {
+		user_cb = rte_zmalloc("EVENT_USER_CALLBACK",
+				sizeof(struct rte_eventdev_callback), 0);
+		if (user_cb != NULL) {
+			user_cb->cb_fn = cb_fn;
+			user_cb->cb_arg = cb_arg;
+			TAILQ_INSERT_TAIL(&(dev->enq_cbs), user_cb, next);
+		}
+	}
+
+	rte_spinlock_unlock(lock);
+
+	return (user_cb == NULL) ? -ENOMEM : 0;
+}
+
+int __rte_experimental
+rte_event_pstdeq_callback_register(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	rte_errno = ENOTSUP;
+	return -ENOTSUP;
+#endif
+
+	struct rte_eventdev *dev;
+	static rte_spinlock_t *lock = &rte_eventdev_pstdeq_cb_lock;
+	struct rte_eventdev_callback *user_cb = NULL;
+
+	if (!cb_fn)
+		return -EINVAL;
+
+	dev = &rte_eventdevs[eventdev_id];
+	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->dev_stop, -EINVAL);
+
+	rte_spinlock_lock(lock);
+
+	TAILQ_FOREACH(user_cb, &(dev->deq_cbs), next)
+	{
+		if (user_cb->cb_fn == cb_fn &&
+				user_cb->cb_arg == cb_arg)
+			break;
+	}
+
+	/* create a new callback. */
+	if (user_cb == NULL) {
+		user_cb = rte_zmalloc("EVENT_USER_CALLBACK",
+				sizeof(struct rte_eventdev_callback), 0);
+		if (user_cb != NULL) {
+			user_cb->cb_fn = cb_fn;
+			user_cb->cb_arg = cb_arg;
+			TAILQ_INSERT_TAIL(&(dev->deq_cbs), user_cb, next);
+		}
+	}
+
+	rte_spinlock_unlock(lock);
+
+	return (user_cb == NULL) ? -ENOMEM : 0;
+}
+
+int __rte_experimental
+rte_event_remove_preenq_callback(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		const struct rte_eventdev_callback *user_cb)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	rte_errno = ENOTSUP;
+	return -ENOTSUP;
+#endif
+
+	int ret = 0;
+	struct rte_eventdev *dev;
+	static rte_spinlock_t *lock = &rte_eventdev_preenq_cb_lock;
+	struct rte_eventdev_callback *cb, *next_cb;
+
+	if (!user_cb)
+		return -EINVAL;
+
+	dev = &rte_eventdevs[eventdev_id];
+	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->dev_stop, -EINVAL);
+
+	rte_spinlock_lock(lock);
+
+	ret = -EINVAL;
+	for (cb = TAILQ_FIRST(&dev->enq_cbs); cb != NULL; cb = next_cb) {
+		next_cb = TAILQ_NEXT(cb, next);
+
+		if (cb == user_cb) {
+			next_cb = TAILQ_NEXT(cb, next);
+			ret = 0;
+			break;
+		}
+	}
+
+	rte_spinlock_unlock(lock);
+
+	return ret;
+}
+
+int __rte_experimental
+rte_event_remove_pstdeq_callback(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		const struct rte_eventdev_callback *user_cb)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	rte_errno = ENOTSUP;
+	return -ENOTSUP;
+#endif
+
+	int ret = 0;
+	struct rte_eventdev *dev;
+	static rte_spinlock_t *lock = &rte_eventdev_pstdeq_cb_lock;
+	struct rte_eventdev_callback *cb, *next_cb;
+
+	if (!user_cb)
+		return -EINVAL;
+
+	dev = &rte_eventdevs[eventdev_id];
+	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->dev_stop, -EINVAL);
+
+	rte_spinlock_lock(lock);
+
+	ret = -EINVAL;
+	for (cb = TAILQ_FIRST(&dev->deq_cbs); cb != NULL; cb = next_cb) {
+		next_cb = TAILQ_NEXT(cb, next);
+
+		if (cb == user_cb) {
+			next_cb = TAILQ_NEXT(cb, next);
+			ret = 0;
+			break;
+		}
+	}
+
+	rte_spinlock_unlock(lock);
+
+	return ret;
+}
+
+struct rte_eventdev_callback * __rte_experimental
+rte_event_add_preenq_callback(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	rte_errno = ENOTSUP;
+	return NULL;
+#endif
+
+	struct rte_eventdev *dev;
+	rte_spinlock_t *lock = &rte_eventdev_preenq_cb_lock;
+	struct rte_eventdev_callback *user_cb = NULL;
+
+	dev = &rte_eventdevs[eventdev_id];
+
+	if ((*dev->dev_ops->dev_stop == NULL) || (!cb_fn)) {
+		rte_errno = EINVAL;
+		return NULL;
+	}
+
+	rte_spinlock_lock(lock);
+
+	TAILQ_FOREACH(user_cb, &(dev->enq_cbs), next)
+	{
+		if (user_cb->cb_fn == cb_fn &&
+				user_cb->cb_arg == cb_arg)
+			break;
+	}
+
+	/* create a new callback. */
+	if (user_cb == NULL) {
+		user_cb = rte_zmalloc("EVENT_USER_CALLBACK",
+				sizeof(struct rte_eventdev_callback), 0);
+		if (user_cb != NULL) {
+			user_cb->cb_fn = cb_fn;
+			user_cb->cb_arg = cb_arg;
+			TAILQ_INSERT_TAIL(&(dev->enq_cbs), user_cb, next);
+		}
+	}
+
+	rte_spinlock_unlock(lock);
+	if (user_cb == NULL) {
+		rte_errno = ENOMEM;
+		return NULL;
+	}
+
+	return user_cb;
+}
+
+struct rte_eventdev_callback * __rte_experimental
+rte_event_add_pstdeq_callback(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	rte_errno = ENOTSUP;
+	return NULL;
+#endif
+
+	struct rte_eventdev *dev;
+	rte_spinlock_t *lock = &rte_eventdev_pstdeq_cb_lock;
+	struct rte_eventdev_callback *user_cb = NULL;
+
+	dev = &rte_eventdevs[eventdev_id];
+
+	if ((*dev->dev_ops->dev_stop == NULL) || (!cb_fn)) {
+		rte_errno = EINVAL;
+		return NULL;
+	}
+
+	rte_spinlock_lock(lock);
+
+	TAILQ_FOREACH(user_cb, &(dev->enq_cbs), next)
+	{
+		if (user_cb->cb_fn == cb_fn &&
+				user_cb->cb_arg == cb_arg)
+			break;
+	}
+
+	/* create a new callback. */
+	if (user_cb == NULL) {
+		user_cb = rte_zmalloc("EVENT_USER_CALLBACK",
+				sizeof(struct rte_eventdev_callback), 0);
+		if (user_cb != NULL) {
+			user_cb->cb_fn = cb_fn;
+			user_cb->cb_arg = cb_arg;
+			TAILQ_INSERT_TAIL(&(dev->enq_cbs), user_cb, next);
+		}
+	}
+
+	rte_spinlock_unlock(lock);
+	if (user_cb == NULL) {
+		rte_errno = ENOMEM;
+		return NULL;
+	}
+
+	return user_cb;
+}
+
+int __rte_experimental
+rte_event_preenq_callback_unregister(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	return -ENOTSUP;
+#endif
+
+	int ret = 0;
+	struct rte_eventdev *dev;
+	static rte_spinlock_t *lock = &rte_eventdev_preenq_cb_lock;
+	struct rte_eventdev_callback *cb, *next;
+
+	if (!cb_fn)
+		return -EINVAL;
+
+	dev = &rte_eventdevs[eventdev_id];
+	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->dev_stop, -EINVAL);
+
+	rte_spinlock_lock(lock);
+
+	ret = -EINVAL;
+	for (cb = TAILQ_FIRST(&dev->enq_cbs); cb != NULL; cb = next) {
+		next = TAILQ_NEXT(cb, next);
+
+		if (cb->cb_fn != cb_fn || cb->cb_arg != cb_arg)
+			continue;
+
+		if (cb->active == 0) {
+			TAILQ_REMOVE(&(dev->enq_cbs), cb, next);
+			rte_free(cb);
+			ret = 0;
+		} else
+			ret = -EAGAIN;
+	}
+
+	rte_spinlock_unlock(lock);
+
+	return ret;
+}
+
+int __rte_experimental
+rte_event_pstdeq_callback_unregister(uint8_t eventdev_id,
+		__rte_unused uint8_t port,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg)
+{
+#ifndef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	return -ENOTSUP;
+#endif
+
+	int ret = 0;
+	struct rte_eventdev *dev;
+	static rte_spinlock_t *lock = &rte_eventdev_pstdeq_cb_lock;
+	struct rte_eventdev_callback *cb, *next;
+
+	if (!cb_fn)
+		return -EINVAL;
+
+	dev = &rte_eventdevs[eventdev_id];
+	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->dev_stop, -EINVAL);
+
+	rte_spinlock_lock(lock);
+
+	ret = -EINVAL;
+	for (cb = TAILQ_FIRST(&dev->deq_cbs); cb != NULL; cb = next) {
+		next = TAILQ_NEXT(cb, next);
+
+		if (cb->cb_fn != cb_fn || cb->cb_arg != cb_arg)
+			continue;
+
+		if (cb->active == 0) {
+			TAILQ_REMOVE(&(dev->deq_cbs), cb, next);
+			rte_free(cb);
+			ret = 0;
+		} else
+			ret = -EAGAIN;
+	}
+
+	rte_spinlock_unlock(lock);
+
+	return ret;
+}
+
 static inline int
 rte_eventdev_data_alloc(uint8_t dev_id, struct rte_eventdev_data **data,
 		int socket_id)
diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
index 517cd8083..30f52e6d5 100644
--- a/lib/librte_eventdev/rte_eventdev.h
+++ b/lib/librte_eventdev/rte_eventdev.h
@@ -215,9 +215,13 @@ extern "C" {
 #include <rte_config.h>
 #include <rte_memory.h>
 #include <rte_errno.h>
+#include <rte_spinlock.h>
+
+#include <sys/queue.h>
 
 struct rte_mbuf; /* we just use mbuf pointers; no need to include rte_mbuf.h */
 struct rte_event;
+struct rte_eventdev_callback;
 
 /* Event device capability bitmap flags */
 #define RTE_EVENT_DEV_CAP_QUEUE_QOS           (1ULL << 0)
@@ -307,6 +311,19 @@ struct rte_event;
  * @see rte_event_port_link()
  */
 
+TAILQ_HEAD(rte_eventdev_enq_cb_list, rte_eventdev_callback);
+TAILQ_HEAD(rte_eventdev_deq_cb_list, rte_eventdev_callback);
+
+typedef uint16_t (*rte_eventdev_cb_fn)(uint8_t dev_id, uint8_t port_id,
+		struct rte_event *ev, uint16_t nb_events, void *cb_arg);
+
+struct rte_eventdev_callback {
+	TAILQ_ENTRY(rte_eventdev_callback) next; /* Callbacks list */
+	rte_eventdev_cb_fn cb_fn; /* Callback address */
+	void *cb_arg; /* Parameter for callback */
+	uint32_t active; /* Callback is executing */
+};
+
 /**
  * Get the total number of event devices that have been successfully
  * initialised.
@@ -1302,6 +1319,9 @@ struct rte_eventdev {
 	struct rte_device *dev;
 	/**< Device info. supplied by probing */
 
+	struct rte_eventdev_enq_cb_list enq_cbs;
+	struct rte_eventdev_deq_cb_list deq_cbs;
+
 	RTE_STD_C11
 	uint8_t attached : 1;
 	/**< Flag indicating the device is attached */
@@ -1310,9 +1330,217 @@ struct rte_eventdev {
 extern struct rte_eventdev *rte_eventdevs;
 /** @internal The pool of rte_eventdev structures. */
 
+/**
+ * Register a callback function for enqueue on device-port Event dev.
+ *
+ * @param dev_id
+ *  Event device id.
+ * @param port
+ *  Event port.
+ * @param cb_fn
+ *  User supplied callback function to be called.
+ * @param cb_arg
+ *  Pointer to the parameters for the registered callback.
+ *
+ * @return
+ *  - On success, zero.
+ *  - On failure, a negative value.
+ */
+int __rte_experimental
+rte_event_preenq_callback_register(uint8_t dev_id, uint8_t port_id,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg);
+
+/**
+ * Register a callback function for dequeue on device-port Event dev.
+ *
+ * @param dev_id
+ *  Event device id.
+ * @param port
+ *  Event port.
+ * @param cb_fn
+ *  User supplied callback function to be called.
+ * @param cb_arg
+ *  Pointer to the parameters for the registered callback.
+ *
+ * @return
+ *  - On success, zero.
+ *  - On failure, a negative value.
+ */
+int __rte_experimental
+rte_event_pstdeq_callback_register(uint8_t dev_id, uint8_t port_id,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg);
+
+/**
+ * Unregister a callback function for enqueue on device-port Event dev.
+ *
+ * @param dev_id
+ *  Event device id.
+ * @param port
+ *  Event port.
+ * @param cb_fn
+ *  User supplied callback function to be called.
+ * @param cb_arg
+ *  Pointer to the parameters for the registered callback. -1 means to
+ *  remove all for the same callback address and same event.
+ *
+ * @return
+ *  - On success, zero.
+ *  - On failure, a negative value.
+ */
+int __rte_experimental
+rte_event_preenq_callback_unregister(uint8_t dev_id, uint8_t port_id,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg);
+
+/**
+ * Unregister a callback function for dequeue on device-port Event dev.
+ *
+ * @param dev_id
+ *  Event device id.
+ * @param port
+ *  Event port.
+ * @param cb_fn
+ *  User supplied callback function to be called.
+ * @param cb_arg
+ *  Pointer to the parameters for the registered callback. -1 means to
+ *  remove all for the same callback address and same event.
+ *
+ * @return
+ *  - On success, zero.
+ *  - On failure, a negative value.
+ */
+int __rte_experimental
+rte_event_pstdeq_callback_unregister(uint8_t dev_id, uint8_t port_id,
+		rte_eventdev_cb_fn cb_fn, void *cb_arg);
+
+/**
+ * Add a callback to be called on event enqueue on a given device & port.
+ *
+ * This API configures a function to be called for each burst of
+ * events enqueued on a given event device-port. The return value is a pointer
+ * that can be used to later remove the callback using
+ * rte_eventdev_remove_preenq_callback().
+ *
+ * Multiple functions are called in the order that they are added.
+ *
+ * @param dev_id
+ *   The device identifier of the Event device.
+ * @param port_id
+ *   The port on the Event device on which the callback is to be added.
+ * @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 port and queue.
+ *
+ * @return
+ *   NULL on error.
+ *   On success, a pointer value which can later be used to remove the callback.
+ */
+struct rte_eventdev_callback * __rte_experimental
+rte_event_add_preenq_callback(uint8_t dev_id,
+		__rte_unused uint8_t port_id, rte_eventdev_cb_fn cb_fn,
+		void *cb_arg);
+
+/**
+ * Add a callback to be called on event dequeue on a given 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_eventdev_remove_pstdeq_callback().
+ *
+ * Multiple functions are called in the order that they are added.
+ *
+ * @param dev_id
+ *   The device identifier of the Event device.
+ * @param port_id
+ *   The port on the Event device on which the callback is to be added.
+ * @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 port and queue.
+ *
+ * @return
+ *   NULL on error.
+ *   On success, a pointer value which can later be used to remove the callback.
+ */
+struct rte_eventdev_callback * __rte_experimental
+rte_event_add_pstdeq_callback(uint8_t dev_id,
+		__rte_unused uint8_t port_id, rte_eventdev_cb_fn cb_fn,
+		void *cb_arg);
+
+/**
+ * Remove an event callback from a given dev and port.
+ *
+ * This function is used to remove callbacks that were added to a event dev-port
+ * using rte_event_add_preenq_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 eventdev is stopped, or the user knows that no
+ *   callbacks are in flight e.g. if called from the thread doing enqueue
+ *   on that eventdev.
+ *
+ * - After a short delay - where the delay is sufficient to allow any
+ *   in-flight callbacks to complete.
+ *
+ * @param dev_id
+ *   The device identifier of the Event device.
+ * @param port_id
+ *   The port on the Event device from which the callback is to be removed.
+ * @param user_cb
+ *   User supplied callback created via rte_event_add_preenq_callback().
+ *
+ * @return
+ *   - 0: Success. Callback was removed.
+ *   - -ENOTSUP: Callback support is not available.
+ *   - -EINVAL:  The dev_id or the port_id is out of range, or the callback
+ *               is NULL or not found for the dev|port.
+ */
+int __rte_experimental
+rte_event_remove_preenq_callback(uint8_t dev_id, uint8_t port_id,
+		const struct rte_eventdev_callback *user_cb);
+
+/**
+ * Remove an event callback from a given dev and port.
+ *
+ * This function is used to remove callbacks that were added to a event dev-port
+ * using rte_event_add_pstdeq_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 eventdev is stopped, or the user knows that no
+ *   callbacks are in flight e.g. if called from the thread doing dequeue
+ *   on that eventdev.
+ *
+ * - After a short delay - where the delay is sufficient to allow any
+ *   in-flight callbacks to complete.
+ *
+ * @param dev_id
+ *   The device identifier of the Event device.
+ * @param port_id
+ *   The port on the Event device from which the callback is to be removed.
+ * @param user_cb
+ *   User supplied callback created via rte_event_add_pstdeq_callback().
+ *
+ * @return
+ *   - 0: Success. Callback was removed.
+ *   - -ENOTSUP: Callback support is not available.
+ *   - -EINVAL:  The dev_id or the port_id is out of range, or the callback
+ *               is NULL or not found for the dev|port.
+ */
+int __rte_experimental
+rte_event_remove_pstdeq_callback(uint8_t dev_id, uint8_t port_id,
+		const struct rte_eventdev_callback *user_cb);
+
 static __rte_always_inline uint16_t
 __rte_event_enqueue_burst(uint8_t dev_id, uint8_t port_id,
-			const struct rte_event ev[], uint16_t nb_events,
+			struct rte_event ev[], uint16_t nb_events,
 			const event_enqueue_burst_t fn)
 {
 	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
@@ -1328,6 +1556,22 @@ __rte_event_enqueue_burst(uint8_t dev_id, uint8_t port_id,
 		return 0;
 	}
 #endif
+
+#ifdef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	if (likely(rte_eventdevs[dev_id].attached)) {
+		struct rte_eventdev_callback *cb = NULL;
+		TAILQ_FOREACH(cb, &(dev->enq_cbs), next)
+		{
+			if (cb->cb_fn == NULL)
+				continue;
+
+			cb->active = 1;
+			nb_events = cb->cb_fn(dev_id, port_id, ev, nb_events, cb->cb_arg);
+			cb->active = 0;
+		}
+	}
+#endif
+
 	/*
 	 * Allow zero cost non burst mode routine invocation if application
 	 * requests nb_events as const one
@@ -1383,7 +1627,7 @@ __rte_event_enqueue_burst(uint8_t dev_id, uint8_t port_id,
  */
 static inline uint16_t
 rte_event_enqueue_burst(uint8_t dev_id, uint8_t port_id,
-			const struct rte_event ev[], uint16_t nb_events)
+			struct rte_event ev[], uint16_t nb_events)
 {
 	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
 
@@ -1434,7 +1678,7 @@ rte_event_enqueue_burst(uint8_t dev_id, uint8_t port_id,
  */
 static inline uint16_t
 rte_event_enqueue_new_burst(uint8_t dev_id, uint8_t port_id,
-			const struct rte_event ev[], uint16_t nb_events)
+			struct rte_event ev[], uint16_t nb_events)
 {
 	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
 
@@ -1485,7 +1729,7 @@ rte_event_enqueue_new_burst(uint8_t dev_id, uint8_t port_id,
  */
 static inline uint16_t
 rte_event_enqueue_forward_burst(uint8_t dev_id, uint8_t port_id,
-			const struct rte_event ev[], uint16_t nb_events)
+			struct rte_event ev[], uint16_t nb_events)
 {
 	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
 
@@ -1606,6 +1850,21 @@ rte_event_dequeue_burst(uint8_t dev_id, uint8_t port_id, struct rte_event ev[],
 	}
 #endif
 
+#ifdef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	if (likely(rte_eventdevs[dev_id].attached)) {
+		struct rte_eventdev_callback *cb = NULL;
+		TAILQ_FOREACH(cb, &(dev->deq_cbs), next)
+		{
+			if (cb->cb_fn == NULL)
+				continue;
+
+			cb->active = 1;
+			nb_events = cb->cb_fn(dev_id, port_id, ev, nb_events, cb->cb_arg);
+			cb->active = 0;
+		}
+	}
+#endif
+
 	/*
 	 * Allow zero cost non burst mode routine invocation if application
 	 * requests nb_events as const one
diff --git a/lib/librte_eventdev/rte_eventdev_version.map b/lib/librte_eventdev/rte_eventdev_version.map
index 95fd089f9..8d52fc839 100644
--- a/lib/librte_eventdev/rte_eventdev_version.map
+++ b/lib/librte_eventdev/rte_eventdev_version.map
@@ -126,6 +126,14 @@ DPDK_19.05 {
 EXPERIMENTAL {
 	global:
 
+	rte_event_add_preenq_callback;
+	rte_event_add_pstdeq_callback;
 	rte_event_eth_rx_adapter_cb_register;
 	rte_event_eth_rx_adapter_stats_get;
+	rte_event_preenq_callback_register;
+	rte_event_preenq_callback_unregister;
+	rte_event_pstdeq_callback_register;
+	rte_event_pstdeq_callback_unregister;
+	rte_event_remove_preenq_callback;
+	rte_event_remove_pstdeq_callback;
 };
-- 
2.17.1

[dpdk-dev] [PATCH v1 2/2] examples/event: add callback handle

[Vipin Varghese]

To add register and unregister for event pre-enqueue and
post-dequeue callback handler.

Signed-off-by: Vipin Varghese <vipin.varghese@intel.com>
---
 examples/eventdev_pipeline/main.c | 49 +++++++++++++++++++++++++++++++
 1 file changed, 49 insertions(+)

diff --git a/examples/eventdev_pipeline/main.c b/examples/eventdev_pipeline/main.c
index f4e57f541..469951463 100644
--- a/examples/eventdev_pipeline/main.c
+++ b/examples/eventdev_pipeline/main.c
@@ -20,6 +20,37 @@ struct config_data cdata = {
 	.worker_cq_depth = 16
 };
 
+#ifdef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+uint16_t
+dump_flowid_events(uint8_t dev_id, uint8_t port_id,
+		__rte_unused struct rte_event *ev,
+		__rte_unused uint16_t nb_events,
+		__rte_unused void *cb_arg);
+
+uint16_t
+dump_flowid_events(uint8_t dev_id, uint8_t port_id,
+		__rte_unused struct rte_event *ev,
+		__rte_unused uint16_t nb_events,
+		__rte_unused void *cb_arg)
+{
+	if (unlikely(nb_events == 0))
+		return 0;
+
+	for (int i = 0; i < nb_events; i++)
+		printf(" dev_id (%u) port_id (%u)\n "
+				"- ev(%p) nb_events (%u)\n"
+				"- flow_id (%u)\n"
+				"- sub_event_type (%u) event_type (%u)\n",
+				dev_id, port_id,
+				ev, nb_events,
+				ev[i].flow_id,
+				ev[i].sub_event_type,
+				ev[i].event_type);
+
+	return nb_events;
+}
+#endif
+
 static bool
 core_in_use(unsigned int lcore_id) {
 	return (fdata->rx_core[lcore_id] || fdata->sched_core[lcore_id] ||
@@ -398,6 +429,14 @@ signal_handler(int signum)
 	if ((signum == SIGINT || signum == SIGTERM) && !once) {
 		printf("\n\nSignal %d received, preparing to exit...\n",
 				signum);
+
+#ifdef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+		rte_eventdev_preenq_callback_unregister(0, 0,
+				dump_flowid_events, NULL);
+		rte_eventdev_pstdeq_callback_unregister(0, 0,
+				dump_flowid_events, NULL);
+#endif
+
 		if (cdata.dump_dev)
 			rte_event_dev_dump(0, stdout);
 		once = 1;
@@ -559,6 +598,16 @@ main(int argc, char **argv)
 	if (core_in_use(lcore_id))
 		fdata->cap.worker(&worker_data[worker_idx++]);
 
+#ifdef RTE_EVENTDEV_ENQDEQ_CALLBACKS
+	if (rte_eventdev_preenq_callback_register(0, 0,
+			dump_flowid_events, NULL))
+		printf(" Failed to register enq callback\n");
+
+	if (rte_eventdev_pstdeq_callback_register(0, 0,
+			dump_flowid_events, NULL))
+		printf(" Failed to register deq callback\n");
+#endif
+
 	rte_eal_mp_wait_lcore();
 
 	if (!cdata.quiet && (port_stat(dev_id, worker_data[0].port_id) !=
-- 
2.17.1