worker.cpp

Go to the documentation of this file.

// Copyright (C) 2026 Douglas Quigg (dstroy0) <dquigg123@gmail.com>
// SPDX-License-Identifier: AGPL-3.0-or-later
 
/**
 * @file worker.cpp
 * @brief Server worker identity - implementation.
 *
 * The id lives in thread-local storage so each worker task resolves its own
 * per-worker state with no lock and no shared lookup. The block is part of task
 * creation (no heap after begin()); an unbound context reads the zero default.
 */
 
#include "network_drivers/session/worker.h"
#include <atomic>
 
namespace
{
// Per-task worker id. Default 0: the user loop(), the lwIP thread, and unit tests
// all read worker 0, which is the sole worker in the default build.
thread_local int t_worker_id = 0;
} // namespace
 
int detws_worker_count(void)
{
    return DETWS_WORKER_COUNT;
}
 
int detws_worker_self(void)
{
    return t_worker_id;
}
 
void detws_worker_set_self(int id)
{
    t_worker_id = id;
}
 
// ---------------------------------------------------------------------------
// Worker tasks
// ---------------------------------------------------------------------------
 
#ifdef ARDUINO
 
#include "freertos/FreeRTOS.h"
#include "freertos/queue.h"
#include "freertos/task.h"
 
namespace
{
// All worker-task state, owned by one instance (internal linkage): the pump callback, the task
// handles, and the run flag. One named owner, unreachable from any other translation unit.
struct WorkerCtx
{
    detws_worker_pump_fn pump = nullptr;
    TaskHandle_t tasks[DETWS_WORKER_COUNT] = {nullptr};
    std::atomic<bool> run{false}; // release on start publishes pump; acquire in the task
};
WorkerCtx s_worker;
 
// Each worker binds its id, then pumps until asked to stop. Between iterations it
// blocks on its task notification instead of free-running the poll: a producer
// (listener_enqueue, detws_defer) nudges it the moment work arrives, so events are
// serviced immediately rather than on the next tick. The block still times out
// after DETWS_WORKER_POLL_TICKS so the idle timeout sweep (check_timeouts) keeps
// reaping stale connections with no events in flight; raising that knob now lowers
// idle wakeups without costing event latency. A nudge that races the pump is
// latched in the notify count, so ulTaskNotifyTake returns at once - no lost wake.
void worker_task(void *arg)
{
    int id = (int)(intptr_t)arg;
    detws_worker_set_self(id);
    while (s_worker.run.load(std::memory_order_acquire))
    {
        if (s_worker.pump)
            s_worker.pump(id);
        ulTaskNotifyTake(pdTRUE, DETWS_WORKER_POLL_TICKS); // wake on event, else idle-sweep timeout
    }
    s_worker.tasks[id] = nullptr;
    vTaskDelete(nullptr);
}
} // namespace
 
// Per-worker deferred-callback queues: app code on any task hands a {fn, arg} to
// the owning worker, which runs it in its own context (race-free push path).
namespace
{
struct DeferCmd
{
    detws_deferred_fn fn;
    void *arg;
};
// The per-worker deferred-callback queue HANDLES, owned by one instance. The hot path
// (detws_defer / run_deferred / wake) touches only these, so this stays small and live.
struct DeferCtx
{
    QueueHandle_t dq[DETWS_WORKER_COUNT] = {nullptr};
};
DeferCtx s_defer;
 
// The FreeRTOS static-queue backing store (control blocks + byte storage), in its OWN
// owned instance. Only detws_workers_start() references it (to create the queues), so a
// firmware that never starts workers (e.g. a pure client sketch) garbage-collects this
// multi-hundred-byte store instead of anchoring it through the always-live handle path.
struct DeferStorageCtx
{
    StaticQueue_t dq_struct[DETWS_WORKER_COUNT];
    uint8_t dq_storage[DETWS_WORKER_COUNT][DETWS_DEFER_QUEUE_DEPTH * sizeof(DeferCmd)];
};
DeferStorageCtx s_defer_store;
} // namespace
 
void detws_workers_start(detws_worker_pump_fn pump)
{
    if (s_worker.run.load(std::memory_order_acquire))
        return; // already running
    s_worker.pump = pump;
    for (int i = 0; i < DETWS_WORKER_COUNT; i++)
        if (!s_defer.dq[i])
            s_defer.dq[i] = xQueueCreateStatic(DETWS_DEFER_QUEUE_DEPTH, sizeof(DeferCmd), s_defer_store.dq_storage[i],
                                               &s_defer_store.dq_struct[i]);
    s_worker.run.store(true, std::memory_order_release);
    for (int i = 0; i < DETWS_WORKER_COUNT; i++)
    {
        int core = (DETWS_WORKER_CORE + i) % portNUM_PROCESSORS;
        xTaskCreatePinnedToCore(worker_task, "detws_worker", DETWS_WORKER_TASK_STACK, (void *)(intptr_t)i,
                                DETWS_WORKER_TASK_PRIORITY, &s_worker.tasks[i], core);
    }
}
 
bool detws_defer(int worker_id, detws_deferred_fn fn, void *arg)
{
    if (!fn)
        return false;
    if (worker_id < 0 || worker_id >= DETWS_WORKER_COUNT || !s_defer.dq[worker_id])
        return false;
    DeferCmd cmd = {fn, arg};
    if (xQueueSend(s_defer.dq[worker_id], &cmd, 0) != pdTRUE)
        return false;
    detws_worker_wake(worker_id); // run the callback now, not on the next idle sweep
    return true;
}
 
void detws_worker_wake(int worker_id)
{
    if (worker_id < 0 || worker_id >= DETWS_WORKER_COUNT)
        return;
    TaskHandle_t t = s_worker.tasks[worker_id];
    if (t)
        xTaskNotifyGive(t);
}
 
void detws_worker_run_deferred(int worker_id)
{
    if (worker_id < 0 || worker_id >= DETWS_WORKER_COUNT || !s_defer.dq[worker_id])
        return;
    DeferCmd cmd;
    while (xQueueReceive(s_defer.dq[worker_id], &cmd, 0) == pdTRUE)
        if (cmd.fn)
            cmd.fn(cmd.arg);
}
 
void detws_workers_stop(void)
{
    if (!s_worker.run.load(std::memory_order_acquire))
        return;
    s_worker.run.store(false, std::memory_order_release);
    // Tasks self-delete on their next iteration; give them a few ticks to exit
    // before the caller tears down the slots they were servicing.
    vTaskDelay(3);
}
 
bool detws_workers_running(void)
{
    return s_worker.run.load(std::memory_order_acquire);
}
 
#else // host build - no tasks; handle()/tests drive the pipeline inline
 
void detws_workers_start(detws_worker_pump_fn)
{
}
void detws_workers_stop(void)
{
}
bool detws_workers_running(void)
{
    return false;
}
void detws_worker_wake(int)
{
} // no worker task on host - nothing to wake
 
// No worker task on host: the caller and the pipeline are the same thread, so a
// deferred callback can run inline immediately (same observable effect, race-free).
bool detws_defer(int, detws_deferred_fn fn, void *arg)
{
    if (!fn)
        return false;
    fn(arg);
    return true;
}
void detws_worker_run_deferred(int)
{
}
 
#endif // ARDUINO