rte_posix_osadapter_task.c

Go to the documentation of this file.

#define _POSIX_C_SOURCE 200809L

#include "rte/posix_osadapter/rte_posix_osadapter.h"
#include "rte_posix_osadapter_period.h"
#include "rte_posix_osadapter_report.h"
 
#include <errno.h>
#include <pthread.h>
#include <sched.h>
#include <signal.h>
#include <stdbool.h>
#include "rte/oal/memory/rte_mem_util.h"
 
typedef struct
{
    pthread_t thread;
    rte_task_config_t config;
    bool joinable;                     /* controlling thread only, see file header */
    volatile sig_atomic_t should_stop; /* the only field both threads write, see file header */
    rte_task_handle_t self_handle;
} posix_task_state_t;
 
/* C99-compatible static assert: posix_task_state_t must fit inside
 * rte_task_storage_t's reserved bytes. */
typedef char posix_task_storage_fits_[(sizeof(posix_task_state_t) <= sizeof(rte_task_storage_t)) ? 1 : -1];
 
/* The thread-specific-data key of the self check (file header): created once per process by
 * create_task_key() through the pthread_once() in task_key_ready(), never deleted. s_task_key_ready is
 * written only inside pthread_once() and read after it returned. */
static pthread_key_t s_task_key;
static bool s_task_key_ready = false;

static bool set_fifo_attributes(pthread_attr_t *attr, uint32_t priority)
{
    bool applied = false;
    int min_prio = sched_get_priority_min(SCHED_FIFO);
    int max_prio = sched_get_priority_max(SCHED_FIFO);
 
    if ((min_prio >= 0) && (max_prio >= min_prio))
    {
        struct sched_param sched;
        int wanted;
 
        if (priority < (uint32_t)min_prio)
        {
            wanted = min_prio;
        }
        else if (priority > (uint32_t)max_prio)
        {
            wanted = max_prio;
        }
        else
        {
            wanted = (int)priority;
        }
 
        rte_mem_set(&sched, 0, sizeof(sched));
        sched.sched_priority = wanted;
        if (pthread_attr_setinheritsched(attr, PTHREAD_EXPLICIT_SCHED) == 0)
        {
            if (pthread_attr_setschedpolicy(attr, SCHED_FIFO) == 0)
            {
                if (pthread_attr_setschedparam(attr, &sched) == 0)
                {
                    applied = true;
                }
            }
        }
    }
    return applied;
}

static void create_task_key(void)
{
    s_task_key_ready = (pthread_key_create(&s_task_key, NULL) == 0);
}

static bool task_key_ready(void)
{
    static pthread_once_t key_once = PTHREAD_ONCE_INIT;
    bool ready = false;
 
    if (pthread_once(&key_once, create_task_key) == 0)
    {
        ready = s_task_key_ready;
    }
    return ready;
}

static void run_periodic_task(posix_task_state_t *state)
{
    struct timespec deadline;
    bool running = (posix_period_init(&deadline) == RTE_STATUS_OK);
    bool clock_failed = !running;
 
    while (running && (state->should_stop == 0))
    {
        posix_period_advance(&deadline, state->config.period_ms);
        if (posix_period_sleep_until(&deadline, &state->should_stop) != RTE_STATUS_OK)
        {
            clock_failed = true;
            running = false;
        }
        else if (state->should_stop == 0)
        {
            state->config.entry(state->config.user_ctx);
        }
        else
        {
            running = false;
        }
    }
 
    if (clock_failed)
    {
        /* OSA rule R3/R6: no monotonic time base - never pace from a guessed time 0. Report once; the controlling
         * thread still joins this thread. */
        POSIX_OSA_REPORT_RT_ATTRIBUTE("task monotonic clock failed - task stopped");
    }
}

static void run_task(posix_task_state_t *state)
{
    if (state->config.period_ms == 0U)
    {
        /* Run-once task: entry() runs exactly once per start (once the
         * self-check mark is set, see file header). should_stop is
         * NOT consulted here - it exists to interrupt a *periodic* task
         * between cycles. A one-shot task has no "between cycles" to
         * interrupt, and a caller cannot meaningfully cancel work that
         * hasn't started yet: rte_task_start() has already committed to
         * running entry() once. suspend()/destroy() on a run-once task
         * therefore just blocks (pthread_join) until this single call
         * completes, rather than racing to skip it (and joins the
         * thread even after entry() has returned).
         *
         * (Previously this branch also checked should_stop==0 before
         * calling entry(), which created a genuine race: if suspend()/
         * destroy() ran on another thread before this thread was first
         * scheduled by the OS, should_stop could already be 1 here and
         * entry() would be silently skipped - the task would appear to
         * "do nothing" with no error reported anywhere.) */
        state->config.entry(state->config.user_ctx);
    }
    else
    {
        run_periodic_task(state);
    }
}

static void *task_thread_main(void *arg)
{
    posix_task_state_t *state = (posix_task_state_t *)arg;
 
    if (pthread_setspecific(s_task_key, state) == 0)
    {
        run_task(state);
    }
    else
    {
        /* OSA rule R2/R6: no self check possible - an entry calling suspend/start on its own task could not be told
         * apart from another thread. Report and call no entry; the controlling thread still joins this thread. */
        POSIX_OSA_REPORT_RT_ATTRIBUTE("task thread-specific data not set - task not run");
    }
    /* The controlling thread learns that this thread ended from pthread_join(). */
    return NULL;
}

