event_machine_queue_group.c

/*
 *   Copyright (c) 2015-2023, 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 "em_core.h"
#include "em_error.h"
#include "em_internal_event.h"
#include "em_internal_event_types.h"
#include "em_mem.h"
#include "em_queue.h"
#include "em_queue_group.h"
#include "em_queue_group_types.h"
#include "em_queue_inline.h"
#include "em_queue_types.h"
 
/* per core (thread) state for em_queue_group_next() */
static ENV_LOCAL unsigned int _qgrp_tbl_iter_idx;
/* Per core (thread) state of em_queue_group_queue_next() */
static ENV_LOCAL unsigned int _qgrp_q_iter_idx;
static ENV_LOCAL em_queue_group_t _qgrp_q_iter_qgrp;
 
em_queue_group_t
em_queue_group_create(const char *name, const em_core_mask_t *mask,
                      int num_notif, const em_notif_t notif_tbl[])
{
        em_queue_group_t queue_group;
        em_status_t err;
 
        if (unlikely(!mask)) {
                INTERNAL_ERROR(EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_CREATE,
                               "Core mask NULL");
                return EM_QUEUE_GROUP_UNDEF;
        }
 
        err = check_notif_tbl(num_notif, notif_tbl);
        if (unlikely(err != EM_OK)) {
                INTERNAL_ERROR(EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_CREATE,
                               "Invalid notif cfg given:%" PRI_STAT "!", err);
                return EM_QUEUE_GROUP_UNDEF;
        }
 
        queue_group = queue_group_create(name, mask, num_notif, notif_tbl,
                                         EM_QUEUE_GROUP_UNDEF/*any free qgrp*/);
        return queue_group;
}
 
em_queue_group_t
em_queue_group_create_sync(const char *name, const em_core_mask_t *mask)
{
        em_queue_group_t queue_group;
 
        if (unlikely(!mask)) {
                INTERNAL_ERROR(EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_CREATE_SYNC,
                               "Core mask NULL");
                return EM_QUEUE_GROUP_UNDEF;
        }
 
        queue_group = queue_group_create_sync(name, mask, EM_QUEUE_GROUP_UNDEF
                                              /* any free queue group */);
        return queue_group;
}
 
em_status_t
em_queue_group_delete(em_queue_group_t queue_group,
                      int num_notif, const em_notif_t notif_tbl[])
{
        em_core_mask_t zero_mask;
        em_status_t err;
 
        queue_group_elem_t *const qgrp_elem = queue_group_elem_get(queue_group);
 
        RETURN_ERROR_IF(!qgrp_elem || !queue_group_allocated(qgrp_elem),
                        EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_DELETE,
                        "Invalid queue group: %" PRI_QGRP "", queue_group);
 
        err = check_notif_tbl(num_notif, notif_tbl);
        RETURN_ERROR_IF(err != EM_OK, EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_DELETE,
                        "Invalid notif cfg given, err=%" PRI_STAT "", err);
 
        em_core_mask_zero(&zero_mask);
 
        /* Use modify and the notif mechanism to set the core mask to zero */
        err = queue_group_modify(qgrp_elem, &zero_mask, num_notif, notif_tbl,
                                 true /*is_delete*/);
 
        RETURN_ERROR_IF(err != EM_OK, err, EM_ESCOPE_QUEUE_GROUP_DELETE,
                        "Queue group:%" PRI_QGRP " modify for delete failed!",
                        queue_group);
 
        return EM_OK;
}
 
