xtc_exec.h

/*-
 * Copyright (c) 2026, The XTC Project
 * @stability stable
 * Use of this source code is governed by the ISC License.
 *
 * src/inc/xtc_exec.h
 *      The L2 multi-loop executor.  Owns N xtc_loop instances, each
 *      running on its own OS thread.  Tasks may be spawned on any loop
 *      from any thread; cross-thread wakers go through a per-loop MPSC
 *      inbox + xtc_io_wakeup pingback.
 *
 *      See M5_CLAIMS.md.
 */
 
#ifndef XTC_EXEC_H
#define XTC_EXEC_H
 
#include "xtc_export.h"
 
#include <stdint.h>
 
#include "xtc_loop.h"
#include "xtc_async.h"
 
typedef struct xtc_exec xtc_exec_t;
 
/*
 * L1 proportional-share (weighted-fair) scheduler handle.  INSPIRED BY
 * Glommio (Glauber Costa / ScyllaDB): a scheduling CLASS carries SHARES
 * (1..1000) and an optional LATENCY bound, and a CFS-style vruntime
 * pick gives each class a weighted CPU fraction on its loop.  Opaque:
 * created by xtc_exec_class_create, placed on a proc via
 * xtc_proc_opts_t.sched_class or xtc_proc_set_class.  A NULL handle
 * (the default) means the implicit plain-FIFO class -- zero overhead
 * until at least one class exists on a loop.
 */
typedef struct xtc_run_class *xtc_exec_class_t;
 
/*
 * PUBLIC: int  xtc_exec_init __P((xtc_exec_t **, int));
 * PUBLIC: int  xtc_exec_fini __P((xtc_exec_t *));
 * PUBLIC: int  xtc_exec_run __P((xtc_exec_t *));
 * PUBLIC: void xtc_exec_set_service_mode __P((xtc_exec_t *, int));
 * PUBLIC: int  xtc_exec_get_service_mode __P((xtc_exec_t *));
 * PUBLIC: void xtc_exec_set_eager_rebalance __P((xtc_exec_t *, int));
 * PUBLIC: int  xtc_exec_get_eager_rebalance __P((xtc_exec_t *));
 * PUBLIC: void xtc_exec_set_steal_backoff __P((xtc_exec_t *, int));
 * PUBLIC: int  xtc_exec_get_steal_backoff __P((xtc_exec_t *));
 * PUBLIC: int  xtc_exec_stop __P((xtc_exec_t *));
 * PUBLIC: int  xtc_exec_n_loops __P((xtc_exec_t *));
 * PUBLIC: int  xtc_exec_loop_id __P((void));
 * PUBLIC: int  xtc_shard_id __P((void));
 * PUBLIC: int  xtc_shard_count __P((void));
 * PUBLIC: xtc_loop_t *xtc_exec_loop __P((xtc_exec_t *, int));
 *
 * PUBLIC: int  xtc_exec_spawn __P((xtc_exec_t *, xtc_task_fn, void *, xtc_task_t **));
 * PUBLIC: int  xtc_exec_spawn_on __P((xtc_exec_t *, int, xtc_task_fn, void *, xtc_task_t **));
 * PUBLIC: int  xtc_exec_async __P((xtc_exec_t *, xtc_coro_fn, void *, xtc_task_t **));
 * PUBLIC: int  xtc_exec_async_on __P((xtc_exec_t *, int, xtc_coro_fn, void *, xtc_task_t **));
 */
 
/*
 * Lifecycle.  n_loops <= 0 selects __os_ncpus().
 *
 * `xtc_exec_run` blocks the calling thread until the executor stops:
 *   - all spawned tasks DONE and all timers fired/cancelled, or
 *   - xtc_exec_stop() called from any thread.
 *
 * On return all worker threads have been joined.
 */
XTC_API int  xtc_exec_init(xtc_exec_t **out, int n_loops);
XTC_API int  xtc_exec_fini(xtc_exec_t *exec);
XTC_API int  xtc_exec_run(xtc_exec_t *exec);
 
/*
 * POLICY CONVENTION for every per-xtc_exec_t scheduler knob below
 * (service mode, eager rebalance, and any future addition -- steal
 * policy, NUMA affinity mode, migration eagerness, ...):
 *
 *   - Named xtc_exec_set_<name> / xtc_exec_get_<name>, boolean knobs
 *     take/return a plain `int` (0/1).
 *   - Takes effect immediately; no xtc_exec_init/_fini needed to
 *     change it, and it is safe to call from any thread at any time
 *     (each knob is a single atomic field on the exec).
 *   - The doc comment states not just the EFFECT but the COST: a
 *     policy that trades CPU/syscalls for responsiveness (like eager
 *     rebalance) says so, so a caller can judge whether to opt in.
 *   - Distinct from xtc_proc_opts_t.migratable, which is a per-PROC
 *     ELIGIBILITY bit decided once at spawn (can this proc's coro be
 *     moved at all), not an executor-wide policy (how hard the
 *     executor tries to move eligible work).  Keeping eligibility and
 *     policy as separate knobs means a consumer who wants some procs
 *     movable does not have to pay an executor-wide policy's runtime
 *     cost, and vice versa.
 *   - Distinct from xtc_preempt_set_involuntary, which is a
 *     PROCESS-WIDE toggle (no xtc_exec_t argument) governing the
 *     signal-context preemption redirect, not scoped to one executor.
 *
 * xtc_cfg (xtc_cfg.h) is a separate, PROCESS-GLOBAL named-tunable
 * registry (log levels, backpressure thresholds, ops-facing settings
 * meant to survive a config-file reload) with no per-instance scoping
 * yet, so it is not used to back a per-exec knob like these; the two
 * mechanisms serve different scopes on purpose.
 */
 
