fuchsia_async/runtime/fuchsia/executor/
send.rs

1// Copyright 2021 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 super::common::{Executor, ExecutorTime};
6use super::scope::ScopeHandle;
7use fuchsia_sync::{Condvar, Mutex};
8
9use crate::runtime::instrument::TaskInstrument;
10use futures::FutureExt;
11use std::future::Future;
12use std::sync::Arc;
13use std::sync::atomic::Ordering;
14use std::time::Duration;
15use std::{fmt, thread};
16
17/// A multi-threaded port-based executor for Fuchsia. Requires that tasks scheduled on it
18/// implement `Send` so they can be load balanced between worker threads.
19///
20/// Having a `SendExecutor` in scope allows the creation and polling of zircon objects, such as
21/// [`fuchsia_async::Channel`].
22///
23/// # Panics
24///
25/// `SendExecutor` will panic on drop if any zircon objects attached to it are still alive. In other
26/// words, zircon objects backed by a `SendExecutor` must be dropped before it.
27pub struct SendExecutor {
28    /// The inner executor state.
29    inner: Arc<Executor>,
30    // LINT.IfChange
31    /// The root scope.
32    root_scope: ScopeHandle,
33    // LINT.ThenChange(//src/developer/debug/zxdb/console/commands/verb_async_backtrace.cc)
34    /// Worker thread handles
35    threads: Vec<thread::JoinHandle<()>>,
36    worker_init: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
37}
38
39impl fmt::Debug for SendExecutor {
40    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
41        f.debug_struct("SendExecutor").field("port", &self.inner.port).finish()
42    }
43}
44
45impl SendExecutor {
46    fn new_inner(
47        num_threads: u8,
48        worker_init: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
49        instrument: Option<Arc<dyn TaskInstrument>>,
50        allow_interrupts: bool,
51    ) -> Self {
52        let port = if allow_interrupts {
53            zx::Port::create_with_opts(zx::PortOptions::BIND_TO_INTERRUPT)
54        } else {
55            zx::Port::create()
56        };
57        let inner = Arc::new(Executor::new_with_port(
58            ExecutorTime::RealTime,
59            /* is_local */ false,
60            num_threads,
61            port,
62            instrument,
63        ));
64        let root_scope = ScopeHandle::root(inner.clone());
65        Executor::set_local(root_scope.clone());
66        Self { inner, root_scope, threads: Vec::default(), worker_init }
67    }
68
69    /// Get a reference to the Fuchsia `zx::Port` being used to listen for events.
70    pub fn port(&self) -> &zx::Port {
71        &self.inner.port
72    }
73
74    /// Run `future` to completion, using this thread and `num_threads` workers in a pool to
75    /// poll active tasks.
76    // The debugger looks for this function on the stack, so if its (fully-qualified) name changes,
77    // the debugger needs to be updated.
78    // LINT.IfChange
79    pub fn run<F>(&mut self, future: F) -> F::Output
80    // LINT.ThenChange(//src/developer/debug/zxdb/console/commands/verb_async_backtrace.cc)
81    where
82        F: Future + Send + 'static,
83        F::Output: Send + 'static,
84    {
85        assert!(self.inner.is_real_time(), "Error: called `run` on an executor using fake time");
86
87        let pair = Arc::new((Mutex::new(None), Condvar::new()));
88        let pair2 = pair.clone();
89
90        // Spawn a future which will set the result upon completion.
91        let task = self.root_scope.new_task(future.map(move |fut_result| {
92            let (lock, cvar) = &*pair2;
93            let mut result = lock.lock();
94            *result = Some(fut_result);
95            cvar.notify_one();
96        }));
97        task.detach();
98        assert!(self.root_scope.insert_task(task, false));
99
100        // Start worker threads, handing off timers from the current thread.
101        self.inner.done.store(false, Ordering::SeqCst);
102        self.create_worker_threads();
103
104        // Wait until the signal the future has completed.
105        let (lock, cvar) = &*pair;
106        let mut result = lock.lock();
107        if result.is_none() {
108            let mut last_polled = 0;
109            let mut last_tasks_ready = false;
110            loop {
111                // This timeout is chosen to be quite high since it impacts all processes that have
112                // multi-threaded async executors, and it exists to workaround arguably misbehaving
113                // users (see the comment below).
114                cvar.wait_for(&mut result, Duration::from_millis(250));
115                if result.is_some() {
116                    break;
117                }
118                let polled = self.inner.polled.load(Ordering::Relaxed);
119                let tasks_ready = !self.inner.ready_tasks.is_empty();
120                if polled == last_polled && last_tasks_ready && tasks_ready {
121                    // If this log message is printed, it most likely means that a task has blocked
122                    // making a reentrant synchronous call that doesn't involve a port message being
123                    // processed by this same executor. This can arise even if you would expect
124                    // there to normally be other port messages involved. One example (that has
125                    // actually happened): spawn a task to service a fuchsia.io connection, then try
126                    // and synchronously connect to that service. If the task hasn't had a chance to
127                    // run, then the async channel might not be registered with the executor, and so
128                    // sending messages to the channel doesn't trigger a port message. Typically,
129                    // the way to solve these issues is to run the service in a different executor
130                    // (which could be the same or a different process).
131                    eprintln!("Tasks might be stalled!");
132                    self.inner.wake_one_thread();
133                }
134                last_polled = polled;
135                last_tasks_ready = tasks_ready;
136            }
137        }
138
139        // Spin down worker threads
140        self.join_all();
141
142        // Unwrap is fine because of the check to `is_none` above.
143        result.take().unwrap()
144    }
145
146    #[doc(hidden)]
147    /// Returns the root scope of the executor.
148    pub fn root_scope(&self) -> &ScopeHandle {
149        &self.root_scope
150    }
151
152    /// Add `self.num_threads` worker threads to the executor's thread pool.
153    /// `timers`: timers from the "main" thread which would otherwise be lost.
154    fn create_worker_threads(&mut self) {
155        for _ in 0..self.inner.num_threads {
156            let inner = self.inner.clone();
157            let root_scope = self.root_scope.clone();
158            let worker_init = self.worker_init.clone();
159            let thread = thread::Builder::new()
160                .name("executor_worker".to_string())
161                .spawn(move || {
162                    Executor::set_local(root_scope);
163                    if let Some(init) = worker_init.as_ref() {
164                        init();
165                    }
166                    inner.worker_lifecycle::</* UNTIL_STALLED: */ false>(None);
167                })
168                .expect("must be able to spawn threads");
169            self.threads.push(thread);
170        }
171    }
172
173    fn join_all(&mut self) {
174        self.inner.mark_done();
175
176        // Join the worker threads
177        for thread in self.threads.drain(..) {
178            thread.join().expect("Couldn't join worker thread.");
179        }
180    }
181}
182
183impl Drop for SendExecutor {
184    fn drop(&mut self) {
185        self.join_all();
186        self.inner.on_parent_drop(&self.root_scope);
187    }
188}
189
190/// A builder for `SendExecutor`.
191#[derive(Default)]
192pub struct SendExecutorBuilder {
193    num_threads: Option<u8>,
194    worker_init: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
195    instrument: Option<Arc<dyn TaskInstrument>>,
196    allow_interrupts: bool,
197}
198
199impl SendExecutorBuilder {
200    /// Creates a new builder used for constructing a `SendExecutor`.
201    pub fn new() -> Self {
202        Self::default()
203    }
204
205    /// Sets the number of threads for the executor.
206    pub fn num_threads(mut self, num_threads: u8) -> Self {
207        self.num_threads = Some(num_threads);
208        self
209    }
210
211    /// Sets the worker initialization function.
212    pub fn worker_init(mut self, worker_init: impl Fn() + Send + Sync + 'static) -> Self {
213        self.worker_init = Some(Arc::new(worker_init));
214        self
215    }
216
217    /// Sets whether the executor should support binding interrupts.
218    pub fn allow_interrupts(mut self, allow_interrupts: bool) -> Self {
219        self.allow_interrupts = allow_interrupts;
220        self
221    }
222
223    /// Sets the instrumentation hook.
224    pub fn instrument(mut self, instrument: Option<Arc<dyn TaskInstrument>>) -> Self {
225        self.instrument = instrument;
226        self
227    }
228
229    /// Builds the `SendExecutor`, consuming this `SendExecutorBuilder`.
230    pub fn build(self) -> SendExecutor {
231        SendExecutor::new_inner(
232            self.num_threads.unwrap_or(1),
233            self.worker_init,
234            self.instrument,
235            self.allow_interrupts,
236        )
237    }
238}
239
240// TODO(https://fxbug.dev/42156503) test SendExecutor with unit tests
241
242#[cfg(test)]
243mod tests {
244    use super::SendExecutorBuilder;
245    use crate::{Task, Timer};
246
247    use fuchsia_sync::{Condvar, Mutex};
248    use futures::channel::oneshot;
249    use std::sync::Arc;
250    use std::sync::atomic::{AtomicU64, Ordering};
251
252    #[test]
253    fn test_stalled_triggers_wake_up() {
254        SendExecutorBuilder::new().num_threads(2).build().run(async {
255            // The timer will only fire on one thread, so use one so we can get to a point where
256            // only one thread is running.
257            Timer::new(zx::MonotonicDuration::from_millis(10)).await;
258
259            let (tx, rx) = oneshot::channel();
260            let pair = Arc::new((Mutex::new(false), Condvar::new()));
261            let pair2 = pair.clone();
262
263            let _task = Task::spawn(async move {
264                // Send a notification to the other task.
265                tx.send(()).unwrap();
266                // Now block the thread waiting for the result.
267                let (lock, cvar) = &*pair;
268                let mut done = lock.lock();
269                while !*done {
270                    cvar.wait(&mut done);
271                }
272            });
273
274            rx.await.unwrap();
275            let (lock, cvar) = &*pair2;
276            *lock.lock() = true;
277            cvar.notify_one();
278        });
279    }
280
281    #[test]
282    fn worker_init_called_once_per_worker() {
283        static NUM_INIT_CALLS: AtomicU64 = AtomicU64::new(0);
284        fn initialize_test_worker() {
285            NUM_INIT_CALLS.fetch_add(1, Ordering::SeqCst);
286        }
287
288        let mut exec =
289            SendExecutorBuilder::new().num_threads(2).worker_init(initialize_test_worker).build();
290        exec.run(async {});
291        assert_eq!(NUM_INIT_CALLS.load(Ordering::SeqCst), 2);
292        exec.run(async {});
293        assert_eq!(NUM_INIT_CALLS.load(Ordering::SeqCst), 4);
294    }
295
296    #[test]
297    fn test_allow_interrupts() {
298        use crate::OnInterrupt;
299        use futures::StreamExt;
300
301        SendExecutorBuilder::new().num_threads(2).allow_interrupts(true).build().run(async {
302            let irq_raw = zx::VirtualInterrupt::create_virtual().unwrap();
303            let irq_clone = irq_raw.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
304            let mut irq = std::pin::pin!(OnInterrupt::new(irq_raw));
305
306            let timestamp = zx::BootInstant::from_nanos(42);
307
308            let task = Task::spawn(async move {
309                Timer::new(zx::MonotonicDuration::from_millis(10)).await;
310                irq_clone.trigger(timestamp).unwrap();
311            });
312
313            let result = irq.next().await.unwrap().unwrap();
314            assert_eq!(result, timestamp);
315            task.await;
316        });
317    }
318}
fuchsia_async/runtime/fuchsia/executor/send.rs

fuchsia_async/runtime/fuchsia/executor/
send.rs
