examples/ipsec-secgw/event_helper.c

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright (C) 2020 Marvell International Ltd.
 */
#include <rte_bitmap.h>
#include <rte_ethdev.h>
#include <rte_eventdev.h>
#include <rte_event_eth_rx_adapter.h>
#include <rte_event_eth_tx_adapter.h>
#include <rte_malloc.h>
#include <stdbool.h>

#include "event_helper.h"

static volatile bool eth_core_running;

static int
eh_get_enabled_cores(struct rte_bitmap *eth_core_mask)
{
    int i, count = 0;

    RTE_LCORE_FOREACH(i) {
        /* Check if this core is enabled in core mask*/
        if (rte_bitmap_get(eth_core_mask, i)) {
            /* Found enabled core */
            count++;
        }
    }
    return count;
}

static inline unsigned int
eh_get_next_eth_core(struct eventmode_conf *em_conf)
{
    static unsigned int prev_core = -1;
    unsigned int next_core;

    /*
     * Make sure we have at least one eth core running, else the following
     * logic would lead to an infinite loop.
     */
    if (eh_get_enabled_cores(em_conf->eth_core_mask) == 0) {
        EH_LOG_ERR("No enabled eth core found");
        return RTE_MAX_LCORE;
    }

    /* Only some cores are marked as eth cores, skip others */
    do {
        /* Get the next core */
        next_core = rte_get_next_lcore(prev_core, 0, 1);

        /* Check if we have reached max lcores */
        if (next_core == RTE_MAX_LCORE)
            return next_core;

        /* Update prev_core */
        prev_core = next_core;
    } while (!(rte_bitmap_get(em_conf->eth_core_mask, next_core)));

    return next_core;
}

static inline unsigned int
eh_get_next_active_core(struct eventmode_conf *em_conf, unsigned int prev_core)
{
    unsigned int next_core;

    /* Get next active core skipping cores reserved as eth cores */
    do {
        /* Get the next core */
        next_core = rte_get_next_lcore(prev_core, 0, 0);

        /* Check if we have reached max lcores */
        if (next_core == RTE_MAX_LCORE)
            return next_core;

        prev_core = next_core;
    } while (rte_bitmap_get(em_conf->eth_core_mask, next_core));

    return next_core;
}

static struct eventdev_params *
eh_get_eventdev_params(struct eventmode_conf *em_conf, uint8_t eventdev_id)
{
    int i;

    for (i = 0; i < em_conf->nb_eventdev; i++) {
        if (em_conf->eventdev_config[i].eventdev_id == eventdev_id)
            break;
    }

    /* No match */
    if (i == em_conf->nb_eventdev)
        return NULL;

    return &(em_conf->eventdev_config[i]);
}

static inline bool
eh_dev_has_rx_internal_port(uint8_t eventdev_id)
{
    bool flag = true;
    int j, ret;

    RTE_ETH_FOREACH_DEV(j) {
        uint32_t caps = 0;

        ret = rte_event_eth_rx_adapter_caps_get(eventdev_id, j, &caps);
        if (ret < 0)
            return false;

        if (!(caps & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT))
            flag = false;
    }
    return flag;
}

static inline bool
eh_dev_has_tx_internal_port(uint8_t eventdev_id)
{
    bool flag = true;
    int j, ret;

    RTE_ETH_FOREACH_DEV(j) {
        uint32_t caps = 0;

        ret = rte_event_eth_tx_adapter_caps_get(eventdev_id, j, &caps);
        if (ret < 0)
            return false;

        if (!(caps & RTE_EVENT_ETH_TX_ADAPTER_CAP_INTERNAL_PORT))
            flag = false;
    }
    return flag;
}

static inline bool
eh_dev_has_burst_mode(uint8_t dev_id)
{
    struct rte_event_dev_info dev_info;

    rte_event_dev_info_get(dev_id, &dev_info);
    return (dev_info.event_dev_cap & RTE_EVENT_DEV_CAP_BURST_MODE) ?
            true : false;
}

static int
eh_set_default_conf_eventdev(struct eventmode_conf *em_conf)
{
    int lcore_count, nb_eventdev, nb_eth_dev, ret;
    struct eventdev_params *eventdev_config;
    struct rte_event_dev_info dev_info;

    /* Get the number of event devices */
    nb_eventdev = rte_event_dev_count();
    if (nb_eventdev == 0) {
        EH_LOG_ERR("No event devices detected");
        return -EINVAL;
    }

    if (nb_eventdev != 1) {
        EH_LOG_ERR("Event mode does not support multiple event devices. "
               "Please provide only one event device.");
        return -EINVAL;
    }

    /* Get the number of eth devs */
    nb_eth_dev = rte_eth_dev_count_avail();
    if (nb_eth_dev == 0) {
        EH_LOG_ERR("No eth devices detected");
        return -EINVAL;
    }

    /* Get the number of lcores */
    lcore_count = rte_lcore_count();

    /* Read event device info */
    ret = rte_event_dev_info_get(0, &dev_info);
    if (ret < 0) {
        EH_LOG_ERR("Failed to read event device info %d", ret);
        return ret;
    }

    /* Check if enough ports are available */
    if (dev_info.max_event_ports < 2) {
        EH_LOG_ERR("Not enough event ports available");
        return -EINVAL;
    }

    /* Get the first event dev conf */
    eventdev_config = &(em_conf->eventdev_config[0]);

    /* Save number of queues & ports available */
    eventdev_config->eventdev_id = 0;
    eventdev_config->nb_eventqueue = dev_info.max_event_queues;
    eventdev_config->nb_eventport = dev_info.max_event_ports;
    eventdev_config->ev_queue_mode = RTE_EVENT_QUEUE_CFG_ALL_TYPES;

    /* Check if there are more queues than required */
    if (eventdev_config->nb_eventqueue > nb_eth_dev + 1) {
        /* One queue is reserved for Tx */
        eventdev_config->nb_eventqueue = nb_eth_dev + 1;
    }

    /* Check if there are more ports than required */
    if (eventdev_config->nb_eventport > lcore_count) {
        /* One port per lcore is enough */
        eventdev_config->nb_eventport = lcore_count;
    }

    /* Update the number of event devices */
    em_conf->nb_eventdev++;

    return 0;
}

static void
eh_do_capability_check(struct eventmode_conf *em_conf)
{
    struct eventdev_params *eventdev_config;
    int all_internal_ports = 1;
    uint32_t eventdev_id;
    int i;

    for (i = 0; i < em_conf->nb_eventdev; i++) {

        /* Get the event dev conf */
        eventdev_config = &(em_conf->eventdev_config[i]);
        eventdev_id = eventdev_config->eventdev_id;

        /* Check if event device has internal port for Rx & Tx */
        if (eh_dev_has_rx_internal_port(eventdev_id) &&
            eh_dev_has_tx_internal_port(eventdev_id)) {
            eventdev_config->all_internal_ports = 1;
        } else {
            all_internal_ports = 0;
        }
    }

    /*
     * If Rx & Tx internal ports are supported by all event devices then
     * eth cores won't be required. Override the eth core mask requested
     * and decrement number of event queues by one as it won't be needed
     * for Tx.
     */
    if (all_internal_ports) {
        rte_bitmap_reset(em_conf->eth_core_mask);
        for (i = 0; i < em_conf->nb_eventdev; i++)
            em_conf->eventdev_config[i].nb_eventqueue--;
    }
}

