libasync_scope_dispatcher/
scope_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::ptr::{NonNull, null};
6use core::sync::atomic::{self, AtomicBool};
7use fuchsia_async::{EHandle, MonotonicInstant, Scope, Timer, WakeupTime};
8use fuchsia_sync::Mutex;
9use futures::task::AtomicWaker;
10use libasync_dispatcher::{AsAsyncDispatcherRef, AsyncDispatcherRef};
11use libasync_sys::async_dispatcher_t;
12use pin_project_lite::pin_project;
13use std::pin::Pin;
14use std::sync::Arc;
15use std::task::{Context, Poll};
16use zx::Status;
17
18use crate::ops;
19use crate::ops::v1::{PendingWaits, Task, TaskQueue};
20use crate::ops::v2::PendingIrqs;
21
22/// Implements a C++-compatible [`async_dispatcher_t`] around a [`fuchsia_async::Scope`].
23#[derive(Debug)]
24#[repr(C)]
25pub struct ScopeDispatcher {
26    // Safety Note: this must go first in this struct for the callbacks to work correctly.
27    dispatcher: async_dispatcher_t,
28    task_queue: Mutex<TaskQueue>,
29    pub(crate) pending_waits: Mutex<PendingWaits>,
30    pub(crate) pending_irqs: Mutex<PendingIrqs>,
31    shutting_down: AtomicBool,
32    shutdown_complete_waker: AtomicWaker,
33    service_waker: AtomicWaker,
34    shutdown_guard: AtomicBool,
35    executor: EHandle,
36    scope: Scope,
37}
38
39// SAFETY: The API of async_dispatcher_t is expected to be thread safe.
40unsafe impl Send for ScopeDispatcher {}
41// SAFETY: The API of async_dispatcher_t is expected to be thread safe.
42unsafe impl Sync for ScopeDispatcher {}
43
44impl ScopeDispatcher {
45    /// Creates a new [`ScopeDispatcher`] on the currently running fuchsia-async executor.
46    ///
47    /// # Panics
48    ///
49    /// Panics if this is not run on a fuchsia-async executor context.
50    pub fn new() -> Arc<Self> {
51        Self::new_on_executor(EHandle::local())
52    }
53
54    /// Creates a new [`ScopeDispatcher`] with a new [`Scope`] on the given `executor`.
55    pub fn new_on_executor(executor: EHandle) -> Arc<Self> {
56        let scope = executor.global_scope().new_child();
57        let scope_handle = scope.as_handle().clone();
58        let dispatcher = async_dispatcher_t { ops: &ops::ASYNC_OPS };
59        let task_queue = Mutex::new(TaskQueue::default());
60        let pending_waits = Mutex::new(PendingWaits::default());
61        let pending_irqs = Mutex::new(PendingIrqs::default());
62        let shutting_down = AtomicBool::new(false);
63        let service_waker = AtomicWaker::new();
64        let shutdown_complete_waker = AtomicWaker::new();
65        let shutdown_guard = AtomicBool::new(false);
66        let this = Arc::new(Self {
67            dispatcher,
68            task_queue,
69            pending_waits,
70            pending_irqs,
71            shutting_down,
72            shutdown_complete_waker,
73            service_waker,
74            shutdown_guard,
75            executor,
76            scope,
77        });
78
79        scope_handle.spawn(this.clone().service_loop());
80
81        this
82    }
83
84    /// Get the pointer to the dispatcher callback struct for passing through FFI layers.
85    pub fn as_ptr(&self) -> *const async_dispatcher_t {
86        (self as *const Self).cast()
87    }
88
89    /// Gets the global executor handle for this dispatcher.
90    pub fn global_handle(&self) -> &EHandle {
91        &self.executor
92    }
93
94    /// Gets the [`fuchsia_async::Scope`] of this dispatcher.
95    pub fn as_scope(&self) -> &Scope {
96        &self.scope
97    }
98
99    /// Returns true if the dispatcher is currently shutting down.
100    pub fn is_shutting_down(&self) -> bool {
101        self.shutting_down.load(atomic::Ordering::Acquire)
102    }
103
104    /// Starts the dispatcher shutdown. Resolves when all outstanding tasks have been completed or
105    /// canceled.
106    pub fn shutdown(&self) -> ShutdownCompletionFuture<'_> {
107        // note: we might want to do more to prevent multiple attempts to shut the dispatcher down.
108        self.shutting_down.store(true, atomic::Ordering::Release);
109        self.service_waker.wake();
110        ShutdownCompletionFuture(self)
111    }
112
113    /// Gets the Scope from a dispatcher pointer. Used in the callbacks.
114    ///
115    /// # Safety
116    ///
117    /// The caller must ensure that the dispatcher pointer is a valid pointer originally obtained
118    /// through [`ScopeDispatcher::as_ptr`], and is still alive.
119    pub(crate) unsafe fn from_ptr<'a>(dispatcher_ptr: *mut async_dispatcher_t) -> &'a Self {
120        let this = dispatcher_ptr.cast::<ScopeDispatcher>();
121        // Safety: the caller promises that this is a valid pointer to what was originally a
122        // ScopeDispatcher object.
123        unsafe { this.as_ref() }.expect("null dispatcher pointer")
124    }
125
126    /// Gets the Scope from a dispatcher pointer. Used in the callbacks.
127    ///
128    /// # Safety
129    ///
130    /// The caller must ensure that the dispatcher pointer is a valid pointer originally obtained
131    /// through [`ScopeDispatcher::as_ptr`].
132    pub(crate) unsafe fn arc_from_ptr(dispatcher_ptr: *mut async_dispatcher_t) -> Arc<Self> {
133        let this = dispatcher_ptr.cast::<ScopeDispatcher>();
134        // Safety: the caller promises that this is a valid pointer to what was originally a
135        // ScopeDispatcher object.
136        unsafe {
137            Arc::increment_strong_count(this);
138            Arc::from_raw(this)
139        }
140    }
141
142    /// Posts a task to the dispatcher
143    pub(crate) fn post_task(&self, task: Task) -> Result<(), Status> {
144        // don't queue new tasks if we're shutting down.
145        if self.is_shutting_down() {
146            return Err(Status::BAD_STATE);
147        }
148        self.task_queue.lock().queue_task(task);
149        self.service_waker.wake();
150        Ok(())
151    }
152
153    /// Cancels a task queued on the dispatcher
154    pub(crate) fn cancel_task(&self, task: Task) -> Result<(), Status> {
155        // If we succeed at cancelling, we won't call the callback so this can be fairly simple.
156        if self.task_queue.lock().take_pending_task(&task).is_some() {
157            self.service_waker.wake();
158            Ok(())
159        } else {
160            Err(Status::NOT_FOUND)
161        }
162    }
163
164    async fn service_loop(self: Arc<Self>) {
165        while let Some(next_task) = NextTaskFuture::new(&self).await {
166            next_task.run(self.clone(), Ok(()));
167        }
168        // we're shutting down, so drain the queues of all outstanding tasks with
169        // a status of CANCELED. Note that we don't really care about fanning these out to all
170        // threads, so we just run them directly here.
171        // Note also that we will not allow any new items to be added to the queues after the
172        // shutdown flag has been set, so we don't have to worry about new things being added at
173        // this point.
174        let waits_to_cancel = self.pending_waits.lock().drain_all();
175        for next_wait in waits_to_cancel {
176            next_wait.run(self.clone(), null(), Err(Status::CANCELED));
177        }
178        let irqs_to_cancel = self.pending_irqs.lock().drain_all();
179        for irq in irqs_to_cancel {
180            irq.run(self.clone(), null(), Err(Status::CANCELED));
181        }
182        while let Some(next_task) = self.task_queue.lock().next_task(MonotonicInstant::INFINITE) {
183            next_task.run(self.clone(), Err(Status::CANCELED));
184        }
185        self.shutdown_guard.store(true, atomic::Ordering::Release);
186        self.shutdown_complete_waker.wake();
187    }
188}
189
190impl AsAsyncDispatcherRef for ScopeDispatcher {
191    fn as_async_dispatcher_ref(&self) -> AsyncDispatcherRef<'_> {
192        // SAFETY: We know this pointer is valid because it is a member of `&self`, which is a valid
193        // reference.
194        let ptr = unsafe { NonNull::new_unchecked(self.as_ptr().cast_mut()) };
195        // SAFETY: The dispatcher ref's lifetime is tied to `self`, of which the dispatcher
196        // structure and callbacks are members, so will not outlive them.
197        unsafe { AsyncDispatcherRef::from_raw(ptr) }
198    }
199}
200
201impl Drop for ScopeDispatcher {
202    fn drop(&mut self) {
203        assert!(
204            self.shutdown_guard.load(atomic::Ordering::Acquire),
205            "Dispatcher not properly shut down before dropping. Call ScopeDispatcher::shutdown()."
206        );
207    }
208}
209
210/// A future which resolves when the dispatcher has been shut down by [`ScopeDispatcher::shutdown`].
211#[must_use = "a future that is never awaited on will never run"]
212pub struct ShutdownCompletionFuture<'a>(&'a ScopeDispatcher);
213
214impl<'a> Future for ShutdownCompletionFuture<'a> {
215    type Output = ();
216
217    fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
218        self.0.shutdown_complete_waker.register(ctx.waker());
219        if self.0.shutdown_guard.load(atomic::Ordering::Acquire) {
220            Poll::Ready(())
221        } else {
222            Poll::Pending
223        }
224    }
225}
226
227pin_project! {
228    #[must_use = "a future that is never awaited on will never run"]
229    struct NextTaskFuture<'a> {
230        dispatcher: &'a Arc<ScopeDispatcher>,
231        #[pin]
232        next_timeout: Timer,
233    }
234}
235
236impl<'a> NextTaskFuture<'a> {
237    fn new(dispatcher: &'a Arc<ScopeDispatcher>) -> Self {
238        let next_timeout = MonotonicInstant::INFINITE.into_timer();
239        Self { dispatcher, next_timeout }
240    }
241}
242
243impl<'a> Future for NextTaskFuture<'a> {
244    type Output = Option<Task>;
245
246    fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
247        let mut task_queue = self.dispatcher.task_queue.lock();
248        // if we are shutting down, the service handler will do the work of canceling the remaining
249        // tasks, so return None to indicate that it should start doing that.
250        if self.dispatcher.shutting_down.load(atomic::Ordering::Acquire) {
251            return Poll::Ready(None);
252        }
253        let now = self.dispatcher.executor.now();
254        if let Some(task) = task_queue.next_task(now) {
255            Poll::Ready(Some(task))
256        } else {
257            let next_deadline = if let Some(task) = task_queue.peek_next_task() {
258                task.deadline().unwrap_or(MonotonicInstant::INFINITE)
259            } else {
260                MonotonicInstant::INFINITE
261            };
262            self.dispatcher.service_waker.register(ctx.waker());
263            let mut this = self.project();
264            this.next_timeout.as_mut().reset(next_deadline);
265            // Note that we don't really care about resolving the timer, we just want to use its
266            // waker to re-awaken this future when we're ready.
267            let _: Poll<()> = this.next_timeout.poll(ctx);
268            Poll::Pending
269        }
270    }
271}
libasync_scope_dispatcher/scope_dispatcher.rs

libasync_scope_dispatcher/
scope_dispatcher.rs
