logicaffeine_compile/concurrency/
driver.rs

1//! The interpreter task driver.
2//!
3//! [`InterpreterTask`] wraps one suspended interpreter future (an `async`
4//! execution that owns its own [`crate::interpreter::Interpreter`]) as a
5//! scheduler [`Task`]. On each `poll` it delivers the scheduler's resume value
6//! into the side-channel, drives the future to its next suspension, and maps the
7//! interpreter's [`BlockingRequest`] to a scheduler [`TaskStep`].
8
9use std::cell::RefCell;
10use std::future::Future;
11use std::pin::Pin;
12use std::rc::Rc;
13use std::task::{Context, Poll};
14
15use logicaffeine_runtime::{RtPayload, Task, TaskCtx, TaskStep};
16
17use super::bridge::{BlockingRequest, ResumeValue, Yield};
18
19/// A shared cell the first failing task writes its error into.
20pub type ErrSink = Rc<RefCell<Option<String>>>;
21
22/// Drives one interpreter task on the scheduler.
23pub struct InterpreterTask<'a> {
24    fut: Pin<Box<dyn Future<Output = Result<(), String>> + 'a>>,
25    ys: Yield<'a>,
26    err_sink: Option<ErrSink>,
27}
28
29impl<'a> InterpreterTask<'a> {
30    /// Wrap a suspended interpreter future + its side-channel. `err_sink`, when
31    /// set, receives the task's error if it fails (first error wins).
32    pub fn new(
33        fut: Pin<Box<dyn Future<Output = Result<(), String>> + 'a>>,
34        ys: Yield<'a>,
35        err_sink: Option<ErrSink>,
36    ) -> Self {
37        InterpreterTask { fut, ys, err_sink }
38    }
39}
40
41impl<'a> Task<'a> for InterpreterTask<'a> {
42    fn poll(&mut self, ctx: &mut TaskCtx) -> TaskStep<'a> {
43        // Deliver whatever the scheduler resumed us with into the side-channel,
44        // so the `.await`ing concurrency op resolves to it.
45        {
46            let mut st = self.ys.borrow_mut();
47            st.resume = ctx_to_resume(ctx);
48        }
49        let waker = futures::task::noop_waker();
50        let mut cx = Context::from_waker(&waker);
51        match self.fut.as_mut().poll(&mut cx) {
52            Poll::Ready(result) => {
53                if let Err(e) = result {
54                    if let Some(sink) = &self.err_sink {
55                        let mut slot = sink.borrow_mut();
56                        if slot.is_none() {
57                            *slot = Some(e);
58                        }
59                    }
60                }
61                TaskStep::Exit(RtPayload::Nothing)
62            }
63            Poll::Pending => {
64                // A `Some` request is a deliberate concurrency yield (channel/timer/select).
65                // A `None` means the interpreter is awaiting real external I/O — a network
66                // op (`Connect`/`Listen`/peer messaging) whose future is pending on the host
67                // reactor. The scheduler can't service that itself, so park the task
68                // `IoPending`; the async drive loop yields to the reactor and re-polls it.
69                match self.ys.borrow_mut().request.take() {
70                    Some(req) => request_to_taskstep(req),
71                    None => TaskStep::IoPending,
72                }
73            }
74        }
75    }
76}
77
78/// Translate the scheduler's resume context into the interpreter's resume value.
79fn ctx_to_resume(ctx: &mut TaskCtx) -> ResumeValue {
80    if let Some(arm) = ctx.selected_arm.take() {
81        ResumeValue::Select {
82            arm,
83            payload: std::mem::replace(&mut ctx.resumed_with, RtPayload::Nothing),
84        }
85    } else if let Some(id) = ctx.new_chan {
86        ResumeValue::Chan(id)
87    } else if let Some(id) = ctx.spawned {
88        ResumeValue::Task(id)
89    } else {
90        ResumeValue::Payload(std::mem::replace(&mut ctx.resumed_with, RtPayload::Nothing))
91    }
92}
93
94/// Map an interpreter blocking request to the scheduler step that services it.
95fn request_to_taskstep<'a>(req: BlockingRequest<'a>) -> TaskStep<'a> {
96    match req {
97        BlockingRequest::NewChan(c) => TaskStep::NewChan(c),
98        BlockingRequest::Send(ch, p) => TaskStep::Send(ch, p),
99        BlockingRequest::TrySend(ch, p) => TaskStep::TrySend(ch, p),
100        BlockingRequest::Recv(ch) => TaskStep::Recv(ch),
101        BlockingRequest::TryRecv(ch) => TaskStep::TryRecv(ch),
102        BlockingRequest::Select(arms) => TaskStep::Select(arms),
103        BlockingRequest::Sleep(d) => TaskStep::Sleep(d),
104        BlockingRequest::Spawn(t) => TaskStep::Spawn(t),
105        BlockingRequest::Await(t) => TaskStep::Await(t),
106        BlockingRequest::Abort(t) => TaskStep::Abort(t),
107        BlockingRequest::Close(ch) => TaskStep::Close(ch),
108    }
109}
logicaffeine_compile/concurrency/driver.rs

logicaffeine_compile/concurrency/
driver.rs
