kunlun/common/utils/iot_task.c

/****************************************************************************

Copyright(c) 2019 by Aerospace C.Power (Chongqing) Microelectronics. ALL RIGHTS RESERVED.

This Information is proprietary to Aerospace C.Power (Chongqing) Microelectronics and MAY NOT
be copied by any method or incorporated into another program without
the express written consent of Aerospace C.Power. This Information or any portion
thereof remains the property of Aerospace C.Power. The Information contained herein
is believed to be accurate and Aerospace C.Power assumes no responsibility or
liability for its use in any way and conveys no license or title under
any patent or copyright and makes no representation or warranty that this
Information is free from patent or copyright infringement.

****************************************************************************/

/* os_shim header files */
#include "os_types.h"
#include "os_task.h"
#include "os_event.h"
#include "os_timer.h"

/* iot common header files */
#include "iot_module.h"
#include "iot_errno.h"
#include "iot_queue.h"
#include "iot_mem_pool.h"
#include "iot_dbglog_parser.h"
#include "iot_dbglog_api.h"
#include "iot_task.h"
#include "iot_utils.h"
#include "iot_io.h"

/* task message queue */
typedef struct _iot_task_msg_queue {
    /* queue to link the message */
    iot_msg_queue_t q;
    /* current quota of the queue */
    uint8_t         curr_quota;
    /* assgined quota of the queue */
    uint8_t         quota;
    /* flag to mark if quota feature enabled for this queue */
    uint8_t         quota_en;
    /* reserved for future */
    uint8_t         reserved;
} iot_task_msg_queue_t;


typedef struct _iot_task {
    /* task configuration */
    iot_task_config_t       cfg;
    /* task handle */
    os_task_h               task_h;
    /* pointer to task message queues */
    iot_task_msg_queue_t    msg_q[IOT_TASK_QUEUE_CNT_MAX];
    /* task messages pool */
    iot_mem_pool_t          *msg_p;
} iot_task_t;

static void iot_task_func(void* arg)
{
    uint8_t i;
    uint32_t events;
    iot_task_t *task = (iot_task_t *)arg;
    iot_msg_entry_t *entry;
    iot_task_msg_t *msg;
    iot_task_msg_queue_t *pq;
    uint8_t quota_disbled_q_cnt = 0;
    uint8_t quota_enabled_q_cnt = 0;
    uint8_t quota_exhausted_q_cnt = 0;
    iot_task_msg_queue_t *quota_disbled_q[IOT_TASK_QUEUE_CNT_MAX];
    iot_task_msg_queue_t *quota_enabled_q[IOT_TASK_QUEUE_CNT_MAX];
    iot_task_msg_queue_t *quota_exhausted_q[IOT_TASK_QUEUE_CNT_MAX];

    iot_printf("%s task %p, exe func %p\n", __FUNCTION__, task,
        task->cfg.msg_exe_func);
    iot_dbglog_input(IOT_TASK_MID, DBGLOG_INFO_LVL_2,
        IOT_TASK_FUNC_ID, 2, task, task->cfg.msg_exe_func);

    /* check if quota enabled and disabled queue exist */
    for (i = 0; i < task->cfg.queue_cnt; i++) {
        pq = &task->msg_q[i];
        if (pq->quota_en) {
            quota_enabled_q[quota_enabled_q_cnt] = pq;
            quota_enabled_q_cnt++;
        } else {
            quota_disbled_q[quota_disbled_q_cnt] = pq;
            quota_disbled_q_cnt++;
        }
    }

    for (;;) {
        /* waiting for messages and task events.
         * always handle taks event firstly.
         */
        events = os_wait_task_event_with_v(MAX_TIME);

handle_event:
        if (events && task->cfg.task_event_func)
            task->cfg.task_event_func(task, events);

handle_msg:
        /* check quota disabled queue */
        if (quota_disbled_q_cnt) {
            /* check message queues based on priority */
            for (i = 0; i < quota_disbled_q_cnt; i++) {
                pq = quota_disbled_q[i];
                entry = iot_msg_queue_get(&pq->q);
                if (entry) {
                    /* get message and handle it */
                    msg = container_of(entry, iot_task_msg_t, link);
                    task->cfg.msg_exe_func(task, msg);
                    /* check if any task event pending */
                    events = os_wait_task_event_with_v(0);
                    if (events)
                        goto handle_event;
                    /* always search from highest priority queue with quota
                     * disabled.
                     */
                    goto handle_msg;
                }
            }
        }

        /* check quota enabled queue */
        if (quota_enabled_q_cnt) {
            quota_exhausted_q_cnt = 0;
            /* check message queues based on priority */
            for (i = 0; i < quota_enabled_q_cnt; i++) {
                pq = quota_enabled_q[i];
                /* check if quota available for this queue */
                if (pq->curr_quota == 0) {
                    quota_exhausted_q[quota_exhausted_q_cnt] = pq;
                    quota_exhausted_q_cnt++;
                    continue;
                }
                entry = iot_msg_queue_get(&pq->q);
                if (entry) {
                    /* get message and handle it */
                    msg = container_of(entry, iot_task_msg_t, link);
                    task->cfg.msg_exe_func(task, msg);
                    pq->curr_quota--;
                    /* check if any task event pending */
                    events = os_wait_task_event_with_v(0);
                    if (events)
                        goto handle_event;
                    /* always search from highest priority queue with quota
                     * disabled.
                     */
                    goto handle_msg;
                }
            }

            if (quota_exhausted_q_cnt == quota_enabled_q_cnt) {
                /* no quota available queue exist, re-assign quota */
                for (i = 0; i < quota_enabled_q_cnt; i++) {
                    pq = quota_enabled_q[i];
                    pq->curr_quota = pq->quota;
                }
                goto handle_msg;
            } else {
                /* check if any quota exhausted queue exist */
                for (i = 0; i < quota_exhausted_q_cnt; i++) {
                    pq = quota_exhausted_q[i];
                    entry = iot_msg_queue_get(&pq->q);
                    if (entry) {
                        /* get message and handle it */
                        msg = container_of(entry, iot_task_msg_t, link);
                        task->cfg.msg_exe_func(task, msg);
                        /* check if any task event pending */
                        events = os_wait_task_event_with_v(0);
                        if (events)
                            goto handle_event;
                        /* always search from highest priority queue with quota
                         * disabled.
                         */
                        goto handle_msg;
                    }
                }
            }
        }
    }
}