static int
eh_set_default_conf_link(struct eventmode_conf *em_conf)
{
    struct eventdev_params *eventdev_config;
    struct eh_event_link_info *link;
    unsigned int lcore_id = -1;
    int i, link_index;

    /*
     * Create a 1:1 mapping from event ports to cores. If the number
     * of event ports is lesser than the cores, some cores won't
     * execute worker. If there are more event ports, then some ports
     * won't be used.
     *
     */

    /*
     * The event queue-port mapping is done according to the link. Since
     * we are falling back to the default link config, enabling
     * "all_ev_queue_to_ev_port" mode flag. This will map all queues
     * to the port.
     */
    em_conf->ext_params.all_ev_queue_to_ev_port = 1;

    /* Get first event dev conf */
    eventdev_config = &(em_conf->eventdev_config[0]);

    /* Loop through the ports */
    for (i = 0; i < eventdev_config->nb_eventport; i++) {

        /* Get next active core id */
        lcore_id = eh_get_next_active_core(em_conf,
                lcore_id);

        if (lcore_id == RTE_MAX_LCORE) {
            /* Reached max cores */
            return 0;
        }

        /* Save the current combination as one link */

        /* Get the index */
        link_index = em_conf->nb_link;

        /* Get the corresponding link */
        link = &(em_conf->link[link_index]);

        /* Save link */
        link->eventdev_id = eventdev_config->eventdev_id;
        link->event_port_id = i;
        link->lcore_id = lcore_id;

        /*
         * Don't set eventq_id as by default all queues
         * need to be mapped to the port, which is controlled
         * by the operating mode.
         */

        /* Update number of links */
        em_conf->nb_link++;
    }

    return 0;
}

static int
eh_set_default_conf_rx_adapter(struct eventmode_conf *em_conf)
{
    struct rx_adapter_connection_info *conn;
    struct eventdev_params *eventdev_config;
    struct rx_adapter_conf *adapter;
    bool rx_internal_port = true;
    bool single_ev_queue = false;
    int nb_eventqueue;
    uint32_t caps = 0;
    int eventdev_id;
    int nb_eth_dev;
    int adapter_id;
    int conn_id;
    int ret;
    int i;

    /* Create one adapter with eth queues mapped to event queue(s) */

    if (em_conf->nb_eventdev == 0) {
        EH_LOG_ERR("No event devs registered");
        return -EINVAL;
    }

    /* Get the number of eth devs */
    nb_eth_dev = rte_eth_dev_count_avail();

    /* Use the first event dev */
    eventdev_config = &(em_conf->eventdev_config[0]);

    /* Get eventdev ID */
    eventdev_id = eventdev_config->eventdev_id;
    adapter_id = 0;

    /* Get adapter conf */
    adapter = &(em_conf->rx_adapter[adapter_id]);

    /* Set adapter conf */
    adapter->eventdev_id = eventdev_id;
    adapter->adapter_id = adapter_id;

    /*
     * If event device does not have internal ports for passing
     * packets then reserved one queue for Tx path
     */
    nb_eventqueue = eventdev_config->all_internal_ports ?
            eventdev_config->nb_eventqueue :
            eventdev_config->nb_eventqueue - 1;

    /*
     * Map all queues of eth device (port) to an event queue. If there
     * are more event queues than eth ports then create 1:1 mapping.
     * Otherwise map all eth ports to a single event queue.
     */
    if (nb_eth_dev > nb_eventqueue)
        single_ev_queue = true;

    for (i = 0; i < nb_eth_dev; i++) {

        /* Use only the ports enabled */
        if ((em_conf->eth_portmask & (1 << i)) == 0)
            continue;

        /* Get the connection id */
        conn_id = adapter->nb_connections;

        /* Get the connection */
        conn = &(adapter->conn[conn_id]);

        /* Set mapping between eth ports & event queues*/
        conn->ethdev_id = i;
        conn->eventq_id = single_ev_queue ? 0 : i;

        /* Add all eth queues eth port to event queue */
        conn->ethdev_rx_qid = -1;

        /* Get Rx adapter capabilities */
        ret = rte_event_eth_rx_adapter_caps_get(eventdev_id, i, &caps);
        if (ret < 0) {
            EH_LOG_ERR("Failed to get event device %d eth rx adapter"
                   " capabilities for port %d", eventdev_id, i);
            return ret;
        }
        if (!(caps & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT))
            rx_internal_port = false;

        /* Update no of connections */
        adapter->nb_connections++;

    }

    if (rx_internal_port) {
        /* Rx core is not required */
        adapter->rx_core_id = -1;
    } else {
        /* Rx core is required */
        adapter->rx_core_id = eh_get_next_eth_core(em_conf);
    }

    /* We have setup one adapter */
    em_conf->nb_rx_adapter = 1;

    return 0;
}

