em_eo.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 <stdio.h>
#include <string.h>
 
#include <odp_api.h>
 
#include <event_machine.h>
#include <event_machine/platform/env/environment.h>
 
#include "em_atomic_group.h"
#include "em_atomic_group_types.h"
#include "em_eo.h"
#include "em_eo_types.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_internal_event.h"
#include "em_libconfig.h"
#include "em_internal_event_types.h"
#include "em_mem.h"
#include "em_queue.h"
#include "em_queue_inline.h"
#include "em_queue_types.h"
#include "misc/list.h"
#include "misc/objpool.h"
 
/**
 * Params for eo_local_func_call_req().
 * Init params with eo_local_func_call_param_init() before usage.
 */
typedef struct {
        eo_elem_t *eo_elem;
        queue_elem_t *q_elem;
        int delete_queues;
        uint64_t ev_id;
        void (*f_done_callback)(void *arg_ptr);
        int num_notif;
        const em_notif_t *notif_tbl; /* notif_tbl[num_notif] */
        bool sync_operation;
} eo_local_func_call_param_t;
 
static void
eo_local_func_call_param_init(eo_local_func_call_param_t *param);
static int
eo_local_func_call_req(const eo_local_func_call_param_t *param);
 
static em_status_t
check_eo_local_status(const loc_func_retval_t *loc_func_retvals);
 
static void
eo_start_done_callback(void *args);
static void
eo_start_sync_done_callback(void *args);
 
static void
eo_stop_done_callback(void *args);
static void
eo_stop_sync_done_callback(void *args);
 
static em_status_t
eo_remove_queue_local(const eo_elem_t *eo_elem, const queue_elem_t *q_elem);
static void
eo_remove_queue_done_callback(void *args);
 
static em_status_t
eo_remove_queue_sync_local(const eo_elem_t *eo_elem,
                           const queue_elem_t *q_elem);
static void
eo_remove_queue_sync_done_callback(void *args);
 
static em_status_t
eo_remove_queue_all_local(const eo_elem_t *eo_elem, int delete_queues);
static void
eo_remove_queue_all_done_callback(void *args);
 
static em_status_t
eo_remove_queue_all_sync_local(const eo_elem_t *eo_elem, int delete_queues);
static void
eo_remove_queue_all_sync_done_callback(void *args);
 
static int read_config_file(void)
{
        const char *conf_str;
        bool val_bool = false;
        int ret;
 
        EM_PRINT("EM EO config:\n");
 
        /*
         * Option: eo.start_local_fn_at_init
         */
        conf_str = "eo.start_local_fn_at_init";
        ret = em_libconfig_lookup_bool(&em_shm->libconfig, conf_str, &val_bool);
        if (unlikely(!ret)) {
                EM_LOG(EM_LOG_ERR, "Config option '%s' not found\n", conf_str);
                return -1;
        }
        /* store & print the value */
        em_shm->opt.eo.start_local_fn_at_init = val_bool;
        EM_PRINT("  %s: %s(%d)\n", conf_str, val_bool ? "true" : "false",
                 val_bool);
 
        /*
         * Option: eo.stop_local_fn_at_term
         */
        conf_str = "eo.stop_local_fn_at_term";
        ret = em_libconfig_lookup_bool(&em_shm->libconfig, conf_str, &val_bool);
        if (unlikely(!ret)) {
                EM_LOG(EM_LOG_ERR, "Config option '%s' not found\n", conf_str);
                return -1;
        }
        /* store & print the value */
        em_shm->opt.eo.stop_local_fn_at_term = val_bool;
        EM_PRINT("  %s: %s(%d)\n", conf_str, val_bool ? "true" : "false",
                 val_bool);
 
        return 0;
}
 
static inline eo_elem_t *
eo_poolelem2eo(const objpool_elem_t *const eo_pool_elem)
{
        return (eo_elem_t *)((uintptr_t)eo_pool_elem -
                             offsetof(eo_elem_t, eo_pool_elem));
}
 
em_status_t
eo_init(eo_tbl_t eo_tbl[], eo_pool_t *eo_pool)
{
        int ret;
        const uint32_t objpool_subpools = MIN(4, OBJSUBPOOLS_MAX);
 
        memset(eo_tbl, 0, sizeof(eo_tbl_t));
        memset(eo_pool, 0, sizeof(eo_pool_t));
 
        if (read_config_file())
                return EM_ERR_LIB_FAILED;
 
        for (int i = 0; i < EM_MAX_EOS; i++) {
                eo_elem_t *const eo_elem = &eo_tbl->eo_elem[i];
                /* Store EO handle */
                eo_elem->eo = eo_idx2hdl(i);
                /* Initialize empty EO-queue list */
                odp_ticketlock_init(&eo_elem->lock);
                list_init(&eo_elem->queue_list);
                eo_elem->stash = ODP_STASH_INVALID;
        }
 
        ret = objpool_init(&eo_pool->objpool, objpool_subpools);
        if (ret != 0)
                return EM_ERR_LIB_FAILED;
 
        for (uint32_t i = 0; i < EM_MAX_EOS; i++)
                objpool_add(&eo_pool->objpool, i % objpool_subpools,
                            &eo_tbl->eo_elem[i].eo_pool_elem);
 
        odp_atomic_init_u32(&em_shm->eo_count, 0);
 
        return EM_OK;
}
 
em_eo_t
eo_alloc(void)
{
        const eo_elem_t *eo_elem;
        const objpool_elem_t *eo_pool_elem;
 
        eo_pool_elem = objpool_rem(&em_shm->eo_pool.objpool, em_core_id());
        if (unlikely(eo_pool_elem == NULL))
                return EM_EO_UNDEF;
 
        eo_elem = eo_poolelem2eo(eo_pool_elem);
        odp_atomic_inc_u32(&em_shm->eo_count);
 
        return eo_elem->eo;
}
 
em_status_t
eo_free(em_eo_t eo)
{
        eo_elem_t *eo_elem = eo_elem_get(eo);
 
        if (unlikely(eo_elem == NULL))
                return EM_ERR_BAD_ID;
 
        eo_elem->state = EM_EO_STATE_UNDEF;
 
        objpool_add(&em_shm->eo_pool.objpool,
                    eo_elem->eo_pool_elem.subpool_idx, &eo_elem->eo_pool_elem);
        odp_atomic_dec_u32(&em_shm->eo_count);
 
        return EM_OK;
}
 
/**
 * Add a queue to an EO
 */