iot_task_h iot_task_create_with_name(module_id_t module, iot_task_config_t *cfg,
    const char *name)
{
    uint8_t i;
    uint32_t ret;
    iot_task_t *task = NULL;

    if (cfg->msg_exe_func == NULL || cfg->msg_cancel_func == NULL)
        goto out;

    if (cfg->msg_cnt == 0 || cfg->msg_size == 0)
        goto out;

    if (cfg->queue_cnt == 0 || cfg->queue_cnt > IOT_TASK_QUEUE_CNT_MAX)
        goto out;

    if (cfg->task_prio < OS_TASK_PRIO_LOWEST ||
        cfg->task_prio > OS_TASK_PRIO_HIGHEST)
        goto out;

    task = os_mem_malloc(module, sizeof(*task));
    if (task == NULL)
        goto out;

    task->cfg = *cfg;

    /* allocate global messages pool */
    ret = iot_mem_pool_new(module, cfg->msg_cnt,
        cfg->msg_size, &task->msg_p, 1);
    if (ret)
        goto err_pool;


    /* initialize message queues */
    for (i = 0; i < cfg->queue_cnt; i++) {
        ret = iot_msg_queue_init(&task->msg_q[i].q);
        if (cfg->queue_cfg[i].quota) {
            /* enable quota for this queue */
            task->msg_q[i].quota_en = 1;
            task->msg_q[i].quota = cfg->queue_cfg[i].quota;
            task->msg_q[i].curr_quota = task->msg_q[i].quota;
        } else {
            /* disable quota for this queue */
            task->msg_q[i].quota_en = 0;
            task->msg_q[i].quota = 0;
            task->msg_q[i].curr_quota = 0;
        }
        if (ret) {
            for (; i > 0; i--) {
                iot_msg_queue_deinit(&task->msg_q[i - 1].q);
            }
            goto err_queue;
        }
    }
    /* start task */
    task->task_h = os_create_task_smp_ext(iot_task_func, task,
        cfg->task_prio, cfg->stack_size, name, cfg->core_id);
    iot_printf("%s task %p, exe func %p\n", __FUNCTION__,
        task, cfg->msg_exe_func);
    iot_dbglog_input(IOT_TASK_MID, DBGLOG_INFO_LVL_2,
        IOT_TASK_CREATE_ID, 2, (uint32_t)task, (uint32_t)cfg->msg_exe_func);

    if (task->task_h == NULL)
        goto err_task;

    goto out;

err_task:
    for (i = 0; i < cfg->queue_cnt; i++)
        iot_msg_queue_deinit(&task->msg_q[i].q);
err_queue:
    iot_mem_pool_destroy(task->msg_p);
err_pool:
    os_mem_free(task);
    task = NULL;
out:
    return task;
}