static int
eh_set_default_conf_tx_adapter(struct eventmode_conf *em_conf)
{
    struct tx_adapter_connection_info *conn;
    struct eventdev_params *eventdev_config;
    struct tx_adapter_conf *tx_adapter;
    bool tx_internal_port = true;
    uint32_t caps = 0;
    int eventdev_id;
    int adapter_id;
    int nb_eth_dev;
    int conn_id;
    int ret;
    int i;

    /*
     * Create one Tx adapter with all eth queues mapped to event queues
     * 1:1.
     */

    if (em_conf->nb_eventdev == 0) {
        EH_LOG_ERR("No event devs registered");
        return -EINVAL;
    }

    /* Get the number of eth devs */
    nb_eth_dev = rte_eth_dev_count_avail();

    /* Use the first event dev */
    eventdev_config = &(em_conf->eventdev_config[0]);

    /* Get eventdev ID */
    eventdev_id = eventdev_config->eventdev_id;
    adapter_id = 0;

    /* Get adapter conf */
    tx_adapter = &(em_conf->tx_adapter[adapter_id]);

    /* Set adapter conf */
    tx_adapter->eventdev_id = eventdev_id;
    tx_adapter->adapter_id = adapter_id;

    /*
     * Map all Tx queues of the eth device (port) to the event device.
     */

    /* Set defaults for connections */

    /*
     * One eth device (port) is one connection. Map all Tx queues
     * of the device to the Tx adapter.
     */

    for (i = 0; i < nb_eth_dev; i++) {

        /* Use only the ports enabled */
        if ((em_conf->eth_portmask & (1 << i)) == 0)
            continue;

        /* Get the connection id */
        conn_id = tx_adapter->nb_connections;

        /* Get the connection */
        conn = &(tx_adapter->conn[conn_id]);

        /* Add ethdev to connections */
        conn->ethdev_id = i;

        /* Add all eth tx queues to adapter */
        conn->ethdev_tx_qid = -1;

        /* Get Tx adapter capabilities */
        ret = rte_event_eth_tx_adapter_caps_get(eventdev_id, i, &caps);
        if (ret < 0) {
            EH_LOG_ERR("Failed to get event device %d eth tx adapter"
                   " capabilities for port %d", eventdev_id, i);
            return ret;
        }
        if (!(caps & RTE_EVENT_ETH_TX_ADAPTER_CAP_INTERNAL_PORT))
            tx_internal_port = false;

        /* Update no of connections */
        tx_adapter->nb_connections++;
    }

    if (tx_internal_port) {
        /* Tx core is not required */
        tx_adapter->tx_core_id = -1;
    } else {
        /* Tx core is required */
        tx_adapter->tx_core_id = eh_get_next_eth_core(em_conf);

        /*
         * Use one event queue per adapter for submitting packets
         * for Tx. Reserving the last queue available
         */
        /* Queue numbers start at 0 */
        tx_adapter->tx_ev_queue = eventdev_config->nb_eventqueue - 1;
    }

    /* We have setup one adapter */
    em_conf->nb_tx_adapter = 1;
    return 0;
}

static int
eh_validate_conf(struct eventmode_conf *em_conf)
{
    int ret;

    /*
     * Check if event devs are specified. Else probe the event devices
     * and initialize the config with all ports & queues available
     */
    if (em_conf->nb_eventdev == 0) {
        ret = eh_set_default_conf_eventdev(em_conf);
        if (ret != 0)
            return ret;
    }

    /* Perform capability check for the selected event devices */
    eh_do_capability_check(em_conf);

    /*
     * Check if links are specified. Else generate a default config for
     * the event ports used.
     */
    if (em_conf->nb_link == 0) {
        ret = eh_set_default_conf_link(em_conf);
        if (ret != 0)
            return ret;
    }

    /*
     * Check if rx adapters are specified. Else generate a default config
     * with one rx adapter and all eth queues - event queue mapped.
     */
    if (em_conf->nb_rx_adapter == 0) {
        ret = eh_set_default_conf_rx_adapter(em_conf);
        if (ret != 0)
            return ret;
    }

    /*
     * Check if tx adapters are specified. Else generate a default config
     * with one tx adapter.
     */
    if (em_conf->nb_tx_adapter == 0) {
        ret = eh_set_default_conf_tx_adapter(em_conf);
        if (ret != 0)
            return ret;
    }

    return 0;
}

static int
eh_initialize_eventdev(struct eventmode_conf *em_conf)
{
    struct rte_event_queue_conf eventq_conf = {0};
    struct rte_event_dev_info evdev_default_conf;
    struct rte_event_dev_config eventdev_conf;
    struct eventdev_params *eventdev_config;
    int nb_eventdev = em_conf->nb_eventdev;
    struct eh_event_link_info *link;
    uint8_t *queue = NULL;
    uint8_t eventdev_id;
    int nb_eventqueue;
    uint8_t i, j;
    int ret;

    for (i = 0; i < nb_eventdev; i++) {

        /* Get eventdev config */
        eventdev_config = &(em_conf->eventdev_config[i]);

        /* Get event dev ID */
        eventdev_id = eventdev_config->eventdev_id;

        /* Get the number of queues */
        nb_eventqueue = eventdev_config->nb_eventqueue;

        /* Reset the default conf */
        memset(&evdev_default_conf, 0,
            sizeof(struct rte_event_dev_info));

        /* Get default conf of eventdev */
        ret = rte_event_dev_info_get(eventdev_id, &evdev_default_conf);
        if (ret < 0) {
            EH_LOG_ERR(
                "Error in getting event device info[devID:%d]",
                eventdev_id);
            return ret;
        }

        memset(&eventdev_conf, 0, sizeof(struct rte_event_dev_config));
        eventdev_conf.nb_events_limit =
                evdev_default_conf.max_num_events;
        eventdev_conf.nb_event_queues = nb_eventqueue;
        eventdev_conf.nb_event_ports =
                eventdev_config->nb_eventport;
        eventdev_conf.nb_event_queue_flows =
                evdev_default_conf.max_event_queue_flows;
        eventdev_conf.nb_event_port_dequeue_depth =
                evdev_default_conf.max_event_port_dequeue_depth;
        eventdev_conf.nb_event_port_enqueue_depth =
                evdev_default_conf.max_event_port_enqueue_depth;

        /* Configure event device */
        ret = rte_event_dev_configure(eventdev_id, &eventdev_conf);
        if (ret < 0) {
            EH_LOG_ERR("Error in configuring event device");
            return ret;
        }

        /* Configure event queues */
        for (j = 0; j < nb_eventqueue; j++) {

            memset(&eventq_conf, 0,
                    sizeof(struct rte_event_queue_conf));

            /* Per event dev queues can be ATQ or SINGLE LINK */
            eventq_conf.event_queue_cfg =
                    eventdev_config->ev_queue_mode;
            /*
             * All queues need to be set with sched_type as
             * schedule type for the application stage. One
             * queue would be reserved for the final eth tx
             * stage if event device does not have internal
             * ports. This will be an atomic queue.
             */
            if (!eventdev_config->all_internal_ports &&
                j == nb_eventqueue-1) {
                eventq_conf.schedule_type =
                    RTE_SCHED_TYPE_ATOMIC;
            } else {
                eventq_conf.schedule_type =
                    em_conf->ext_params.sched_type;
            }

            /* Set max atomic flows to 1024 */
            eventq_conf.nb_atomic_flows = 1024;
            eventq_conf.nb_atomic_order_sequences = 1024;

            /* Setup the queue */
            ret = rte_event_queue_setup(eventdev_id, j,
                    &eventq_conf);
            if (ret < 0) {
                EH_LOG_ERR("Failed to setup event queue %d",
                       ret);
                return ret;
            }
        }

        /* Configure event ports */
        for (j = 0; j <  eventdev_config->nb_eventport; j++) {
            ret = rte_event_port_setup(eventdev_id, j, NULL);
            if (ret < 0) {
                EH_LOG_ERR("Failed to setup event port %d",
                       ret);
                return ret;
            }
        }
    }

    /* Make event queue - event port link */
    for (j = 0; j <  em_conf->nb_link; j++) {

        /* Get link info */
        link = &(em_conf->link[j]);

        /* Get event dev ID */
        eventdev_id = link->eventdev_id;

        /*
         * If "all_ev_queue_to_ev_port" params flag is selected, all
         * queues need to be mapped to the port.
         */
        if (em_conf->ext_params.all_ev_queue_to_ev_port)
            queue = NULL;
        else
            queue = &(link->eventq_id);

        /* Link queue to port */
        ret = rte_event_port_link(eventdev_id, link->event_port_id,
                queue, NULL, 1);
        if (ret < 0) {
            EH_LOG_ERR("Failed to link event port %d", ret);
            return ret;
        }
    }

    return 0;
}

