rustdv_sim/
combinators.rs

1//! Ports of `First`/`Combine` (design-doc mapping row 16): future
2//! combinators with drop-based cancellation of the losers — the cleanup
3//! cocotb does manually with kill-on-completion tasks falls out of RAII.
4
5use std::fmt;
6use std::future::Future;
7use std::pin::Pin;
8use std::task::{Context, Poll};
9
10use crate::time::SimDuration;
11use crate::triggers::Timer;
12
13pub enum Either<A, B> {
14    First(A),
15    Second(B),
16}
17
18/// First of two futures; the loser is dropped (unsubscribing its trigger).
19///
20/// The futures are boxed behind lifetime `'a` rather than `'static` (D82), so
21/// a future that *borrows* — a component's `run` borrowing the tree, a
22/// sub-sequence borrowing its parent sequence — can be composed here. `'static`
23/// futures satisfy any `'a`, so every earlier caller is unaffected.
24pub struct First2<'a, A, B> {
25    a: Pin<Box<dyn Future<Output = A> + 'a>>,
26    b: Pin<Box<dyn Future<Output = B> + 'a>>,
27}
28
29pub fn first2<'a, FA, FB>(a: FA, b: FB) -> First2<'a, FA::Output, FB::Output>
30where
31    FA: Future + 'a,
32    FB: Future + 'a,
33{
34    First2 { a: Box::pin(a), b: Box::pin(b) }
35}
36
37impl<A, B> Future for First2<'_, A, B> {
38    type Output = Either<A, B>;
39    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
40        if let Poll::Ready(v) = self.a.as_mut().poll(cx) {
41            return Poll::Ready(Either::First(v));
42        }
43        if let Poll::Ready(v) = self.b.as_mut().poll(cx) {
44            return Poll::Ready(Either::Second(v));
45        }
46        Poll::Pending
47    }
48}
49
50/// Join of two futures (port of `Combine`; SystemVerilog's `fork...join`).
51pub struct Join2<'a, A, B> {
52    a: Pin<Box<dyn Future<Output = A> + 'a>>,
53    b: Pin<Box<dyn Future<Output = B> + 'a>>,
54    ra: Option<A>,
55    rb: Option<B>,
56}
57
58pub fn join2<'a, FA, FB>(a: FA, b: FB) -> Join2<'a, FA::Output, FB::Output>
59where
60    FA: Future + 'a,
61    FB: Future + 'a,
62{
63    Join2 { a: Box::pin(a), b: Box::pin(b), ra: None, rb: None }
64}
65
66impl<A, B> Future for Join2<'_, A, B> {
67    type Output = (A, B);
68    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<(A, B)> {
69        // Sound: the inner futures are boxed (their pinning is their own),
70        // and ra/rb are plain values we intentionally move on completion.
71        let this = unsafe { self.get_unchecked_mut() };
72        if this.ra.is_none() {
73            if let Poll::Ready(v) = this.a.as_mut().poll(cx) {
74                this.ra = Some(v);
75            }
76        }
77        if this.rb.is_none() {
78            if let Poll::Ready(v) = this.b.as_mut().poll(cx) {
79                this.rb = Some(v);
80            }
81        }
82        if this.ra.is_some() && this.rb.is_some() {
83            Poll::Ready((this.ra.take().unwrap(), this.rb.take().unwrap()))
84        } else {
85            Poll::Pending
86        }
87    }
88}
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91pub struct TimeoutError;
92
93impl fmt::Display for TimeoutError {
94    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
95        write!(f, "operation timed out")
96    }
97}
98impl std::error::Error for TimeoutError {}
99
100/// Run `fut` with a simulation-time timeout.
101pub async fn with_timeout<'a, F>(fut: F, d: SimDuration) -> Result<F::Output, TimeoutError>
102where
103    F: Future + 'a,
104{
105    match first2(fut, Timer::new(d)).await {
106        Either::First(v) => Ok(v),
107        Either::Second(()) => Err(TimeoutError),
108    }
109}
110
111/// Join *N* futures, where N is known only at run time (D82).
112///
113/// `join2` covers the fixed-arity case the `join!` macro expands to; this
114/// covers a `Vec` built at run time — a parent joining however many children
115/// it has, or a virtual sequence joining a list of sub-sequences. Every future
116/// is polled on each wake until all have completed; results come back in the
117/// original order. Like `Join2`, the futures may borrow (`'a`).
118pub struct JoinAll<'a, T> {
119    futs: Vec<Option<Pin<Box<dyn Future<Output = T> + 'a>>>>,
120    out: Vec<Option<T>>,
121}
122
123pub fn join_all<'a, T>(futs: Vec<Pin<Box<dyn Future<Output = T> + 'a>>>) -> JoinAll<'a, T> {
124    let n = futs.len();
125    JoinAll { futs: futs.into_iter().map(Some).collect(), out: (0..n).map(|_| None).collect() }
126}
127
128impl<T> Future for JoinAll<'_, T> {
129    type Output = Vec<T>;
130    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<T>> {
131        // Sound: the inner futures are boxed (they own their pinning), and
132        // `out` holds plain values we move out once every future is done.
133        let this = unsafe { self.get_unchecked_mut() };
134        let mut all_done = true;
135        for (i, slot) in this.futs.iter_mut().enumerate() {
136            if let Some(f) = slot {
137                match f.as_mut().poll(cx) {
138                    Poll::Ready(v) => {
139                        this.out[i] = Some(v);
140                        *slot = None; // drop the finished future
141                    }
142                    Poll::Pending => all_done = false,
143                }
144            }
145        }
146        if all_done {
147            Poll::Ready(this.out.iter_mut().map(|o| o.take().unwrap()).collect())
148        } else {
149            Poll::Pending
150        }
151    }
152}
153
154/// `first!(a, b, ...)` — first completed future wins; losers are dropped.
155#[macro_export]
156macro_rules! first {
157    ($a:expr, $b:expr $(,)?) => {
158        $crate::combinators::first2($a, $b)
159    };
160    ($a:expr, $b:expr, $($rest:expr),+ $(,)?) => {
161        $crate::combinators::first2($a, $crate::first!($b, $($rest),+))
162    };
163}
164
165/// `join!(a, b, ...)` — wait for all.
166#[macro_export]
167macro_rules! join {
168    ($a:expr, $b:expr $(,)?) => {
169        $crate::combinators::join2($a, $b)
170    };
171    ($a:expr, $b:expr, $($rest:expr),+ $(,)?) => {
172        $crate::combinators::join2($a, $crate::join!($b, $($rest),+))
173    };
174}
175
176// ===========================================================================
177// Tests — no simulator.
178// ===========================================================================
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183    use crate::testing::block_on;
184    use std::cell::RefCell;
185    use std::rc::Rc;
186
187    #[test]
188    fn join2_yields_both() {
189        block_on(async {
190            let (a, b) = join2(async { 1u8 }, async { "two" }).await;
191            assert_eq!(a, 1);
192            assert_eq!(b, "two");
193        });
194    }
195
196    #[test]
197    fn join_all_preserves_input_order() {
198        block_on(async {
199            let futs: Vec<Pin<Box<dyn Future<Output = u8>>>> =
200                vec![Box::pin(async { 1 }), Box::pin(async { 2 }), Box::pin(async { 3 })];
201            assert_eq!(join_all(futs).await, vec![1, 2, 3]);
202        });
203    }
204
205    #[test]
206    fn first2_returns_the_winner() {
207        block_on(async {
208            let ev = crate::sync::Event::new();
209            let waiter = ev.clone();
210            crate::executor::spawn(async move {
211                ev.set();
212            });
213            match first2(async { 7u8 }, async move { waiter.wait().await }).await {
214                Either::First(v) => assert_eq!(v, 7),
215                Either::Second(()) => panic!("the ready future should have won"),
216            }
217        });
218    }
219
220    /// D82c in miniature: **losing a race means being dropped.** Racing the
221    /// whole run tree instead of each component dropped the tree mid-phase
222    /// and a test passed with its scoreboard never running.
223    #[test]
224    fn first2_drops_the_loser() {
225        struct Tattle(Rc<RefCell<bool>>);
226        impl Drop for Tattle {
227            fn drop(&mut self) {
228                *self.0.borrow_mut() = true;
229            }
230        }
231
232        let dropped = Rc::new(RefCell::new(false));
233        let flag = dropped.clone();
234        block_on(async move {
235            let never = crate::sync::Event::new();
236            // The tattle is moved *into* the future from outside, so it is
237            // dropped when the future is dropped — whether or not the future
238            // ever got polled.
239            let tattle = Tattle(flag);
240            let loser = async move {
241                let _t = tattle;
242                never.wait().await;
243            };
244            let _ = first2(async { 1u8 }, loser).await;
245        });
246        assert!(*dropped.borrow(), "the losing future was dropped, not left running");
247    }
248
249    /// The whole point of D82: a joined future may **borrow**, so a
250    /// sub-sequence can use its parent's state. If this stops compiling, the
251    /// combinators have regained a `'static` bound.
252    #[test]
253    fn joined_futures_may_borrow() {
254        block_on(async {
255            let owned = vec![1u8, 2, 3];
256            let borrow_a = async { owned.len() };
257            let borrow_b = async { owned[0] as usize };
258            let (a, b) = join2(borrow_a, borrow_b).await;
259            assert_eq!((a, b), (3, 1));
260        });
261    }
262}
rustdv_sim/combinators.rs

rustdv_sim/
combinators.rs