void iot_task_delete(iot_task_h task_h)
{
    uint8_t i;
    iot_msg_queue_t *pq;
    iot_msg_entry_t *entry;
    iot_task_msg_t *msg;
    iot_task_t *task = (iot_task_t *)task_h;

    if (task->task_h) {
        /* delete iot task */
        os_delete_task(task->task_h);
        task->task_h = NULL;

        for (i = 0; i < task->cfg.queue_cnt; i++) {
            pq = &task->msg_q[i].q;
            entry = iot_msg_queue_get(pq);
            while (entry) {
                /* get messages */
                msg = container_of(entry, iot_task_msg_t, link);
                task->cfg.msg_cancel_func(task, msg);
                entry = iot_msg_queue_get(pq);
            }

            /* deinit the sg message queue */
            iot_msg_queue_deinit(pq);
        }
    }

    iot_printf("%s task %p, exe func %p\n", __FUNCTION__, task,
        task->cfg.msg_exe_func);
    iot_dbglog_input(IOT_TASK_MID, DBGLOG_INFO_LVL_2,
        IOT_TASK_DELETE_ID, 2, task, task->cfg.msg_exe_func);

    iot_mem_pool_destroy(task->msg_p);

    os_mem_free(task);
}

iot_task_msg_t *iot_task_alloc_msg_with_reserved(iot_task_h task_h,
    uint32_t reserved)
{
    iot_task_t *task = (iot_task_t *)task_h;
    iot_task_msg_t *msg;

    msg = iot_mem_pool_alloc_with_reserve(task->msg_p, reserved);
    if (msg)
        iot_msg_entry_init(&msg->link);

    return msg;
}

void iot_task_free_msg(iot_task_h task_h, iot_task_msg_t *msg)
{
    iot_task_t *task = (iot_task_t *)task_h;
    iot_mem_pool_free(task->msg_p, msg);
}

void iot_task_queue_msg(iot_task_h task_h, iot_task_msg_t *msg, uint8_t prio)
{
    iot_task_t *task = (iot_task_t *)task_h;

    IOT_ASSERT(prio < task->cfg.queue_cnt);
    iot_msg_queue_put(&task->msg_q[prio].q, &msg->link);
    os_set_task_event(task->task_h);
}

typedef struct iot_task_msg_check_param {
    iot_msg_entry_t *head;
    uint16_t type;
    uint16_t id;
} iot_task_msg_check_param_t;

static uint32_t iot_task_msg_check(iot_msg_entry_t *msg, void* param)
{
    iot_task_msg_check_param_t *check;
    iot_task_msg_t *task_msg;

    check = (iot_task_msg_check_param_t *)param;
    task_msg = container_of(msg, iot_task_msg_t, link);

    if (task_msg->type == check->type
        && (check->id == IOT_TASK_MSG_ID_ALL || task_msg->id == check->id)) {
        msg->next = check->head;
        check->head = msg;
        return 1;
    }

    return 0;
}

void iot_task_clean_msg(iot_task_h task_h, uint16_t type, uint16_t id)
{
    uint8_t i;
    iot_msg_queue_t *pq;
    iot_task_msg_t *task_msg;
    iot_msg_entry_t *msg, *next_msg;
    iot_task_msg_check_param_t param;
    iot_task_t *task = (iot_task_t *)task_h;

    param.head = NULL;
    param.type = type;
    param.id = id;

    /* delete the message to be cleaned up from the message queue */
    for (i = 0; i < task->cfg.queue_cnt; i++) {
        pq = &task->msg_q[i].q;
        iot_msg_queue_loop(pq, iot_task_msg_check, &param);
    }

    /* clean messages */
    for (msg = param.head; msg != NULL; msg = next_msg) {
        next_msg = msg->next;
        task_msg = container_of(msg, iot_task_msg_t, link);
        task->cfg.msg_cancel_func(task, task_msg);
    }
}

os_task_h IRAM_ATTR iot_task_get_os_task_h(iot_task_h task_h)
{
    iot_task_t *task = (iot_task_t *)task_h;
    return task->task_h;
}