Hello Mattias,
On Thu, Sep 28, 2023 at 9:36 AM Mattias Rönnblom
<mattias.ronnb...@ericsson.com> wrote:
>
> The purpose of the dispatcher library is to help reduce coupling in an
> Eventdev-based DPDK application.
>
> In addition, the dispatcher also provides a convenient and flexible
> way for the application to use service cores for application-level
> processing.
>
> Signed-off-by: Mattias Rönnblom <mattias.ronnb...@ericsson.com>
> Tested-by: Peter Nilsson <peter.j.nils...@ericsson.com>
> Reviewed-by: Heng Wang <heng.w...@ericsson.com>
>
I have some simple comments.
> --
>
> PATCH v5:
> o Move from using an integer id to a pointer to reference a dispatcher
> instance, to simplify the API.
> o Fix bug where dispatcher stats retrieval function erroneously depended
> on the user-supplied stats buffer being all-zero.
>
> PATCH v4:
> o Fix bugs in handler and finalizer unregistration. (Naga Harish)
> o Return -EINVAL in cases where NULL pointers were provided in
> calls requiring non-NULL pointers. (Naga Harish)
> o Add experimental warning for the whole API. (Jerin Jacob)
>
> PATCH v3:
> o To underline its optional character and since it does not provide
> hardware abstraction, the event dispatcher is now a separate
> library.
> o Change name from rte_event_dispatcher -> rte_dispatcher, to make it
> shorter and to avoid the rte_event_* namespace.
>
> PATCH v2:
> o Add dequeue batch count statistic.
> o Add statistics reset function to API.
> o Clarify MT safety guarantees (or lack thereof) in the API documentation.
> o Change loop variable type in evd_lcore_get_handler_by_id() to uint16_t,
> to be consistent with similar loops elsewhere in the dispatcher.
> o Fix variable names in finalizer unregister function.
>
> PATCH:
> o Change prefix from RED to EVD, to avoid confusion with random
> early detection.
>
> RFC v4:
> o Move handlers to per-lcore data structures.
> o Introduce mechanism which rearranges handlers so that often-used
> handlers tend to be tried first.
> o Terminate dispatch loop in case all events are delivered.
> o To avoid the dispatcher's service function hogging the CPU, process
> only one batch per call.
> o Have service function return -EAGAIN if no work is performed.
> o Events delivered in the process function is no longer marked 'const',
> since modifying them may be useful for the application and cause
> no difficulties for the dispatcher.
> o Various minor API documentation improvements.
>
> RFC v3:
> o Add stats_get() function to the version.map file.
> ---
> MAINTAINERS | 3 +
> doc/api/doxy-api-index.md | 1 +
> doc/api/doxy-api.conf.in | 1 +
> lib/dispatcher/meson.build | 17 +
> lib/dispatcher/rte_dispatcher.c | 708 ++++++++++++++++++++++++++++++++
> lib/dispatcher/rte_dispatcher.h | 468 +++++++++++++++++++++
> lib/dispatcher/version.map | 20 +
> lib/meson.build | 2 +
> 8 files changed, 1220 insertions(+)
> create mode 100644 lib/dispatcher/meson.build
> create mode 100644 lib/dispatcher/rte_dispatcher.c
> create mode 100644 lib/dispatcher/rte_dispatcher.h
> create mode 100644 lib/dispatcher/version.map
>
> diff --git a/MAINTAINERS b/MAINTAINERS
> index a926155f26..6704cd5b2c 100644
> --- a/MAINTAINERS
> +++ b/MAINTAINERS
> @@ -1726,6 +1726,9 @@ M: Nithin Dabilpuram <ndabilpu...@marvell.com>
> M: Pavan Nikhilesh <pbhagavat...@marvell.com>
> F: lib/node/
>
> +Dispatcher - EXPERIMENTAL
> +M: Mattias Rönnblom <mattias.ronnb...@ericsson.com>
> +F: lib/dispatcher/
Double empty line between sections in MAINTAINERS file, please.
>
> Test Applications
> -----------------
> diff --git a/doc/api/doxy-api-index.md b/doc/api/doxy-api-index.md
> index fdeda13932..7d0cad9fed 100644
> --- a/doc/api/doxy-api-index.md
> +++ b/doc/api/doxy-api-index.md
> @@ -155,6 +155,7 @@ The public API headers are grouped by topics:
>
> - **classification**
> [reorder](@ref rte_reorder.h),
> + [dispatcher](@ref rte_dispatcher.h),
> [distributor](@ref rte_distributor.h),
> [EFD](@ref rte_efd.h),
> [ACL](@ref rte_acl.h),
> diff --git a/doc/api/doxy-api.conf.in b/doc/api/doxy-api.conf.in
> index a88accd907..59c679e621 100644
> --- a/doc/api/doxy-api.conf.in
> +++ b/doc/api/doxy-api.conf.in
> @@ -34,6 +34,7 @@ INPUT =
> @TOPDIR@/doc/api/doxy-api-index.md \
> @TOPDIR@/lib/cmdline \
> @TOPDIR@/lib/compressdev \
> @TOPDIR@/lib/cryptodev \
> + @TOPDIR@/lib/dispatcher \
> @TOPDIR@/lib/distributor \
> @TOPDIR@/lib/dmadev \
> @TOPDIR@/lib/efd \
I see no release note updates, please add one entry (in the "New
features" section) to announce this new library.
> diff --git a/lib/dispatcher/meson.build b/lib/dispatcher/meson.build
> new file mode 100644
> index 0000000000..c6054a3a5d
> --- /dev/null
> +++ b/lib/dispatcher/meson.build
> @@ -0,0 +1,17 @@
> +# SPDX-License-Identifier: BSD-3-Clause
> +# Copyright(c) 2023 Ericsson AB
> +
> +if is_windows
> + build = false
> + reason = 'not supported on Windows'
> + subdir_done()
> +endif
> +
> +sources = files(
> + 'rte_dispatcher.c',
> +)
> +headers = files(
> + 'rte_dispatcher.h',
> +)
For a single $file, you can go with a single line: files('$file')
> +
> +deps += ['eventdev']
> diff --git a/lib/dispatcher/rte_dispatcher.c b/lib/dispatcher/rte_dispatcher.c
> new file mode 100644
> index 0000000000..0e69db2b9b
> --- /dev/null
> +++ b/lib/dispatcher/rte_dispatcher.c
> @@ -0,0 +1,708 @@
> +/* SPDX-License-Identifier: BSD-3-Clause
> + * Copyright(c) 2023 Ericsson AB
> + */
> +
> +#include <stdbool.h>
> +#include <stdint.h>
> +
> +#include <rte_branch_prediction.h>
> +#include <rte_common.h>
> +#include <rte_lcore.h>
> +#include <rte_random.h>
> +#include <rte_service_component.h>
> +
> +#include "eventdev_pmd.h"
> +
> +#include <rte_dispatcher.h>
> +
> +#define EVD_MAX_PORTS_PER_LCORE 4
> +#define EVD_MAX_HANDLERS 32
> +#define EVD_MAX_FINALIZERS 16
> +#define EVD_AVG_PRIO_INTERVAL 2000
> +#define EVD_SERVICE_NAME "dispatcher"
> +
> +struct rte_dispatcher_lcore_port {
> + uint8_t port_id;
> + uint16_t batch_size;
> + uint64_t timeout;
> +};
> +
> +struct rte_dispatcher_handler {
> + int id;
> + rte_dispatcher_match_t match_fun;
> + void *match_data;
> + rte_dispatcher_process_t process_fun;
> + void *process_data;
> +};
> +
> +struct rte_dispatcher_finalizer {
> + int id;
> + rte_dispatcher_finalize_t finalize_fun;
> + void *finalize_data;
> +};
> +
> +struct rte_dispatcher_lcore {
> + uint8_t num_ports;
> + uint16_t num_handlers;
> + int32_t prio_count;
> + struct rte_dispatcher_lcore_port ports[EVD_MAX_PORTS_PER_LCORE];
> + struct rte_dispatcher_handler handlers[EVD_MAX_HANDLERS];
> + struct rte_dispatcher_stats stats;
> +} __rte_cache_aligned;
> +
> +struct rte_dispatcher {
> + uint8_t event_dev_id;
> + int socket_id;
> + uint32_t service_id;
> + struct rte_dispatcher_lcore lcores[RTE_MAX_LCORE];
> + uint16_t num_finalizers;
> + struct rte_dispatcher_finalizer finalizers[EVD_MAX_FINALIZERS];
> +};
> +
> +static int
> +evd_lookup_handler_idx(struct rte_dispatcher_lcore *lcore,
> + const struct rte_event *event)
Wrt DPDK coding tyle, indent is a single tab.
Adding an extra tab is recommended when continuing control statements
like if()/for()/..
On the other hand, max accepted length for a line is 100 columns.
Wdyt of a single line for this specific case?
And please check the indentation in the rest of the file.
> +{
> + uint16_t i;
> +
> + for (i = 0; i < lcore->num_handlers; i++) {
> + struct rte_dispatcher_handler *handler =
> + &lcore->handlers[i];
> +
> + if (handler->match_fun(event, handler->match_data))
> + return i;
> + }
> +
> + return -1;
> +}
> +
> +static void
> +evd_prioritize_handler(struct rte_dispatcher_lcore *lcore,
> + int handler_idx)
> +{
> + struct rte_dispatcher_handler tmp;
> +
> + if (handler_idx == 0)
> + return;
> +
> + /* Let the lucky handler "bubble" up the list */
> +
> + tmp = lcore->handlers[handler_idx - 1];
> +
> + lcore->handlers[handler_idx - 1] = lcore->handlers[handler_idx];
> +
> + lcore->handlers[handler_idx] = tmp;
We don't need so many blank lines.
> +}
> +
> +static inline void
> +evd_consider_prioritize_handler(struct rte_dispatcher_lcore *lcore,
> + int handler_idx, uint16_t handler_events)
> +{
> + lcore->prio_count -= handler_events;
> +
> + if (unlikely(lcore->prio_count <= 0)) {
> + evd_prioritize_handler(lcore, handler_idx);
> +
> + /*
> + * Randomize the interval in the unlikely case
> + * the traffic follow some very strict pattern.
> + */
> + lcore->prio_count =
> + rte_rand_max(EVD_AVG_PRIO_INTERVAL) +
> + EVD_AVG_PRIO_INTERVAL / 2;
> + }
> +}
> +
> +static inline void
> +evd_dispatch_events(struct rte_dispatcher *dispatcher,
> + struct rte_dispatcher_lcore *lcore,
> + struct rte_dispatcher_lcore_port *port,
> + struct rte_event *events, uint16_t num_events)
> +{
> + int i;
> + struct rte_event bursts[EVD_MAX_HANDLERS][num_events];
> + uint16_t burst_lens[EVD_MAX_HANDLERS] = { 0 };
> + uint16_t drop_count = 0;
> + uint16_t dispatch_count;
> + uint16_t dispatched = 0;
> +
> + for (i = 0; i < num_events; i++) {
> + struct rte_event *event = &events[i];
> + int handler_idx;
> +
> + handler_idx = evd_lookup_handler_idx(lcore, event);
> +
> + if (unlikely(handler_idx < 0)) {
> + drop_count++;
> + continue;
> + }
> +
> + bursts[handler_idx][burst_lens[handler_idx]] = *event;
> + burst_lens[handler_idx]++;
> + }
> +
> + dispatch_count = num_events - drop_count;
> +
> + for (i = 0; i < lcore->num_handlers &&
> + dispatched < dispatch_count; i++) {
> + struct rte_dispatcher_handler *handler =
> + &lcore->handlers[i];
> + uint16_t len = burst_lens[i];
> +
> + if (len == 0)
> + continue;
> +
> + handler->process_fun(dispatcher->event_dev_id, port->port_id,
> + bursts[i], len, handler->process_data);
> +
> + dispatched += len;
> +
> + /*
> + * Safe, since any reshuffling will only involve
> + * already-processed handlers.
> + */
> + evd_consider_prioritize_handler(lcore, i, len);
> + }
> +
> + lcore->stats.ev_batch_count++;
> + lcore->stats.ev_dispatch_count += dispatch_count;
> + lcore->stats.ev_drop_count += drop_count;
> +
> + for (i = 0; i < dispatcher->num_finalizers; i++) {
> + struct rte_dispatcher_finalizer *finalizer =
> + &dispatcher->finalizers[i];
> +
> + finalizer->finalize_fun(dispatcher->event_dev_id,
> + port->port_id,
> + finalizer->finalize_data);
> + }
> +}
> +
> +static __rte_always_inline uint16_t
> +evd_port_dequeue(struct rte_dispatcher *dispatcher,
> + struct rte_dispatcher_lcore *lcore,
> + struct rte_dispatcher_lcore_port *port)
> +{
> + uint16_t batch_size = port->batch_size;
> + struct rte_event events[batch_size];
> + uint16_t n;
> +
> + n = rte_event_dequeue_burst(dispatcher->event_dev_id, port->port_id,
> + events, batch_size, port->timeout);
> +
> + if (likely(n > 0))
> + evd_dispatch_events(dispatcher, lcore, port, events, n);
> +
> + lcore->stats.poll_count++;
> +
> + return n;
> +}
> +
> +static __rte_always_inline uint16_t
> +evd_lcore_process(struct rte_dispatcher *dispatcher,
> + struct rte_dispatcher_lcore *lcore)
> +{
> + uint16_t i;
> + uint16_t event_count = 0;
> +
> + for (i = 0; i < lcore->num_ports; i++) {
> + struct rte_dispatcher_lcore_port *port =
> + &lcore->ports[i];
> +
> + event_count += evd_port_dequeue(dispatcher, lcore, port);
> + }
> +
> + return event_count;
> +}
> +
> +static int32_t
> +evd_process(void *userdata)
> +{
> + struct rte_dispatcher *dispatcher = userdata;
> + unsigned int lcore_id = rte_lcore_id();
> + struct rte_dispatcher_lcore *lcore =
> + &dispatcher->lcores[lcore_id];
> + uint64_t event_count;
> +
> + event_count = evd_lcore_process(dispatcher, lcore);
> +
> + if (unlikely(event_count == 0))
> + return -EAGAIN;
> +
> + return 0;
> +}
> +
> +static int
> +evd_service_register(struct rte_dispatcher *dispatcher)
> +{
> + struct rte_service_spec service = {
> + .callback = evd_process,
> + .callback_userdata = dispatcher,
> + .capabilities = RTE_SERVICE_CAP_MT_SAFE,
> + .socket_id = dispatcher->socket_id
> + };
> + int rc;
> +
> + snprintf(service.name, sizeof(service.name), EVD_SERVICE_NAME);
> +
> + rc = rte_service_component_register(&service,
> &dispatcher->service_id);
> +
No need for blank line.
> + if (rc)
if (rc != 0)
> + RTE_EDEV_LOG_ERR("Registration of dispatcher service "
> + "%s failed with error code %d\n",
> + service.name, rc);
> +
> + return rc;
> +}
> +
> +static int
> +evd_service_unregister(struct rte_dispatcher *dispatcher)
> +{
> + int rc;
> +
> + rc = rte_service_component_unregister(dispatcher->service_id);
> +
> + if (rc)
> + RTE_EDEV_LOG_ERR("Unregistration of dispatcher service "
> + "failed with error code %d\n", rc);
> +
> + return rc;
> +}
> +
> +struct rte_dispatcher *
> +rte_dispatcher_create(uint8_t event_dev_id)
> +{
> + int socket_id;
> + struct rte_dispatcher *dispatcher;
> + int rc;
> +
> + socket_id = rte_event_dev_socket_id(event_dev_id);
> +
> + dispatcher =
> + rte_malloc_socket("dispatcher", sizeof(struct rte_dispatcher),
> + RTE_CACHE_LINE_SIZE, socket_id);
> +
> + if (dispatcher == NULL) {
> + RTE_EDEV_LOG_ERR("Unable to allocate memory for
> dispatcher\n");
> + rte_errno = ENOMEM;
> + return NULL;
> + }
> +
> + *dispatcher = (struct rte_dispatcher) {
> + .event_dev_id = event_dev_id,
> + .socket_id = socket_id
> + };
> +
> + rc = evd_service_register(dispatcher);
> +
> + if (rc < 0) {
> + rte_free(dispatcher);
> + rte_errno = -rc;
> + return NULL;
> + }
> +
> + return dispatcher;
> +}
> +
> +int
> +rte_dispatcher_free(struct rte_dispatcher *dispatcher)
> +{
> + int rc;
> +
> + if (dispatcher == NULL)
> + return 0;
> +
> + rc = evd_service_unregister(dispatcher);
> +
> + if (rc)
> + return rc;
> +
> + rte_free(dispatcher);
> +
> + return 0;
> +}
> +
> +uint32_t
> +rte_dispatcher_service_id_get(const struct rte_dispatcher *dispatcher)
> +{
> + return dispatcher->service_id;
> +}
> +
> +static int
> +lcore_port_index(struct rte_dispatcher_lcore *lcore,
> + uint8_t event_port_id)
> +{
> + uint16_t i;
> +
> + for (i = 0; i < lcore->num_ports; i++) {
> + struct rte_dispatcher_lcore_port *port =
> + &lcore->ports[i];
> +
> + if (port->port_id == event_port_id)
> + return i;
> + }
> +
> + return -1;
> +}
> +
> +int
> +rte_dispatcher_bind_port_to_lcore(struct rte_dispatcher *dispatcher,
> + uint8_t event_port_id, uint16_t batch_size,
> + uint64_t timeout, unsigned int lcore_id)
> +{
> + struct rte_dispatcher_lcore *lcore;
> + struct rte_dispatcher_lcore_port *port;
> +
> + lcore = &dispatcher->lcores[lcore_id];
> +
> + if (lcore->num_ports == EVD_MAX_PORTS_PER_LCORE)
> + return -ENOMEM;
> +
> + if (lcore_port_index(lcore, event_port_id) >= 0)
> + return -EEXIST;
> +
> + port = &lcore->ports[lcore->num_ports];
> +
> + *port = (struct rte_dispatcher_lcore_port) {
> + .port_id = event_port_id,
> + .batch_size = batch_size,
> + .timeout = timeout
> + };
> +
> + lcore->num_ports++;
> +
> + return 0;
> +}
> +
> +int
> +rte_dispatcher_unbind_port_from_lcore(struct rte_dispatcher *dispatcher,
> + uint8_t event_port_id,
> + unsigned int lcore_id)
> +{
> + struct rte_dispatcher_lcore *lcore;
> + int port_idx;
> + struct rte_dispatcher_lcore_port *port;
> + struct rte_dispatcher_lcore_port *last;
> +
> + lcore = &dispatcher->lcores[lcore_id];
> +
> + port_idx = lcore_port_index(lcore, event_port_id);
> +
> + if (port_idx < 0)
> + return -ENOENT;
> +
> + port = &lcore->ports[port_idx];
> + last = &lcore->ports[lcore->num_ports - 1];
> +
> + if (port != last)
> + *port = *last;
> +
> + lcore->num_ports--;
> +
> + return 0;
> +}
> +
> +static struct rte_dispatcher_handler*
Missing a space before *.
> +evd_lcore_get_handler_by_id(struct rte_dispatcher_lcore *lcore,
> + int handler_id)
> +{
> + uint16_t i;
> +
> + for (i = 0; i < lcore->num_handlers; i++) {
> + struct rte_dispatcher_handler *handler =
> + &lcore->handlers[i];
> +
> + if (handler->id == handler_id)
> + return handler;
> + }
> +
> + return NULL;
> +}
> +
> +static int
> +evd_alloc_handler_id(struct rte_dispatcher *dispatcher)
> +{
> + int handler_id = 0;
> + struct rte_dispatcher_lcore *reference_lcore =
> + &dispatcher->lcores[0];
> +
> + if (reference_lcore->num_handlers == EVD_MAX_HANDLERS)
> + return -1;
> +
> + while (evd_lcore_get_handler_by_id(reference_lcore, handler_id) !=
> NULL)
> + handler_id++;
> +
> + return handler_id;
> +}
> +
> +static void
> +evd_lcore_install_handler(struct rte_dispatcher_lcore *lcore,
> + const struct rte_dispatcher_handler *handler)
> +{
> + int handler_idx = lcore->num_handlers;
> +
> + lcore->handlers[handler_idx] = *handler;
> + lcore->num_handlers++;
> +}
> +
> +static void
> +evd_install_handler(struct rte_dispatcher *dispatcher,
> + const struct rte_dispatcher_handler *handler)
> +{
> + int i;
> +
> + for (i = 0; i < RTE_MAX_LCORE; i++) {
> + struct rte_dispatcher_lcore *lcore =
> + &dispatcher->lcores[i];
> + evd_lcore_install_handler(lcore, handler);
> + }
> +}
> +
> +int
> +rte_dispatcher_register(struct rte_dispatcher *dispatcher,
> + rte_dispatcher_match_t match_fun, void *match_data,
> + rte_dispatcher_process_t process_fun,
> + void *process_data)
> +{
> + struct rte_dispatcher_handler handler = {
> + .match_fun = match_fun,
> + .match_data = match_data,
> + .process_fun = process_fun,
> + .process_data = process_data
> + };
> +
> + handler.id = evd_alloc_handler_id(dispatcher);
> +
> + if (handler.id < 0)
> + return -ENOMEM;
> +
> + evd_install_handler(dispatcher, &handler);
> +
> + return handler.id;
> +}
> +
> +static int
> +evd_lcore_uninstall_handler(struct rte_dispatcher_lcore *lcore,
> + int handler_id)
> +{
> + struct rte_dispatcher_handler *unreg_handler;
> + int handler_idx;
> + uint16_t last_idx;
> +
> + unreg_handler = evd_lcore_get_handler_by_id(lcore, handler_id);
> +
> + if (unreg_handler == NULL) {
> + RTE_EDEV_LOG_ERR("Invalid handler id %d\n", handler_id);
> + return -EINVAL;
> + }
> +
> + handler_idx = unreg_handler - &lcore->handlers[0];
> +
> + last_idx = lcore->num_handlers - 1;
> +
> + if (handler_idx != last_idx) {
> + /* move all handlers to maintain handler order */
> + int n = last_idx - handler_idx;
> + memmove(unreg_handler, unreg_handler + 1,
> + sizeof(struct rte_dispatcher_handler) * n);
> + }
> +
> + lcore->num_handlers--;
> +
> + return 0;
> +}
> +
> +static int
> +evd_uninstall_handler(struct rte_dispatcher *dispatcher,
> + int handler_id)
> +{
> + unsigned int lcore_id;
> +
> + for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
> + struct rte_dispatcher_lcore *lcore =
> + &dispatcher->lcores[lcore_id];
> + int rc;
> +
> + rc = evd_lcore_uninstall_handler(lcore, handler_id);
> +
> + if (rc < 0)
> + return rc;
> + }
> +
> + return 0;
> +}
> +
> +int
> +rte_dispatcher_unregister(struct rte_dispatcher *dispatcher, int handler_id)
> +{
> + int rc;
No need for rc.
> +
> + rc = evd_uninstall_handler(dispatcher, handler_id);
> +
> + return rc;
> +}
> +
> +static struct rte_dispatcher_finalizer*
Missing a space before *.
> +evd_get_finalizer_by_id(struct rte_dispatcher *dispatcher,
> + int handler_id)
> +{
> + int i;
> +
> + for (i = 0; i < dispatcher->num_finalizers; i++) {
> + struct rte_dispatcher_finalizer *finalizer =
> + &dispatcher->finalizers[i];
> +
> + if (finalizer->id == handler_id)
> + return finalizer;
> + }
> +
> + return NULL;
> +}
> +
> +static int
> +evd_alloc_finalizer_id(struct rte_dispatcher *dispatcher)
> +{
> + int finalizer_id = 0;
> +
> + while (evd_get_finalizer_by_id(dispatcher, finalizer_id) != NULL)
> + finalizer_id++;
> +
> + return finalizer_id;
> +}
> +
> +static struct rte_dispatcher_finalizer *
> +evd_alloc_finalizer(struct rte_dispatcher *dispatcher)
> +{
> + int finalizer_idx;
> + struct rte_dispatcher_finalizer *finalizer;
> +
> + if (dispatcher->num_finalizers == EVD_MAX_FINALIZERS)
> + return NULL;
> +
> + finalizer_idx = dispatcher->num_finalizers;
> + finalizer = &dispatcher->finalizers[finalizer_idx];
> +
> + finalizer->id = evd_alloc_finalizer_id(dispatcher);
> +
> + dispatcher->num_finalizers++;
> +
> + return finalizer;
> +}
> +
> +int
> +rte_dispatcher_finalize_register(struct rte_dispatcher *dispatcher,
> + rte_dispatcher_finalize_t finalize_fun,
> + void *finalize_data)
> +{
> + struct rte_dispatcher_finalizer *finalizer;
> +
> + finalizer = evd_alloc_finalizer(dispatcher);
> +
> + if (finalizer == NULL)
> + return -ENOMEM;
> +
> + finalizer->finalize_fun = finalize_fun;
> + finalizer->finalize_data = finalize_data;
> +
> + return finalizer->id;
> +}
> +
> +int
> +rte_dispatcher_finalize_unregister(struct rte_dispatcher *dispatcher,
> + int finalizer_id)
> +{
> + struct rte_dispatcher_finalizer *unreg_finalizer;
> + int finalizer_idx;
> + uint16_t last_idx;
> +
> + unreg_finalizer = evd_get_finalizer_by_id(dispatcher, finalizer_id);
> +
> + if (unreg_finalizer == NULL) {
> + RTE_EDEV_LOG_ERR("Invalid finalizer id %d\n", finalizer_id);
> + return -EINVAL;
> + }
> +
> + finalizer_idx = unreg_finalizer - &dispatcher->finalizers[0];
> +
> + last_idx = dispatcher->num_finalizers - 1;
> +
> + if (finalizer_idx != last_idx) {
> + /* move all finalizers to maintain order */
> + int n = last_idx - finalizer_idx;
> + memmove(unreg_finalizer, unreg_finalizer + 1,
> + sizeof(struct rte_dispatcher_finalizer) * n);
> + }
> +
> + dispatcher->num_finalizers--;
> +
> + return 0;
> +}
> +
> +static int
> +evd_set_service_runstate(struct rte_dispatcher *dispatcher, int state)
> +{
> + int rc;
> +
> + rc = rte_service_component_runstate_set(dispatcher->service_id,
> + state);
> +
> + if (rc != 0) {
> + RTE_EDEV_LOG_ERR("Unexpected error %d occurred while setting "
> + "service component run state to %d\n", rc,
> + state);
> + RTE_ASSERT(0);
Why not propagating the error to callers?
> + }
> +
> + return 0;
> +}
> +
> +int
> +rte_dispatcher_start(struct rte_dispatcher *dispatcher)
> +{
> + return evd_set_service_runstate(dispatcher, 1);
> +}
> +
> +int
> +rte_dispatcher_stop(struct rte_dispatcher *dispatcher)
> +{
> + return evd_set_service_runstate(dispatcher, 0);
> +}
> +
> +static void
> +evd_aggregate_stats(struct rte_dispatcher_stats *result,
> + const struct rte_dispatcher_stats *part)
> +{
> + result->poll_count += part->poll_count;
> + result->ev_batch_count += part->ev_batch_count;
> + result->ev_dispatch_count += part->ev_dispatch_count;
> + result->ev_drop_count += part->ev_drop_count;
> +}
> +
> +void
> +rte_dispatcher_stats_get(const struct rte_dispatcher *dispatcher,
> + struct rte_dispatcher_stats *stats)
> +{
> + unsigned int lcore_id;
> +
> + *stats = (struct rte_dispatcher_stats) {};
> +
> + for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
> + const struct rte_dispatcher_lcore *lcore =
> + &dispatcher->lcores[lcore_id];
> +
> + evd_aggregate_stats(stats, &lcore->stats);
> + }
> +}
> +
> +void
> +rte_dispatcher_stats_reset(struct rte_dispatcher *dispatcher)
> +{
> + unsigned int lcore_id;
> +
> + for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
> + struct rte_dispatcher_lcore *lcore =
> + &dispatcher->lcores[lcore_id];
> +
> + lcore->stats = (struct rte_dispatcher_stats) {};
> + }
> +}
> diff --git a/lib/dispatcher/rte_dispatcher.h b/lib/dispatcher/rte_dispatcher.h
> new file mode 100644
> index 0000000000..0387316d7b
> --- /dev/null
> +++ b/lib/dispatcher/rte_dispatcher.h
> @@ -0,0 +1,468 @@
> +/* SPDX-License-Identifier: BSD-3-Clause
> + * Copyright(c) 2023 Ericsson AB
> + */
> +
> +#ifndef __RTE_DISPATCHER_H__
> +#define __RTE_DISPATCHER_H__
> +
> +/**
> + * @file
> + *
> + * RTE Dispatcher
> + *
> + * @warning
> + * @b EXPERIMENTAL:
> + * All functions in this file may be changed or removed without prior notice.
> + *
> + * The purpose of the dispatcher is to help decouple different parts
> + * of an application (e.g., modules), sharing the same underlying
> + * event device.
> + */
> +
> +#ifdef __cplusplus
> +extern "C" {
> +#endif
> +
> +#include <rte_eventdev.h>
The headers check does not complain, yet this header as a dependency
on stdbool (for bool type), stdint.h (for uintX_t types) and
rte_compat.h (for __rte_experimental).
I prefer we have explicit includes here rather than rely on implicit
rte_eventdev.h dependencies.
> +
> +/**
> + * Function prototype for match callbacks.
> + *
> + * Match callbacks are used by an application to decide how the
> + * dispatcher distributes events to different parts of the
> + * application.
> + *
> + * The application is not expected to process the event at the point
> + * of the match call. Such matters should be deferred to the process
> + * callback invocation.
> + *
> + * The match callback may be used as an opportunity to prefetch data.
> + *
> + * @param event
> + * Pointer to event
> + *
> + * @param cb_data
> + * The pointer supplied by the application in
> + * rte_dispatcher_register().
> + *
> + * @return
> + * Returns true in case this events should be delivered (via
event*
> + * the process callback), and false otherwise.
> + */
> +typedef bool
> +(*rte_dispatcher_match_t)(const struct rte_event *event, void *cb_data);
> +
> +/**
> + * Function prototype for process callbacks.
> + *
> + * The process callbacks are used by the dispatcher to deliver
> + * events for processing.
> + *
> + * @param event_dev_id
> + * The originating event device id.
> + *
> + * @param event_port_id
> + * The originating event port.
> + *
> + * @param events
> + * Pointer to an array of events.
> + *
> + * @param num
> + * The number of events in the @p events array.
> + *
> + * @param cb_data
> + * The pointer supplied by the application in
> + * rte_dispatcher_register().
> + */
> +
> +typedef void
> +(*rte_dispatcher_process_t)(uint8_t event_dev_id, uint8_t event_port_id,
> + struct rte_event *events, uint16_t num,
> + void *cb_data);
> +
> +/**
> + * Function prototype for finalize callbacks.
> + *
> + * The finalize callbacks are used by the dispatcher to notify the
> + * application it has delivered all events from a particular batch
> + * dequeued from the event device.
> + *
> + * @param event_dev_id
> + * The originating event device id.
> + *
> + * @param event_port_id
> + * The originating event port.
> + *
> + * @param cb_data
> + * The pointer supplied by the application in
> + * rte_dispatcher_finalize_register().
> + */
> +
> +typedef void
> +(*rte_dispatcher_finalize_t)(uint8_t event_dev_id, uint8_t event_port_id,
> + void *cb_data);
> +
> +/**
> + * Dispatcher statistics
> + */
> +struct rte_dispatcher_stats {
> + uint64_t poll_count;
> + /**< Number of event dequeue calls made toward the event device. */
We had a number of issues with doxygen post annotations.
Prefer the prefixed ones.
+ /** Number of event dequeue calls made toward the event device. */
+ uint64_t poll_count;
> + uint64_t ev_batch_count;
> + /**< Number of non-empty event batches dequeued from event device.*/
> + uint64_t ev_dispatch_count;
> + /**< Number of events dispatched to a handler.*/
> + uint64_t ev_drop_count;
> + /**< Number of events dropped because no handler was found. */
> +};
> +
> +/**
> + * Create a dispatcher with the specified id.
> + *
> + * @param event_dev_id
> + * The identifier of the event device from which this dispatcher
> + * will dequeue events.
> + *
> + * @return
> + * A pointer to a new dispatcher instance, or NULL on failure, in which
> + * case rte_errno is set.
> + */
> +__rte_experimental
> +struct rte_dispatcher *
> +rte_dispatcher_create(uint8_t event_dev_id);
> +
> +/**
> + * Free a dispatcher.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @return
> + * - 0: Success
> + * - <0: Error code on failure
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_free(struct rte_dispatcher *dispatcher);
> +
> +/**
> + * Retrieve the service identifier of a dispatcher.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @return
> + * The dispatcher service's id.
> + */
> +__rte_experimental
> +uint32_t
> +rte_dispatcher_service_id_get(const struct rte_dispatcher *dispatcher);
> +
> +/**
> + * Binds an event device port to a specific lcore on the specified
> + * dispatcher.
> + *
> + * This function configures the event port id to be used by the event
> + * dispatcher service, if run on the specified lcore.
> + *
> + * Multiple event device ports may be bound to the same lcore. A
> + * particular port must not be bound to more than one lcore.
> + *
> + * If the dispatcher service is mapped (with rte_service_map_lcore_set())
> + * to a lcore to which no ports are bound, the service function will be a
> + * no-operation.
> + *
> + * This function may be called by any thread (including unregistered
> + * non-EAL threads), but not while the dispatcher is running on lcore
> + * specified by @c lcore_id.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @param event_port_id
> + * The event device port identifier.
> + *
> + * @param batch_size
> + * The batch size to use in rte_event_dequeue_burst(), for the
> + * configured event device port and lcore.
> + *
> + * @param timeout
> + * The timeout parameter to use in rte_event_dequeue_burst(), for the
> + * configured event device port and lcore.
> + *
> + * @param lcore_id
> + * The lcore by which this event port will be used.
> + *
> + * @return
> + * - 0: Success
> + * - -ENOMEM: Unable to allocate sufficient resources.
> + * - -EEXISTS: Event port is already configured.
> + * - -EINVAL: Invalid arguments.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_bind_port_to_lcore(struct rte_dispatcher *dispatcher,
> + uint8_t event_port_id, uint16_t batch_size,
> + uint64_t timeout, unsigned int lcore_id);
> +
> +/**
> + * Unbind an event device port from a specific lcore.
> + *
> + * This function may be called by any thread (including unregistered
> + * non-EAL threads), but not while the dispatcher is running on
> + * lcore specified by @c lcore_id.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @param event_port_id
> + * The event device port identifier.
> + *
> + * @param lcore_id
> + * The lcore which was using this event port.
> + *
> + * @return
> + * - 0: Success
> + * - -ENOENT: Event port id not bound to this @c lcore_id.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_unbind_port_from_lcore(struct rte_dispatcher *dispatcher,
> + uint8_t event_port_id,
> + unsigned int lcore_id);
> +
> +/**
> + * Register an event handler.
> + *
> + * The match callback function is used to select if a particular event
> + * should be delivered, using the corresponding process callback
> + * function.
> + *
> + * The reason for having two distinct steps is to allow the dispatcher
> + * to deliver all events as a batch. This in turn will cause
> + * processing of a particular kind of events to happen in a
> + * back-to-back manner, improving cache locality.
> + *
> + * The list of handler callback functions is shared among all lcores,
> + * but will only be executed on lcores which has an eventdev port
> + * bound to them, and which are running the dispatcher service.
> + *
> + * An event is delivered to at most one handler. Events where no
> + * handler is found are dropped.
> + *
> + * The application must not depend on the order of which the match
> + * functions are invoked.
> + *
> + * Ordering of events is not guaranteed to be maintained between
> + * different deliver callbacks. For example, suppose there are two
> + * callbacks registered, matching different subsets of events arriving
> + * on an atomic queue. A batch of events [ev0, ev1, ev2] are dequeued
> + * on a particular port, all pertaining to the same flow. The match
> + * callback for registration A returns true for ev0 and ev2, and the
> + * matching function for registration B for ev1. In that scenario, the
> + * dispatcher may choose to deliver first [ev0, ev2] using A's deliver
> + * function, and then [ev1] to B - or vice versa.
> + *
> + * rte_dispatcher_register() may be called by any thread
> + * (including unregistered non-EAL threads), but not while the event
> + * dispatcher is running on any service lcore.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @param match_fun
> + * The match callback function.
> + *
> + * @param match_cb_data
> + * A pointer to some application-specific opaque data (or NULL),
> + * which is supplied back to the application when match_fun is
> + * called.
> + *
> + * @param process_fun
> + * The process callback function.
> + *
> + * @param process_cb_data
> + * A pointer to some application-specific opaque data (or NULL),
> + * which is supplied back to the application when process_fun is
> + * called.
> + *
> + * @return
> + * - >= 0: The identifier for this registration.
> + * - -ENOMEM: Unable to allocate sufficient resources.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_register(struct rte_dispatcher *dispatcher,
> + rte_dispatcher_match_t match_fun, void *match_cb_data,
> + rte_dispatcher_process_t process_fun,
> + void *process_cb_data);
> +
> +/**
> + * Unregister an event handler.
> + *
> + * This function may be called by any thread (including unregistered
> + * non-EAL threads), but not while the dispatcher is running on
> + * any service lcore.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @param handler_id
> + * The handler registration id returned by the original
> + * rte_dispatcher_register() call.
> + *
> + * @return
> + * - 0: Success
> + * - -EINVAL: The @c handler_id parameter was invalid.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_unregister(struct rte_dispatcher *dispatcher, int handler_id);
> +
> +/**
> + * Register a finalize callback function.
> + *
> + * An application may optionally install one or more finalize
> + * callbacks.
> + *
> + * All finalize callbacks are invoked by the dispatcher when a
> + * complete batch of events (retrieve using rte_event_dequeue_burst())
> + * have been delivered to the application (or have been dropped).
> + *
> + * The finalize callback is not tied to any particular handler.
> + *
> + * The finalize callback provides an opportunity for the application
> + * to do per-batch processing. One case where this may be useful is if
> + * an event output buffer is used, and is shared among several
> + * handlers. In such a case, proper output buffer flushing may be
> + * assured using a finalize callback.
> + *
> + * rte_dispatcher_finalize_register() may be called by any thread
> + * (including unregistered non-EAL threads), but not while the
> + * dispatcher is running on any service lcore.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @param finalize_fun
> + * The function called after completing the processing of a
> + * dequeue batch.
> + *
> + * @param finalize_data
> + * A pointer to some application-specific opaque data (or NULL),
> + * which is supplied back to the application when @c finalize_fun is
> + * called.
> + *
> + * @return
> + * - >= 0: The identifier for this registration.
> + * - -ENOMEM: Unable to allocate sufficient resources.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_finalize_register(struct rte_dispatcher *dispatcher,
> + rte_dispatcher_finalize_t finalize_fun,
> + void *finalize_data);
> +
> +/**
> + * Unregister a finalize callback.
> + *
> + * This function may be called by any thread (including unregistered
> + * non-EAL threads), but not while the dispatcher is running on
> + * any service lcore.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @param reg_id
> + * The finalize registration id returned by the original
> + * rte_dispatcher_finalize_register() call.
> + *
> + * @return
> + * - 0: Success
> + * - -EINVAL: The @c reg_id parameter was invalid.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_finalize_unregister(struct rte_dispatcher *dispatcher,
> + int reg_id);
> +
> +/**
> + * Start a dispatcher instance.
> + *
> + * Enables the dispatcher service.
> + *
> + * The underlying event device must have been started prior to calling
> + * rte_dispatcher_start().
> + *
> + * For the dispatcher to actually perform work (i.e., dispatch
> + * events), its service must have been mapped to one or more service
> + * lcores, and its service run state set to '1'. A dispatcher's
> + * service is retrieved using rte_dispatcher_service_id_get().
> + *
> + * Each service lcore to which the dispatcher is mapped should
> + * have at least one event port configured. Such configuration is
> + * performed by calling rte_dispatcher_bind_port_to_lcore(), prior to
> + * starting the dispatcher.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @return
> + * - 0: Success
> + * - <0: Error code on failure
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_start(struct rte_dispatcher *dispatcher);
> +
> +/**
> + * Stop a running dispatcher instance.
> + *
> + * Disables the dispatcher service.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + *
> + * @return
> + * - 0: Success
> + * - -EINVAL: Invalid @c id.
> + */
> +__rte_experimental
> +int
> +rte_dispatcher_stop(struct rte_dispatcher *dispatcher);
> +
> +/**
> + * Retrieve statistics for a dispatcher instance.
> + *
> + * This function is MT safe and may be called by any thread
> + * (including unregistered non-EAL threads).
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + * @param[out] stats
> + * A pointer to a structure to fill with statistics.
> + */
> +__rte_experimental
> +void
> +rte_dispatcher_stats_get(const struct rte_dispatcher *dispatcher,
> + struct rte_dispatcher_stats *stats);
> +
> +/**
> + * Reset statistics for a dispatcher instance.
> + *
> + * This function may be called by any thread (including unregistered
> + * non-EAL threads), but may not produce the correct result if the
> + * dispatcher is running on any service lcore.
> + *
> + * @param dispatcher
> + * The dispatcher instance.
> + */
> +__rte_experimental
> +void
> +rte_dispatcher_stats_reset(struct rte_dispatcher *dispatcher);
> +
> +#ifdef __cplusplus
> +}
> +#endif
> +
> +#endif /* __RTE_DISPATCHER__ */
> diff --git a/lib/dispatcher/version.map b/lib/dispatcher/version.map
> new file mode 100644
> index 0000000000..8f9ad96522
> --- /dev/null
> +++ b/lib/dispatcher/version.map
> @@ -0,0 +1,20 @@
> +EXPERIMENTAL {
> + global:
> +
> + # added in 23.11
> + rte_dispatcher_create;
> + rte_dispatcher_free;
> + rte_dispatcher_service_id_get;
> + rte_dispatcher_bind_port_to_lcore;
> + rte_dispatcher_unbind_port_from_lcore;
> + rte_dispatcher_register;
> + rte_dispatcher_unregister;
> + rte_dispatcher_finalize_register;
> + rte_dispatcher_finalize_unregister;
> + rte_dispatcher_start;
> + rte_dispatcher_stop;
> + rte_dispatcher_stats_get;
> + rte_dispatcher_stats_reset;
Sort alphabetically please.
> +
> + local: *;
> +};
> diff --git a/lib/meson.build b/lib/meson.build
> index 099b0ed18a..3093b338d2 100644
> --- a/lib/meson.build
> +++ b/lib/meson.build
> @@ -35,6 +35,7 @@ libraries = [
> 'distributor',
> 'efd',
> 'eventdev',
> + 'dispatcher', # dispatcher depends on eventdev
> 'gpudev',
> 'gro',
> 'gso',
> @@ -81,6 +82,7 @@ optional_libs = [
> 'cfgfile',
> 'compressdev',
> 'cryptodev',
> + 'dispatcher',
> 'distributor',
> 'dmadev',
> 'efd',
> --
> 2.34.1
>
--
David Marchand
