cros_async/
lib.rs

1// Copyright 2020 The ChromiumOS Authors
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! An Executor and future combinators based on operations that block on file descriptors.
6//!
7//! This crate is meant to be used with the `futures-rs` crate that provides further combinators
8//! and utility functions to combine and manage futures. All futures will run until they block on a
9//! file descriptor becoming readable or writable. Facilities are provided to register future
10//! wakers based on such events.
11//!
12//! # Running top-level futures.
13//!
14//! Use helper functions based the desired behavior of your application.
15//!
16//! ## Completing one of several futures.
17//!
18//! If there are several top level tasks that should run until any one completes, use the "select"
19//! family of executor constructors. These return an [`Executor`](trait.Executor.html) whose `run`
20//! function will return when the first future completes. The uncompleted futures will also be
21//! returned so they can be run further or otherwise cleaned up. These functions are inspired by
22//! the `select_all` function from futures-rs, but built to be run inside an FD based executor and
23//! to poll only when necessary. See the docs for [`select2`](fn.select2.html),
24//! [`select3`](fn.select3.html), [`select4`](fn.select4.html), and [`select5`](fn.select5.html).
25//!
26//! ## Completing all of several futures.
27//!
28//! If there are several top level tasks that all need to be completed, use the "complete" family
29//! of executor constructors. These return an [`Executor`](trait.Executor.html) whose `run`
30//! function will return only once all the futures passed to it have completed. These functions are
31//! inspired by the `join_all` function from futures-rs, but built to be run inside an FD based
32//! executor and to poll only when necessary. See the docs for [`complete2`](fn.complete2.html),
33//! [`complete3`](fn.complete3.html), [`complete4`](fn.complete4.html), and
34//! [`complete5`](fn.complete5.html).
35//!
36//! # Implementing new FD-based futures.
37//!
38//! For URing implementations should provide an implementation of the `IoSource` trait.
39//! For the FD executor, new futures can use the existing ability to poll a source to build async
40//! functionality on top of.
41//!
42//! # Implementations
43//!
44//! Currently there are two paths for using the asynchronous IO. One uses a WaitContext and drives
45//! futures based on the FDs signaling they are ready for the opteration. This method will exist so
46//! long as kernels < 5.4 are supported.
47//! The other method submits operations to io_uring and is signaled when they complete. This is more
48//! efficient, but only supported on kernel 5.4+.
49//! If `IoSource::new` is used to interface with async IO, then the correct backend will be chosen
50//! automatically.
51//!
52//! # Examples
53//!
54//! See the docs for `IoSource` if support for kernels <5.4 is required. Focus on `UringSource` if
55//! all systems have support for io_uring.
56
57mod async_types;
58pub mod audio_streams_async;
59mod blocking;
60mod common_executor;
61mod complete;
62mod event;
63mod executor;
64mod io_ext;
65mod io_source;
66pub mod mem;
67mod queue;
68mod select;
69pub mod sync;
70pub mod sys;
71mod timer;
72#[cfg(feature = "tokio")]
73mod tokio_executor;
74mod waker;
75
76use std::future::Future;
77use std::pin::Pin;
78use std::task::Poll;
79
80pub use async_types::*;
81pub use base::Event;
82#[cfg(any(target_os = "android", target_os = "linux"))]
83pub use blocking::sys::linux::block_on::block_on;
84pub use blocking::unblock;
85pub use blocking::unblock_disarm;
86pub use blocking::BlockingPool;
87pub use blocking::CancellableBlockingPool;
88pub use blocking::TimeoutAction;
89pub use event::EventAsync;
90pub use executor::Executor;
91pub use executor::ExecutorKind;
92pub(crate) use executor::ExecutorTrait;
93pub use executor::TaskHandle;
94#[cfg(windows)]
95pub use futures::executor::block_on;
96use futures::stream::FuturesUnordered;
97pub use io_ext::AsyncError;
98pub use io_ext::AsyncResult;
99pub use io_ext::AsyncWrapper;
100pub use io_ext::IntoAsync;
101pub use io_source::IoOptions;
102pub use io_source::IoSource;
103pub use mem::BackingMemory;
104pub use mem::MemRegion;
105pub use mem::MemRegionIter;
106pub use mem::VecIoWrapper;
107use remain::sorted;
108pub use select::SelectResult;
109#[cfg(any(target_os = "android", target_os = "linux"))]
110pub use sys::linux::uring_executor::is_uring_stable;
111use thiserror::Error as ThisError;
112pub use timer::TimerAsync;
113
114#[sorted]
115#[derive(ThisError, Debug)]
116pub enum Error {
117    /// Error from EventAsync
118    #[error("Failure in EventAsync: {0}")]
119    EventAsync(base::Error),
120    /// Error from the handle executor.
121    #[cfg(windows)]
122    #[error("Failure in the handle executor: {0}")]
123    HandleExecutor(sys::windows::handle_executor::Error),
124    #[error("IO error: {0}")]
125    Io(std::io::Error),
126    /// Error from the polled(FD) source, which includes error from the FD executor.
127    #[cfg(any(target_os = "android", target_os = "linux"))]
128    #[error("An error with a poll source: {0}")]
129    PollSource(sys::linux::poll_source::Error),
130    /// Error from Timer.
131    #[error("Failure in Timer: {0}")]
132    Timer(base::Error),
133    /// Error from TimerFd.
134    #[error("Failure in TimerAsync: {0}")]
135    TimerAsync(AsyncError),
136    /// Error from the uring executor.
137    #[cfg(any(target_os = "android", target_os = "linux"))]
138    #[error("Failure in the uring executor: {0}")]
139    URingExecutor(sys::linux::uring_executor::Error),
140}
141pub type Result<T> = std::result::Result<T, Error>;
142
143/// Heterogeneous collection of `async_task:Task` that are running in a "detached" state.
144///
145/// We keep them around to ensure they are dropped before the executor they are running on.
146pub(crate) struct DetachedTasks(FuturesUnordered<Pin<Box<dyn Future<Output = ()> + Send>>>);
147
148impl DetachedTasks {
149    pub(crate) fn new() -> Self {
150        DetachedTasks(FuturesUnordered::new())
151    }
152
153    pub(crate) fn push<R: Send + 'static>(&self, task: async_task::Task<R>) {
154        // Convert to fallible, otherwise poll could panic if the `Runnable` is dropped early.
155        let task = task.fallible();
156        self.0.push(Box::pin(async {
157            let _ = task.await;
158        }));
159    }
160
161    /// Polls all the tasks, dropping any that complete.
162    pub(crate) fn poll(&mut self, cx: &mut std::task::Context) {
163        use futures::Stream;
164        while let Poll::Ready(Some(_)) = Pin::new(&mut self.0).poll_next(cx) {}
165    }
166}
167
168// Select helpers to run until any future completes.
169
170/// Creates a combinator that runs the two given futures until one completes, returning a tuple
171/// containing the result of the finished future and the still pending future.
172///
173///  # Example
174///
175///    ```
176///    use cros_async::{SelectResult, select2, block_on};
177///    use futures::future::pending;
178///    use futures::pin_mut;
179///
180///    let first = async {5};
181///    let second = async {let () = pending().await;};
182///    pin_mut!(first);
183///    pin_mut!(second);
184///    match block_on(select2(first, second)) {
185///        (SelectResult::Finished(5), SelectResult::Pending(_second)) => (),
186///        _ => panic!("Select didn't return the first future"),
187///    };
188///    ```
189pub async fn select2<F1: Future + Unpin, F2: Future + Unpin>(
190    f1: F1,
191    f2: F2,
192) -> (SelectResult<F1>, SelectResult<F2>) {
193    select::Select2::new(f1, f2).await
194}
195
196/// Creates a combinator that runs the three given futures until one or more completes, returning a
197/// tuple containing the result of the finished future(s) and the still pending future(s).
198///
199///  # Example
200///
201///    ```
202///    use cros_async::{SelectResult, select3, block_on};
203///    use futures::future::pending;
204///    use futures::pin_mut;
205///
206///    let first = async {4};
207///    let second = async {let () = pending().await;};
208///    let third = async {5};
209///    pin_mut!(first);
210///    pin_mut!(second);
211///    pin_mut!(third);
212///    match block_on(select3(first, second, third)) {
213///        (SelectResult::Finished(4),
214///         SelectResult::Pending(_second),
215///         SelectResult::Finished(5)) => (),
216///        _ => panic!("Select didn't return the futures"),
217///    };
218///    ```
219pub async fn select3<F1: Future + Unpin, F2: Future + Unpin, F3: Future + Unpin>(
220    f1: F1,
221    f2: F2,
222    f3: F3,
223) -> (SelectResult<F1>, SelectResult<F2>, SelectResult<F3>) {
224    select::Select3::new(f1, f2, f3).await
225}
226
227/// Creates a combinator that runs the four given futures until one or more completes, returning a
228/// tuple containing the result of the finished future(s) and the still pending future(s).
229///
230///  # Example
231///
232///    ```
233///    use cros_async::{SelectResult, select4, block_on};
234///    use futures::future::pending;
235///    use futures::pin_mut;
236///
237///    let first = async {4};
238///    let second = async {let () = pending().await;};
239///    let third = async {5};
240///    let fourth = async {let () = pending().await;};
241///    pin_mut!(first);
242///    pin_mut!(second);
243///    pin_mut!(third);
244///    pin_mut!(fourth);
245///    match block_on(select4(first, second, third, fourth)) {
246///        (SelectResult::Finished(4), SelectResult::Pending(_second),
247///         SelectResult::Finished(5), SelectResult::Pending(_fourth)) => (),
248///        _ => panic!("Select didn't return the futures"),
249///    };
250///    ```
251pub async fn select4<
252    F1: Future + Unpin,
253    F2: Future + Unpin,
254    F3: Future + Unpin,
255    F4: Future + Unpin,
256>(
257    f1: F1,
258    f2: F2,
259    f3: F3,
260    f4: F4,
261) -> (
262    SelectResult<F1>,
263    SelectResult<F2>,
264    SelectResult<F3>,
265    SelectResult<F4>,
266) {
267    select::Select4::new(f1, f2, f3, f4).await
268}
269
270/// Creates a combinator that runs the five given futures until one or more completes, returning a
271/// tuple containing the result of the finished future(s) and the still pending future(s).
272///
273///  # Example
274///
275///    ```
276///    use cros_async::{SelectResult, select5, block_on};
277///    use futures::future::pending;
278///    use futures::pin_mut;
279///
280///    let first = async {4};
281///    let second = async {let () = pending().await;};
282///    let third = async {5};
283///    let fourth = async {let () = pending().await;};
284///    let fifth = async {6};
285///    pin_mut!(first);
286///    pin_mut!(second);
287///    pin_mut!(third);
288///    pin_mut!(fourth);
289///    pin_mut!(fifth);
290///    match block_on(select5(first, second, third, fourth, fifth)) {
291///        (SelectResult::Finished(4), SelectResult::Pending(_second),
292///         SelectResult::Finished(5), SelectResult::Pending(_fourth),
293///         SelectResult::Finished(6)) => (),
294///        _ => panic!("Select didn't return the futures"),
295///    };
296///    ```
297pub async fn select5<
298    F1: Future + Unpin,
299    F2: Future + Unpin,
300    F3: Future + Unpin,
301    F4: Future + Unpin,
302    F5: Future + Unpin,
303>(
304    f1: F1,
305    f2: F2,
306    f3: F3,
307    f4: F4,
308    f5: F5,
309) -> (
310    SelectResult<F1>,
311    SelectResult<F2>,
312    SelectResult<F3>,
313    SelectResult<F4>,
314    SelectResult<F5>,
315) {
316    select::Select5::new(f1, f2, f3, f4, f5).await
317}
318
319/// Creates a combinator that runs the six given futures until one or more completes, returning a
320/// tuple containing the result of the finished future(s) and the still pending future(s).
321///
322///  # Example
323///
324///    ```
325///    use cros_async::{SelectResult, select6, block_on};
326///    use futures::future::pending;
327///    use futures::pin_mut;
328///
329///    let first = async {1};
330///    let second = async {let () = pending().await;};
331///    let third = async {3};
332///    let fourth = async {let () = pending().await;};
333///    let fifth = async {5};
334///    let sixth = async {6};
335///    pin_mut!(first);
336///    pin_mut!(second);
337///    pin_mut!(third);
338///    pin_mut!(fourth);
339///    pin_mut!(fifth);
340///    pin_mut!(sixth);
341///    match block_on(select6(first, second, third, fourth, fifth, sixth)) {
342///        (SelectResult::Finished(1), SelectResult::Pending(_second),
343///         SelectResult::Finished(3), SelectResult::Pending(_fourth),
344///         SelectResult::Finished(5), SelectResult::Finished(6)) => (),
345///        _ => panic!("Select didn't return the futures"),
346///    };
347///    ```
348pub async fn select6<
349    F1: Future + Unpin,
350    F2: Future + Unpin,
351    F3: Future + Unpin,
352    F4: Future + Unpin,
353    F5: Future + Unpin,
354    F6: Future + Unpin,
355>(
356    f1: F1,
357    f2: F2,
358    f3: F3,
359    f4: F4,
360    f5: F5,
361    f6: F6,
362) -> (
363    SelectResult<F1>,
364    SelectResult<F2>,
365    SelectResult<F3>,
366    SelectResult<F4>,
367    SelectResult<F5>,
368    SelectResult<F6>,
369) {
370    select::Select6::new(f1, f2, f3, f4, f5, f6).await
371}
372
373pub async fn select7<
374    F1: Future + Unpin,
375    F2: Future + Unpin,
376    F3: Future + Unpin,
377    F4: Future + Unpin,
378    F5: Future + Unpin,
379    F6: Future + Unpin,
380    F7: Future + Unpin,
381>(
382    f1: F1,
383    f2: F2,
384    f3: F3,
385    f4: F4,
386    f5: F5,
387    f6: F6,
388    f7: F7,
389) -> (
390    SelectResult<F1>,
391    SelectResult<F2>,
392    SelectResult<F3>,
393    SelectResult<F4>,
394    SelectResult<F5>,
395    SelectResult<F6>,
396    SelectResult<F7>,
397) {
398    select::Select7::new(f1, f2, f3, f4, f5, f6, f7).await
399}
400
401pub async fn select8<
402    F1: Future + Unpin,
403    F2: Future + Unpin,
404    F3: Future + Unpin,
405    F4: Future + Unpin,
406    F5: Future + Unpin,
407    F6: Future + Unpin,
408    F7: Future + Unpin,
409    F8: Future + Unpin,
410>(
411    f1: F1,
412    f2: F2,
413    f3: F3,
414    f4: F4,
415    f5: F5,
416    f6: F6,
417    f7: F7,
418    f8: F8,
419) -> (
420    SelectResult<F1>,
421    SelectResult<F2>,
422    SelectResult<F3>,
423    SelectResult<F4>,
424    SelectResult<F5>,
425    SelectResult<F6>,
426    SelectResult<F7>,
427    SelectResult<F8>,
428) {
429    select::Select8::new(f1, f2, f3, f4, f5, f6, f7, f8).await
430}
431
432pub async fn select9<
433    F1: Future + Unpin,
434    F2: Future + Unpin,
435    F3: Future + Unpin,
436    F4: Future + Unpin,
437    F5: Future + Unpin,
438    F6: Future + Unpin,
439    F7: Future + Unpin,
440    F8: Future + Unpin,
441    F9: Future + Unpin,
442>(
443    f1: F1,
444    f2: F2,
445    f3: F3,
446    f4: F4,
447    f5: F5,
448    f6: F6,
449    f7: F7,
450    f8: F8,
451    f9: F9,
452) -> (
453    SelectResult<F1>,
454    SelectResult<F2>,
455    SelectResult<F3>,
456    SelectResult<F4>,
457    SelectResult<F5>,
458    SelectResult<F6>,
459    SelectResult<F7>,
460    SelectResult<F8>,
461    SelectResult<F9>,
462) {
463    select::Select9::new(f1, f2, f3, f4, f5, f6, f7, f8, f9).await
464}
465
466pub async fn select10<
467    F1: Future + Unpin,
468    F2: Future + Unpin,
469    F3: Future + Unpin,
470    F4: Future + Unpin,
471    F5: Future + Unpin,
472    F6: Future + Unpin,
473    F7: Future + Unpin,
474    F8: Future + Unpin,
475    F9: Future + Unpin,
476    F10: Future + Unpin,
477>(
478    f1: F1,
479    f2: F2,
480    f3: F3,
481    f4: F4,
482    f5: F5,
483    f6: F6,
484    f7: F7,
485    f8: F8,
486    f9: F9,
487    f10: F10,
488) -> (
489    SelectResult<F1>,
490    SelectResult<F2>,
491    SelectResult<F3>,
492    SelectResult<F4>,
493    SelectResult<F5>,
494    SelectResult<F6>,
495    SelectResult<F7>,
496    SelectResult<F8>,
497    SelectResult<F9>,
498    SelectResult<F10>,
499) {
500    select::Select10::new(f1, f2, f3, f4, f5, f6, f7, f8, f9, f10).await
501}
502
503pub async fn select11<
504    F1: Future + Unpin,
505    F2: Future + Unpin,
506    F3: Future + Unpin,
507    F4: Future + Unpin,
508    F5: Future + Unpin,
509    F6: Future + Unpin,
510    F7: Future + Unpin,
511    F8: Future + Unpin,
512    F9: Future + Unpin,
513    F10: Future + Unpin,
514    F11: Future + Unpin,
515>(
516    f1: F1,
517    f2: F2,
518    f3: F3,
519    f4: F4,
520    f5: F5,
521    f6: F6,
522    f7: F7,
523    f8: F8,
524    f9: F9,
525    f10: F10,
526    f11: F11,
527) -> (
528    SelectResult<F1>,
529    SelectResult<F2>,
530    SelectResult<F3>,
531    SelectResult<F4>,
532    SelectResult<F5>,
533    SelectResult<F6>,
534    SelectResult<F7>,
535    SelectResult<F8>,
536    SelectResult<F9>,
537    SelectResult<F10>,
538    SelectResult<F11>,
539) {
540    select::Select11::new(f1, f2, f3, f4, f5, f6, f7, f8, f9, f10, f11).await
541}
542
543pub async fn select12<
544    F1: Future + Unpin,
545    F2: Future + Unpin,
546    F3: Future + Unpin,
547    F4: Future + Unpin,
548    F5: Future + Unpin,
549    F6: Future + Unpin,
550    F7: Future + Unpin,
551    F8: Future + Unpin,
552    F9: Future + Unpin,
553    F10: Future + Unpin,
554    F11: Future + Unpin,
555    F12: Future + Unpin,
556>(
557    f1: F1,
558    f2: F2,
559    f3: F3,
560    f4: F4,
561    f5: F5,
562    f6: F6,
563    f7: F7,
564    f8: F8,
565    f9: F9,
566    f10: F10,
567    f11: F11,
568    f12: F12,
569) -> (
570    SelectResult<F1>,
571    SelectResult<F2>,
572    SelectResult<F3>,
573    SelectResult<F4>,
574    SelectResult<F5>,
575    SelectResult<F6>,
576    SelectResult<F7>,
577    SelectResult<F8>,
578    SelectResult<F9>,
579    SelectResult<F10>,
580    SelectResult<F11>,
581    SelectResult<F12>,
582) {
583    select::Select12::new(f1, f2, f3, f4, f5, f6, f7, f8, f9, f10, f11, f12).await
584}
585
586// Combination helpers to run until all futures are complete.
587
588/// Creates a combinator that runs the two given futures to completion, returning a tuple of the
589/// outputs each yields.
590///
591///  # Example
592///
593///    ```
594///    use cros_async::{complete2, block_on};
595///
596///    let first = async {5};
597///    let second = async {6};
598///    assert_eq!(block_on(complete2(first, second)), (5,6));
599///    ```
600pub async fn complete2<F1, F2>(f1: F1, f2: F2) -> (F1::Output, F2::Output)
601where
602    F1: Future,
603    F2: Future,
604{
605    complete::Complete2::new(f1, f2).await
606}
607
608/// Creates a combinator that runs the three given futures to completion, returning a tuple of the
609/// outputs each yields.
610///
611///  # Example
612///
613///    ```
614///    use cros_async::{complete3, block_on};
615///
616///    let first = async {5};
617///    let second = async {6};
618///    let third = async {7};
619///    assert_eq!(block_on(complete3(first, second, third)), (5,6,7));
620///    ```
621pub async fn complete3<F1, F2, F3>(f1: F1, f2: F2, f3: F3) -> (F1::Output, F2::Output, F3::Output)
622where
623    F1: Future,
624    F2: Future,
625    F3: Future,
626{
627    complete::Complete3::new(f1, f2, f3).await
628}
629
630/// Creates a combinator that runs the four given futures to completion, returning a tuple of the
631/// outputs each yields.
632///
633///  # Example
634///
635///    ```
636///    use cros_async::{complete4, block_on};
637///
638///    let first = async {5};
639///    let second = async {6};
640///    let third = async {7};
641///    let fourth = async {8};
642///    assert_eq!(block_on(complete4(first, second, third, fourth)), (5,6,7,8));
643///    ```
644pub async fn complete4<F1, F2, F3, F4>(
645    f1: F1,
646    f2: F2,
647    f3: F3,
648    f4: F4,
649) -> (F1::Output, F2::Output, F3::Output, F4::Output)
650where
651    F1: Future,
652    F2: Future,
653    F3: Future,
654    F4: Future,
655{
656    complete::Complete4::new(f1, f2, f3, f4).await
657}
658
659/// Creates a combinator that runs the five given futures to completion, returning a tuple of the
660/// outputs each yields.
661///
662///  # Example
663///
664///    ```
665///    use cros_async::{complete5, block_on};
666///
667///    let first = async {5};
668///    let second = async {6};
669///    let third = async {7};
670///    let fourth = async {8};
671///    let fifth = async {9};
672///    assert_eq!(block_on(complete5(first, second, third, fourth, fifth)),
673///               (5,6,7,8,9));
674///    ```
675pub async fn complete5<F1, F2, F3, F4, F5>(
676    f1: F1,
677    f2: F2,
678    f3: F3,
679    f4: F4,
680    f5: F5,
681) -> (F1::Output, F2::Output, F3::Output, F4::Output, F5::Output)
682where
683    F1: Future,
684    F2: Future,
685    F3: Future,
686    F4: Future,
687    F5: Future,
688{
689    complete::Complete5::new(f1, f2, f3, f4, f5).await
690}
cros_async/lib.rs

cros_async/
lib.rs
