trillium_server_common/
runtime.rs

1use futures_lite::Stream;
2use std::{
3    fmt::{self, Debug, Formatter},
4    future::Future,
5    pin::Pin,
6    sync::Arc,
7    task::{Context, Poll, ready},
8    time::Duration,
9};
10
11mod droppable_future;
12pub use droppable_future::DroppableFuture;
13
14mod runtime_trait;
15pub use runtime_trait::RuntimeTrait;
16
17mod fan_out;
18pub use fan_out::FanOut;
19
20mod object_safe_runtime;
21use object_safe_runtime::ObjectSafeRuntime;
22
23/// A type-erased [`RuntimeTrait`] implementation. Think of this as an `Arc<dyn RuntimeTrait>`
24#[derive(Clone)]
25pub struct Runtime(Arc<dyn ObjectSafeRuntime>);
26
27impl Debug for Runtime {
28    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
29        f.debug_tuple("Runtime").field(&format_args!("..")).finish()
30    }
31}
32
33impl<R: RuntimeTrait> From<Arc<R>> for Runtime {
34    fn from(value: Arc<R>) -> Self {
35        Self(value)
36    }
37}
38
39impl Runtime {
40    /// Construct a new type-erased runtime object from any [`RuntimeTrait`] implementation.
41    pub fn new(runtime: impl RuntimeTrait) -> Self {
42        runtime.into() // we avoid re-arcing a Runtime by using Into::into
43    }
44
45    /// Spawn a future on the runtime, returning a future that has detach-on-drop semantics
46    ///
47    /// Spawned tasks conform to the following behavior:
48    ///
49    /// * detach on drop: If the returned [`DroppableFuture`] is dropped immediately, the task will
50    ///   continue to execute until completion.
51    ///
52    /// * unwinding: If the spawned future panics, this must not propagate to the join handle.
53    ///   Instead, the awaiting the join handle returns None in case of panic.
54    pub fn spawn<Output: Send + 'static>(
55        &self,
56        fut: impl Future<Output = Output> + Send + 'static,
57    ) -> DroppableFuture<Pin<Box<dyn Future<Output = Option<Output>> + Send + 'static>>> {
58        let fut = RuntimeTrait::spawn(self, fut).into_inner();
59        DroppableFuture::new(Box::pin(fut))
60    }
61
62    /// Spawn a future on the runtime without a join handle.
63    ///
64    /// Cheaper than [`spawn`][Self::spawn] when the caller doesn't need the output or
65    /// completion signal: no channel or join-handle allocation is made.
66    pub fn spawn_detached<Fut>(&self, fut: Fut)
67    where
68        Fut: Future<Output = ()> + Send + 'static,
69    {
70        self.0.spawn_detached(Box::pin(fut));
71    }
72
73    /// Wake in this amount of wall time
74    pub async fn delay(&self, duration: Duration) {
75        RuntimeTrait::delay(self, duration).await
76    }
77
78    /// Returns a [`Stream`] that yields a `()` on the provided period
79    pub fn interval(&self, period: Duration) -> impl Stream<Item = ()> + Send + '_ {
80        RuntimeTrait::interval(self, period)
81    }
82
83    /// Runtime implementation hook for blocking on a top level future.
84    pub fn block_on<Fut>(&self, fut: Fut) -> Fut::Output
85    where
86        Fut: Future,
87    {
88        RuntimeTrait::block_on(self, fut)
89    }
90
91    /// Race a future against the provided duration, returning None in case of timeout.
92    pub async fn timeout<Fut>(&self, duration: Duration, fut: Fut) -> Option<Fut::Output>
93    where
94        Fut: Future + Send,
95        Fut::Output: Send + 'static,
96    {
97        RuntimeTrait::timeout(self, duration, fut).await
98    }
99}
100
101impl RuntimeTrait for Runtime {
102    async fn delay(&self, duration: Duration) {
103        self.0.delay(duration).await
104    }
105
106    fn interval(&self, period: Duration) -> impl Stream<Item = ()> + Send + 'static {
107        self.0.interval(period)
108    }
109
110    fn spawn<Fut>(
111        &self,
112        fut: Fut,
113    ) -> DroppableFuture<impl Future<Output = Option<Fut::Output>> + Send + 'static>
114    where
115        Fut: Future + Send + 'static,
116        Fut::Output: Send + 'static,
117    {
118        let (send, receive) = async_channel::bounded(1);
119        let spawn_fut = self.0.spawn(Box::pin(SendOnComplete { fut, send }));
120        DroppableFuture::new(Box::pin(async move {
121            spawn_fut.await;
122            receive.try_recv().ok()
123        }))
124    }
125
126    fn spawn_detached<Fut>(&self, fut: Fut)
127    where
128        Fut: Future<Output = ()> + Send + 'static,
129    {
130        self.0.spawn_detached(Box::pin(fut));
131    }
132
133    fn block_on<Fut>(&self, fut: Fut) -> Fut::Output
134    where
135        Fut: Future,
136    {
137        let (send, receive) = std::sync::mpsc::channel();
138        self.0.block_on(Box::pin(async move {
139            let _ = send.send(fut.await);
140        }));
141        receive.recv().unwrap()
142    }
143
144    fn hook_signals(
145        &self,
146        signals: impl IntoIterator<Item = i32>,
147    ) -> impl Stream<Item = i32> + Send + 'static {
148        self.0.hook_signals(signals.into_iter().collect())
149    }
150}
151
152pin_project_lite::pin_project! {
153    /// Sends the inner future's output on completion. A hand-written combinator rather than an
154    /// `async` block because a generator that captures `fut` and awaits it stores the future
155    /// twice (capture slot + await slot), doubling the task allocation for every spawn.
156    struct SendOnComplete<Fut: Future> {
157        #[pin]
158        fut: Fut,
159        send: async_channel::Sender<Fut::Output>,
160    }
161}
162
163impl<Fut: Future> Future for SendOnComplete<Fut> {
164    type Output = ();
165
166    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
167        let this = self.project();
168        let output = ready!(this.fut.poll(cx));
169        let _ = this.send.try_send(output);
170        Poll::Ready(())
171    }
172}
trillium_server_common/runtime.rs

trillium_server_common/
runtime.rs
