logicaffeine_compile/concurrency/
bridge.rs

1//! The async↔scheduler bridge.
2//!
3//! The tree-walker executes statements in an `async fn`. To run concurrent tasks
4//! on the (synchronous, poll-based) [`logicaffeine_runtime`] scheduler, each
5//! concurrency op writes a [`BlockingRequest`] into a per-task side-channel
6//! ([`YieldState`]) and `.await`s a [`YieldFuture`], which returns `Poll::Pending`
7//! once — suspending the interpreter's async stack at exactly that point. The
8//! task's driver ([`super::driver::InterpreterTask`]) reads the request, maps it
9//! to a scheduler `TaskStep`, and on the next poll delivers the scheduler's
10//! resume value back through the same channel.
11
12use std::cell::RefCell;
13use std::future::Future;
14use std::pin::Pin;
15use std::rc::Rc;
16use std::task::{Context, Poll};
17
18use logicaffeine_runtime::{ChanId, RtPayload, SelectArm, Task, TaskId};
19
20/// A blocking (or non-blocking-but-scheduler-routed) request the interpreter
21/// hands to the scheduler.
22pub enum BlockingRequest<'a> {
23    /// Create a channel of the given capacity (`None` = config default).
24    NewChan(Option<usize>),
25    /// Send a value into a channel (blocks if full).
26    Send(ChanId, RtPayload),
27    /// Non-blocking send; resume with `Payload(Bool(success))`.
28    TrySend(ChanId, RtPayload),
29    /// Receive a value from a channel (blocks if empty).
30    Recv(ChanId),
31    /// Non-blocking receive; resume with the value, or `Payload(Nothing)` if empty.
32    TryRecv(ChanId),
33    /// Await the first ready arm of a select.
34    Select(Vec<SelectArm>),
35    /// Sleep for some logical ticks.
36    Sleep(u64),
37    /// Spawn a child task; resume with its `TaskId`.
38    Spawn(Box<dyn Task<'a> + 'a>),
39    /// Await a task's completion; resume with its result payload.
40    Await(TaskId),
41    /// Abort a task.
42    Abort(TaskId),
43    /// Close a channel.
44    Close(ChanId),
45}
46
47/// What the scheduler delivers back to the interpreter on resume.
48pub enum ResumeValue {
49    /// Nothing meaningful (e.g. after `Send`/`Sleep`/`Spawn`-fire-and-forget).
50    None,
51    /// A received value (after `Recv` / a winning select recv arm).
52    Payload(RtPayload),
53    /// A freshly-created channel id (after `NewChan`).
54    Chan(ChanId),
55    /// A spawned task's id (after `Spawn`).
56    Task(TaskId),
57    /// The winning arm of a `Select`: its index plus the received payload
58    /// (`Nothing` for a timeout arm).
59    Select { arm: usize, payload: RtPayload },
60}
61
62impl ResumeValue {
63    /// The received payload, or `Nothing`.
64    pub fn into_payload(self) -> RtPayload {
65        match self {
66            ResumeValue::Payload(p) => p,
67            _ => RtPayload::Nothing,
68        }
69    }
70    /// The created channel id, if this resume was from a `NewChan`.
71    pub fn into_chan(self) -> Option<ChanId> {
72        match self {
73            ResumeValue::Chan(c) => Some(c),
74            _ => None,
75        }
76    }
77    /// The spawned task id, if this resume was from a `Spawn`.
78    pub fn into_task(self) -> Option<TaskId> {
79        match self {
80            ResumeValue::Task(t) => Some(t),
81            _ => None,
82        }
83    }
84    /// The winning arm index + its payload, if this resume was from a `Select`.
85    pub fn into_select(self) -> Option<(usize, RtPayload)> {
86        match self {
87            ResumeValue::Select { arm, payload } => Some((arm, payload)),
88            _ => None,
89        }
90    }
91}
92
93/// The per-task side-channel: the interpreter writes a `request`, the driver
94/// delivers a `resume`.
95pub struct YieldState<'a> {
96    pub request: Option<BlockingRequest<'a>>,
97    pub resume: ResumeValue,
98}
99
100impl<'a> YieldState<'a> {
101    pub fn new() -> Self {
102        YieldState { request: None, resume: ResumeValue::None }
103    }
104}
105
106impl<'a> Default for YieldState<'a> {
107    fn default() -> Self {
108        YieldState::new()
109    }
110}
111
112/// A shared handle to a task's side-channel (cloned between the interpreter and
113/// its driving `InterpreterTask`).
114pub type Yield<'a> = Rc<RefCell<YieldState<'a>>>;
115
116/// The future a concurrency op `.await`s: it returns `Poll::Pending` exactly once
117/// (the request was written just before), suspending the interpreter's async
118/// stack; the next poll (driven by the scheduler) resolves to the resume value.
119pub struct YieldFuture<'a> {
120    yielded: bool,
121    ys: Yield<'a>,
122}
123
124impl<'a> YieldFuture<'a> {
125    pub fn new(ys: Yield<'a>) -> Self {
126        YieldFuture { yielded: false, ys }
127    }
128}
129
130impl<'a> Future for YieldFuture<'a> {
131    type Output = ResumeValue;
132    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<ResumeValue> {
133        if !self.yielded {
134            self.yielded = true;
135            Poll::Pending
136        } else {
137            let mut st = self.ys.borrow_mut();
138            Poll::Ready(std::mem::replace(&mut st.resume, ResumeValue::None))
139        }
140    }
141}
logicaffeine_compile/concurrency/bridge.rs

logicaffeine_compile/concurrency/
bridge.rs