static int
eh_start_eventdev(struct eventmode_conf *em_conf)
{
    struct eventdev_params *eventdev_config;
    int nb_eventdev = em_conf->nb_eventdev;
    int i, ret;

    /* Start event devices */
    for (i = 0; i < nb_eventdev; i++) {

        /* Get eventdev config */
        eventdev_config = &(em_conf->eventdev_config[i]);

        ret = rte_event_dev_start(eventdev_config->eventdev_id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to start event device %d, %d",
                   i, ret);
            return ret;
        }
    }
    return 0;
}

static int
eh_rx_adapter_configure(struct eventmode_conf *em_conf,
        struct rx_adapter_conf *adapter)
{
    struct rte_event_eth_rx_adapter_queue_conf queue_conf = {0};
    struct rte_event_dev_info evdev_default_conf = {0};
    struct rte_event_port_conf port_conf = {0};
    struct rx_adapter_connection_info *conn;
    uint8_t eventdev_id;
    uint32_t service_id;
    int ret;
    int j;

    /* Get event dev ID */
    eventdev_id = adapter->eventdev_id;

    /* Get default configuration of event dev */
    ret = rte_event_dev_info_get(eventdev_id, &evdev_default_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to get event dev info %d", ret);
        return ret;
    }

    /* Setup port conf */
    port_conf.new_event_threshold = 1200;
    port_conf.dequeue_depth =
            evdev_default_conf.max_event_port_dequeue_depth;
    port_conf.enqueue_depth =
            evdev_default_conf.max_event_port_enqueue_depth;

    /* Create Rx adapter */
    ret = rte_event_eth_rx_adapter_create(adapter->adapter_id,
            adapter->eventdev_id, &port_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to create rx adapter %d", ret);
        return ret;
    }

    /* Setup various connections in the adapter */
    for (j = 0; j < adapter->nb_connections; j++) {
        /* Get connection */
        conn = &(adapter->conn[j]);

        /* Setup queue conf */
        queue_conf.ev.queue_id = conn->eventq_id;
        queue_conf.ev.sched_type = em_conf->ext_params.sched_type;
        queue_conf.ev.event_type = RTE_EVENT_TYPE_ETHDEV;

        /* Add queue to the adapter */
        ret = rte_event_eth_rx_adapter_queue_add(adapter->adapter_id,
                conn->ethdev_id, conn->ethdev_rx_qid,
                &queue_conf);
        if (ret < 0) {
            EH_LOG_ERR("Failed to add eth queue to rx adapter %d",
                   ret);
            return ret;
        }
    }

    /* Get the service ID used by rx adapter */
    ret = rte_event_eth_rx_adapter_service_id_get(adapter->adapter_id,
                              &service_id);
    if (ret != -ESRCH && ret < 0) {
        EH_LOG_ERR("Failed to get service id used by rx adapter %d",
               ret);
        return ret;
    }

    rte_service_set_runstate_mapped_check(service_id, 0);

    /* Start adapter */
    ret = rte_event_eth_rx_adapter_start(adapter->adapter_id);
    if (ret < 0) {
        EH_LOG_ERR("Failed to start rx adapter %d", ret);
        return ret;
    }

    return 0;
}

static int
eh_initialize_rx_adapter(struct eventmode_conf *em_conf)
{
    struct rx_adapter_conf *adapter;
    int i, ret;

    /* Configure rx adapters */
    for (i = 0; i < em_conf->nb_rx_adapter; i++) {
        adapter = &(em_conf->rx_adapter[i]);
        ret = eh_rx_adapter_configure(em_conf, adapter);
        if (ret < 0) {
            EH_LOG_ERR("Failed to configure rx adapter %d", ret);
            return ret;
        }
    }
    return 0;
}

static int32_t
eh_start_worker_eth_core(struct eventmode_conf *conf, uint32_t lcore_id)
{
    uint32_t service_id[EVENT_MODE_MAX_ADAPTERS_PER_RX_CORE];
    struct rx_adapter_conf *rx_adapter;
    struct tx_adapter_conf *tx_adapter;
    int service_count = 0;
    int adapter_id;
    int32_t ret;
    int i;

    EH_LOG_INFO("Entering eth_core processing on lcore %u", lcore_id);

    /*
     * Parse adapter config to check which of all Rx adapters need
     * to be handled by this core.
     */
    for (i = 0; i < conf->nb_rx_adapter; i++) {
        /* Check if we have exceeded the max allowed */
        if (service_count > EVENT_MODE_MAX_ADAPTERS_PER_RX_CORE) {
            EH_LOG_ERR(
                  "Exceeded the max allowed adapters per rx core");
            break;
        }

        rx_adapter = &(conf->rx_adapter[i]);
        if (rx_adapter->rx_core_id != lcore_id)
            continue;

        /* Adapter is handled by this core */
        adapter_id = rx_adapter->adapter_id;

        /* Get the service ID for the adapters */
        ret = rte_event_eth_rx_adapter_service_id_get(adapter_id,
                &(service_id[service_count]));

        if (ret != -ESRCH && ret < 0) {
            EH_LOG_ERR(
                "Failed to get service id used by rx adapter");
            return ret;
        }

        /* Update service count */
        service_count++;
    }

    /*
     * Parse adapter config to see which of all Tx adapters need
     * to be handled by this core.
     */
    for (i = 0; i < conf->nb_tx_adapter; i++) {
        /* Check if we have exceeded the max allowed */
        if (service_count > EVENT_MODE_MAX_ADAPTERS_PER_TX_CORE) {
            EH_LOG_ERR(
                "Exceeded the max allowed adapters per tx core");
            break;
        }

        tx_adapter = &conf->tx_adapter[i];
        if (tx_adapter->tx_core_id != lcore_id)
            continue;

        /* Adapter is handled by this core */
        adapter_id = tx_adapter->adapter_id;

        /* Get the service ID for the adapters */
        ret = rte_event_eth_tx_adapter_service_id_get(adapter_id,
                &(service_id[service_count]));

        if (ret != -ESRCH && ret < 0) {
            EH_LOG_ERR(
                "Failed to get service id used by tx adapter");
            return ret;
        }

        /* Update service count */
        service_count++;
    }

    eth_core_running = true;

    while (eth_core_running) {
        for (i = 0; i < service_count; i++) {
            /* Initiate adapter service */
            rte_service_run_iter_on_app_lcore(service_id[i], 0);
        }
    }

    return 0;
}

