em_event.c

/*
 *   Copyright (c) 2015-2026, Nokia Solutions and Networks
 *   All rights reserved.
 *
 *   Redistribution and use in source and binary forms, with or without
 *   modification, are permitted provided that the following conditions
 *   are met:
 *
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the copyright holder nor the names of its
 *       contributors may be used to endorse or promote products derived
 *       from this software without specific prior written permission.
 *
 *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
 
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
 
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
 
#include <stdbool.h>
#include <stdint.h>
#include <string.h>
 
#include <odp_api.h>
 
#include <event_machine.h>
#include <event_machine/platform/env/environment.h>
 
#include "em_chaining.h"
#include "em_error.h"
#include "em_event.h"
#include "em_event_inline.h"
#include "em_event_state.h"
#include "em_event_types.h"
#include "em_libconfig.h"
#include "em_mem.h"
#include "em_pool.h"
#include "em_pool_inline.h"
#include "em_pool_types.h"
#include "em_queue.h"
#include "em_queue_inline.h"
#include "em_queue_types.h"
#include "misc/list.h"
 
/*
 * Sanity check that no extra padding is added to the event_hdr_t by
 * alignment directives etc.
 */
typedef event_hdr_t _ev_hdr__size_check__arr_t[3];
COMPILE_TIME_ASSERT(sizeof(_ev_hdr__size_check__arr_t) ==
                    3 * sizeof(event_hdr_t), EVENT_HDR_SIZE_ERROR2);
 
/*
 * Verify the value set for EM_CHECK_LEVEL - this define is set either from
 * the include/event_machine/api/event_machine_config.h file or by the
 * configure.ac option --enable-check-level=N.
 */
COMPILE_TIME_ASSERT(EM_CHECK_LEVEL >= 0 && EM_CHECK_LEVEL <= 3,
                    EM_CHECK_LEVEL__BAD_VALUE);
 
static int read_config_file(void)
{
        EM_PRINT("EM event vector config:\n");
 
        const char *conf_str = "vector.backend";
        int ret = em_libconfig_lookup_string(&em_shm->libconfig, conf_str,
                                             &em_shm->opt.vector.backend_str);
        if (unlikely(!ret)) {
                EM_LOG(EM_LOG_ERR, "Config option '%s' not found\n", conf_str);
                return -1;
        }
 
        const char *backend_str = em_shm->opt.vector.backend_str;
 
        /* store & print the value */
        if (strcmp(backend_str, "packet") == 0) {
                em_shm->opt.vector.backend = EM_VECTOR_BACKEND_PACKET;
        } else if (strcmp(backend_str, "event") == 0) {
                const odp_pool_capability_t *capa = &em_shm->mpool_tbl.odp_pool_capability;
 
                if (unlikely(capa->event_vector.max_pools == 0)) {
                        EM_LOG(EM_LOG_ERR,
                               "Config option '%s = \"event\"' not supported by ODP\n",
                               conf_str);
                        return -1;
                }
                em_shm->opt.vector.backend = EM_VECTOR_BACKEND_EVENT;
        } else {
                EM_LOG(EM_LOG_ERR,
                       "Invalid value '%s' for %s (expected \"packet\" or \"event\")\n",
                       backend_str, conf_str);
                return -1;
        }
 
        EM_PRINT("  %s: %s\n", conf_str, backend_str);
 
        return 0;
}
 
em_status_t event_init(void)
{
        if (read_config_file())
                return EM_ERR_LIB_FAILED;
 
        return EM_OK;
}
 