em_status_t
em_queue_group_delete_sync(em_queue_group_t queue_group)
{
        em_core_mask_t zero_mask;
        em_status_t err;
 
        queue_group_elem_t *const qgrp_elem = queue_group_elem_get(queue_group);
 
        RETURN_ERROR_IF(!qgrp_elem || !queue_group_allocated(qgrp_elem),
                        EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_DELETE_SYNC,
                        "Invalid queue group:%" PRI_QGRP "", queue_group);
 
        em_core_mask_zero(&zero_mask);
 
        /* Use modify and the notif mechanism to set the core mask to zero */
        err = queue_group_modify_sync(qgrp_elem, &zero_mask,
                                      true /*is_delete*/);
 
        RETURN_ERROR_IF(err != EM_OK, err, EM_ESCOPE_QUEUE_GROUP_DELETE_SYNC,
                        "Queue group:%" PRI_QGRP " modify for delete failed!",
                        queue_group);
 
        return EM_OK;
}
 
em_status_t
em_queue_group_modify(em_queue_group_t queue_group,
                      const em_core_mask_t *new_mask,
                      int num_notif, const em_notif_t notif_tbl[])
{
        queue_group_elem_t *const qgrp_elem = queue_group_elem_get(queue_group);
        em_status_t err;
 
        RETURN_ERROR_IF(!qgrp_elem || !queue_group_allocated(qgrp_elem),
                        EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_MODIFY,
                        "Invalid queue group:%" PRI_QGRP "", queue_group);
 
        RETURN_ERROR_IF(!new_mask, EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_MODIFY,
                        "Queue group mask NULL! Queue group:%" PRI_QGRP "",
                        queue_group);
 
        err = check_notif_tbl(num_notif, notif_tbl);
        RETURN_ERROR_IF(err != EM_OK, EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_MODIFY,
                        "Invalid notif cfg given, err=%" PRI_STAT "", err);
 
        err = queue_group_modify(qgrp_elem, new_mask, num_notif, notif_tbl,
                                 false /*!is_delete*/);
        RETURN_ERROR_IF(err != EM_OK, err, EM_ESCOPE_QUEUE_GROUP_MODIFY,
                        "Queue group:%" PRI_QGRP " modify failed!",
                        queue_group);
        return EM_OK;
}
 
em_status_t
em_queue_group_modify_sync(em_queue_group_t queue_group,
                           const em_core_mask_t *new_mask)
{
        queue_group_elem_t *const qgrp_elem = queue_group_elem_get(queue_group);
        em_status_t err;
 
        RETURN_ERROR_IF(!qgrp_elem || !queue_group_allocated(qgrp_elem),
                        EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_MODIFY_SYNC,
                        "Invalid queue group: %" PRI_QGRP "", queue_group);
 
        RETURN_ERROR_IF(!new_mask, EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_MODIFY_SYNC,
                        "Queue group mask NULL! Queue group:%" PRI_QGRP "",
                        queue_group);
 
        err = queue_group_modify_sync(qgrp_elem, new_mask, false/*!is_delete*/);
 
        RETURN_ERROR_IF(err != EM_OK, err, EM_ESCOPE_QUEUE_GROUP_MODIFY_SYNC,
                        "Queue group:%" PRI_QGRP " modify sync failed!",
                        queue_group);
        return EM_OK;
}
 
em_queue_group_t
em_queue_group_find(const char *name)
{
        odp_schedule_group_t odp_group;
 
        if (!name || name[0] == '\0')
                return EM_QUEUE_GROUP_UNDEF;
 
        odp_group = odp_schedule_group_lookup(name);
        if (odp_group == ODP_SCHED_GROUP_INVALID)
                return EM_QUEUE_GROUP_UNDEF;
 
        for (int i = 0; i < EM_MAX_QUEUE_GROUPS; i++) {
                const queue_group_elem_t *qgrp_elem =
                        &em_shm->queue_group_tbl.queue_group_elem[i];
 
                if (qgrp_elem->odp_sched_group == odp_group &&
                    queue_group_allocated(qgrp_elem))
                        return qgrp_idx2hdl(i);
        }
 
        return EM_QUEUE_GROUP_UNDEF;
}
 