em_status_t
eo_add_queue(eo_elem_t *const eo_elem, queue_elem_t *const q_elem)
{
        queue_state_t old_state = q_elem->state;
        queue_state_t new_state = EM_QUEUE_STATE_BIND;
        em_status_t err;
 
        err = queue_state_change__check(old_state, new_state, 1/*is_setup*/);
        if (unlikely(err != EM_OK))
                return err;
 
        q_elem->max_events = (uint16_t)eo_elem->max_events;
 
        q_elem->flags.use_multi_rcv = eo_elem->use_multi_rcv ? true : false;
        if (eo_elem->use_multi_rcv)
                q_elem->receive_multi_func = eo_elem->receive_multi_func;
        else
                q_elem->receive_func = eo_elem->receive_func;
 
        q_elem->eo = (uint16_t)(uintptr_t)eo_elem->eo;
        q_elem->eo_ctx = eo_elem->eo_ctx;
        q_elem->eo_elem = eo_elem;
        q_elem->state = new_state;
 
        /* Link the new queue into the EO's queue-list */
        odp_ticketlock_lock(&eo_elem->lock);
        list_add(&eo_elem->queue_list, &q_elem->eo_queue_node);
        odp_atomic_inc_u32(&eo_elem->num_queues);
        odp_ticketlock_unlock(&eo_elem->lock);
 
        return EM_OK;
}
 
static inline em_status_t
eo_rem_queue_locked(eo_elem_t *const eo_elem, queue_elem_t *const q_elem)
{
        queue_state_t old_state = q_elem->state;
        queue_state_t new_state = EM_QUEUE_STATE_INIT;
        em_status_t err;
 
        err = queue_state_change__check(old_state, new_state, 0/*!is_setup*/);
        if (unlikely(err != EM_OK))
                return err;
 
        list_rem(&eo_elem->queue_list, &q_elem->eo_queue_node);
        odp_atomic_dec_u32(&eo_elem->num_queues);
 
        q_elem->state = new_state;
        q_elem->eo = (uint16_t)(uintptr_t)EM_EO_UNDEF;
        q_elem->eo_elem = NULL;
 
        return EM_OK;
}
 
/**
 * Remove a queue from an EO
 */
em_status_t
eo_rem_queue(eo_elem_t *const eo_elem, queue_elem_t *const q_elem)
{
        em_status_t err;
 
        odp_ticketlock_lock(&eo_elem->lock);
        err = eo_rem_queue_locked(eo_elem, q_elem);
        odp_ticketlock_unlock(&eo_elem->lock);
 
        if (unlikely(err != EM_OK))
                return err;
 
        return EM_OK;
}
 
/*
 * Remove all queues associated with the EO.
 * Note: does not delete the queues.
 */
em_status_t
eo_rem_queue_all(eo_elem_t *const eo_elem)
{
        em_status_t err = EM_OK;
        queue_elem_t *q_elem;
        const list_node_t *list_node;
 
        odp_ticketlock_lock(&eo_elem->lock);
 
        /* Loop through all queues associated with the EO */
        list_for_each(&eo_elem->queue_list, list_node) {
                q_elem = eo_queue_node_to_queue_elem(list_node);
                /* remove the queue from the EO */
                err = eo_rem_queue_locked(eo_elem, q_elem);
                if (unlikely(err != EM_OK))
                        break;
        } /* end loop */
 
        odp_ticketlock_unlock(&eo_elem->lock);
 
        return err;
}
 
/*
 * Delete all queues associated with the EO.
 * The queue needs to be removed from the EO before the actual delete.
 */
em_status_t
eo_delete_queue_all(eo_elem_t *const eo_elem)
{
        em_status_t err = EM_OK;
        queue_elem_t *q_elem;
        const list_node_t *list_node;
        const char *fail_op = "eo_rem_queue_locked()";
        const char *err_str = "remove queue from EO failed";
 
        odp_ticketlock_lock(&eo_elem->lock);
 
        /* Loop through all queues associated with the EO */
        list_for_each(&eo_elem->queue_list, list_node) {
                q_elem = eo_queue_node_to_queue_elem(list_node);
                /* remove the queue from the EO */
                err = eo_rem_queue_locked(eo_elem, q_elem);
                if (unlikely(err != EM_OK)) {
                        /* uses default fail_op and err_str */
                        break;
                }
                /* delete the queue */
                err = queue_delete(q_elem, &err_str);
                if (unlikely(err != EM_OK)) {
                        fail_op = "queue_delete()";
                        break;
                }
        } /* end loop */
 
        odp_ticketlock_unlock(&eo_elem->lock);
 
        if (unlikely(err != EM_OK))
                EM_LOG(EM_LOG_ERR,
                       "EO:%" PRI_EO " %s(): %s failed: %s (0x%08X(%s))\n",
                       eo_elem->eo, __func__, fail_op, err_str, err, err_status_str(err));
 
        return err;
}
 
/**
 * Request all EM-cores (worker or control) to run an EO's local start function
 * as triggered from em_eo_start()
 *
 * @param eo_elem  EO element pointer
 * @param num_notif Number of notifications in notif_tbl
 * @param notif_tbl Notification table, can be NULL if num_notif is 0
 *
 * @return the number of requests sent or a negative value on error
 * @retval  0 if no requests were sent
 * @retval >0 the number of requests sent
 * @retval <0 on error
 */
int eo_start_local_req(eo_elem_t *const eo_elem,
                       int num_notif, const em_notif_t notif_tbl[])
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = NULL; /* no q_elem */
        param.delete_queues = EM_FALSE;
        param.ev_id = EO_START_LOCAL_REQ;
        param.f_done_callback = eo_start_done_callback;
        param.num_notif = num_notif;
        param.notif_tbl = notif_tbl;
        param.sync_operation = false;
 
        return eo_local_func_call_req(&param);
}
 
/**
 * Callback function run when all start_local functions are finished,
 * triggered by calling em_eo_start() when using local-start functions
 */
static void
eo_start_done_callback(void *args)
{
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
        em_status_t ret;
 
        if (unlikely(eo_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_START_DONE_CB,
                               "eo_elem is NULL!");
                return;
        }
 
        if (check_eo_local_status(loc_func_retvals) == EM_OK) {
                ret = queue_enable_all(eo_elem); /* local starts OK */
                if (ret == EM_OK)
                        eo_elem->state = EM_EO_STATE_RUNNING;
        }
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
 
        /* Send events buffered during the EO-start/local-start functions */
        eo_start_send_buffered_events(eo_elem);
}
 
/**
 * Request all EM-cores (worker or control) to run an EO's local start function
 * as triggered from em_eo_start_sync()
 *
 * @param eo_elem  EO element pointer
 *
 * @return the number of requests sent or a negative value on error
 * @retval  0 if no requests were sent
 * @retval >0 the number of requests sent
 * @retval <0 on error
 */
int eo_start_sync_local_req(eo_elem_t *const eo_elem)
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = NULL; /* no q_elem */
        param.delete_queues = EM_FALSE;
        param.ev_id = EO_START_SYNC_LOCAL_REQ;
        param.f_done_callback = eo_start_sync_done_callback;
        param.num_notif = 0;
        param.notif_tbl = NULL;
        param.sync_operation = true;
 
        return eo_local_func_call_req(&param);
}
 
/**
 * Callback function run when all start_local functions are finished,
 * triggered by calling em_eo_start_sync() when using local-start functions
 */