static bool is_task_thread(const posix_task_state_t *state)
{
    return (pthread_getspecific(s_task_key) == (const void *)state);
}

static rte_status_t join_task_thread(posix_task_state_t *state)
{
    rte_status_t status = RTE_STATUS_OK;
 
    if (state->joinable)
    {
        posix_period_wake_all(&state->should_stop);
        if (pthread_join(state->thread, NULL) == 0)
        {
            state->joinable = false;
        }
        else
        {
            status = RTE_STATUS_INTERNAL_ERROR;
        }
    }
    return status;
}

static rte_status_t create_task_thread(posix_task_state_t *state)
{
    rte_status_t status = RTE_STATUS_INTERNAL_ERROR;
    pthread_attr_t attr;
 
    if (pthread_attr_init(&attr) == 0)
    {
        bool inherit_fallback = false;
        int rc = -1;
 
        if (state->config.stack_size > 0U)
        {
            if (pthread_attr_setstacksize(&attr, state->config.stack_size) != 0)
            {
                /* OSA rule R1: fixed stack size - report that the configured size was refused. */
                POSIX_OSA_REPORT_RT_ATTRIBUTE("task stack size not applied");
            }
        }
 
        if (set_fifo_attributes(&attr, state->config.priority))
        {
            rc = pthread_create(&state->thread, &attr, task_thread_main, state);
            if (rc == EPERM)
            {
                /* No privilege for the explicit SCHED_FIFO attributes; no thread was created. */
                inherit_fallback = true;
            }
        }
        else
        {
            inherit_fallback = true;
        }
 
        if (inherit_fallback)
        {
            /* OSA rule R1/R6: static real-time priority not applied - report once per start, then create the thread
             * with PTHREAD_INHERIT_SCHED (ADR-018 fallback, stack size kept; see file header for Linux vs macOS). */
            POSIX_OSA_REPORT_RT_ATTRIBUTE("task SCHED_FIFO priority not applied");
            rc = -1;
            if (pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED) == 0)
            {
                rc = pthread_create(&state->thread, &attr, task_thread_main, state);
            }
        }
 
        if (rc == 0)
        {
            status = RTE_STATUS_OK;
        }
        (void)pthread_attr_destroy(&attr);
    }
    return status;
}
 
static rte_status_t osadapter_create(rte_task_storage_t *storage, const rte_task_config_t *config,
                                     rte_task_handle_t *out_handle)
{
    posix_task_state_t *state;
    bool ready;
 
    if ((storage == NULL) || (config == NULL) || (out_handle == NULL) || (config->entry == NULL))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    ready = task_key_ready();
    if (ready)
    {
        ready = posix_period_wake_ready();
    }
    if (!ready)
    {
        /* The self-check key or the period wake object could not be created (file header): no task without them;
         * storage left untouched. */
        return RTE_STATUS_INTERNAL_ERROR;
    }
 
    state = (posix_task_state_t *)(void *)storage;
    rte_mem_set(state, 0, sizeof(*state));
    state->config = *config;
    state->self_handle = (rte_task_handle_t)(void *)storage;
    *out_handle = state->self_handle;
    return RTE_STATUS_OK;
}
 
static rte_status_t osadapter_suspend(rte_task_handle_t handle)
{
    posix_task_state_t *state = (posix_task_state_t *)(void *)handle;
    rte_status_t status;
 
    if (state == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    /* Set (and the waiting thread woken) before the self check, so a periodic task's loop ends whichever branch is
     * taken. On the join branch join_task_thread() sets it and wakes once more, which is harmless. */
    posix_period_wake_all(&state->should_stop);
    if (is_task_thread(state))
    {
        /* Self-call rule (file header): the task thread cannot join itself. It ends after this entry call; the next
         * start/suspend/destroy from another thread joins it. */
        status = RTE_STATUS_OK;
    }
    else
    {
        status = join_task_thread(state);
    }
    return status;
}
 
static rte_status_t osadapter_start(rte_task_handle_t handle)
{
    posix_task_state_t *state = (posix_task_state_t *)(void *)handle;
    rte_status_t status;
 
    if (state == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    if (is_task_thread(state))
    {
        /* Self-call rule (file header): a restart from the task's own entry would leave this thread running next to
         * the new one. Refused, nothing changed. */
        status = RTE_STATUS_NOT_SUPPORTED;
    }
    else
    {
        /* Starting a task whose thread exists (running, or a run-once task whose entry has returned) means "restart
         * from the top" - see file header: join the old thread before starting a fresh one. */
        status = join_task_thread(state);
        if (status == RTE_STATUS_OK)
        {
            state->should_stop = 0;
            status = create_task_thread(state);
            if (status == RTE_STATUS_OK)
            {
                state->joinable = true;
            }
        }
    }
    return status;
}
 
static rte_status_t osadapter_destroy(rte_task_handle_t handle)
{
    return osadapter_suspend(handle);
}
 
static const rte_osadapter_task_t s_posix_task_osadapter = { osadapter_create, osadapter_start, osadapter_suspend,
                                                            osadapter_destroy };
 
const rte_osadapter_task_t *rte_posix_osadapter_task(void)
{
    return &s_posix_task_osadapter;
}