em_status_t em_queue_group_mask(em_queue_group_t queue_group,
                                em_core_mask_t *mask /*out*/)
{
        int allocated;
        bool ongoing_delete;
        queue_group_elem_t *const qgrp_elem = queue_group_elem_get(queue_group);
 
        RETURN_ERROR_IF(!qgrp_elem, EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_MASK,
                        "Invalid queue group:%" PRI_QGRP "", queue_group);
 
        odp_ticketlock_lock(&qgrp_elem->lock);
 
        allocated = queue_group_allocated(qgrp_elem);
        ongoing_delete = qgrp_elem->ongoing_delete;
        em_core_mask_copy(mask, &qgrp_elem->core_mask);
 
        odp_ticketlock_unlock(&qgrp_elem->lock);
 
        RETURN_ERROR_IF(!allocated || ongoing_delete,
                        EM_ERR_BAD_STATE, EM_ESCOPE_QUEUE_GROUP_MASK,
                        "Queue group:%" PRI_QGRP " in bad state:\t"
                        "allocated=%s, ongoing_delete=%s",
                        queue_group, allocated ? "true" : "false(!)",
                        ongoing_delete ? "true(!)" : "false");
 
        return EM_OK;
}
 
size_t em_queue_group_name(em_queue_group_t queue_group,
                           char *name, size_t maxlen)
{
        const queue_group_elem_t *qgrp_elem = queue_group_elem_get(queue_group);
        odp_schedule_group_info_t info;
        size_t len;
        int ret;
 
        if (unlikely(!name || maxlen == 0)) {
                INTERNAL_ERROR(EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_NAME,
                               "Invalid name=0x%" PRIx64 " or maxlen=%zu",
                               name, maxlen);
                return 0;
        }
 
        name[0] = '\0';
 
        if (unlikely(!qgrp_elem || !queue_group_allocated(qgrp_elem))) {
                INTERNAL_ERROR(EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_NAME,
                               "Invalid queue group:%" PRI_QGRP "",
                               queue_group);
                return 0;
        }
 
        ret = odp_schedule_group_info(qgrp_elem->odp_sched_group, &info);
        if (unlikely(ret != 0)) {
                INTERNAL_ERROR(EM_ERR_LIB_FAILED, EM_ESCOPE_QUEUE_GROUP_NAME,
                               "Failed to retrieve queue group info");
                return 0;
        }
 
        if (unlikely(!info.name))
                return 0;
 
        len = strnlen(info.name, ODP_SCHED_GROUP_NAME_LEN - 1);
        if (maxlen - 1 < len)
                len = maxlen - 1;
 
        memcpy(name, info.name, len);
        name[len] = '\0';
 
        return len;
}
 
em_queue_group_t em_queue_group_first(unsigned int *num)
{
        const queue_group_elem_t *const qgrp_elem_tbl =
                em_shm->queue_group_tbl.queue_group_elem;
        const queue_group_elem_t *qgrp_elem = &qgrp_elem_tbl[0];
        const unsigned int max_qgrps = EM_MAX_QUEUE_GROUPS;
        const unsigned int qgrp_cnt = queue_group_count();
 
        _qgrp_tbl_iter_idx = 0; /* reset iteration */
 
        if (num)
                *num = qgrp_cnt;
 
        if (qgrp_cnt == 0) {
                _qgrp_tbl_iter_idx = max_qgrps; /* UNDEF = _next() */
                return EM_QUEUE_GROUP_UNDEF;
        }
 
        /* find first */
        while (!queue_group_allocated(qgrp_elem)) {
                _qgrp_tbl_iter_idx++;
                if (_qgrp_tbl_iter_idx >= max_qgrps)
                        return EM_QUEUE_GROUP_UNDEF;
                qgrp_elem = &qgrp_elem_tbl[_qgrp_tbl_iter_idx];
        }
 
        return qgrp_idx2hdl(_qgrp_tbl_iter_idx);
}
 
