input_pipeline_dso/
dispatcher.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use core::task::Context;
6use fidl_next::{ClientEnd, ServerEnd};
7use futures::prelude::*;
8use futures::task::Poll;
9use pin_project_lite::pin_project;
10use std::pin::Pin;
11
12#[cfg(feature = "dso")]
13pub use dso::*;
14
15#[cfg(not(feature = "dso"))]
16pub use elf::*;
17
18pin_project! {
19    #[derive(Debug)]
20    #[must_use = "futures do nothing unless polled"]
21    pub struct OnTimeout<F, T, OT> {
22        #[pin]
23        timer: T,
24        #[pin]
25        future: F,
26        on_timeout: Option<OT>,
27    }
28}
29
30impl<F: Future, T, OT> Future for OnTimeout<F, T, OT>
31where
32    T: Future<Output = ()> + 'static,
33    OT: FnOnce() -> F::Output,
34{
35    type Output = F::Output;
36
37    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
38        let this = self.project();
39        if let Poll::Ready(item) = this.future.poll(cx) {
40            return Poll::Ready(item);
41        }
42        if let Poll::Ready(()) = this.timer.poll(cx) {
43            let ot = this.on_timeout.take().expect("polled with timeout after completion");
44            let item = (ot)();
45            return Poll::Ready(item);
46        }
47        Poll::Pending
48    }
49}
50
51/// A wrapper for a future which will complete with a provided closure when a timeout occurs. This
52/// is forked from [`fuchsia_async::OnTimeout`] because that has a fixed dependency on
53/// [`fuchsia_async::Timer`] which driver dispatcher does not support.
54pub trait TimeoutExt: Future + Sized {
55    fn on_timeout<T, OT>(self, timer: T, on_timeout: OT) -> OnTimeout<Self, T, OT>
56    where
57        T: Future<Output = ()> + 'static,
58        OT: FnOnce() -> Self::Output,
59    {
60        OnTimeout { timer, future: self, on_timeout: Some(on_timeout) }
61    }
62}
63
64impl<F: Future + Sized> TimeoutExt for F {}
65
66#[derive(Clone, Default)]
67pub struct Dispatcher {}
68
69mod dso {
70    #![cfg(feature = "dso")]
71
72    pub use super::*;
73    use fdf::{AsyncDispatcher, OnDriverDispatcher};
74    use libasync::DispatcherTimerExt;
75
76    #[derive(Debug, Clone, Copy, Default, Hash, PartialEq, Eq, PartialOrd, Ord)]
77    #[repr(transparent)]
78    pub struct MonotonicInstant(zx::MonotonicInstant);
79
80    impl From<zx::MonotonicInstant> for MonotonicInstant {
81        fn from(o: zx::MonotonicInstant) -> Self {
82            Self(o)
83        }
84    }
85
86    impl From<MonotonicInstant> for zx::MonotonicInstant {
87        fn from(o: MonotonicInstant) -> Self {
88            o.0
89        }
90    }
91
92    impl MonotonicInstant {
93        pub fn now() -> Self {
94            Self(zx::MonotonicInstant::get())
95        }
96
97        pub fn into_nanos(&self) -> i64 {
98            self.0.into_nanos()
99        }
100
101        pub fn into_zx(self) -> zx::MonotonicInstant {
102            self.0
103        }
104
105        pub fn after(duration: zx::MonotonicDuration) -> Self {
106            Self(zx::MonotonicInstant::after(duration))
107        }
108    }
109
110    pub type Transport = libasync_fidl::AsyncChannel<Dispatcher>;
111    pub type DriverTransport = fdf_fidl::DriverChannel<fdf::CurrentDispatcher>;
112
113    #[derive(Debug)]
114    enum TaskHandleInner<T> {
115        Join(::libasync::JoinHandle<T>),
116        Local(::libasync::JoinHandle<()>, std::rc::Rc<std::cell::RefCell<Option<T>>>),
117    }
118
119    #[derive(Debug)]
120    pub struct TaskHandle<T> {
121        inner: Option<TaskHandleInner<T>>,
122        detached: bool,
123    }
124
125    impl<T> Drop for TaskHandle<T> {
126        fn drop(&mut self) {
127            if !self.detached {
128                if let Some(inner) = self.inner.as_mut() {
129                    match inner {
130                        TaskHandleInner::Join(h) => {
131                            _ = h.abort();
132                        }
133                        TaskHandleInner::Local(h, _) => {
134                            _ = h.abort();
135                        }
136                    }
137                }
138            }
139        }
140    }
141
142    impl TaskHandle<()> {
143        pub fn detach(mut self) {
144            self.detached = true;
145        }
146    }
147
148    impl<T: 'static> Future for TaskHandle<T> {
149        type Output = T;
150
151        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
152            match self.inner.as_mut().unwrap() {
153                TaskHandleInner::Join(h) => match h.poll_unpin(cx) {
154                    Poll::Pending => Poll::Pending,
155                    Poll::Ready(Ok(t)) => Poll::Ready(t),
156                    Poll::Ready(Err(e)) => panic!("TaskHandle: polled unexpected error {e:?}"),
157                },
158                TaskHandleInner::Local(h, result) => match h.poll_unpin(cx) {
159                    Poll::Pending => Poll::Pending,
160                    Poll::Ready(Ok(())) => {
161                        let res = result
162                            .borrow_mut()
163                            .take()
164                            .expect("TaskHandle completed but result missing");
165                        Poll::Ready(res)
166                    }
167                    Poll::Ready(Err(e)) => panic!("TaskHandle: polled unexpected error {e:?}"),
168                },
169            }
170        }
171    }
172
173    impl Dispatcher {
174        #[must_use]
175        pub fn spawn_local(future: impl Future<Output = ()> + 'static) -> TaskHandle<()>
176        where
177            Self: 'static,
178        {
179            // This should never panic if the dispatcher is valid.
180            TaskHandle {
181                inner: Some(TaskHandleInner::Join(fdf::CurrentDispatcher.spawn_local(future))),
182                detached: false,
183            }
184        }
185
186        pub fn after_deadline(deadline: MonotonicInstant) -> impl Future<Output = ()> + 'static {
187            let f = fdf::CurrentDispatcher.after_deadline(deadline.into());
188            async move {
189                // This should never panic if the dispatcher is valid.
190                f.await.expect("Dispatcher::after_deadline");
191            }
192        }
193
194        pub fn client_from_zx_channel<P>(
195            client_end: ClientEnd<P, zx::Channel>,
196        ) -> ClientEnd<P, Transport> {
197            libasync_fidl::AsyncChannel::<Dispatcher>::client_from_zx_channel(client_end)
198        }
199        pub fn server_from_zx_channel<P>(
200            server_end: ServerEnd<P, zx::Channel>,
201        ) -> ServerEnd<P, Transport> {
202            libasync_fidl::AsyncChannel::<Dispatcher>::server_from_zx_channel(server_end)
203        }
204    }
205
206    #[derive(Clone, Copy, Default, Debug)]
207    pub struct LocalDriverExecutor;
208
209    impl fidl_next::Executor for LocalDriverExecutor {
210        type JoinHandle<T: 'static> = TaskHandle<T>;
211
212        fn spawn<F>(&self, future: F) -> Self::JoinHandle<F::Output>
213        where
214            F: Future + Send + 'static,
215            F::Output: Send + 'static,
216        {
217            use fdf::OnDispatcher;
218            TaskHandle {
219                inner: Some(TaskHandleInner::Join(
220                    fdf::CurrentDispatcher.compute(future).detach_on_drop(),
221                )),
222                detached: false,
223            }
224        }
225    }
226
227    impl fidl_next::LocalExecutor for LocalDriverExecutor {
228        fn spawn_local<F>(&self, future: F) -> Self::JoinHandle<F::Output>
229        where
230            F: Future + 'static,
231            F::Output: 'static,
232        {
233            use fdf::OnDriverDispatcher;
234            let result = std::rc::Rc::new(std::cell::RefCell::new(None));
235            let result_clone = result.clone();
236            let handle = fdf::CurrentDispatcher.spawn_local(async move {
237                *result_clone.borrow_mut() = Some(future.await);
238            });
239            TaskHandle { inner: Some(TaskHandleInner::Local(handle, result)), detached: false }
240        }
241    }
242
243    impl fidl_next::RunsTransport<Transport> for LocalDriverExecutor {}
244
245    impl fdf::GetAsyncDispatcher for Dispatcher {
246        fn try_get_async_dispatcher(&self) -> Option<AsyncDispatcher> {
247            fdf::CurrentDispatcher.try_get_async_dispatcher()
248        }
249    }
250}
251
252mod elf {
253    #![cfg(not(feature = "dso"))]
254
255    pub use super::*;
256
257    pub type MonotonicInstant = fuchsia_async::MonotonicInstant;
258
259    pub type Transport = zx::Channel;
260
261    #[derive(Debug)]
262    pub struct TaskHandle<T>(fuchsia_async::Task<T>);
263
264    impl TaskHandle<()> {
265        pub fn detach(self) {
266            self.0.detach();
267        }
268    }
269
270    #[cfg(test)]
271    impl<T: 'static> From<fuchsia_async::Task<T>> for TaskHandle<T> {
272        fn from(task: fuchsia_async::Task<T>) -> Self {
273            Self(task)
274        }
275    }
276
277    impl<T: 'static> Future for TaskHandle<T> {
278        type Output = T;
279
280        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
281            match self.0.poll_unpin(cx) {
282                Poll::Ready(t) => Poll::Ready(t),
283                Poll::Pending => Poll::Pending,
284            }
285        }
286    }
287
288    impl Dispatcher {
289        #[must_use]
290        pub fn spawn_local(future: impl Future<Output = ()> + 'static) -> TaskHandle<()>
291        where
292            Self: 'static,
293        {
294            TaskHandle(fuchsia_async::Task::local(future))
295        }
296
297        pub fn after_deadline(deadline: MonotonicInstant) -> impl Future<Output = ()> + 'static {
298            fuchsia_async::Timer::new(deadline)
299        }
300
301        pub fn client_from_zx_channel<P>(
302            client_end: fidl_next::ClientEnd<P, zx::Channel>,
303        ) -> ClientEnd<P, Transport> {
304            client_end
305        }
306        pub fn server_from_zx_channel<P>(
307            server_end: ServerEnd<P, zx::Channel>,
308        ) -> ServerEnd<P, Transport> {
309            server_end
310        }
311    }
312
313    #[derive(Clone, Copy, Default, Debug)]
314    pub struct LocalDriverExecutor;
315
316    impl fidl_next::Executor for LocalDriverExecutor {
317        type JoinHandle<T: 'static> = TaskHandle<T>;
318
319        fn spawn<F>(&self, future: F) -> Self::JoinHandle<F::Output>
320        where
321            F: Future + Send + 'static,
322            F::Output: Send + 'static,
323        {
324            use fidl_next::LocalExecutor;
325            self.spawn_local(future)
326        }
327    }
328
329    impl fidl_next::LocalExecutor for LocalDriverExecutor {
330        fn spawn_local<F>(&self, future: F) -> Self::JoinHandle<F::Output>
331        where
332            F: Future + 'static,
333            F::Output: 'static,
334        {
335            TaskHandle(fuchsia_async::Task::local(future))
336        }
337    }
338
339    impl fidl_next::RunsTransport<Transport> for LocalDriverExecutor {}
340}
input_pipeline_dso/dispatcher.rs

input_pipeline_dso/
dispatcher.rs
