simu/process.rs
1// SPDX-FileCopyrightText: Copyright (c) Siemens 2026 contributed by Christoph Kuhmuench christoph.kuhmuench@gmail.com
2//
3// SPDX-License-Identifier: MIT OR Apache-2.0
4
5//! Observable spawned processes.
6//!
7//! [`SimEnv::spawn`](crate::env::SimEnv::spawn) and
8//! [`EnvHandle::spawn`](crate::env::EnvHandle::spawn) return a
9//! [`ProcessHandle<T>`] that implements `Future<Output = T>`. Awaiting the
10//! handle suspends the calling process until the spawned process finishes
11//! and yields its return value.
12//!
13//! The handle is **not `Clone`** — a single awaiter per process, matching
14//! `tokio::JoinHandle`. Broadcast/completion-signal patterns already have
15//! [`EventTrigger`](crate::EventTrigger).
16
17use std::cell::RefCell;
18use std::future::Future;
19use std::pin::Pin;
20use std::rc::Rc;
21use std::task::{Context, Poll, Waker};
22
23/// Shared slot between the spawn-wrapper and the handle.
24struct ProcessSlot<T> {
25 result: Option<T>,
26 waker: Option<Waker>,
27}
28
29/// Handle to a spawned process. Resolves to the process's return value.
30///
31/// Dropping the handle before awaiting detaches the process — it continues to
32/// run; its return value, if any, is dropped when the process completes.
33///
34/// ```
35/// use simu::SimEnv;
36///
37/// let mut env = SimEnv::with_seed(0);
38/// let h = env.handle();
39/// env.spawn(async move {
40/// let hc = h.clone();
41/// let child = h.spawn(async move {
42/// hc.timeout(3.0).await;
43/// "charged" // the child's return value
44/// });
45/// let result = child.await; // suspend until the child finishes
46/// assert_eq!(result, "charged");
47/// assert_eq!(h.now(), 3.0);
48/// });
49/// env.run();
50/// ```
51pub struct ProcessHandle<T> {
52 slot: Rc<RefCell<ProcessSlot<T>>>,
53}
54
55impl<T> std::fmt::Debug for ProcessHandle<T> {
56 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57 let mut d = f.debug_struct("ProcessHandle");
58 if let Ok(slot) = self.slot.try_borrow() {
59 d.field("ready", &slot.result.is_some());
60 }
61 d.finish_non_exhaustive()
62 }
63}
64
65impl<T: 'static> Future for ProcessHandle<T> {
66 type Output = T;
67
68 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
69 let mut slot = self.slot.borrow_mut();
70 if let Some(value) = slot.result.take() {
71 return Poll::Ready(value);
72 }
73 // Dedup the waker — avoid cloning on every re-poll from the same task.
74 let waker = cx.waker();
75 match &slot.waker {
76 Some(existing) if existing.will_wake(waker) => {}
77 _ => slot.waker = Some(waker.clone()),
78 }
79 Poll::Pending
80 }
81}
82
83impl<T: 'static> ProcessHandle<T> {
84 /// Await the handle and discard the return value.
85 ///
86 /// Useful with [`any_of!`](crate::any_of) / [`all_of!`](crate::all_of),
87 /// whose sub-futures must have `Output = ()`.
88 pub async fn discard(self) {
89 let _ = self.await;
90 }
91}
92
93/// Build the `(wrapper future, handle)` pair used by `EnvHandle::spawn`.
94///
95/// The wrapper type-erases `F::Output` into `()` so the process table can stay
96/// `HashMap<_, Pin<Box<dyn Future<Output = ()>>>>`. When the user's future
97/// resolves, the wrapper stores the value in the shared slot and wakes the
98/// handle's awaiter (if any).
99#[allow(clippy::type_complexity)]
100pub(crate) fn spawn_with_handle<F>(
101 future: F,
102) -> (Pin<Box<dyn Future<Output = ()>>>, ProcessHandle<F::Output>)
103where
104 F: Future + 'static,
105 F::Output: 'static,
106{
107 let slot = Rc::new(RefCell::new(ProcessSlot {
108 result: None,
109 waker: None,
110 }));
111 let slot_for_wrapper = Rc::clone(&slot);
112
113 let wrapped = async move {
114 let result = future.await;
115 let mut s = slot_for_wrapper.borrow_mut();
116 s.result = Some(result);
117 if let Some(w) = s.waker.take() {
118 w.wake();
119 }
120 };
121
122 (Box::pin(wrapped), ProcessHandle { slot })
123}