void print_event_info(void)
{
        EM_PRINT("\n"
                 "EM Events\n"
                 "---------\n"
                 "event-hdr size: %zu B\n",
                 sizeof(event_hdr_t));
 
        DBG_PRINT("\t\toffset\tsize\n"
                  "\t\t------\t----\n"
                  "esv.state_cnt:\t%3zu B\t%2zu B\n"
                  "esv.state:\t%3zu B\t%2zu B\n"
                  "flags:\t\t%3zu B\t%2zu B\n"
                  "align_offset:\t%3zu B\t%2zu B\n"
                  "event_type:\t%3zu B\t%2zu B\n"
                  "event:\t\t%3zu B\t%2zu B\n"
                  "event_size:\t%3zu B\t%2zu B\n"
                  "egrp_gen:\t%3zu B\t%2zu B\n"
                  "egrp:\t\t%3zu B\t%2zu B\n"
                  "tmo:\t\t%3zu B\t%2zu B\n"
                  "user_area info:\t%3zu B\t%2zu B\n"
                  "vector_type:\t%3zu B\t%2zu B\n"
                  "end_hdr_data:\t%3zu B\t%2zu B\n"
                  "  <pad>\t\t%3zu B\n"
                  "end:\t\t%3zu B\t%2zu B\n",
                  offsetof(event_hdr_t, state_cnt), sizeof_field(event_hdr_t, state_cnt),
                  offsetof(event_hdr_t, state), sizeof_field(event_hdr_t, state),
                  offsetof(event_hdr_t, flags), sizeof_field(event_hdr_t, flags),
                  offsetof(event_hdr_t, align_offset), sizeof_field(event_hdr_t, align_offset),
                  offsetof(event_hdr_t, event_type), sizeof_field(event_hdr_t, event_type),
                  /* NOLINTNEXTLINE(bugprone-sizeof-expression) */
                  offsetof(event_hdr_t, event), sizeof_field(event_hdr_t, event),
                  offsetof(event_hdr_t, event_size), sizeof_field(event_hdr_t, event_size),
                  offsetof(event_hdr_t, egrp_gen), sizeof_field(event_hdr_t, egrp_gen),
                  /* NOLINTNEXTLINE(bugprone-sizeof-expression) */
                  offsetof(event_hdr_t, egrp), sizeof_field(event_hdr_t, egrp),
                  /* NOLINTNEXTLINE(bugprone-sizeof-expression) */
                  offsetof(event_hdr_t, tmo), sizeof_field(event_hdr_t, tmo),
                  offsetof(event_hdr_t, user_area), sizeof_field(event_hdr_t, user_area),
                  offsetof(event_hdr_t, vector_type), sizeof_field(event_hdr_t, vector_type),
                  offsetof(event_hdr_t, end_hdr_data), sizeof_field(event_hdr_t, end_hdr_data),
                  offsetof(event_hdr_t, end) - offsetof(event_hdr_t, end_hdr_data),
                  offsetof(event_hdr_t, end), sizeof_field(event_hdr_t, end));
        EM_PRINT("\n");
}
 
/**
 * Start-up helper for pool preallocation
 */
event_prealloc_hdr_t *event_prealloc(const mpool_elem_t *pool_elem, uint32_t size)
{
        /*
         * EM event pools created with type=PKT can support:
         *   - SW events (bufs)
         *   - pkt events.
         *
         * EM event pools created with type=SW can support:
         *   - SW events (bufs) only
         */
        event_hdr_t *ev_hdr = NULL;
 
        if (pool_elem->event_type == EM_EVENT_TYPE_PACKET)
                ev_hdr = event_alloc_pkt(pool_elem, size);
        else if (pool_elem->event_type == EM_EVENT_TYPE_SW)
                ev_hdr = event_alloc_buf(pool_elem, size);
        else if (pool_elem->event_type == EM_EVENT_TYPE_VECTOR)
                ev_hdr = event_alloc_vector(pool_elem, size);
 
        if (unlikely(ev_hdr == NULL))
                return NULL;
 
        /* event now allocated */
 
        if (esv_enabled()) {
                em_event_t event = ev_hdr->event;
 
                (void)evstate_prealloc(event, ev_hdr);
        }
        ev_hdr->flags.all = 0; /* clear only after evstate_alloc() */
        ev_hdr->user_area.all = 0;
 
        event_prealloc_hdr_t *prealloc_hdr = (event_prealloc_hdr_t *)ev_hdr;
 
        return prealloc_hdr;
}
 