static void
eo_start_sync_done_callback(void *args)
{
        em_locm_t *const locm = &em_locm;
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
        em_status_t ret;
 
        if (unlikely(eo_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_START_SYNC_DONE_CB,
                               "eo_elem is NULL!");
                return;
        }
 
        if (check_eo_local_status(loc_func_retvals) == EM_OK) {
                ret = queue_enable_all(eo_elem); /* local starts OK */
                if (ret == EM_OK)
                        eo_elem->state = EM_EO_STATE_RUNNING;
        }
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
 
        /* Enable the caller of the sync API func to proceed (on this core) */
        locm->sync_api.in_progress = false;
 
        /*
         * Events buffered during the EO-start/local-start functions are sent
         * from em_eo_start_sync() after this.
         */
}
 
/**
 * Called by em_send() & variants during an EO start-function.
 *
 * Events sent from within the EO-start functions are buffered and sent
 * after the start-operation has completed. Otherwise it would not be
 * possible to reliably send events from the start-functions to the
 * EO's own queues.
 */
int eo_start_buffer_events(const em_event_t events[], int num, em_queue_t queue)
{
        eo_elem_t *const eo_elem = em_locm.start_eo_elem;
        const uint16_t qidx = queue_hdl2idx(queue);
        const evhdl_t *const evhdl_tbl = (const evhdl_t *const)events;
        stash_entry_t entry_tbl[num];
 
        if (unlikely(eo_elem == NULL))
                return 0;
 
        odp_ticketlock_lock(&eo_elem->lock);
 
        for (int i = 0; i < num; i++) {
                /* ESV evgen dropped */
                entry_tbl[i] = (stash_entry_t){.qidx = qidx,
                                               .evptr = evhdl_tbl[i].evptr};
        }
 
        /* Enqueue events to internal queue */
        int ret = odp_stash_put_u64(eo_elem->stash, &entry_tbl[0].u64, num);
 
        if (unlikely(ret < 0))
                ret = 0;
 
        odp_ticketlock_unlock(&eo_elem->lock);
 
        return ret;
}
 
/**
 * @brief Helper to eo_start_send_buffered_events()
 */
static void eo_start_send_multi(em_event_t ev_tbl[], int num,
                                em_queue_t queue, em_event_group_t event_group)
{
        int num_sent = 0;
 
        /* send events with same destination queue and event group */
        if (event_group == EM_EVENT_GROUP_UNDEF)
                num_sent = em_send_multi(ev_tbl, num, queue);
        else
                num_sent = em_send_group_multi(ev_tbl, num, queue, event_group);
 
        if (unlikely(num_sent != num)) {
                /* User's eo-start saw successful em_send, free here */
                em_free_multi(&ev_tbl[num_sent], num - num_sent);
                INTERNAL_ERROR(EM_ERR_LIB_FAILED, EM_ESCOPE_EO_START,
                               "Q:%" PRI_QUEUE " req:%u sent:%u",
                               queue, num, num_sent);
        }
}
 
/**
 * Send the buffered events at the end of the EO-start operation.
 *
 * Events sent from within the EO-start functions are buffered and sent
 * after the start-operation has completed. Otherwise it would not be
 * possible to reliably send events from the start-functions to the
 * EO's own queues.
 */
void eo_start_send_buffered_events(eo_elem_t *const eo_elem)
{
        /* max events to send in a burst */
        const unsigned int max_ev = 32;
        stash_entry_t entry_tbl[max_ev];
        em_event_t ev_tbl[max_ev];
        event_hdr_t *ev_hdr_tbl[max_ev];
 
        odp_ticketlock_lock(&eo_elem->lock);
 
        /*
         * Send the buffered events in bursts into the destination queue.
         *
         * This is startup: we can use some extra cycles to create the
         * event-arrays to send in bursts.
         */
        int err = 0;
        int num = 0;
 
        do {
                num = odp_stash_get_u64(eo_elem->stash, &entry_tbl[0].u64 /*[out]*/, max_ev);
                if (num <= 0) {
                        if (unlikely(num < 0))
                                INTERNAL_ERROR(EM_ERR_LIB_FAILED, EM_ESCOPE_EO_START,
                                               "odp_stash_get_u64() fails: %d", num);
                        goto buffered_send_exit;
                }
 
                for (int i = 0; i < num; i++)
                        ev_tbl[i] = (em_event_t)(uintptr_t)entry_tbl[i].evptr;
 
                event_to_hdr_multi(ev_tbl, ev_hdr_tbl, num);
 
                if (esv_enabled())
                        evstate_em2usr_multi(ev_tbl/*in/out*/, ev_hdr_tbl, num,
                                             EVSTATE__EO_START_SEND_BUFFERED);
 
                int tbl_idx = 0; /* index into ..._tbl[] */
 
                /*
                 * Send in batches of 'batch_cnt' events.
                 * Each batch contains events from the same queue & evgrp.
                 */
                do {
                        const int qidx = entry_tbl[tbl_idx].qidx;
                        const em_queue_t queue = queue_idx2hdl(qidx);
                        const em_event_group_t event_group = ev_hdr_tbl[tbl_idx]->egrp;
                        int batch_cnt = 1;
 
                        for (int i = tbl_idx + 1; i < num &&
                             entry_tbl[i].qidx == qidx &&
                             ev_hdr_tbl[i]->egrp == event_group; i++) {
                                batch_cnt++;
                        }
 
                        /* send events with same destination queue and event group */
                        eo_start_send_multi(&ev_tbl[tbl_idx], batch_cnt, queue, event_group);
 
                        tbl_idx += batch_cnt;
                } while (tbl_idx < num);
        } while (num > 0);
 
buffered_send_exit:
        err = odp_stash_destroy(eo_elem->stash);
 
        eo_elem->stash = ODP_STASH_INVALID;
        odp_ticketlock_unlock(&eo_elem->lock);
 
        if (unlikely(err)) {
                INTERNAL_ERROR(EM_ERR_LIB_FAILED, EM_ESCOPE_EO_START,
                               "odp_stash_destroy() fails: %d", err);
        }
}
 
/**
 * Request all EM-cores (worker or control) to run an EO's local stop function
 * as triggered from em_eo_stop()
 *
 * @param eo_elem  EO element pointer
 * @param num_notif  Number of notifications to send
 * @param notif_tbl  Array of notifications to send
 *
 * @return the number of requests sent or a negative value on error
 * @retval  0 if no requests were sent
 * @retval >0 the number of requests sent
 * @retval <0 on error
 */
int eo_stop_local_req(eo_elem_t *const eo_elem,
                      int num_notif, const em_notif_t notif_tbl[])
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = NULL; /* no q_elem */
        param.delete_queues = EM_FALSE;
        param.ev_id = EO_STOP_LOCAL_REQ;
        param.f_done_callback = eo_stop_done_callback;
        param.num_notif = num_notif;
        param.notif_tbl = notif_tbl;
        param.sync_operation = false;
 
        return eo_local_func_call_req(&param);
}
 
/**
 * Helper to eo_stop_done_callback() and em_eo_stop()
 *
 * @return EO global stop function return value
 */