static int32_t
eh_stop_worker_eth_core(void)
{
    if (eth_core_running) {
        EH_LOG_INFO("Stopping eth cores");
        eth_core_running = false;
    }
    return 0;
}

static struct eh_app_worker_params *
eh_find_worker(uint32_t lcore_id, struct eh_conf *conf,
        struct eh_app_worker_params *app_wrkrs, uint8_t nb_wrkr_param)
{
    struct eh_app_worker_params curr_conf = { {{0} }, NULL};
    struct eh_event_link_info *link = NULL;
    struct eh_app_worker_params *tmp_wrkr;
    struct eventmode_conf *em_conf;
    uint8_t eventdev_id;
    int i;

    /* Get eventmode config */
    em_conf = conf->mode_params;

    /*
     * Use event device from the first lcore-event link.
     *
     * Assumption: All lcore-event links tied to a core are using the
     * same event device. In other words, one core would be polling on
     * queues of a single event device only.
     */

    /* Get a link for this lcore */
    for (i = 0; i < em_conf->nb_link; i++) {
        link = &(em_conf->link[i]);
        if (link->lcore_id == lcore_id)
            break;
    }

    if (link == NULL) {
        EH_LOG_ERR("No valid link found for lcore %d", lcore_id);
        return NULL;
    }

    /* Get event dev ID */
    eventdev_id = link->eventdev_id;

    /* Populate the curr_conf with the capabilities */

    /* Check for Tx internal port */
    if (eh_dev_has_tx_internal_port(eventdev_id))
        curr_conf.cap.tx_internal_port = EH_TX_TYPE_INTERNAL_PORT;
    else
        curr_conf.cap.tx_internal_port = EH_TX_TYPE_NO_INTERNAL_PORT;

    /* Check for burst mode */
    if (eh_dev_has_burst_mode(eventdev_id))
        curr_conf.cap.burst = EH_RX_TYPE_BURST;
    else
        curr_conf.cap.burst = EH_RX_TYPE_NON_BURST;

    curr_conf.cap.ipsec_mode = conf->ipsec_mode;

    /* Parse the passed list and see if we have matching capabilities */

    /* Initialize the pointer used to traverse the list */
    tmp_wrkr = app_wrkrs;

    for (i = 0; i < nb_wrkr_param; i++, tmp_wrkr++) {

        /* Skip this if capabilities are not matching */
        if (tmp_wrkr->cap.u64 != curr_conf.cap.u64)
            continue;

        /* If the checks pass, we have a match */
        return tmp_wrkr;
    }

    return NULL;
}

static int
eh_verify_match_worker(struct eh_app_worker_params *match_wrkr)
{
    /* Verify registered worker */
    if (match_wrkr->worker_thread == NULL) {
        EH_LOG_ERR("No worker registered");
        return 0;
    }

    /* Success */
    return 1;
}

static uint8_t
eh_get_event_lcore_links(uint32_t lcore_id, struct eh_conf *conf,
        struct eh_event_link_info **links)
{
    struct eh_event_link_info *link_cache;
    struct eventmode_conf *em_conf = NULL;
    struct eh_event_link_info *link;
    uint8_t lcore_nb_link = 0;
    size_t single_link_size;
    size_t cache_size;
    int index = 0;
    int i;

    if (conf == NULL || links == NULL) {
        EH_LOG_ERR("Invalid args");
        return -EINVAL;
    }

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    if (em_conf == NULL) {
        EH_LOG_ERR("Invalid event mode parameters");
        return -EINVAL;
    }

    /* Get the number of links registered */
    for (i = 0; i < em_conf->nb_link; i++) {

        /* Get link */
        link = &(em_conf->link[i]);

        /* Check if we have link intended for this lcore */
        if (link->lcore_id == lcore_id) {

            /* Update the number of links for this core */
            lcore_nb_link++;

        }
    }

    /* Compute size of one entry to be copied */
    single_link_size = sizeof(struct eh_event_link_info);

    /* Compute size of the buffer required */
    cache_size = lcore_nb_link * sizeof(struct eh_event_link_info);

    /* Compute size of the buffer required */
    link_cache = calloc(1, cache_size);

    /* Get the number of links registered */
    for (i = 0; i < em_conf->nb_link; i++) {

        /* Get link */
        link = &(em_conf->link[i]);

        /* Check if we have link intended for this lcore */
        if (link->lcore_id == lcore_id) {

            /* Cache the link */
            memcpy(&link_cache[index], link, single_link_size);

            /* Update index */
            index++;
        }
    }

    /* Update the links for application to use the cached links */
    *links = link_cache;

    /* Return the number of cached links */
    return lcore_nb_link;
}

static int
eh_tx_adapter_configure(struct eventmode_conf *em_conf,
        struct tx_adapter_conf *adapter)
{
    struct rte_event_dev_info evdev_default_conf = {0};
    struct rte_event_port_conf port_conf = {0};
    struct tx_adapter_connection_info *conn;
    struct eventdev_params *eventdev_config;
    uint8_t tx_port_id = 0;
    uint8_t eventdev_id;
    uint32_t service_id;
    int ret, j;

    /* Get event dev ID */
    eventdev_id = adapter->eventdev_id;

    /* Get event device conf */
    eventdev_config = eh_get_eventdev_params(em_conf, eventdev_id);

    /* Create Tx adapter */

    /* Get default configuration of event dev */
    ret = rte_event_dev_info_get(eventdev_id, &evdev_default_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to get event dev info %d", ret);
        return ret;
    }

    /* Setup port conf */
    port_conf.new_event_threshold =
            evdev_default_conf.max_num_events;
    port_conf.dequeue_depth =
            evdev_default_conf.max_event_port_dequeue_depth;
    port_conf.enqueue_depth =
            evdev_default_conf.max_event_port_enqueue_depth;

    /* Create adapter */
    ret = rte_event_eth_tx_adapter_create(adapter->adapter_id,
            adapter->eventdev_id, &port_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to create tx adapter %d", ret);
        return ret;
    }

    /* Setup various connections in the adapter */
    for (j = 0; j < adapter->nb_connections; j++) {

        /* Get connection */
        conn = &(adapter->conn[j]);

        /* Add queue to the adapter */
        ret = rte_event_eth_tx_adapter_queue_add(adapter->adapter_id,
                conn->ethdev_id, conn->ethdev_tx_qid);
        if (ret < 0) {
            EH_LOG_ERR("Failed to add eth queue to tx adapter %d",
                   ret);
            return ret;
        }
    }

    /*
     * Check if Tx core is assigned. If Tx core is not assigned then
     * the adapter has internal port for submitting Tx packets and
     * Tx event queue & port setup is not required
     */
    if (adapter->tx_core_id == (uint32_t) (-1)) {
        /* Internal port is present */
        goto skip_tx_queue_port_setup;
    }