event_prealloc_hdr_t *list_node_to_prealloc_hdr(list_node_t *const list_node)
{
        event_prealloc_hdr_t *const ev_hdr = (event_prealloc_hdr_t *)(uintptr_t)
                ((uint8_t *)list_node - offsetof(event_prealloc_hdr_t, list_node));
 
        return likely(list_node != NULL) ? ev_hdr : NULL;
}
 
/**
 * Helper for em_event_clone().
 *
 * Clone an event originating from an external odp pkt-pool.
 * Initialize the new cloned event as an EM event and return it.
 *
 * Alloc and copy content via ODP.
 * Also the ev_hdr in the odp-pkt user_area is copied.
 */
em_event_t pkt_clone_odp(odp_packet_t pkt, odp_pool_t pkt_pool,
                         uint32_t offset, uint32_t size,
                         bool clone_uarea, bool is_clone_part)
{
        odp_packet_t clone_pkt;
 
        if (is_clone_part) {
                /* only data is copied, ODP-uarea isn't */
                clone_pkt = odp_packet_copy_part(pkt, offset, size, pkt_pool);
                if (unlikely(clone_pkt == ODP_PACKET_INVALID))
                        return EM_EVENT_UNDEF;
 
                const void *src_odp_uarea = odp_packet_user_area(pkt);
                void *dst_odp_uarea = odp_packet_user_area(clone_pkt);
                size_t cpy_size = sizeof(event_hdr_t);
 
                if (clone_uarea) {
                        /* copy ODP-uarea (EM-hdr + EM-uarea) */
                        uint32_t src_uarea_size = odp_packet_user_area_size(pkt);
                        uint32_t dst_uarea_size = odp_packet_user_area_size(clone_pkt);
 
                        if (unlikely(dst_uarea_size < src_uarea_size)) {
                                odp_packet_free(clone_pkt);
                                return EM_EVENT_UNDEF;
                        }
                        /* update 'cpy_size' to include the whole ODP-uarea (EM-hdr + EM-uarea) */
                        cpy_size = src_uarea_size;
                }
                /* copy the EM-hdr and possibly also the EM-uarea if requested */
                memcpy(dst_odp_uarea, src_odp_uarea, cpy_size);
        } else {
                /* identical clone, also ODP-uarea (EM-hdr + EM-uarea) is copied */
                clone_pkt = odp_packet_copy(pkt, pkt_pool);
                if (unlikely(clone_pkt == ODP_PACKET_INVALID))
                        return EM_EVENT_UNDEF;
        }
 
        odp_packet_user_flag_set(clone_pkt, USER_FLAG_SET);
 
        odp_event_t odp_clone_event = odp_packet_to_event(clone_pkt);
        event_hdr_t *clone_hdr = odp_packet_user_area(clone_pkt);
        em_event_t clone_event = event_odp2em(odp_clone_event);
 
        /*
         * Init hdr of event, also ESV init if needed.
         * The clone_hdr is a copy of parent's, update only relevant fields.
         */
        if (esv_enabled())
                clone_event = evstate_init(clone_event, clone_hdr, false);
        else
                clone_hdr->event = clone_event;
 
        clone_hdr->flags.all = 0;
        clone_hdr->egrp = EM_EVENT_GROUP_UNDEF;
        /* other fields: use parent's values as is */
 
        return clone_event;
}
 
void
output_queue_track(queue_elem_t *const output_q_elem)
{
        output_queue_track_t *const track =
                &em_locm.output_queue_track;
        const uint32_t qidx = output_q_elem->output.idx;
 
        if (track->used_queues[qidx] == NULL) {
                track->used_queues[qidx] = output_q_elem;
                track->idx[track->idx_cnt++] = qidx;
        }
}
 