/* Service mode: when set, xtc_exec_run does not idle-auto-stop and runs
 * until xtc_exec_stop is called.  Used by a supervised xtc_app, which is
 * a long-running service rather than a finite work pool.  Default 0. */
XTC_API void xtc_exec_set_service_mode(xtc_exec_t *exec, int on);
XTC_API int  xtc_exec_get_service_mode(xtc_exec_t *exec);
 
/*
 * Eager work-stealing rebalance (OFF by default).
 *
 * Every executor loop whose run queue is empty steals a peer's
 * migratable (xtc_proc_opts_t.migratable) work before it blocks in the
 * poller, whether or not it owns parked fibers or pending timers, and
 * while its run queue stays empty it bounds each blocking poll to 1 ms
 * (32 ms with steal backoff) and looks again.  (Up to 1.55.0 a loop with
 * any pending timer never stole unless eager rebalance was on.)
 *
 * Eager rebalance adds one thing: enqueuing a migratable task nudges one
 * idle peer loop (via the poller wakeup) so it steals at once instead of
 * on its next tick.  This trades a cross-thread wakeup per enqueue (when
 * a peer is idle) and some extra migration (with its cache/NUMA cost)
 * for lower rebalance latency.  A server whose migratable backends park
 * on sockets opts in; a cache-locality-bound workload leaves it off.
 * Only migratable tasks are ever moved; pinned work (the default) is
 * unaffected either way.
 */
XTC_API void xtc_exec_set_eager_rebalance(xtc_exec_t *exec, int on);
XTC_API int  xtc_exec_get_eager_rebalance(xtc_exec_t *exec);
XTC_API void xtc_exec_set_steal_backoff(xtc_exec_t *exec, int on);
XTC_API int  xtc_exec_get_steal_backoff(xtc_exec_t *exec);
 
int  xtc_exec_set_preempt(xtc_exec_t *exec, int64_t interval_ns);
XTC_API int  xtc_exec_stop(xtc_exec_t *exec);
 
XTC_API int  xtc_exec_n_loops(xtc_exec_t *exec);
 
/*
 * From inside a task running on a loop, returns that loop's
 * 0-based index.  From any other thread returns -1.  Tests use this
 * to verify cross-loop spawn placement and steals.
 */
XTC_API int  xtc_exec_loop_id(void);
 
/* Seastar-style per-shard API.  xtc_shard_id() is the 0-based index of
 * the loop the caller runs on (a standalone loop is shard 0 of 1; -1
 * off a loop); xtc_shard_count() is the number of shards (1 for a
 * standalone loop, 0 off a loop).  Index per-core state with these
 * for a shared-nothing design. */
XTC_API int  xtc_shard_id(void);
XTC_API int  xtc_shard_count(void);
 
/* Borrow a loop pointer (for tests; not generally needed). */
XTC_API xtc_loop_t *xtc_exec_loop(xtc_exec_t *exec, int idx);
 
/*
 * Per-loop work statistics, for observability and load-balance
 * diagnosis (e.g. confirming work stealing is distributing under a
 * tail-latency-sensitive workload).
 *
 *   tasks_run -- task steps executed on this loop
 *   steals    -- tasks this loop successfully stole from a peer
 *
 * Reads are lock-free snapshots of relaxed atomics; exactness across
 * a concurrently running executor is not guaranteed.
 */
typedef struct xtc_loop_stats {
        size_t   size;     /* sizeof(xtc_loop_stats_t): XTC_LOOP_STATS_INIT */
        uint64_t tasks_run;
        uint64_t steals;
} xtc_loop_stats_t;
 
#define XTC_LOOP_STATS_INIT { .size = sizeof(xtc_loop_stats_t) }
 
/*
 * PUBLIC: int  xtc_exec_loop_stats __P((xtc_exec_t *, int, xtc_loop_stats_t *));
 */
XTC_API int  xtc_exec_loop_stats(xtc_exec_t *exec, int idx, xtc_loop_stats_t *out);
 
/*
 * Spawns.  These are the multi-loop equivalents of xtc_task_spawn /
 * xtc_async; they may be called from any thread.
 */
XTC_API int  xtc_exec_spawn(xtc_exec_t *exec, xtc_task_fn fn, void *user,
                            xtc_task_t **out_task);
XTC_API int  xtc_exec_spawn_on(xtc_exec_t *exec, int loop_idx,
                               xtc_task_fn fn, void *user, xtc_task_t **out_task);