em_status_t eo_stop_done(eo_elem_t *const eo_elem)
{
        em_locm_t *const locm = &em_locm;
        queue_elem_t *const save_q_elem = locm->current.q_elem;
        queue_elem_t tmp_q_elem;
        em_status_t ret;
        em_eo_t eo = eo_elem->eo;
        void *const eo_ctx = eo_elem->eo_ctx;
 
        /* Change state here to allow em_eo_delete() from EO global stop */
        eo_elem->state = EM_EO_STATE_CREATED; /* == EO_STATE_STOPPED */
 
        /*
         * Use a tmp q_elem as the 'current q_elem' to enable calling
         * em_eo_current() from the EO stop functions.
         * Before returning, restore the original 'current q_elem' from
         * 'save_q_elem'.
         */
        memset(&tmp_q_elem, 0, sizeof(tmp_q_elem));
        tmp_q_elem.eo = (uint16_t)(uintptr_t)eo;
 
        locm->current.q_elem = &tmp_q_elem;
        /*
         * Call the Global EO stop function now that all
         * EO local stop functions are done.
         */
        ret = eo_elem->stop_func(eo_ctx, eo);
        /* Restore the original 'current q_elem' */
        locm->current.q_elem = save_q_elem;
 
        return ret;
}
 
/**
 * Callback function run when all stop_local functions are finished,
 * triggered by calling eo_eo_stop().
 */
static void
eo_stop_done_callback(void *args)
{
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
 
        if (unlikely(eo_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_STOP_DONE_CB,
                               "eo_elem is NULL!");
                return;
        }
 
        em_eo_t eo = eo_elem->eo;
 
        (void)check_eo_local_status(loc_func_retvals);
 
        em_status_t ret = eo_stop_done(eo_elem);
 
        /*
         * Note: the EO might not be available after this if the EO global stop
         * called em_eo_delete()!
         */
 
        if (unlikely(ret != EM_OK))
                INTERNAL_ERROR(EM_ERR, EM_ESCOPE_EO_STOP_DONE_CB,
                               "EO:%" PRI_EO " stop-func failed:%" PRIxSTAT "", eo, ret);
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
}
 
/**
 * Request all EM-cores (worker or control) to run an EO's local stop function
 * as triggered from em_eo_stop_sync()
 *
 * @param eo_elem  EO element pointer
 *
 * @return the number of requests sent or a negative value on error
 * @retval  0 if no requests were sent
 * @retval >0 the number of requests sent
 * @retval <0 on error
 */
int eo_stop_sync_local_req(eo_elem_t *const eo_elem)
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = NULL; /* no q_elem */
        param.delete_queues = EM_FALSE;
        param.ev_id = EO_STOP_SYNC_LOCAL_REQ;
        param.f_done_callback = eo_stop_sync_done_callback;
        param.num_notif = 0;
        param.notif_tbl = NULL;
        param.sync_operation = true;
 
        return eo_local_func_call_req(&param);
}
 
/**
 * Callback function run when all stop_local functions are finished,
 * triggered by calling eo_eo_stop_sync().
 */
static void
eo_stop_sync_done_callback(void *args)
{
        em_locm_t *const locm = &em_locm;
        const loc_func_retval_t *loc_func_retvals = args;
        const eo_elem_t *eo_elem = loc_func_retvals->eo_elem;
 
        if (unlikely(eo_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_STOP_SYNC_DONE_CB,
                               "eo_elem is NULL!");
                /* Enable the caller of the sync API func to proceed */
                locm->sync_api.in_progress = false;
                return;
        }
 
        (void)check_eo_local_status(loc_func_retvals);
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
 
        /* Enable the caller of the sync API func to proceed (on this core) */
        locm->sync_api.in_progress = false;
}
 
int eo_remove_queue_local_req(eo_elem_t *const eo_elem, queue_elem_t *const q_elem,
                              int num_notif, const em_notif_t notif_tbl[])
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = q_elem;
        param.delete_queues = EM_FALSE;
        param.ev_id = EO_REM_QUEUE_LOCAL_REQ;
        param.f_done_callback = eo_remove_queue_done_callback;
        param.num_notif = num_notif;
        param.notif_tbl = notif_tbl;
        param.sync_operation = false;
 
        return eo_local_func_call_req(&param);
}
 
static em_status_t
eo_remove_queue_local(const eo_elem_t *eo_elem, const queue_elem_t *q_elem)
{
        (void)eo_elem;
        (void)q_elem;
 
        return EM_OK;
}
 
static void
eo_remove_queue_done_callback(void *args)
{
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
        queue_elem_t *const q_elem = loc_func_retvals->q_elem;
        em_status_t ret;
 
        if (unlikely(eo_elem == NULL || q_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_REMOVE_QUEUE_DONE_CB,
                               "eo_elem/q_elem is NULL!");
                return;
        }
 
        (void)check_eo_local_status(loc_func_retvals);
 
        /* Remove the queue from the EO */
        ret = eo_rem_queue(eo_elem, q_elem);
 
        if (unlikely(ret != EM_OK))
                INTERNAL_ERROR(ret, EM_ESCOPE_EO_REMOVE_QUEUE_DONE_CB,
                               "EO:%" PRI_EO " remove Q:%" PRI_QUEUE " failed",
                               eo_elem->eo, q_elem->queue);
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
}
 
int eo_remove_queue_sync_local_req(eo_elem_t *const eo_elem,
                                   queue_elem_t *const q_elem)
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = q_elem;
        param.delete_queues = EM_FALSE;
        param.ev_id = EO_REM_QUEUE_SYNC_LOCAL_REQ;
        param.f_done_callback = eo_remove_queue_sync_done_callback;
        param.num_notif = 0;
        param.notif_tbl = NULL;
        param.sync_operation = true;
 
        return eo_local_func_call_req(&param);
}
 
static em_status_t
eo_remove_queue_sync_local(const eo_elem_t *eo_elem, const queue_elem_t *q_elem)
{
        (void)eo_elem;
        (void)q_elem;
 
        return EM_OK;
}
 
static void
eo_remove_queue_sync_done_callback(void *args)
{
        em_locm_t *const locm = &em_locm;
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
        queue_elem_t *const q_elem = loc_func_retvals->q_elem;
        em_status_t ret;
 
        if (unlikely(eo_elem == NULL || q_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_REMOVE_QUEUE_SYNC_DONE_CB,
                               "eo_elem/q_elem is NULL!");
                /* Enable the caller of the sync API func to proceed */
                locm->sync_api.in_progress = false;
                return;
        }
 
        (void)check_eo_local_status(loc_func_retvals);
 
        /* Remove the queue from the EO */
        ret = eo_rem_queue(eo_elem, q_elem);
 
        if (unlikely(ret != EM_OK))
                INTERNAL_ERROR(ret,
                               EM_ESCOPE_EO_REMOVE_QUEUE_SYNC_DONE_CB,
                               "EO:%" PRI_EO " remove Q:%" PRI_QUEUE " failed",
                               eo_elem->eo, q_elem->queue);
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
 
        /* Enable the caller of the sync API func to proceed (on this core) */
        locm->sync_api.in_progress = false;
}
 