void
output_queue_drain(const queue_elem_t *output_q_elem)
{
        const em_queue_t output_queue = (em_queue_t)(uintptr_t)output_q_elem->queue;
        const em_output_func_t output_fn =
                output_q_elem->output.output_conf.output_fn;
        void *const output_fn_args =
                output_q_elem->output.output_conf.output_fn_args;
 
        const int deq_max = 32;
 
        em_event_t output_ev_tbl[deq_max];
        /* use same event-tbl, dequeue odp events into the EM event-tbl */
        odp_event_t *const odp_deq_events = (odp_event_t *)output_ev_tbl;
 
        const odp_queue_t odp_queue = output_q_elem->odp_queue;
        unsigned int output_num;
        int deq;
        int ret;
 
        const bool esv_ena = esv_enabled();
 
        do {
                deq = odp_queue_deq_multi(odp_queue,
                                          odp_deq_events/*out=output_ev_tbl[]*/,
                                          deq_max);
                if (unlikely(deq <= 0))
                        return;
 
                output_num = (unsigned int)deq;
                /* odp_deq_events[] == output_ev_tbl[] */
                if (esv_ena) {
                        event_hdr_t *ev_hdrs[output_num];
 
                        /* Update EM hdls from hdrs, odp-ev conv lost evgen (non-EM events OK) */
                        event_to_hdr_multi(output_ev_tbl, ev_hdrs, output_num);
                        for (unsigned int i = 0; i < output_num; i++)
                                if (likely(ev_hdrs[i]->event_type != EM_EVENT_TYPE_ODP))
                                        output_ev_tbl[i] = ev_hdrs[i]->event;
                }
 
                ret = output_fn(output_ev_tbl, output_num,
                                output_queue, output_fn_args);
 
                if (unlikely((unsigned int)ret != output_num))
                        em_free_multi(&output_ev_tbl[ret], output_num - ret);
        } while (deq > 0);
}
 
void
output_queue_buffering_drain(void)
{
        output_queue_track_t *const track = &em_locm.output_queue_track;
 
        for (unsigned int i = 0; i < track->idx_cnt; i++) {
                int qidx = track->idx[i];
                queue_elem_t *output_q_elem = track->used_queues[qidx];
                odp_ticketlock_t *lock = &output_q_elem->output.lock;
 
                /*
                 * drain if lock available, otherwise another core is already
                 * draining so no need to do anything.
                 */
                if (odp_ticketlock_trylock(lock)) {
                        output_queue_drain(output_q_elem);
                        odp_ticketlock_unlock(lock);
                }
 
                track->idx[i] = 0;
                track->used_queues[qidx] = NULL;
        }
        track->idx_cnt = 0;
}
 
/** Helper to event_vector_tbl() */
uint32_t event_vector_tbl__odp_evvec(odp_event_vector_t evvec,
                                     em_event_t **event_tbl /*out*/)
{
        odp_event_t *odp_evtbl = NULL;
        uint32_t num = odp_event_vector_tbl(evvec, &odp_evtbl/*out*/);
 
        /* set output arg 'event_tbl' even if there are no events in the vector */
        *event_tbl = (em_event_t *)odp_evtbl; /* Careful! Points to same table (aliasing) */
 
        if (!num)
                return 0;
 
        /*
         * Init the event-table as needed, might contain EM events or
         * ODP events depending on source.
         */
        if (esv_enabled()) {
                /*
                 * Drop ESV generation from event handles by converting to
                 * odp-events, then init as needed as EM events.
                 */
                (void)events_em2odp_inplace(*event_tbl, num);
        }
 
        event_hdr_t *ev_hdr_tbl[num];
 
        /* odp_evtbl[] and event_tbl[] alias each other! */
        event_init_odp_multi(odp_evtbl /*in*/, *event_tbl /*in,out*/,
                             ev_hdr_tbl /*out*/, num, false, NULL/*disp_q_elem*/);
 
        return num;
}
 
