logicaffeine_runtime/
task.rs

1//! Tasks — the unit the scheduler multiplexes.
2//!
3//! A task is a small state machine: each `poll` advances it until it next wants
4//! to block (on a channel, a select, or a timer), yield, spawn, or exit. The
5//! scheduler drives `poll`, parks the task on blocking steps, and resumes it
6//! (delivering any value) when the block clears. The interpreter and VM each
7//! implement [`Task`] over their own continuation; the toy tasks in the tests
8//! implement it by hand.
9
10use crate::channel::ChanId;
11use crate::payload::RtPayload;
12
13/// A scheduler-assigned task handle.
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
15pub struct TaskId(pub u64);
16
17/// One arm of a `Select` (`Await the first of:`).
18#[derive(Debug, Clone)]
19pub enum SelectArm {
20    /// `Receive x from ch:` — ready when `ch` has a value (or a waiting sender).
21    Recv(ChanId),
22    /// `After d ticks:` — fires after `d` logical ticks if no earlier arm wins.
23    Timeout(u64),
24}
25
26/// What a task asks the scheduler to do after a `poll`.
27pub enum TaskStep<'t> {
28    /// Cooperative yield — re-enqueue me behind the other ready tasks.
29    Yield,
30    /// Block until a value can be received from `ch`; resume with that value
31    /// (delivered as `TaskCtx::resumed_with`).
32    Recv(ChanId),
33    /// Block until `ch` has room, then hand it `payload`; resume.
34    Send(ChanId, RtPayload),
35    /// Non-blocking send: hand `payload` to `ch` if it can be taken right now
36    /// (a waiting receiver or room in the buffer); resume immediately with
37    /// `RtPayload::Bool(true)` on success, `Bool(false)` if it would have blocked.
38    TrySend(ChanId, RtPayload),
39    /// Non-blocking receive: resume immediately with a value if one is available
40    /// (in `TaskCtx::resumed_with`), or `RtPayload::Nothing` if the channel is empty.
41    TryRecv(ChanId),
42    /// Block on a choice over several arms; resume with the winning arm's index
43    /// in `TaskCtx::selected_arm` (and its value in `resumed_with` for a recv arm).
44    Select(Vec<SelectArm>),
45    /// Block for `ticks` logical ticks.
46    Sleep(u64),
47    /// Create a channel with the given capacity; resume immediately with its
48    /// `ChanId` (delivered in `TaskCtx::new_chan`).
49    NewChan(Option<usize>),
50    /// Spawn a child task; resume immediately with its `TaskId` (in `TaskCtx::spawned`).
51    Spawn(Box<dyn Task<'t> + 't>),
52    /// Spawn a child *by descriptor* (function index + already-materialised
53    /// arguments) rather than a boxed body — resume with its `TaskId`. The boxed
54    /// [`TaskStep::Spawn`] cannot cross an OS-thread boundary (the body is `!Send`);
55    /// the work-stealing driver uses this `Send` form so any worker can build the
56    /// child locally from the descriptor. (The cooperative driver builds inline and
57    /// keeps using `Spawn`.)
58    SpawnDesc { func: u16, args: Vec<RtPayload>, want_handle: bool },
59    /// Block until task `TaskId` finishes; resume with its result payload
60    /// (`Nothing` if that task was aborted). Backs awaiting a task handle.
61    Await(TaskId),
62    /// Cancel task `TaskId` (backs `Stop handle`); resume immediately. The
63    /// aborted task never polls again and its awaiters observe an aborted result.
64    Abort(TaskId),
65    /// Close channel `ChanId`; resume immediately. Subsequent receives on the
66    /// drained channel return `Nothing` rather than blocking.
67    Close(ChanId),
68    /// Blocked on external async I/O (a network `await` — `Connect`/`Listen`/peer
69    /// messaging) that the scheduler itself cannot service: it has no reactor. The task is
70    /// parked `BlockedIo` and re-polled by the async drive loop after it yields to the host
71    /// reactor (tokio natively, the browser event loop on wasm). A purely channel/timer
72    /// program never produces this, so its behavior is byte-identical.
73    IoPending,
74    /// Finished, with a result payload.
75    Exit(RtPayload),
76}
77
78/// Per-`poll` context: what the task was resumed with.
79pub struct TaskCtx {
80    /// The value delivered by the blocking step that just resumed this task
81    /// (the received value for `Recv`/winning `Select` recv arm; `Nothing` otherwise).
82    pub resumed_with: RtPayload,
83    /// For a resumed `Select`, the index of the arm that won; `None` otherwise.
84    pub selected_arm: Option<usize>,
85    /// For a resumed `NewChan`, the id of the freshly-created channel.
86    pub new_chan: Option<ChanId>,
87    /// For a resumed `Spawn`, the id of the spawned task (its handle).
88    pub spawned: Option<TaskId>,
89}
90
91/// A schedulable unit of work. `'t` ties any spawned children to the lifetime of
92/// borrowed data the task closes over (e.g. the interpreter's borrowed AST).
93pub trait Task<'t> {
94    /// Advance the task until its next scheduling point.
95    fn poll(&mut self, ctx: &mut TaskCtx) -> TaskStep<'t>;
96
97    /// Scheduling priority — higher runs first under [`crate::SchedulePolicy::Priority`].
98    /// Defaults to 0 (all tasks equal).
99    fn priority(&self) -> u8 {
100        0
101    }
102
103    /// Output lines the task produced during the *just-finished* `poll` slice, in
104    /// order. The work-stealing driver collects these per slice and flushes them in
105    /// the deterministic pick-order apply, so concurrent output matches the
106    /// cooperative interleaving byte-for-byte. Cooperative tasks write straight to
107    /// their sink and leave this empty (the default).
108    fn take_output(&mut self) -> Vec<String> {
109        Vec::new()
110    }
111}
logicaffeine_runtime/task.rs

logicaffeine_runtime/
task.rs