int eo_remove_queue_all_local_req(eo_elem_t *const eo_elem, int delete_queues,
                                  int num_notif, const em_notif_t notif_tbl[])
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = NULL; /* no q_elem */
        param.delete_queues = delete_queues;
        param.ev_id = EO_REM_QUEUE_ALL_LOCAL_REQ;
        param.f_done_callback = eo_remove_queue_all_done_callback;
        param.num_notif = num_notif;
        param.notif_tbl = notif_tbl;
        param.sync_operation = false;
 
        return eo_local_func_call_req(&param);
}
 
static em_status_t
eo_remove_queue_all_local(const eo_elem_t *eo_elem, int delete_queues)
{
        (void)eo_elem;
        (void)delete_queues;
 
        return EM_OK;
}
 
static void
eo_remove_queue_all_done_callback(void *args)
{
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
        int delete_queues = loc_func_retvals->delete_queues;
        em_status_t ret;
 
        if (unlikely(eo_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_REMOVE_QUEUE_ALL_DONE_CB,
                               "eo_elem is NULL!");
                return;
        }
 
        (void)check_eo_local_status(loc_func_retvals);
 
        /* Remove or delete all the EO's queues */
        if (delete_queues)
                ret = eo_delete_queue_all(eo_elem);
        else
                ret = eo_rem_queue_all(eo_elem);
 
        if (unlikely(ret != EM_OK))
                INTERNAL_ERROR(ret, EM_ESCOPE_EO_REMOVE_QUEUE_ALL_DONE_CB,
                               "EO:%" PRI_EO " removing all queues failed",
                               eo_elem->eo);
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
}
 
int eo_remove_queue_all_sync_local_req(eo_elem_t *const eo_elem, int delete_queues)
{
        eo_local_func_call_param_t param;
 
        eo_local_func_call_param_init(&param);
        param.eo_elem = eo_elem;
        param.q_elem = NULL; /* no q_elem */
        param.delete_queues = delete_queues;
        param.ev_id = EO_REM_QUEUE_ALL_SYNC_LOCAL_REQ;
        param.f_done_callback = eo_remove_queue_all_sync_done_callback;
        param.num_notif = 0;
        param.notif_tbl = NULL;
        param.sync_operation = true;
 
        return eo_local_func_call_req(&param);
}
 
static em_status_t
eo_remove_queue_all_sync_local(const eo_elem_t *eo_elem, int delete_queues)
{
        (void)eo_elem;
        (void)delete_queues;
 
        return EM_OK;
}
 
static void
eo_remove_queue_all_sync_done_callback(void *args)
{
        em_locm_t *const locm = &em_locm;
        const loc_func_retval_t *loc_func_retvals = args;
        eo_elem_t *const eo_elem = loc_func_retvals->eo_elem;
        int delete_queues = loc_func_retvals->delete_queues;
        em_status_t ret;
 
        if (unlikely(eo_elem == NULL)) {
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_POINTER),
                               EM_ESCOPE_EO_REMOVE_QUEUE_ALL_SYNC_DONE_CB,
                               "eo_elem is NULL!");
                /* Enable the caller of the sync API func to proceed */
                locm->sync_api.in_progress = false;
                return;
        }
 
        (void)check_eo_local_status(loc_func_retvals);
 
        /* Remove or delete all the EO's queues */
        if (delete_queues)
                ret = eo_delete_queue_all(eo_elem);
        else
                ret = eo_rem_queue_all(eo_elem);
 
        if (unlikely(ret != EM_OK))
                INTERNAL_ERROR(ret,
                               EM_ESCOPE_EO_REMOVE_QUEUE_ALL_SYNC_DONE_CB,
                               "EO:%" PRI_EO " removing all queues failed",
                               eo_elem->eo);
 
        /* free the storage for local func return values */
        em_free(loc_func_retvals->event);
 
        /* Enable the caller of the sync API func to proceed (on this core) */
        locm->sync_api.in_progress = false;
}
 
static em_status_t
check_eo_local_status(const loc_func_retval_t *loc_func_retvals)
{
        const em_core_mask_t *core_mask =  &loc_func_retvals->core_mask;
        bool local_fail = false;
        em_status_t err;
 
        int idx = em_core_mask_first(core_mask);
 
        while (idx >= 0) {
                err = loc_func_retvals->core[idx];
                if (err != EM_OK) {
                        local_fail = true;
                        break;
                }
                idx = em_core_mask_next(core_mask, idx);
        }
 
        /* All OK - return */
        if (!local_fail)
                return EM_OK;
 
        /*
         * Errors found - report them
         */
        static const char core_err[] = "coreXX:0x12345678 ";
        const int core_count = em_core_mask_count(core_mask);
        const size_t err_sz = core_count * sizeof(core_err);
        char errmsg[err_sz];
        int c = 0;
 
        idx = em_core_mask_first(core_mask);
 
        while (idx >= 0) {
                err = loc_func_retvals->core[idx];
                if (err != EM_OK) {
                        int n = snprintf(&errmsg[c], sizeof(core_err),
                                         "core%02d:0x%08X ", idx, err);
                        if ((unsigned int)n >= sizeof(core_err))
                                break;
                        c += n;
                }
                idx = em_core_mask_next(core_mask, idx);
        }
        errmsg[err_sz - 1] = '\0';
 
        INTERNAL_ERROR(EM_ERR, EM_ESCOPE_EVENT_INTERNAL_LFUNC_CALL,
                       "\nLocal start function failed on cores:\n"
                       "%s", errmsg);
        return EM_ERR;
}
 
static void
eo_local_func_call_param_init(eo_local_func_call_param_t *param)
{
        memset(param, 0, sizeof(*param));
}
 
/**
 * Request a function to be run on each EM-core (worker or control) and call
 * 'f_done_callback(arg_ptr)' when all those functions have completed.
 *
 * @param param  EO local function call parameters
 *
 * @return the number of requests sent or a negative value on error
 * @retval  0 if no requests were sent
 * @retval >0 the number of requests sent
 * @retval <0 on error
 */
