fuchsia_async/
test_support.rs

1// Copyright 2020 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
5#[cfg(target_os = "fuchsia")]
6use crate::TestExecutorBuilder;
7use crate::{LocalExecutorBuilder, SendExecutorBuilder, TimeoutExt};
8use futures::prelude::*;
9use std::pin::Pin;
10use std::sync::atomic::{AtomicUsize, Ordering};
11#[cfg(target_os = "fuchsia")]
12use std::task::Poll;
13use std::time::Duration;
14
15// Apply the timeout from config to test
16// Ideally this would be a function like Config::with_timeout, but we need to handle Send and !Send
17// and it's likely better not to have to duplicate this code.
18macro_rules! apply_timeout {
19    ($config:expr, $test:expr) => {{
20        let timeout = $config.timeout;
21        let test = $test;
22        move |run| {
23            let test = test(run);
24            async move {
25                if let Some(timeout) = timeout {
26                    test.on_timeout(timeout, || panic!("timeout on run {}", run)).await
27                } else {
28                    test.await
29                }
30            }
31        }
32    }};
33}
34
35/// Defines how to compose multiple test runs for a kind of test result.
36pub trait TestResult: Sized {
37    /// How to repeatedly run a test with this result in a single threaded executor.
38    fn run_singlethreaded(
39        test: &(dyn Sync + Fn(usize) -> Pin<Box<dyn Future<Output = Self>>>),
40        options: TestOptions,
41        cfg: EnvironmentConfig,
42    ) -> Self;
43
44    /// Similarly, but use run_until_stalled
45    #[cfg(target_os = "fuchsia")]
46    fn run_until_stalled<
47        F: 'static + Sync + Fn(usize) -> Fut,
48        Fut: 'static + Future<Output = Self>,
49    >(
50        fake_time: bool,
51        test: F,
52        options: TestOptions,
53        cfg: EnvironmentConfig,
54    ) -> Self;
55
56    /// Whether the result is successful.
57    fn is_ok(&self) -> bool;
58}
59
60/// Defines how to compose multiple test runs for a kind of test result in a multithreaded executor.
61pub trait MultithreadedTestResult: Sized {
62    /// How to repeatedly run a test with this result in a multi threaded executor.
63    fn run<F: 'static + Sync + Fn(usize) -> Fut, Fut: 'static + Send + Future<Output = Self>>(
64        test: F,
65        threads: u8,
66        options: TestOptions,
67        cfg: EnvironmentConfig,
68    ) -> Self;
69
70    /// Whether the result is successful.
71    fn is_ok(&self) -> bool;
72}
73
74impl<E: Send + 'static + std::fmt::Debug> TestResult for Result<(), E> {
75    fn run_singlethreaded(
76        test: &(dyn Sync + Fn(usize) -> Pin<Box<dyn Future<Output = Self>>>),
77        options: TestOptions,
78        cfg: EnvironmentConfig,
79    ) -> Self {
80        cfg.run(1, |run| {
81            LocalExecutorBuilder::new()
82                .allow_interrupts(options.allow_interrupts)
83                .build()
84                .run_singlethreaded(test(run))
85        })
86    }
87
88    #[cfg(target_os = "fuchsia")]
89    fn run_until_stalled<
90        F: 'static + Sync + Fn(usize) -> Fut,
91        Fut: 'static + Future<Output = Self>,
92    >(
93        fake_time: bool,
94        test: F,
95        options: TestOptions,
96        cfg: EnvironmentConfig,
97    ) -> Self {
98        let test = apply_timeout!(cfg, |run| test(run));
99        cfg.run(1, |run| {
100            let mut executor = TestExecutorBuilder::new()
101                .fake_time(fake_time)
102                .allow_interrupts(options.allow_interrupts)
103                .build();
104            match executor.run_until_stalled(&mut std::pin::pin!(test(run))) {
105                Poll::Ready(result) => result,
106                Poll::Pending => panic!(
107                    "Stalled without completing. Consider using \"run_singlethreaded\", or check \
108                     for a deadlock."
109                ),
110            }
111        })
112    }
113
114    fn is_ok(&self) -> bool {
115        Result::is_ok(self)
116    }
117}
118
119impl<E: 'static + Send> MultithreadedTestResult for Result<(), E> {
120    fn run<F: 'static + Sync + Fn(usize) -> Fut, Fut: 'static + Send + Future<Output = Self>>(
121        test: F,
122        threads: u8,
123        options: TestOptions,
124        cfg: EnvironmentConfig,
125    ) -> Self {
126        let test = apply_timeout!(cfg, |run| test(run));
127        // Fuchsia's SendExecutor actually uses an extra thread, but it doesn't do anything, so we
128        // don't count it.
129        cfg.run(threads, |run| {
130            SendExecutorBuilder::new()
131                .num_threads(threads)
132                .allow_interrupts(options.allow_interrupts)
133                .build()
134                .run(test(run))
135        })
136    }
137
138    fn is_ok(&self) -> bool {
139        Result::is_ok(self)
140    }
141}
142
143impl TestResult for () {
144    fn run_singlethreaded(
145        test: &(dyn Sync + Fn(usize) -> Pin<Box<dyn Future<Output = Self>>>),
146        options: TestOptions,
147        cfg: EnvironmentConfig,
148    ) -> Self {
149        let _ = cfg.run(1, |run| {
150            LocalExecutorBuilder::new()
151                .allow_interrupts(options.allow_interrupts)
152                .build()
153                .run_singlethreaded(test(run));
154            Ok::<(), ()>(())
155        });
156    }
157
158    #[cfg(target_os = "fuchsia")]
159    fn run_until_stalled<
160        F: Sync + 'static + Fn(usize) -> Fut,
161        Fut: 'static + Future<Output = Self>,
162    >(
163        fake_time: bool,
164        test: F,
165        options: TestOptions,
166        cfg: EnvironmentConfig,
167    ) -> Self {
168        let _ = TestResult::run_until_stalled(
169            fake_time,
170            move |run| {
171                let test = test(run);
172                async move {
173                    test.await;
174                    Ok::<(), ()>(())
175                }
176            },
177            options,
178            cfg,
179        );
180    }
181
182    fn is_ok(&self) -> bool {
183        true
184    }
185}
186
187impl MultithreadedTestResult for () {
188    fn run<F: 'static + Sync + Fn(usize) -> Fut, Fut: 'static + Send + Future<Output = Self>>(
189        test: F,
190        threads: u8,
191        options: TestOptions,
192        cfg: EnvironmentConfig,
193    ) -> Self {
194        // Fuchsia's SendExecutor actually uses an extra thread, but it doesn't do anything, so we
195        // don't count it.
196        let _ = cfg.run(threads, |run| {
197            SendExecutorBuilder::new()
198                .num_threads(threads)
199                .allow_interrupts(options.allow_interrupts)
200                .build()
201                .run(test(run));
202            Ok::<(), ()>(())
203        });
204    }
205
206    fn is_ok(&self) -> bool {
207        true
208    }
209}
210
211/// Environment configurations for a single test run.
212#[derive(Clone)]
213pub struct EnvironmentConfig {
214    repeat_count: usize,
215    max_concurrency: usize,
216    max_threads: u8,
217    timeout: Option<Duration>,
218}
219
220fn env_var<T: std::str::FromStr>(name: &str, default: T) -> T {
221    std::env::var(name).unwrap_or_default().parse().unwrap_or(default)
222}
223
224impl EnvironmentConfig {
225    fn get() -> Self {
226        let repeat_count = std::cmp::max(1, env_var("FASYNC_TEST_REPEAT_COUNT", 1));
227        let max_concurrency = env_var("FASYNC_TEST_MAX_CONCURRENCY", 0);
228        let timeout_seconds = env_var("FASYNC_TEST_TIMEOUT_SECONDS", 0);
229        let max_threads = env_var("FASYNC_TEST_MAX_THREADS", 0);
230        let timeout =
231            if timeout_seconds == 0 { None } else { Some(Duration::from_secs(timeout_seconds)) };
232        Self { repeat_count, max_concurrency, max_threads, timeout }
233    }
234
235    fn in_parallel<E: Send>(
236        &self,
237        threads: u8,
238        f: impl Fn() -> Result<(), E> + Sync,
239    ) -> Result<(), E> {
240        std::thread::scope(|s| {
241            let mut join_handles = Vec::new();
242            for _ in 1..threads {
243                join_handles.push(s.spawn(&f));
244            }
245            f()?;
246            for h in join_handles {
247                if let Ok(result @ Err(_)) = h.join() {
248                    return result;
249                }
250            }
251            Ok(())
252        })
253    }
254
255    fn run<E: Send>(
256        &self,
257        test_threads: u8,
258        f: impl Fn(usize) -> Result<(), E> + Sync,
259    ) -> Result<(), E> {
260        // max_concurrency is the maximum number of runs of the same test to run in parallel, but
261        // each test can run multiple threads.  max_threads is the maximum number of threads.
262        let mut threads = std::cmp::min(std::cmp::max(self.repeat_count, 1), self.max_concurrency);
263        if self.max_threads != 0 {
264            threads =
265                std::cmp::min(threads, std::cmp::max(self.max_threads / test_threads, 1) as usize);
266        }
267        let threads = u8::try_from(threads).unwrap_or(u8::MAX);
268        let run = AtomicUsize::new(0);
269        self.in_parallel(threads, || {
270            loop {
271                let this_run = run.fetch_add(1, Ordering::Relaxed);
272                if this_run >= self.repeat_count {
273                    return Ok(());
274                }
275                let result = f(this_run);
276                if result.is_err() {
277                    // Prevent any more runs from starting.
278                    run.store(self.repeat_count, Ordering::Relaxed);
279                    return result;
280                }
281            }
282        })
283    }
284}
285
286/// Options for running an async test.
287#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
288pub struct TestOptions {
289    /// True iff the executor should support binding Zircon interrupts.
290    pub allow_interrupts: bool,
291}
292
293/// Runs a test in an executor, potentially repeatedly and concurrently.
294pub fn run_singlethreaded_test<F, Fut, R>(test: F, options: TestOptions) -> R
295where
296    F: 'static + Sync + Fn(usize) -> Fut,
297    Fut: 'static + Future<Output = R>,
298    R: TestResult,
299{
300    TestResult::run_singlethreaded(
301        &|run| test(run).boxed_local(),
302        options,
303        EnvironmentConfig::get(),
304    )
305}
306
307/// Runs a test in an executor until it's stalled.
308#[cfg(target_os = "fuchsia")]
309pub fn run_until_stalled_test<F, Fut, R>(fake_time: bool, test: F, options: TestOptions) -> R
310where
311    F: 'static + Sync + Fn(usize) -> Fut,
312    Fut: 'static + Future<Output = R>,
313    R: TestResult,
314{
315    TestResult::run_until_stalled(fake_time, test, options, EnvironmentConfig::get())
316}
317
318/// Runs a test in an executor, potentially repeatedly and concurrently.
319pub fn run_test<F, Fut, R>(test: F, threads: u8, options: TestOptions) -> R
320where
321    F: 'static + Sync + Fn(usize) -> Fut,
322    Fut: 'static + Send + Future<Output = R>,
323    R: MultithreadedTestResult,
324{
325    MultithreadedTestResult::run(test, threads, options, EnvironmentConfig::get())
326}
327
328#[cfg(test)]
329mod tests {
330    use super::{EnvironmentConfig, MultithreadedTestResult, TestOptions, TestResult};
331    use futures::lock::Mutex;
332    use futures::prelude::*;
333    use std::collections::HashSet;
334    use std::sync::Arc;
335    use std::time::Duration;
336
337    #[test]
338    fn run_singlethreaded() {
339        const REPEAT_COUNT: usize = 1000;
340        const MAX_THREADS: u8 = 10;
341        let pending_runs: Arc<Mutex<HashSet<_>>> =
342            Arc::new(Mutex::new((0..REPEAT_COUNT).collect()));
343        let pending_runs_child = pending_runs.clone();
344        TestResult::run_singlethreaded(
345            &move |i| {
346                let pending_runs_child = pending_runs_child.clone();
347                async move {
348                    assert!(pending_runs_child.lock().await.remove(&i));
349                }
350                .boxed_local()
351            },
352            TestOptions::default(),
353            EnvironmentConfig {
354                repeat_count: REPEAT_COUNT,
355                max_concurrency: 0,
356                max_threads: MAX_THREADS,
357                timeout: None,
358            },
359        );
360        assert!(pending_runs.try_lock().unwrap().is_empty());
361    }
362
363    // TODO(https://fxbug.dev/42138715): should_panic tests trigger LSAN
364    #[ignore]
365    #[test]
366    #[should_panic]
367    fn run_singlethreaded_with_timeout() {
368        TestResult::run_singlethreaded(
369            &move |_| {
370                async move {
371                    futures::future::pending::<()>().await;
372                }
373                .boxed_local()
374            },
375            TestOptions::default(),
376            EnvironmentConfig {
377                repeat_count: 1,
378                max_concurrency: 0,
379                max_threads: 0,
380                timeout: Some(Duration::from_millis(1)),
381            },
382        );
383    }
384
385    #[test]
386    #[cfg(target_os = "fuchsia")]
387    fn run_until_stalled() {
388        const REPEAT_COUNT: usize = 1000;
389        let pending_runs: Arc<Mutex<HashSet<_>>> =
390            Arc::new(Mutex::new((0..REPEAT_COUNT).collect()));
391        let pending_runs_child = pending_runs.clone();
392        TestResult::run_until_stalled(
393            false,
394            move |i| {
395                let pending_runs_child = pending_runs_child.clone();
396                async move {
397                    assert!(pending_runs_child.lock().await.remove(&i));
398                }
399            },
400            TestOptions::default(),
401            EnvironmentConfig {
402                repeat_count: REPEAT_COUNT,
403                max_concurrency: 1,
404                max_threads: 1,
405                timeout: None,
406            },
407        );
408        assert!(pending_runs.try_lock().unwrap().is_empty());
409    }
410
411    #[test]
412    fn run() {
413        const REPEAT_COUNT: usize = 1000;
414        const THREADS: u8 = 4;
415        let pending_runs: Arc<Mutex<HashSet<_>>> =
416            Arc::new(Mutex::new((0..REPEAT_COUNT).collect()));
417        let pending_runs_child = pending_runs.clone();
418        MultithreadedTestResult::run(
419            move |i| {
420                let pending_runs_child = pending_runs_child.clone();
421                async move {
422                    assert!(pending_runs_child.lock().await.remove(&i));
423                }
424            },
425            THREADS,
426            TestOptions::default(),
427            EnvironmentConfig {
428                repeat_count: REPEAT_COUNT,
429                max_concurrency: 0,
430                max_threads: THREADS,
431                timeout: None,
432            },
433        );
434        assert!(pending_runs.try_lock().unwrap().is_empty());
435    }
436
437    // TODO(https://fxbug.dev/42138715): should_panic tests trigger LSAN
438    #[ignore]
439    #[test]
440    #[should_panic]
441    fn run_with_timeout() {
442        const THREADS: u8 = 4;
443        MultithreadedTestResult::run(
444            move |_| async move {
445                futures::future::pending::<()>().await;
446            },
447            THREADS,
448            TestOptions::default(),
449            EnvironmentConfig {
450                repeat_count: 1,
451                max_concurrency: 0,
452                max_threads: 0,
453                timeout: Some(Duration::from_millis(1)),
454            },
455        );
456    }
457
458    #[test]
459    fn run_singlethreaded_with_allow_interrupts() {
460        TestResult::run_singlethreaded(
461            &move |_| async move {}.boxed_local(),
462            TestOptions { allow_interrupts: true },
463            EnvironmentConfig {
464                repeat_count: 1,
465                max_concurrency: 0,
466                max_threads: 1,
467                timeout: None,
468            },
469        );
470    }
471
472    #[test]
473    #[cfg(target_os = "fuchsia")]
474    fn run_until_stalled_with_allow_interrupts() {
475        TestResult::run_until_stalled(
476            false,
477            move |_| async move {},
478            TestOptions { allow_interrupts: true },
479            EnvironmentConfig {
480                repeat_count: 1,
481                max_concurrency: 1,
482                max_threads: 1,
483                timeout: None,
484            },
485        );
486    }
487
488    #[test]
489    fn run_with_allow_interrupts() {
490        const THREADS: u8 = 2;
491        MultithreadedTestResult::run(
492            move |_| async move {},
493            THREADS,
494            TestOptions { allow_interrupts: true },
495            EnvironmentConfig {
496                repeat_count: 1,
497                max_concurrency: 0,
498                max_threads: THREADS,
499                timeout: None,
500            },
501        );
502    }
503}
fuchsia_async/test_support.rs

fuchsia_async/
test_support.rs