XTC_API int  xtc_exec_async(xtc_exec_t *exec, xtc_coro_fn fn, void *arg,
                            xtc_task_t **out_task);
XTC_API int  xtc_exec_async_on(xtc_exec_t *exec, int loop_idx,
                               xtc_coro_fn fn, void *arg, xtc_task_t **out_task);
 
/*
 * L1 -- OPT-IN PROPORTIONAL-SHARE (WEIGHTED-FAIR) SCHEDULER.
 *
 * INSPIRED BY Glommio (Glauber Costa / ScyllaDB): Glommio's executor
 * gives each task queue SHARES (1..1000) and a Latency class, and a
 * CFS-style min-vruntime scheduler (account_vruntime: delta *
 * reciprocal_shares >> 12, reciprocal_shares = (1<<22)/shares) hands
 * each queue a weighted CPU fraction while a Latency::Matters(d) class
 * shrinks the preempt interval so it is checked within d.  This is the
 * one capability Glommio had that libxtc lacked.
 *
 * OFF BY DEFAULT WITH ZERO OVERHEAD: a loop with no class created runs
 * the exact plain-FIFO + work-stealing-deque path, byte-for-byte.  The
 * vruntime pick activates only once a class exists on a loop.
 *
 * xtc_exec_class_create tags a scheduling class on `loop` with `shares`
 * (1..1000) and an optional `latency_ns` bound (0 = none; a non-zero
 * bound shrinks the loop's effective yield/preempt interval so the
 * class is serviced promptly).  Returns the handle in *out.  A proc is
 * placed in a class via xtc_proc_opts_t.sched_class at spawn, or via
 * xtc_proc_set_class from inside the proc.  Returns XTC_E_INVAL on bad
 * args, XTC_E_AGAIN if the per-loop class cap is reached.
 *
 * xtc_exec_class_shares / _latency read a handle's parameters back
 * (0 on a NULL handle).
 *
 * PUBLIC: int  xtc_exec_class_create __P((xtc_loop_t *, int, int64_t, xtc_exec_class_t *));
 * PUBLIC: int  xtc_exec_class_shares __P((xtc_exec_class_t));
 * PUBLIC: int64_t xtc_exec_class_latency __P((xtc_exec_class_t));
 * PUBLIC: uint64_t xtc_exec_class_runs __P((xtc_exec_class_t));
 * PUBLIC: uint64_t xtc_exec_class_vruntime __P((xtc_exec_class_t));
 */
XTC_API int  xtc_exec_class_create(xtc_loop_t *loop, int shares,
                                   int64_t latency_ns, xtc_exec_class_t *out);
XTC_API int  xtc_exec_class_shares(xtc_exec_class_t cls);
XTC_API int64_t xtc_exec_class_latency(xtc_exec_class_t cls);
XTC_API uint64_t xtc_exec_class_runs(xtc_exec_class_t cls);
XTC_API uint64_t xtc_exec_class_vruntime(xtc_exec_class_t cls);
 
/*
 * L3 -- OVER-BUDGET STALL WATCHDOG (opt-in, off by default).
 *
 * INSPIRED BY Glommio's stall detector (executor/stall.rs): when a
 * single task run exceeds a budget the runtime reports WHICH code
 * monopolized the core.  libxtc does it with a cheap in-loop
 * wall-clock check at the run-end boundary -- no watcher thread, no
 * signal -- so it is a single branch on a disabled flag when off.
 *
 * The report callback shape: `loop` and `task` name where it happened,
 * `ran_ns` is how long the run took, `budget_ns` the configured budget.
 * When no callback is set the runtime logs a WARN line and emits a
 * backtrace of the loop to stderr.
 *
 * xtc_loop_set_stall_budget arms the per-loop budget (0 = disable).
 * xtc_loop_set_stall_cb installs the report sink (NULL = log default).
 * xtc_exec_set_stall_budget is the convenience that arms every loop of
 * an executor at once.  xtc_loop_stall_count reads the over-budget
 * report count (telemetry).
 *
 * PUBLIC: void xtc_loop_set_stall_budget __P((xtc_loop_t *, int64_t));
 * PUBLIC: void xtc_loop_set_stall_cb __P((xtc_loop_t *, xtc_stall_cb, void *));
 * PUBLIC: void xtc_exec_set_stall_budget __P((xtc_exec_t *, int64_t));
 * PUBLIC: uint64_t xtc_loop_stall_count __P((const xtc_loop_t *));
 */
typedef void (*xtc_stall_cb)(xtc_loop_t *loop, xtc_task_t *task,
                             int64_t ran_ns, int64_t budget_ns, void *user);
XTC_API void xtc_loop_set_stall_budget(xtc_loop_t *loop, int64_t budget_ns);
XTC_API void xtc_loop_set_stall_cb(xtc_loop_t *loop, xtc_stall_cb cb,
                                   void *user);
XTC_API void xtc_exec_set_stall_budget(xtc_exec_t *exec, int64_t budget_ns);
XTC_API uint64_t xtc_loop_stall_count(const xtc_loop_t *loop);
 
#endif /* XTC_EXEC_H */