static int
eo_local_func_call_req(const eo_local_func_call_param_t *param)
{
        core_map_t *const core_map = &em_shm->core_map;
 
        em_event_t event = em_alloc(sizeof(internal_event_t),
                                    EM_EVENT_TYPE_SW, EM_POOL_DEFAULT);
        if (unlikely(event == EM_EVENT_UNDEF)) {
                INTERNAL_ERROR(EM_ERR_ALLOC_FAILED, EM_ESCOPE_EO_LOCAL_FUNC_CALL_REQ,
                               "Internal event (%u) allocation failed", param->ev_id);
                return -1;
        }
 
        em_event_t tmp = em_alloc(sizeof(loc_func_retval_t),
                                  EM_EVENT_TYPE_SW, EM_POOL_DEFAULT);
        if (unlikely(tmp == EM_EVENT_UNDEF)) {
                em_free(event);
                INTERNAL_ERROR(EM_ERR_ALLOC_FAILED, EM_ESCOPE_EO_LOCAL_FUNC_CALL_REQ,
                               "Internal loc_func_retval_t allocation failed");
                return -1;
        }
 
        /*
         * Take the core_map rwlock for reading - can afford to keep it for a
         * "long" time since blocking write-access is only needed in core
         * add/rem scenarios (via em_init/term_local() and the older
         * em_init/term_core()). The core ctrl events should be sent before
         * releasing the lock to prevent add/rem-core actions while sending
         * these ctrl events.
         */
        odp_rwlock_read_lock(&core_map->rwlock);
 
        uint32_t core_count = odp_atomic_load_u32(&core_map->current_core_count);
 
        if (core_count == 0) {
                /* No EM-cores available, free resources and return */
                odp_rwlock_read_unlock(&core_map->rwlock);
                em_free(event);
                em_free(tmp);
                return 0;
        }
 
        uint32_t num_reqs = core_count;
        uint32_t free_count = num_reqs + 1; /* all cores + 'done' event */
        em_core_mask_t core_mask;
        odp_thrmask_t thr_mask;
 
        odp_thrmask_copy(&thr_mask, &core_map->rwlocked.thrmask_emcores);
        em_core_mask_copy(&core_mask, &core_map->rwlocked.logic_mask);
 
        if (unlikely(core_count != (uint32_t)em_core_mask_count(&core_mask) ||
                     core_count != (uint32_t)odp_thrmask_count(&thr_mask))) {
                odp_rwlock_read_unlock(&core_map->rwlock);
                em_free(event);
                em_free(tmp);
                INTERNAL_ERROR(EM_FATAL(EM_ERR_BAD_STATE), EM_ESCOPE_EO_LOCAL_FUNC_CALL_REQ,
                               "Core count mismatch: core_count=%" PRIu32 "\n"
                               "                     count:core_mask=%" PRIu32 "\n"
                               "                     count:thr_mask=%" PRIu32 "",
                               core_count, em_core_mask_count(&core_mask),
                               odp_thrmask_count(&thr_mask));
                return -1;
        }
 
        internal_event_t *i_event = em_event_pointer(event);
 
        i_event->id = param->ev_id;
        i_event->loc_func.eo_elem = param->eo_elem;
        i_event->loc_func.q_elem = param->q_elem;
        i_event->loc_func.delete_queues = param->delete_queues;
 
        loc_func_retval_t *loc_func_retvals = em_event_pointer(tmp);
 
        loc_func_retvals->eo_elem = param->eo_elem;
        loc_func_retvals->q_elem = param->q_elem;
        loc_func_retvals->delete_queues = param->delete_queues;
        loc_func_retvals->event = tmp; /* store event handle for em_free() */
        odp_atomic_init_u32(&loc_func_retvals->free_at_zero, free_count);
        em_core_mask_copy(&loc_func_retvals->core_mask, &core_mask);
        memset(&loc_func_retvals->core[0], EM_OK, sizeof(loc_func_retvals->core));
 
        /* ptr to retval storage so loc func calls can record retval there */
        i_event->loc_func.retvals = loc_func_retvals;
 
        /* Give ptr to retval storage also to 'done' function */
        void *f_done_arg_ptr = loc_func_retvals;
 
        int err = send_core_ctrl_events(&thr_mask, event,
                                        param->f_done_callback, f_done_arg_ptr,
                                        param->num_notif, param->notif_tbl,
                                        param->sync_operation);
 
        odp_rwlock_read_unlock(&core_map->rwlock);
 
        if (unlikely(err)) {
                char mask_str[ODP_THRMASK_STR_SIZE];
                uint32_t unsent_cnt = err;
                uint32_t cnt;
 
                em_free(event);
                cnt = odp_atomic_fetch_sub_u32(&loc_func_retvals->free_at_zero,
                                               unsent_cnt + 1);
                if (cnt == unsent_cnt + 1)
                        em_free(tmp);
 
                odp_thrmask_to_str(&thr_mask, mask_str, ODP_THRMASK_STR_SIZE);
                INTERNAL_ERROR(EM_ERR_LIB_FAILED, EM_ESCOPE_EO_LOCAL_FUNC_CALL_REQ,
                               "send_core_ctrl_events(mask=%s) failed", mask_str);
                return -1;
        }
 
        return num_reqs;
}
 
/**
 * EM internal event handler (see em_internal_event.c&h)
 * Handle the internal event requesting a local function call.
 */
void i_event__eo_local_func_call_req(const internal_event_t *i_ev)
{
        em_locm_t *const locm = &em_locm;
        const uint64_t f_type = i_ev->loc_func.id;
        eo_elem_t *eo_elem = i_ev->loc_func.eo_elem;
        const queue_elem_t *q_elem = i_ev->loc_func.q_elem;
        int delete_queues = i_ev->loc_func.delete_queues;
        loc_func_retval_t *const loc_func_retvals = i_ev->loc_func.retvals;
        em_status_t status = EM_ERR;
        queue_elem_t *const save_q_elem = locm->current.q_elem;
        queue_elem_t tmp_q_elem;
 
        switch (f_type) {
        case EO_START_SYNC_LOCAL_REQ:
        case EO_START_LOCAL_REQ:
                if (!eo_elem->start_local_func) {
                        /* No local start func given */
                        status = EM_ERR_BAD_POINTER;
                        break;
                }
                /*
                 * Use a tmp q_elem as the 'current q_elem' to enable calling
                 * em_eo_current() from the EO start functions.
                 * Before returning, restore the original 'current q_elem' from
                 * 'save_q_elem'.
                 */
                memset(&tmp_q_elem, 0, sizeof(tmp_q_elem));
                tmp_q_elem.eo = (uint16_t)(uintptr_t)eo_elem->eo;
                locm->current.q_elem = &tmp_q_elem;
 
                locm->start_eo_elem = eo_elem;
                status = eo_elem->start_local_func(eo_elem->eo_ctx,
                                                   eo_elem->eo);
                locm->start_eo_elem = NULL;
                /* Restore the original 'current q_elem' */
                locm->current.q_elem = save_q_elem;
                break;
 
        case EO_STOP_SYNC_LOCAL_REQ:
        case EO_STOP_LOCAL_REQ:
                if (eo_elem->stop_local_func) {
                        /*
                         * Use a tmp q_elem as the 'current q_elem' to enable
                         * calling em_eo_current() from the EO start functions.
                         * Before returning, restore the original 'current
                         * q_elem' from 'save_q_elem'.
                         */
                        memset(&tmp_q_elem, 0, sizeof(tmp_q_elem));
                        tmp_q_elem.eo = (uint16_t)(uintptr_t)eo_elem->eo;
                        locm->current.q_elem = &tmp_q_elem;
 
                        status = eo_elem->stop_local_func(eo_elem->eo_ctx,
                                                          eo_elem->eo);
                        /* Restore the original 'current q_elem' */
                        locm->current.q_elem = save_q_elem;
                } else {
                        status = EM_OK; /* No local stop func given */
                }
                break;
 
        case EO_REM_QUEUE_LOCAL_REQ:
                status = eo_remove_queue_local(eo_elem, q_elem);
                break;
        case EO_REM_QUEUE_SYNC_LOCAL_REQ:
                status = eo_remove_queue_sync_local(eo_elem, q_elem);
                break;
        case EO_REM_QUEUE_ALL_LOCAL_REQ:
                status = eo_remove_queue_all_local(eo_elem, delete_queues);
                break;
        case EO_REM_QUEUE_ALL_SYNC_LOCAL_REQ:
                status = eo_remove_queue_all_sync_local(eo_elem, delete_queues);
                break;
        default:
                status = EM_FATAL(EM_ERR_BAD_ID);
                break;
        }
 
        if (status != EM_OK) {
                /* store failing status, egrp 'done' can check if all ok */
                loc_func_retvals->core[em_core_id()] = status;
 
                INTERNAL_ERROR(status, EM_ESCOPE_EVENT_INTERNAL_LFUNC_CALL,
                               "EO:%" PRI_EO "-%s:Local func(%" PRIx64 ")fail",
                               eo_elem->eo, eo_elem->name, f_type);
        }
 
        /*
         * In case of setup error, determine if 'loc_func_retvals' should be
         * freed here, in the setup code in eo_local_func_call_req() or
         * normally in a successful case in the
         * eo_start/stop_local__done_callback() function when the event group
         * completion notif is handled.
         */
        uint32_t cnt = odp_atomic_fetch_sub_u32(&loc_func_retvals->free_at_zero, 1);
 
        if (unlikely(cnt == 1)) {
                (void)check_eo_local_status(loc_func_retvals);
                em_free(loc_func_retvals->event);
        }
}
 
