lenso_kernel/
driver.rs

1use super::{
2    AbortHandle, CancellationToken, Cell, Context, Duration, Either, Future, FutureExt,
3    InvocationContext, LocalBoxFuture, NativeAppRuntime, Pin, Poll, Rc, RefCell,
4    RequestAdmissionPlan, RuntimeFailure, SpawnError, VecDeque, begin_module_supervision, oneshot,
5    pending, schedule_module_supervision, select,
6};
7
8/// A task owned by a Runtime Driver's single-threaded local lane.
9pub type LocalTask = Pin<Box<dyn Future<Output = ()> + 'static>>;
10
11/// Result of joining a Runtime Driver task.
12#[derive(Clone, Copy, Debug, Eq, PartialEq)]
13pub enum TaskOutcome {
14    /// The task completed normally.
15    Completed,
16    /// The task observed cooperative cancellation.
17    Cancelled,
18    /// The task or its Runtime Driver wrapper terminated abnormally.
19    Failed,
20}
21
22/// Driver-owned handle used to cancel and join local work.
23#[derive(Debug)]
24pub struct DriverTask {
25    pub(super) abort: AbortHandle,
26    pub(super) completion: oneshot::Receiver<TaskOutcome>,
27}
28
29impl DriverTask {
30    /// Creates a task handle from Driver-owned cancellation and completion primitives.
31    pub fn new(abort: AbortHandle, completion: oneshot::Receiver<TaskOutcome>) -> Self {
32        Self { abort, completion }
33    }
34
35    /// Requests cooperative cancellation of this task.
36    pub fn cancel(&self) {
37        self.abort.abort();
38    }
39
40    pub(super) fn abort_handle(&self) -> AbortHandle {
41        self.abort.clone()
42    }
43}
44
45impl Future for DriverTask {
46    type Output = TaskOutcome;
47
48    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
49        Pin::new(&mut self.completion)
50            .poll(context)
51            .map(|outcome| outcome.unwrap_or(TaskOutcome::Failed))
52    }
53}
54
55/// Host facilities required to advance the portable Kernel.
56pub trait RuntimeDriver: Clone + 'static {
57    /// Returns the current monotonic instant relative to Driver startup.
58    fn now(&self) -> Duration;
59
60    /// Waits until the supplied monotonic instant.
61    fn sleep_until(&self, deadline: Duration) -> LocalBoxFuture<'static, ()>;
62
63    /// Cooperatively yields to other work on the local task lane.
64    fn yield_now(&self) -> LocalBoxFuture<'static, ()>;
65
66    /// Supplies deterministic or entropy-backed jitter bounded by the policy value.
67    fn jitter(&self, _maximum: Duration) -> Duration {
68        Duration::ZERO
69    }
70
71    /// Schedules Kernel-owned work on the local task lane.
72    fn spawn_local(&self, task: LocalTask) -> Result<DriverTask, SpawnError>;
73
74    /// Reports whether the embedding Runner requested shutdown.
75    fn shutdown_requested(&self) -> bool;
76}
77
78#[derive(Clone)]
79pub(super) struct DriverControl {
80    pub(super) now: Rc<dyn Fn() -> Duration>,
81    pub(super) sleep_until: Rc<dyn Fn(Duration) -> LocalBoxFuture<'static, ()>>,
82    pub(super) yield_now: Rc<dyn Fn() -> LocalBoxFuture<'static, ()>>,
83    pub(super) jitter: Rc<dyn Fn(Duration) -> Duration>,
84    pub(super) spawn_local: Rc<dyn Fn(LocalTask) -> Result<DriverTask, SpawnError>>,
85}
86
87impl std::fmt::Debug for DriverControl {
88    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
89        formatter
90            .debug_struct("DriverControl")
91            .finish_non_exhaustive()
92    }
93}
94
95impl DriverControl {
96    pub(super) fn new<D: RuntimeDriver>(driver: &D) -> Self {
97        let now_driver = driver.clone();
98        let sleep_driver = driver.clone();
99        let yield_driver = driver.clone();
100        let jitter_driver = driver.clone();
101        let spawn_driver = driver.clone();
102        Self {
103            now: Rc::new(move || now_driver.now()),
104            sleep_until: Rc::new(move |deadline| sleep_driver.sleep_until(deadline)),
105            yield_now: Rc::new(move || yield_driver.yield_now()),
106            jitter: Rc::new(move |maximum| jitter_driver.jitter(maximum)),
107            spawn_local: Rc::new(move |task| spawn_driver.spawn_local(task)),
108        }
109    }
110}
111
112pub(super) async fn wait_until<F: Future>(
113    driver: &DriverControl,
114    deadline: Duration,
115    future: F,
116) -> Option<F::Output> {
117    let work = future.fuse();
118    let timer = (driver.sleep_until)(deadline).fuse();
119    futures::pin_mut!(work, timer);
120    match select(work, timer).await {
121        Either::Left((output, _)) => Some(output),
122        Either::Right(((), _)) => None,
123    }
124}
125
126/// A bounded, per-binding Operation admission state.
127#[derive(Clone, Debug)]
128pub(super) struct RequestAdmission {
129    pub(super) limits: RequestAdmissionPlan,
130    pub(super) state: Rc<RequestAdmissionState>,
131}
132
133#[derive(Debug, Default)]
134pub(super) struct RequestAdmissionState {
135    pub(super) active: Cell<usize>,
136    pub(super) queued: Cell<usize>,
137    pub(super) waiters: RefCell<VecDeque<Rc<QueueWaiter>>>,
138}
139
140#[derive(Clone, Copy, Debug, Eq, PartialEq)]
141pub(super) enum QueueWaiterStatus {
142    Waiting,
143    Woken,
144    Acquired,
145    Cancelled,
146}
147
148#[derive(Debug)]
149pub(super) struct QueueWaiter {
150    pub(super) status: Cell<QueueWaiterStatus>,
151    pub(super) wakeup: RefCell<Option<oneshot::Sender<()>>>,
152}
153
154impl RequestAdmission {
155    pub(super) fn new(limits: RequestAdmissionPlan) -> Self {
156        Self {
157            limits,
158            state: Rc::new(RequestAdmissionState::default()),
159        }
160    }
161
162    pub(super) fn queue_depth(&self) -> usize {
163        self.state.queued.get()
164    }
165
166    pub(crate) fn try_acquire(
167        &self,
168        capability: &'static str,
169        operation: &str,
170        context: &InvocationContext,
171        driver: &DriverControl,
172    ) -> Result<RequestPermit, RuntimeFailure> {
173        ensure_context_active(driver, context)?;
174        if self.state.active.get() < self.limits.max_concurrency() {
175            self.state.active.set(self.state.active.get() + 1);
176            return Ok(RequestPermit {
177                state: self.state.clone(),
178            });
179        }
180        Err(RuntimeFailure::ResourceExhausted {
181            capability,
182            operation: operation.to_owned(),
183        })
184    }
185
186    pub(super) fn acquire(
187        &self,
188        capability: &'static str,
189        operation: &str,
190        context: InvocationContext,
191        driver: DriverControl,
192    ) -> LocalBoxFuture<'static, Result<RequestPermit, RuntimeFailure>> {
193        if let Ok(permit) = self.try_acquire(capability, operation, &context, &driver) {
194            return Box::pin(futures::future::ready(Ok(permit)));
195        }
196        if let Err(error) = ensure_context_active(&driver, &context) {
197            return Box::pin(futures::future::ready(Err(error)));
198        }
199
200        if self.state.queued.get() >= self.limits.queue_capacity() {
201            return Box::pin(futures::future::ready(Err(
202                RuntimeFailure::ResourceExhausted {
203                    capability,
204                    operation: operation.to_owned(),
205                },
206            )));
207        }
208
209        let (wakeup, waiter) = oneshot::channel();
210        let waiter_state = Rc::new(QueueWaiter {
211            status: Cell::new(QueueWaiterStatus::Waiting),
212            wakeup: RefCell::new(Some(wakeup)),
213        });
214        self.state.queued.set(self.state.queued.get() + 1);
215        self.state
216            .waiters
217            .borrow_mut()
218            .push_back(waiter_state.clone());
219        let queued = QueuedAdmission {
220            state: self.state.clone(),
221            waiter_state,
222            waiter,
223        };
224        Box::pin(async move { queued.wait(&driver, &context).await })
225    }
226}
227
228#[derive(Debug)]
229pub(super) struct QueuedAdmission {
230    pub(super) state: Rc<RequestAdmissionState>,
231    pub(super) waiter_state: Rc<QueueWaiter>,
232    pub(super) waiter: oneshot::Receiver<()>,
233}
234
235impl QueuedAdmission {
236    pub(super) async fn wait(
237        mut self,
238        driver: &DriverControl,
239        context: &InvocationContext,
240    ) -> Result<RequestPermit, RuntimeFailure> {
241        let result = await_with_context(driver, context, &mut self.waiter).await;
242        match result {
243            Ok(Ok(())) => {
244                if self.waiter_state.status.get() == QueueWaiterStatus::Woken {
245                    self.waiter_state.status.set(QueueWaiterStatus::Acquired);
246                    self.state.queued.set(self.state.queued.get() - 1);
247                    Ok(RequestPermit {
248                        state: self.state.clone(),
249                    })
250                } else {
251                    Err(RuntimeFailure::Cancelled {
252                        request_id: context.request_id(),
253                    })
254                }
255            }
256            Ok(Err(_)) => Err(RuntimeFailure::Cancelled {
257                request_id: context.request_id(),
258            }),
259            Err(error) => Err(error),
260        }
261    }
262}
263
264impl Drop for QueuedAdmission {
265    fn drop(&mut self) {
266        let previous = self
267            .waiter_state
268            .status
269            .replace(QueueWaiterStatus::Cancelled);
270        match previous {
271            QueueWaiterStatus::Waiting => {
272                self.state.queued.set(self.state.queued.get() - 1);
273            }
274            QueueWaiterStatus::Woken => {
275                self.state.queued.set(self.state.queued.get() - 1);
276                self.state.active.set(self.state.active.get() - 1);
277                wake_next(&self.state);
278            }
279            QueueWaiterStatus::Acquired | QueueWaiterStatus::Cancelled => {}
280        }
281        self.state
282            .waiters
283            .borrow_mut()
284            .retain(|waiter| !Rc::ptr_eq(waiter, &self.waiter_state));
285    }
286}
287
288#[derive(Debug)]
289pub(super) struct RequestPermit {
290    pub(super) state: Rc<RequestAdmissionState>,
291}
292
293impl Drop for RequestPermit {
294    fn drop(&mut self) {
295        self.state.active.set(self.state.active.get() - 1);
296        wake_next(&self.state);
297    }
298}
299
300pub(super) fn wake_next(state: &Rc<RequestAdmissionState>) {
301    loop {
302        let Some(waiter) = state.waiters.borrow_mut().pop_front() else {
303            return;
304        };
305        if waiter.status.replace(QueueWaiterStatus::Woken) != QueueWaiterStatus::Waiting {
306            continue;
307        }
308        state.active.set(state.active.get() + 1);
309        let sent = waiter
310            .wakeup
311            .borrow_mut()
312            .take()
313            .is_some_and(|wakeup| wakeup.send(()).is_ok());
314        if sent {
315            return;
316        }
317        waiter.status.set(QueueWaiterStatus::Cancelled);
318        state.active.set(state.active.get() - 1);
319        state.queued.set(state.queued.get() - 1);
320    }
321}
322
323pub(super) async fn await_with_context<F: Future>(
324    driver: &DriverControl,
325    context: &InvocationContext,
326    future: F,
327) -> Result<F::Output, RuntimeFailure> {
328    ensure_context_active(driver, context)?;
329
330    let work = future.fuse();
331    let cancellation = context.cancellation.cancelled().fuse();
332    let deadline: LocalBoxFuture<'static, ()> = context.deadline().map_or_else(
333        || Box::pin(pending::<()>()) as LocalBoxFuture<'static, ()>,
334        |deadline| (driver.sleep_until)(deadline),
335    );
336    let deadline = deadline.fuse();
337    futures::pin_mut!(work, cancellation, deadline);
338
339    match select(select(work, cancellation), deadline).await {
340        Either::Left((Either::Left((output, _)), _)) => Ok(output),
341        Either::Left((Either::Right(((), _)), _)) => Err(RuntimeFailure::Cancelled {
342            request_id: context.request_id(),
343        }),
344        Either::Right(((), _)) => Err(RuntimeFailure::DeadlineExceeded {
345            request_id: context.request_id(),
346        }),
347    }
348}
349
350pub(super) async fn await_with_generation_context<F: Future>(
351    driver: &DriverControl,
352    context: &InvocationContext,
353    generation_cancellation: CancellationToken,
354    capability: &'static str,
355    future: F,
356) -> Result<F::Output, RuntimeFailure> {
357    ensure_context_active(driver, context)?;
358    if generation_cancellation.is_cancelled() {
359        return Err(RuntimeFailure::Unavailable { capability });
360    }
361
362    let work = future.fuse();
363    let cancellation = context.cancellation.cancelled().fuse();
364    let generation_cancellation = generation_cancellation.cancelled().fuse();
365    let deadline: LocalBoxFuture<'static, ()> = context.deadline().map_or_else(
366        || Box::pin(pending::<()>()) as LocalBoxFuture<'static, ()>,
367        |deadline| (driver.sleep_until)(deadline),
368    );
369    let deadline = deadline.fuse();
370    futures::pin_mut!(work, cancellation, generation_cancellation, deadline);
371
372    match select(
373        select(select(work, cancellation), generation_cancellation),
374        deadline,
375    )
376    .await
377    {
378        Either::Left((Either::Left((Either::Left((output, _)), _)), _)) => Ok(output),
379        Either::Left((Either::Left((Either::Right(((), _)), _)), _)) => {
380            Err(RuntimeFailure::Cancelled {
381                request_id: context.request_id(),
382            })
383        }
384        Either::Left((Either::Right(((), _)), _)) => {
385            Err(RuntimeFailure::Unavailable { capability })
386        }
387        Either::Right(((), _)) => Err(RuntimeFailure::DeadlineExceeded {
388            request_id: context.request_id(),
389        }),
390    }
391}
392
393pub(super) fn is_module_failure(error: &RuntimeFailure) -> bool {
394    matches!(error, RuntimeFailure::ModuleFailure { .. })
395}
396
397pub(super) fn schedule_module_supervision_after_failure(
398    runtime: &Rc<NativeAppRuntime>,
399    instance_key: &str,
400    error: RuntimeFailure,
401) -> RuntimeFailure {
402    if is_module_failure(&error)
403        && begin_module_supervision(runtime, instance_key).unwrap_or(false)
404        && let Err(schedule_error) = schedule_module_supervision(runtime, instance_key)
405    {
406        return handle_supervision_schedule_failure(runtime, instance_key, schedule_error);
407    }
408    error
409}
410
411pub(super) fn handle_supervision_schedule_failure(
412    runtime: &Rc<NativeAppRuntime>,
413    instance_key: &str,
414    error: RuntimeFailure,
415) -> RuntimeFailure {
416    let must_fail = runtime
417        .supervision
418        .borrow()
419        .get(instance_key)
420        .is_some_and(|state| state.criticality.is_critical() || state.required_path);
421    if must_fail {
422        runtime.terminal_failure.replace(Some(error.clone()));
423        runtime.begin_shutdown();
424    }
425    error
426}
427
428pub(super) fn ensure_context_active(
429    driver: &DriverControl,
430    context: &InvocationContext,
431) -> Result<(), RuntimeFailure> {
432    if context.is_cancelled() {
433        return Err(RuntimeFailure::Cancelled {
434            request_id: context.request_id(),
435        });
436    }
437    if context.is_expired((driver.now)()) {
438        return Err(RuntimeFailure::DeadlineExceeded {
439            request_id: context.request_id(),
440        });
441    }
442    Ok(())
443}
444
445/// The result of bounded cleanup after a graceful shutdown request.
446#[derive(Clone, Debug, Eq, PartialEq)]
447pub enum ShutdownOutcome {
448    /// Every managed task, resource, and Module generation was cleaned up.
449    Clean,
450    /// Cleanup completed but a Module or resource reported a Runtime Failure.
451    RuntimeFailure { error: RuntimeFailure },
452    /// The global shutdown deadline expired; remaining work was terminated.
453    Timeout,
454}
455
456/// A successful terminal result returned to the embedding Runner.
457#[derive(Clone, Debug, Eq, PartialEq)]
458pub enum TerminalOutcome {
459    /// The App completed a bounded clean shutdown.
460    CleanShutdown,
461    /// The App could not start because a Module reported a startup failure.
462    StartupFailure { error: RuntimeFailure },
463    /// The running App reported a Runtime Failure during terminal cleanup.
464    RuntimeFailure { error: RuntimeFailure },
465    /// The running App failed and cleanup reported a second Runtime Failure.
466    RuntimeFailureDuringShutdown {
467        error: RuntimeFailure,
468        cleanup_error: RuntimeFailure,
469    },
470    /// The running App failed and cleanup then exceeded its global deadline.
471    RuntimeFailureWithShutdownTimeout { error: RuntimeFailure },
472    /// The App exceeded its one global shutdown deadline.
473    ShutdownTimeout,
474}
lenso_kernel/driver.rs

lenso_kernel/
driver.rs