/** Helper to event_vector_tbl() */
uint32_t event_vector_tbl__odp_pktvec(odp_packet_vector_t pkt_vec,
                                      em_event_t **event_tbl /*out*/)
{
        odp_packet_t *pkt_tbl = NULL;
        uint32_t num = odp_packet_vector_tbl(pkt_vec, &pkt_tbl/*out*/);
 
        /* set output arg 'event_tbl' even if there are no pkts in the vector */
        *event_tbl = (em_event_t *)pkt_tbl; /* Careful! Points to same table (aliasing)*/
 
        if (!num)
                return 0;
 
        /*
         * Init the event-table as needed, might contain EM events or
         * ODP packets depending on source.
         */
        if (esv_enabled()) {
                /*
                 * Drop ESV generation from event handles by converting to
                 * odp-packets, then init as needed as EM events.
                 */
                (void)events_em2pkt_inplace(*event_tbl, num);
        }
 
        event_hdr_t *ev_hdr_tbl[num];
 
        /* pkt_tbl[] and event_tbl[] may alias each other! */
        event_init_pkt_multi(pkt_tbl /*in*/, *event_tbl /*in,out*/,
                             ev_hdr_tbl /*out*/, num, false);
 
        return num;
}
 
uint32_t event_vector_tbl(em_event_t vector_event,
                          em_event_t **event_tbl /*out*/)
{
        odp_event_t odp_event = event_em2odp(vector_event);
        odp_event_type_t odp_etype = odp_event_type(odp_event);
        uint32_t num;
 
        if (odp_etype == ODP_EVENT_VECTOR) {
                odp_event_vector_t evvec = odp_event_vector_from_event(odp_event);
 
                num = event_vector_tbl__odp_evvec(evvec, event_tbl /*out*/);
        } else {
                /* odp_etype == ODP_EVENT_PACKET_VECTOR */
                odp_packet_vector_t pkt_vec = odp_packet_vector_from_event(odp_event);
 
                num = event_vector_tbl__odp_pktvec(pkt_vec, event_tbl /*out*/);
        }
 
        return num;
}
 
em_status_t event_vector_max_size(em_event_t vector_event, uint32_t *max_size /*out*/,
                                  em_escope_t escope)
{
        odp_event_t odp_event = event_em2odp(vector_event);
        odp_event_type_t odp_etype = odp_event_type(odp_event);
        odp_pool_t odp_pool = ODP_POOL_INVALID;
 
        if (odp_etype == ODP_EVENT_VECTOR) {
                odp_event_vector_t evvec = odp_event_vector_from_event(odp_event);
 
                odp_pool = odp_event_vector_pool(evvec);
        } else {
                /* odp_etype == ODP_EVENT_PACKET_VECTOR */
                odp_packet_vector_t pktvec = odp_packet_vector_from_event(odp_event);
 
                odp_pool = odp_packet_vector_pool(pktvec);
        }
 
        pool_subpool_t pool_subpool = pool_subpool_odp2em(odp_pool);
        em_pool_t pool = (em_pool_t)(uintptr_t)pool_subpool.pool;
        int subpool = pool_subpool.subpool;
 
        if (unlikely(pool == EM_POOL_UNDEF)) {
                /*
                 * Don't report an error if 'pool == EM_POOL_UNDEF' since that
                 * might happen if the vector is input from pktio that is using
                 * external (to EM) odp vector pools.
                 */
                *max_size = 0;
                return EM_OK; /* EM does not have the max_size info */
        }
 
        const mpool_elem_t *pool_elem = pool_elem_get(pool);
 
        if (unlikely(!pool_elem ||
                     (EM_CHECK_LEVEL > 2 && !pool_allocated(pool_elem)))) {
                *max_size = 0;
                return INTERNAL_ERROR(EM_ERR_BAD_STATE, escope,
                                      "Invalid pool:%" PRI_POOL "", pool);
        }
 
        if (unlikely(subpool >= pool_elem->num_subpools)) {
                /* not found */
                *max_size = 0;
                return INTERNAL_ERROR(EM_ERR_NOT_FOUND, escope,
                                      "Subpool not found, pool:%" PRI_POOL "", pool);
        }
 
        /* subpool index found, store corresponding size */
        *max_size = pool_elem->size[subpool];
 
        return EM_OK;
}