unsigned int eo_count(void)
{
        return odp_atomic_load_u32(&em_shm->eo_count);
}
 
size_t eo_name(const eo_elem_t *const eo_elem,
               char name[/*out*/], const size_t maxlen)
{
        size_t len;
 
        len = strnlen(eo_elem->name, sizeof(eo_elem->name) - 1);
        if (maxlen - 1 < len)
                len = maxlen - 1;
 
        memcpy(name, eo_elem->name, len);
        name[len] = '\0';
 
        return len;
}
 
static const char *state_to_str(em_eo_state_t state)
{
        const char *state_str;
 
        switch (state) {
        case EM_EO_STATE_UNDEF:
                state_str = "UNDEF";
                break;
        case EM_EO_STATE_CREATED:
                state_str = "CREATED";
                break;
        case EM_EO_STATE_STARTING:
                state_str = "STARTING";
                break;
        case EM_EO_STATE_RUNNING:
                state_str = "RUNNING";
                break;
        case EM_EO_STATE_STOPPING:
                state_str = "STOPPING";
                break;
        case EM_EO_STATE_ERROR:
                state_str = "ERROR";
                break;
        default:
                state_str = "UNKNOWN";
                break;
        }
 
        return state_str;
}
 
#define EO_INFO_HDR_FMT \
"Number of EOs: %d\n\n" \
"ID        Name                            State     Start-local  Stop-local" \
"  Multi-rcv  Max-events  Err-hdl  Q-num  EO-ctx\n" \
"---------------------------------------------------------------------------" \
"-----------------------------------------------\n%s\n"
 
#define EO_INFO_LEN 123
#define EO_INFO_FMT "%-10" PRI_EO "%-32s%-10s%-13c%-12c%-11c%-12d%-9c%-7d%-6c\n"
 
void eo_info_print_all(void)
{
        unsigned int num_eo;
        eo_elem_t *eo_elem;
        int len = 0;
        int n_print = 0;
        em_eo_t eo = em_eo_first(&num_eo);
 
        /*
         * num_eo may not match the amount of EOs actually returned by iterating
         * using em_eo_next() if EOs are added or removed in parallel by
         * another core. Thus space for 10 extra EOs is reserved. If more than 10
         * EOs are added by other cores in parallel, we only print information of
         * the (num_eo + 10) EOs.
         *
         * The extra 1 byte is reserved for the terminating null byte.
         */
        const int eo_info_str_len = (num_eo + 10) * EO_INFO_LEN + 1;
        char eo_info_str[eo_info_str_len];
 
        while (eo != EM_EO_UNDEF) {
                eo_elem = eo_elem_get(eo);
                if (unlikely(eo_elem == NULL || !eo_allocated(eo_elem))) {
                        eo = em_eo_next();
                        continue;
                }
 
                n_print = snprintf(eo_info_str + len,
                                   eo_info_str_len - len,
                                   EO_INFO_FMT, eo, eo_elem->name,
                                   state_to_str(eo_elem->state),
                                   eo_elem->start_local_func ? 'Y' : 'N',
                                   eo_elem->stop_local_func ? 'Y' : 'N',
                                   eo_elem->use_multi_rcv ? 'Y' : 'N',
                                   eo_elem->max_events,
                                   eo_elem->error_handler_func ? 'Y' : 'N',
                                   odp_atomic_load_u32(&eo_elem->num_queues),
                                   eo_elem->eo_ctx ? 'Y' : 'N');
 
                /* Not enough space to hold more eo info */
                if (n_print >= eo_info_str_len - len)
                        break;
 
                len += n_print;
                eo = em_eo_next();
        }
 
        /* No EO */
        if (!len) {
                EM_PRINT("No EO has been created!\n");
                return;
        }
 
        /*
         * To prevent printing incomplete information of the last eo when there
         * is not enough space to hold all eo info.
         */
        eo_info_str[len] = '\0';
        EM_PRINT(EO_INFO_HDR_FMT, num_eo, eo_info_str);
}
 
#define EO_Q_INFO_HDR_FMT \
"EO %" PRI_EO "(%s) has %d queue(s):\n\n" \
"Handle    Name                            Priority  Type      State    Qgrp" \
"      Ctx\n" \
"---------------------------------------------------------------------------" \
"---------\n" \
"%s\n"
 
#define EO_Q_INFO_LEN 85
#define EO_Q_INFO_FMT \
"%-10" PRI_QUEUE "%-32s%-10d%-10s%-9s%-10" PRI_QGRP "%-3c\n" /*85 characters*/
 