    /* Setup Tx queue & port */

    /* Get event port used by the adapter */
    ret = rte_event_eth_tx_adapter_event_port_get(
            adapter->adapter_id, &tx_port_id);
    if (ret) {
        EH_LOG_ERR("Failed to get tx adapter port id %d", ret);
        return ret;
    }

    /*
     * Tx event queue is reserved for Tx adapter. Unlink this queue
     * from all other ports
     *
     */
    for (j = 0; j < eventdev_config->nb_eventport; j++) {
        rte_event_port_unlink(eventdev_id, j,
                      &(adapter->tx_ev_queue), 1);
    }

    /* Link Tx event queue to Tx port */
    ret = rte_event_port_link(eventdev_id, tx_port_id,
            &(adapter->tx_ev_queue), NULL, 1);
    if (ret != 1) {
        EH_LOG_ERR("Failed to link event queue to port");
        return ret;
    }

    /* Get the service ID used by Tx adapter */
    ret = rte_event_eth_tx_adapter_service_id_get(adapter->adapter_id,
                              &service_id);
    if (ret != -ESRCH && ret < 0) {
        EH_LOG_ERR("Failed to get service id used by tx adapter %d",
               ret);
        return ret;
    }

    rte_service_set_runstate_mapped_check(service_id, 0);

skip_tx_queue_port_setup:
    /* Start adapter */
    ret = rte_event_eth_tx_adapter_start(adapter->adapter_id);
    if (ret < 0) {
        EH_LOG_ERR("Failed to start tx adapter %d", ret);
        return ret;
    }

    return 0;
}

static int
eh_initialize_tx_adapter(struct eventmode_conf *em_conf)
{
    struct tx_adapter_conf *adapter;
    int i, ret;

    /* Configure Tx adapters */
    for (i = 0; i < em_conf->nb_tx_adapter; i++) {
        adapter = &(em_conf->tx_adapter[i]);
        ret = eh_tx_adapter_configure(em_conf, adapter);
        if (ret < 0) {
            EH_LOG_ERR("Failed to configure tx adapter %d", ret);
            return ret;
        }
    }
    return 0;
}

static void
eh_display_operating_mode(struct eventmode_conf *em_conf)
{
    char sched_types[][32] = {
        "RTE_SCHED_TYPE_ORDERED",
        "RTE_SCHED_TYPE_ATOMIC",
        "RTE_SCHED_TYPE_PARALLEL",
    };
    EH_LOG_INFO("Operating mode:");

    EH_LOG_INFO("\tScheduling type: \t%s",
        sched_types[em_conf->ext_params.sched_type]);

    EH_LOG_INFO("");
}

static void
eh_display_event_dev_conf(struct eventmode_conf *em_conf)
{
    char queue_mode[][32] = {
        "",
        "ATQ (ALL TYPE QUEUE)",
        "SINGLE LINK",
    };
    char print_buf[256] = { 0 };
    int i;

    EH_LOG_INFO("Event Device Configuration:");

    for (i = 0; i < em_conf->nb_eventdev; i++) {
        sprintf(print_buf,
            "\tDev ID: %-2d \tQueues: %-2d \tPorts: %-2d",
            em_conf->eventdev_config[i].eventdev_id,
            em_conf->eventdev_config[i].nb_eventqueue,
            em_conf->eventdev_config[i].nb_eventport);
        sprintf(print_buf + strlen(print_buf),
            "\tQueue mode: %s",
            queue_mode[em_conf->eventdev_config[i].ev_queue_mode]);
        EH_LOG_INFO("%s", print_buf);
    }
    EH_LOG_INFO("");
}

static void
eh_display_rx_adapter_conf(struct eventmode_conf *em_conf)
{
    int nb_rx_adapter = em_conf->nb_rx_adapter;
    struct rx_adapter_connection_info *conn;
    struct rx_adapter_conf *adapter;
    char print_buf[256] = { 0 };
    int i, j;

    EH_LOG_INFO("Rx adapters configured: %d", nb_rx_adapter);

    for (i = 0; i < nb_rx_adapter; i++) {
        adapter = &(em_conf->rx_adapter[i]);
        sprintf(print_buf,
            "\tRx adapter ID: %-2d\tConnections: %-2d\tEvent dev ID: %-2d",
            adapter->adapter_id,
            adapter->nb_connections,
            adapter->eventdev_id);
        if (adapter->rx_core_id == (uint32_t)-1)
            sprintf(print_buf + strlen(print_buf),
                "\tRx core: %-2s", "[INTERNAL PORT]");
        else if (adapter->rx_core_id == RTE_MAX_LCORE)
            sprintf(print_buf + strlen(print_buf),
                "\tRx core: %-2s", "[NONE]");
        else
            sprintf(print_buf + strlen(print_buf),
                "\tRx core: %-2d", adapter->rx_core_id);

        EH_LOG_INFO("%s", print_buf);

        for (j = 0; j < adapter->nb_connections; j++) {
            conn = &(adapter->conn[j]);

            sprintf(print_buf,
                "\t\tEthdev ID: %-2d", conn->ethdev_id);

            if (conn->ethdev_rx_qid == -1)
                sprintf(print_buf + strlen(print_buf),
                    "\tEth rx queue: %-2s", "ALL");
            else
                sprintf(print_buf + strlen(print_buf),
                    "\tEth rx queue: %-2d",
                    conn->ethdev_rx_qid);

            sprintf(print_buf + strlen(print_buf),
                "\tEvent queue: %-2d", conn->eventq_id);
            EH_LOG_INFO("%s", print_buf);
        }
    }
    EH_LOG_INFO("");
}

static void
eh_display_tx_adapter_conf(struct eventmode_conf *em_conf)
{
    int nb_tx_adapter = em_conf->nb_tx_adapter;
    struct tx_adapter_connection_info *conn;
    struct tx_adapter_conf *adapter;
    char print_buf[256] = { 0 };
    int i, j;

    EH_LOG_INFO("Tx adapters configured: %d", nb_tx_adapter);

    for (i = 0; i < nb_tx_adapter; i++) {
        adapter = &(em_conf->tx_adapter[i]);
        sprintf(print_buf,
            "\tTx adapter ID: %-2d\tConnections: %-2d\tEvent dev ID: %-2d",
            adapter->adapter_id,
            adapter->nb_connections,
            adapter->eventdev_id);
        if (adapter->tx_core_id == (uint32_t)-1)
            sprintf(print_buf + strlen(print_buf),
                "\tTx core: %-2s", "[INTERNAL PORT]");
        else if (adapter->tx_core_id == RTE_MAX_LCORE)
            sprintf(print_buf + strlen(print_buf),
                "\tTx core: %-2s", "[NONE]");
        else
            sprintf(print_buf + strlen(print_buf),
                "\tTx core: %-2d,\tInput event queue: %-2d",
                adapter->tx_core_id, adapter->tx_ev_queue);

        EH_LOG_INFO("%s", print_buf);

        for (j = 0; j < adapter->nb_connections; j++) {
            conn = &(adapter->conn[j]);

            sprintf(print_buf,
                "\t\tEthdev ID: %-2d", conn->ethdev_id);

            if (conn->ethdev_tx_qid == -1)
                sprintf(print_buf + strlen(print_buf),
                    "\tEth tx queue: %-2s", "ALL");
            else
                sprintf(print_buf + strlen(print_buf),
                    "\tEth tx queue: %-2d",
                    conn->ethdev_tx_qid);
            EH_LOG_INFO("%s", print_buf);
        }
    }
    EH_LOG_INFO("");
}