em_queue_group_t em_queue_group_next(void)
{
        const unsigned int max_qgrps = EM_MAX_QUEUE_GROUPS;
 
        if (_qgrp_tbl_iter_idx >= max_qgrps - 1)
                return EM_QUEUE_GROUP_UNDEF;
 
        _qgrp_tbl_iter_idx++;
 
        const queue_group_elem_t *const qgrp_elem_tbl =
                em_shm->queue_group_tbl.queue_group_elem;
        const queue_group_elem_t *qgrp_elem =
                &qgrp_elem_tbl[_qgrp_tbl_iter_idx];
 
        /* find next */
        while (!queue_group_allocated(qgrp_elem)) {
                _qgrp_tbl_iter_idx++;
                if (_qgrp_tbl_iter_idx >= max_qgrps)
                        return EM_QUEUE_GROUP_UNDEF;
                qgrp_elem = &qgrp_elem_tbl[_qgrp_tbl_iter_idx];
        }
 
        return qgrp_idx2hdl(_qgrp_tbl_iter_idx);
}
 
em_queue_t em_queue_group_queue_first(unsigned int *num,
                                      em_queue_group_t queue_group)
{
        queue_group_elem_t *qgrp_elem = queue_group_elem_get(queue_group);
        const unsigned int max_queues = em_shm->queue_tbl.max_queue_num;
 
        if (unlikely(!qgrp_elem || !queue_group_allocated(qgrp_elem))) {
                INTERNAL_ERROR(EM_ERR_BAD_ARG, EM_ESCOPE_QUEUE_GROUP_QUEUE_FIRST,
                               "Invalid queue group:%" PRI_QGRP "",
                               queue_group);
                if (num)
                        *num = 0;
                return EM_QUEUE_UNDEF;
        }
 
        const unsigned int num_queues = odp_atomic_load_u32(&qgrp_elem->num_queues);
 
        if (num)
                *num = num_queues;
 
        if (num_queues == 0) {
                _qgrp_q_iter_idx = max_queues; /* UNDEF = _next() */
                return EM_QUEUE_UNDEF;
        }
 
        /*
         * A 'qgrp_elem' contains a linked list with all it's queues. That list
         * might be modified while processing this iteration, so instead we just
         * go through the whole queue table.
         * This is potentially a slow implementation and perhaps worth
         * re-thinking?
         */
        const queue_elem_t *const q_elem_tbl = em_shm->queue_tbl.queue_elem;
        const queue_elem_t *q_elem = &q_elem_tbl[0];
 
        _qgrp_q_iter_idx = 0; /* reset list */
        _qgrp_q_iter_qgrp = queue_group;
 
        /* find first */
        while (!queue_allocated(q_elem) ||
               q_elem->queue_group != _qgrp_q_iter_qgrp) {
                _qgrp_q_iter_idx++;
                if (_qgrp_q_iter_idx >= max_queues)
                        return EM_QUEUE_UNDEF;
                q_elem = &q_elem_tbl[_qgrp_q_iter_idx];
        }
 
        return queue_idx2hdl(_qgrp_q_iter_idx);
}
 
em_queue_t em_queue_group_queue_next(void)
{
        const unsigned int max_queues = em_shm->queue_tbl.max_queue_num;
 
        if (_qgrp_q_iter_idx >= max_queues - 1)
                return EM_QUEUE_UNDEF;
 
        _qgrp_q_iter_idx++;
 
        const queue_elem_t *const q_elem_tbl = em_shm->queue_tbl.queue_elem;
        const queue_elem_t *q_elem = &q_elem_tbl[_qgrp_q_iter_idx];
 
        /* find next */
        while (!queue_allocated(q_elem) ||
               q_elem->queue_group != _qgrp_q_iter_qgrp) {
                _qgrp_q_iter_idx++;
                if (_qgrp_q_iter_idx >= max_queues)
                        return EM_QUEUE_UNDEF;
                q_elem = &q_elem_tbl[_qgrp_q_iter_idx];
        }
 
        return queue_idx2hdl(_qgrp_q_iter_idx);
}
 
uint64_t em_queue_group_to_u64(em_queue_group_t queue_group)
{
        return (uint64_t)queue_group;
}