void eo_queue_info_print(em_eo_t eo)
{
        unsigned int q_num;
        em_queue_t q;
        const queue_elem_t *q_elem;
        char q_name[EM_QUEUE_NAME_LEN];
        int len = 0;
        int n_print = 0;
        const eo_elem_t *eo_elem = eo_elem_get(eo);
 
        if (unlikely(eo_elem == NULL || !eo_allocated(eo_elem))) {
                EM_PRINT("EO %" PRI_EO " is not created!\n", eo);
                return;
        }
 
        q = em_eo_queue_first(&q_num, eo);
 
        /*
         * q_num may not match the amount of queues actually returned by iterating
         * using em_eo_queue_next() if queues are added or removed in parallel
         * by another core. Thus space for 10 extra queues is reserved. If more
         * than 10 queues are added to this EO by other cores, we only print info
         * of the (q_num + 10) queues.
         *
         * The extra 1 byte is reserved for the terminating null byte.
         */
        const int eo_q_info_str_len = (q_num + 10) * EO_Q_INFO_LEN + 1;
        char eo_q_info_str[eo_q_info_str_len];
 
        while (q != EM_QUEUE_UNDEF) {
                q_elem = queue_elem_get(q);
                if (unlikely(q_elem == NULL || !queue_allocated(q_elem))) {
                        q = em_eo_queue_next();
                        continue;
                }
 
                queue_name(q_elem, q_name, EM_QUEUE_NAME_LEN - 1);
 
                n_print = snprintf(eo_q_info_str + len,
                                   eo_q_info_str_len - len,
                                   EO_Q_INFO_FMT,
                                   q, q_name, q_elem->priority,
                                   queue_type_str(q_elem->type),
                                   queue_state_str(q_elem->state),
                                   q_elem->queue_group,
                                   q_elem->context ? 'Y' : 'N');
 
                /* Not enough space to hold more queue info */
                if (n_print >= eo_q_info_str_len - len)
                        break;
 
                len += n_print;
                q = em_eo_queue_next();
        }
 
        /* EO has no queue */
        if (!len) {
                EM_PRINT("EO %" PRI_EO "(%s) has no queue!\n", eo, eo_elem->name);
                return;
        }
 
        /*
         * To prevent printing incomplete information of the last queue when
         * there is not enough space to hold all queue info.
         */
        eo_q_info_str[len] = '\0';
        EM_PRINT(EO_Q_INFO_HDR_FMT, eo, eo_elem->name, q_num, eo_q_info_str);
}
 
/**
 * @brief Create a stash used to buffer events sent during EO-start
 */
odp_stash_t eo_start_stash_create(void)
{
        unsigned int num_obj = 0;
        odp_stash_capability_t stash_capa;
        odp_stash_param_t stash_param;
        odp_stash_t stash = ODP_STASH_INVALID;
 
        int ret = odp_stash_capability(&stash_capa, ODP_STASH_TYPE_FIFO);
 
        if (ret != 0)
                return ODP_STASH_INVALID;
 
        odp_stash_param_init(&stash_param);
 
        stash_param.type = ODP_STASH_TYPE_FIFO;
        stash_param.put_mode = ODP_STASH_OP_MT;
        stash_param.get_mode = ODP_STASH_OP_MT;
 
        /* Stash size: use EM default queue size value from config file: */
        num_obj = em_shm->opt.queue.min_events_default;
        if (num_obj != 0)
                stash_param.num_obj = num_obj;
        /* else: use odp default as set by odp_stash_param_init() */
 
        stash_param.obj_size = sizeof(uint64_t);
        if (stash_param.num_obj > stash_capa.max_num.u64) {
                EM_LOG(EM_LOG_PRINT,
                       "%s(): req stash.num_obj(%" PRIu64 ") > capa.max_num.u64(%" PRIu64 ").\n"
                       "      ==> using max value:%" PRIu64 "\n", __func__,
                       stash_param.num_obj, stash_capa.max_num.u64, stash_capa.max_num.u64);
                stash_param.num_obj = stash_capa.max_num.u64;
        }
 
        stash_param.cache_size = 0; /* No core local caching */
 
        stash = odp_stash_create(NULL, &stash_param);
        if (unlikely(stash == ODP_STASH_INVALID))
                return ODP_STASH_INVALID;
 
        return stash;
}
 
/**
 * @brief Run EO local-start functions for all EOs on a new core (if configured)
 *
 * If the EM-core is added after initial startup and EO:s are already
 * created and running: check all running EO's and call the local start
 * functions (if available).
 */
void eo_start_local_fn_at_init(void)
{
        unsigned int num = 0;
        em_eo_t eo = em_eo_first(&num);
 
        if (num == 0)
                return;
 
        DBG_PRINT("%s():%d EOs available, EM-core:%02d\n", __func__, num, em_core_id());
 
        while (eo != EM_EO_UNDEF) {
                eo_elem_t *eo_elem = eo_elem_get(eo);
 
                if (likely(eo_elem)) {
                        odp_ticketlock_lock(&eo_elem->lock);
                        bool is_locked = true;
                        bool cfg_file_opt = em_shm->opt.eo.start_local_fn_at_init;
                        em_eo_start_local_mode_t mode = eo_elem->start_local_mode;
 
                        bool has_local_fn = eo_elem->start_local_func &&
                                            eo_elem->state == EM_EO_STATE_RUNNING;
                        bool has_run_mode = mode == EM_EO_START_LOCAL_MODE_INIT_RUN ||
                                            (mode == EM_EO_START_LOCAL_MODE_CONFIG_FILE &&
                                             cfg_file_opt);
 
                        if (has_local_fn && has_run_mode) {
                                em_start_local_func_t loc_fn = eo_elem->start_local_func;
                                void *eo_ctx = eo_elem->eo_ctx;
 
                                odp_ticketlock_unlock(&eo_elem->lock);
                                is_locked = false;
 
                                /* Run the EO local start function (unlocked) */
                                loc_fn(eo_ctx, eo);
                        }
 
                        if (is_locked)
                                odp_ticketlock_unlock(&eo_elem->lock);
                }
                eo = em_eo_next();
        }
        DBG_PRINT("%s(): EO local starts done, EM-core:%02d\n", __func__, em_core_id());
}
 
/**
 * @brief Run EO local-stop funcs for all EOs during term-core (if configured)
 *
 * If the EM-core is removed while EOs are running:
 * check all running EO's and call the local stop function (if available).
 */
void eo_stop_local_fn_at_term(void)
{
        unsigned int num = 0;
        em_eo_t eo = em_eo_first(&num);
 
        if (num == 0)
                return;
 
        DBG_PRINT("%s():%d EOs available, EM-core:%02d\n", __func__, num, em_core_id());
 
        while (eo != EM_EO_UNDEF) {
                eo_elem_t *eo_elem = eo_elem_get(eo);
 
                if (likely(eo_elem)) {
                        odp_ticketlock_lock(&eo_elem->lock);
                        bool is_locked = true;
                        bool cfg_file_opt = em_shm->opt.eo.stop_local_fn_at_term;
                        em_eo_stop_local_mode_t mode = eo_elem->stop_local_mode;
 
                        bool has_local_fn = eo_elem->stop_local_func &&
                                            eo_elem->state == EM_EO_STATE_RUNNING;
                        bool has_run_mode = mode == EM_EO_STOP_LOCAL_MODE_TERM_RUN ||
                                            (mode == EM_EO_STOP_LOCAL_MODE_CONFIG_FILE &&
                                             cfg_file_opt);
 
                        if (has_local_fn && has_run_mode) {
                                em_stop_local_func_t loc_fn = eo_elem->stop_local_func;
                                void *eo_ctx = eo_elem->eo_ctx;
 
                                odp_ticketlock_unlock(&eo_elem->lock);
                                is_locked = false;
 
                                /* Run the EO local stop function (unlocked) */
                                loc_fn(eo_ctx, eo);
                        }
 
                        if (is_locked)
                                odp_ticketlock_unlock(&eo_elem->lock);
                }
                eo = em_eo_next();
        }
        DBG_PRINT("%s(): EO local stops done, EM-core:%02d\n", __func__, em_core_id());
}