static void
eh_display_link_conf(struct eventmode_conf *em_conf)
{
    struct eh_event_link_info *link;
    char print_buf[256] = { 0 };
    int i;

    EH_LOG_INFO("Links configured: %d", em_conf->nb_link);

    for (i = 0; i < em_conf->nb_link; i++) {
        link = &(em_conf->link[i]);

        sprintf(print_buf,
            "\tEvent dev ID: %-2d\tEvent port: %-2d",
            link->eventdev_id,
            link->event_port_id);

        if (em_conf->ext_params.all_ev_queue_to_ev_port)
            sprintf(print_buf + strlen(print_buf),
                "Event queue: %-2s\t", "ALL");
        else
            sprintf(print_buf + strlen(print_buf),
                "Event queue: %-2d\t", link->eventq_id);

        sprintf(print_buf + strlen(print_buf),
            "Lcore: %-2d", link->lcore_id);
        EH_LOG_INFO("%s", print_buf);
    }
    EH_LOG_INFO("");
}

struct eh_conf *
eh_conf_init(void)
{
    struct eventmode_conf *em_conf = NULL;
    struct eh_conf *conf = NULL;
    unsigned int eth_core_id;
    void *bitmap = NULL;
    uint32_t nb_bytes;

    /* Allocate memory for config */
    conf = calloc(1, sizeof(struct eh_conf));
    if (conf == NULL) {
        EH_LOG_ERR("Failed to allocate memory for eventmode helper "
               "config");
        return NULL;
    }

    /* Set default conf */

    /* Packet transfer mode: poll */
    conf->mode = EH_PKT_TRANSFER_MODE_POLL;
    conf->ipsec_mode = EH_IPSEC_MODE_TYPE_APP;

    /* Keep all ethernet ports enabled by default */
    conf->eth_portmask = -1;

    /* Allocate memory for event mode params */
    conf->mode_params = calloc(1, sizeof(struct eventmode_conf));
    if (conf->mode_params == NULL) {
        EH_LOG_ERR("Failed to allocate memory for event mode params");
        goto free_conf;
    }

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    /* Allocate and initialize bitmap for eth cores */
    nb_bytes = rte_bitmap_get_memory_footprint(RTE_MAX_LCORE);
    if (!nb_bytes) {
        EH_LOG_ERR("Failed to get bitmap footprint");
        goto free_em_conf;
    }

    bitmap = rte_zmalloc("event-helper-ethcore-bitmap", nb_bytes,
                 RTE_CACHE_LINE_SIZE);
    if (!bitmap) {
        EH_LOG_ERR("Failed to allocate memory for eth cores bitmap\n");
        goto free_em_conf;
    }

    em_conf->eth_core_mask = rte_bitmap_init(RTE_MAX_LCORE, bitmap,
                         nb_bytes);
    if (!em_conf->eth_core_mask) {
        EH_LOG_ERR("Failed to initialize bitmap");
        goto free_bitmap;
    }

    /* Set schedule type as not set */
    em_conf->ext_params.sched_type = SCHED_TYPE_NOT_SET;

    /* Set two cores as eth cores for Rx & Tx */

    /* Use first core other than main core as Rx core */
    eth_core_id = rte_get_next_lcore(0, /* curr core */
                     1, /* skip main core */
                     0  /* wrap */);

    rte_bitmap_set(em_conf->eth_core_mask, eth_core_id);

    /* Use next core as Tx core */
    eth_core_id = rte_get_next_lcore(eth_core_id,   /* curr core */
                     1,     /* skip main core */
                     0      /* wrap */);

    rte_bitmap_set(em_conf->eth_core_mask, eth_core_id);

    return conf;

free_bitmap:
    rte_free(bitmap);
free_em_conf:
    free(em_conf);
free_conf:
    free(conf);
    return NULL;
}

void
eh_conf_uninit(struct eh_conf *conf)
{
    struct eventmode_conf *em_conf = NULL;

    if (!conf || !conf->mode_params)
        return;

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    /* Free evenmode configuration memory */
    rte_free(em_conf->eth_core_mask);
    free(em_conf);
    free(conf);
}

void
eh_display_conf(struct eh_conf *conf)
{
    struct eventmode_conf *em_conf;

    if (conf == NULL) {
        EH_LOG_ERR("Invalid event helper configuration");
        return;
    }

    if (conf->mode != EH_PKT_TRANSFER_MODE_EVENT)
        return;

    if (conf->mode_params == NULL) {
        EH_LOG_ERR("Invalid event mode parameters");
        return;
    }

    /* Get eventmode conf */
    em_conf = (struct eventmode_conf *)(conf->mode_params);

    /* Display user exposed operating modes */
    eh_display_operating_mode(em_conf);

    /* Display event device conf */
    eh_display_event_dev_conf(em_conf);

    /* Display Rx adapter conf */
    eh_display_rx_adapter_conf(em_conf);

    /* Display Tx adapter conf */
    eh_display_tx_adapter_conf(em_conf);

    /* Display event-lcore link */
    eh_display_link_conf(em_conf);
}

int32_t
eh_devs_init(struct eh_conf *conf)
{
    struct eventmode_conf *em_conf;
    uint16_t port_id;
    int ret;

    if (conf == NULL) {
        EH_LOG_ERR("Invalid event helper configuration");
        return -EINVAL;
    }

    if (conf->mode != EH_PKT_TRANSFER_MODE_EVENT)
        return 0;

    if (conf->mode_params == NULL) {
        EH_LOG_ERR("Invalid event mode parameters");
        return -EINVAL;
    }

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    /* Eventmode conf would need eth portmask */
    em_conf->eth_portmask = conf->eth_portmask;

    /* Validate the requested config */
    ret = eh_validate_conf(em_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to validate the requested config %d", ret);
        return ret;
    }

    /* Display the current configuration */
    eh_display_conf(conf);

    /* Stop eth devices before setting up adapter */
    RTE_ETH_FOREACH_DEV(port_id) {

        /* Use only the ports enabled */
        if ((conf->eth_portmask & (1 << port_id)) == 0)
            continue;

        ret = rte_eth_dev_stop(port_id);
        if (ret != 0) {
            EH_LOG_ERR("Failed to stop port %u, err: %d",
                    port_id, ret);
            return ret;
        }
    }

    /* Setup eventdev */
    ret = eh_initialize_eventdev(em_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to initialize event dev %d", ret);
        return ret;
    }

    /* Setup Rx adapter */
    ret = eh_initialize_rx_adapter(em_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to initialize rx adapter %d", ret);
        return ret;
    }

    /* Setup Tx adapter */
    ret = eh_initialize_tx_adapter(em_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to initialize tx adapter %d", ret);
        return ret;
    }

    /* Start eventdev */
    ret = eh_start_eventdev(em_conf);
    if (ret < 0) {
        EH_LOG_ERR("Failed to start event dev %d", ret);
        return ret;
    }

    /* Start eth devices after setting up adapter */
    RTE_ETH_FOREACH_DEV(port_id) {

        /* Use only the ports enabled */
        if ((conf->eth_portmask & (1 << port_id)) == 0)
            continue;

        ret = rte_eth_dev_start(port_id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to start eth dev %d, %d",
                   port_id, ret);
            return ret;
        }
    }

    return 0;
}

int32_t
eh_devs_uninit(struct eh_conf *conf)
{
    struct eventmode_conf *em_conf;
    int ret, i, j;
    uint16_t id;

    if (conf == NULL) {
        EH_LOG_ERR("Invalid event helper configuration");
        return -EINVAL;
    }

    if (conf->mode != EH_PKT_TRANSFER_MODE_EVENT)
        return 0;

    if (conf->mode_params == NULL) {
        EH_LOG_ERR("Invalid event mode parameters");
        return -EINVAL;
    }

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    /* Stop and release rx adapters */
    for (i = 0; i < em_conf->nb_rx_adapter; i++) {

        id = em_conf->rx_adapter[i].adapter_id;
        ret = rte_event_eth_rx_adapter_stop(id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to stop rx adapter %d", ret);
            return ret;
        }

        for (j = 0; j < em_conf->rx_adapter[i].nb_connections; j++) {

            ret = rte_event_eth_rx_adapter_queue_del(id,
                em_conf->rx_adapter[i].conn[j].ethdev_id, -1);
            if (ret < 0) {
                EH_LOG_ERR(
                       "Failed to remove rx adapter queues %d",
                       ret);
                return ret;
            }
        }

        ret = rte_event_eth_rx_adapter_free(id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to free rx adapter %d", ret);
            return ret;
        }
    }

    /* Stop and release event devices */
    for (i = 0; i < em_conf->nb_eventdev; i++) {

        id = em_conf->eventdev_config[i].eventdev_id;
        rte_event_dev_stop(id);

        ret = rte_event_dev_close(id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to close event dev %d, %d", id, ret);
            return ret;
        }
    }

    /* Stop and release tx adapters */
    for (i = 0; i < em_conf->nb_tx_adapter; i++) {

        id = em_conf->tx_adapter[i].adapter_id;
        ret = rte_event_eth_tx_adapter_stop(id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to stop tx adapter %d", ret);
            return ret;
        }

        for (j = 0; j < em_conf->tx_adapter[i].nb_connections; j++) {

            ret = rte_event_eth_tx_adapter_queue_del(id,
                em_conf->tx_adapter[i].conn[j].ethdev_id, -1);
            if (ret < 0) {
                EH_LOG_ERR(
                    "Failed to remove tx adapter queues %d",
                    ret);
                return ret;
            }
        }

        ret = rte_event_eth_tx_adapter_free(id);
        if (ret < 0) {
            EH_LOG_ERR("Failed to free tx adapter %d", ret);
            return ret;
        }
    }

    return 0;
}

void
eh_launch_worker(struct eh_conf *conf, struct eh_app_worker_params *app_wrkr,
        uint8_t nb_wrkr_param)
{
    struct eh_app_worker_params *match_wrkr;
    struct eh_event_link_info *links = NULL;
    struct eventmode_conf *em_conf;
    uint32_t lcore_id;
    uint8_t nb_links;

    if (conf == NULL) {
        EH_LOG_ERR("Invalid event helper configuration");
        return;
    }

    if (conf->mode_params == NULL) {
        EH_LOG_ERR("Invalid event mode parameters");
        return;
    }

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    /* Get core ID */
    lcore_id = rte_lcore_id();

    /* Check if this is eth core */
    if (rte_bitmap_get(em_conf->eth_core_mask, lcore_id)) {
        eh_start_worker_eth_core(em_conf, lcore_id);
        return;
    }

    if (app_wrkr == NULL || nb_wrkr_param == 0) {
        EH_LOG_ERR("Invalid args");
        return;
    }

    /*
     * This is a regular worker thread. The application registers
     * multiple workers with various capabilities. Run worker
     * based on the selected capabilities of the event
     * device configured.
     */

    /* Get the first matching worker for the event device */
    match_wrkr = eh_find_worker(lcore_id, conf, app_wrkr, nb_wrkr_param);
    if (match_wrkr == NULL) {
        EH_LOG_ERR("Failed to match worker registered for lcore %d",
               lcore_id);
        goto clean_and_exit;
    }

    /* Verify sanity of the matched worker */
    if (eh_verify_match_worker(match_wrkr) != 1) {
        EH_LOG_ERR("Failed to validate the matched worker");
        goto clean_and_exit;
    }

    /* Get worker links */
    nb_links = eh_get_event_lcore_links(lcore_id, conf, &links);

    /* Launch the worker thread */
    match_wrkr->worker_thread(links, nb_links);

    /* Free links info memory */
    free(links);

clean_and_exit:

    /* Flag eth_cores to stop, if started */
    eh_stop_worker_eth_core();
}

uint8_t
eh_get_tx_queue(struct eh_conf *conf, uint8_t eventdev_id)
{
    struct eventdev_params *eventdev_config;
    struct eventmode_conf *em_conf;

    if (conf == NULL) {
        EH_LOG_ERR("Invalid event helper configuration");
        return -EINVAL;
    }

    if (conf->mode_params == NULL) {
        EH_LOG_ERR("Invalid event mode parameters");
        return -EINVAL;
    }

    /* Get eventmode conf */
    em_conf = conf->mode_params;

    /* Get event device conf */
    eventdev_config = eh_get_eventdev_params(em_conf, eventdev_id);

    if (eventdev_config == NULL) {
        EH_LOG_ERR("Failed to read eventdev config");
        return -EINVAL;
    }

    /*
     * The last queue is reserved to be used as atomic queue for the
     * last stage (eth packet tx stage)
     */
    return eventdev_config->nb_eventqueue - 1;
}