Merge remote-tracking branch 'upstream/master' into 342-stream-throttle
Wouter Geraedts
5 years ago
commit
14d7d3bf9c
@ -0,0 +1,42 @@
|
||||
#![feature(test)]
|
||||
|
||||
extern crate test;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_std::sync::Mutex;
|
||||
use async_std::task;
|
||||
use test::Bencher;
|
||||
|
||||
#[bench]
|
||||
fn create(b: &mut Bencher) {
|
||||
b.iter(|| Mutex::new(()));
|
||||
}
|
||||
|
||||
#[bench]
|
||||
fn contention(b: &mut Bencher) {
|
||||
b.iter(|| task::block_on(run(10, 1000)));
|
||||
}
|
||||
|
||||
#[bench]
|
||||
fn no_contention(b: &mut Bencher) {
|
||||
b.iter(|| task::block_on(run(1, 10000)));
|
||||
}
|
||||
|
||||
async fn run(task: usize, iter: usize) {
|
||||
let m = Arc::new(Mutex::new(()));
|
||||
let mut tasks = Vec::new();
|
||||
|
||||
for _ in 0..task {
|
||||
let m = m.clone();
|
||||
tasks.push(task::spawn(async move {
|
||||
for _ in 0..iter {
|
||||
let _ = m.lock().await;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for t in tasks {
|
||||
t.await;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,11 @@
|
||||
#![feature(test)]
|
||||
|
||||
extern crate test;
|
||||
|
||||
use async_std::task;
|
||||
use test::Bencher;
|
||||
|
||||
#[bench]
|
||||
fn block_on(b: &mut Bencher) {
|
||||
b.iter(|| task::block_on(async {}));
|
||||
}
|
||||
@ -0,0 +1,44 @@
|
||||
//! TCP echo server, accepting connections both on both ipv4 and ipv6 sockets.
|
||||
//!
|
||||
//! To send messages, do:
|
||||
//!
|
||||
//! ```sh
|
||||
//! $ nc 127.0.0.1 8080
|
||||
//! $ nc ::1 8080
|
||||
//! ```
|
||||
|
||||
use async_std::io;
|
||||
use async_std::net::{TcpListener, TcpStream};
|
||||
use async_std::prelude::*;
|
||||
use async_std::task;
|
||||
|
||||
async fn process(stream: TcpStream) -> io::Result<()> {
|
||||
println!("Accepted from: {}", stream.peer_addr()?);
|
||||
|
||||
let (reader, writer) = &mut (&stream, &stream);
|
||||
io::copy(reader, writer).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn main() -> io::Result<()> {
|
||||
task::block_on(async {
|
||||
let ipv4_listener = TcpListener::bind("127.0.0.1:8080").await?;
|
||||
println!("Listening on {}", ipv4_listener.local_addr()?);
|
||||
let ipv6_listener = TcpListener::bind("[::1]:8080").await?;
|
||||
println!("Listening on {}", ipv6_listener.local_addr()?);
|
||||
|
||||
let ipv4_incoming = ipv4_listener.incoming();
|
||||
let ipv6_incoming = ipv6_listener.incoming();
|
||||
|
||||
let mut incoming = ipv4_incoming.merge(ipv6_incoming);
|
||||
|
||||
while let Some(stream) = incoming.next().await {
|
||||
let stream = stream?;
|
||||
task::spawn(async {
|
||||
process(stream).await.unwrap();
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
@ -1,139 +0,0 @@
|
||||
extension_trait! {
|
||||
use std::pin::Pin;
|
||||
use std::ops::{Deref, DerefMut};
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
#[doc = r#"
|
||||
A future represents an asynchronous computation.
|
||||
|
||||
A future is a value that may not have finished computing yet. This kind of
|
||||
"asynchronous value" makes it possible for a thread to continue doing useful
|
||||
work while it waits for the value to become available.
|
||||
|
||||
# The `poll` method
|
||||
|
||||
The core method of future, `poll`, *attempts* to resolve the future into a
|
||||
final value. This method does not block if the value is not ready. Instead,
|
||||
the current task is scheduled to be woken up when it's possible to make
|
||||
further progress by `poll`ing again. The `context` passed to the `poll`
|
||||
method can provide a [`Waker`], which is a handle for waking up the current
|
||||
task.
|
||||
|
||||
When using a future, you generally won't call `poll` directly, but instead
|
||||
`.await` the value.
|
||||
|
||||
[`Waker`]: ../task/struct.Waker.html
|
||||
"#]
|
||||
pub trait Future {
|
||||
#[doc = r#"
|
||||
The type of value produced on completion.
|
||||
"#]
|
||||
type Output;
|
||||
|
||||
#[doc = r#"
|
||||
Attempt to resolve the future to a final value, registering
|
||||
the current task for wakeup if the value is not yet available.
|
||||
|
||||
# Return value
|
||||
|
||||
This function returns:
|
||||
|
||||
- [`Poll::Pending`] if the future is not ready yet
|
||||
- [`Poll::Ready(val)`] with the result `val` of this future if it
|
||||
finished successfully.
|
||||
|
||||
Once a future has finished, clients should not `poll` it again.
|
||||
|
||||
When a future is not ready yet, `poll` returns `Poll::Pending` and
|
||||
stores a clone of the [`Waker`] copied from the current [`Context`].
|
||||
This [`Waker`] is then woken once the future can make progress.
|
||||
For example, a future waiting for a socket to become
|
||||
readable would call `.clone()` on the [`Waker`] and store it.
|
||||
When a signal arrives elsewhere indicating that the socket is readable,
|
||||
[`Waker::wake`] is called and the socket future's task is awoken.
|
||||
Once a task has been woken up, it should attempt to `poll` the future
|
||||
again, which may or may not produce a final value.
|
||||
|
||||
Note that on multiple calls to `poll`, only the [`Waker`] from the
|
||||
[`Context`] passed to the most recent call should be scheduled to
|
||||
receive a wakeup.
|
||||
|
||||
# Runtime characteristics
|
||||
|
||||
Futures alone are *inert*; they must be *actively* `poll`ed to make
|
||||
progress, meaning that each time the current task is woken up, it should
|
||||
actively re-`poll` pending futures that it still has an interest in.
|
||||
|
||||
The `poll` function is not called repeatedly in a tight loop -- instead,
|
||||
it should only be called when the future indicates that it is ready to
|
||||
make progress (by calling `wake()`). If you're familiar with the
|
||||
`poll(2)` or `select(2)` syscalls on Unix it's worth noting that futures
|
||||
typically do *not* suffer the same problems of "all wakeups must poll
|
||||
all events"; they are more like `epoll(4)`.
|
||||
|
||||
An implementation of `poll` should strive to return quickly, and should
|
||||
not block. Returning quickly prevents unnecessarily clogging up
|
||||
threads or event loops. If it is known ahead of time that a call to
|
||||
`poll` may end up taking awhile, the work should be offloaded to a
|
||||
thread pool (or something similar) to ensure that `poll` can return
|
||||
quickly.
|
||||
|
||||
# Panics
|
||||
|
||||
Once a future has completed (returned `Ready` from `poll`), calling its
|
||||
`poll` method again may panic, block forever, or cause other kinds of
|
||||
problems; the `Future` trait places no requirements on the effects of
|
||||
such a call. However, as the `poll` method is not marked `unsafe`,
|
||||
Rust's usual rules apply: calls must never cause undefined behavior
|
||||
(memory corruption, incorrect use of `unsafe` functions, or the like),
|
||||
regardless of the future's state.
|
||||
|
||||
[`Poll::Pending`]: ../task/enum.Poll.html#variant.Pending
|
||||
[`Poll::Ready(val)`]: ../task/enum.Poll.html#variant.Ready
|
||||
[`Context`]: ../task/struct.Context.html
|
||||
[`Waker`]: ../task/struct.Waker.html
|
||||
[`Waker::wake`]: ../task/struct.Waker.html#method.wake
|
||||
"#]
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output>;
|
||||
}
|
||||
|
||||
pub trait FutureExt: std::future::Future {
|
||||
}
|
||||
|
||||
impl<F: Future + Unpin + ?Sized> Future for Box<F> {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future + Unpin + ?Sized> Future for &mut F {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
|
||||
impl<P> Future for Pin<P>
|
||||
where
|
||||
P: DerefMut + Unpin,
|
||||
<P as Deref>::Target: Future,
|
||||
{
|
||||
type Output = <<P as Deref>::Target as Future>::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future> Future for std::panic::AssertUnwindSafe<F> {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_timer::Delay;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
#[doc(hidden)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct DelayFuture<F> {
|
||||
#[pin]
|
||||
future: F,
|
||||
#[pin]
|
||||
delay: Delay,
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> DelayFuture<F> {
|
||||
pub fn new(future: F, dur: Duration) -> DelayFuture<F> {
|
||||
let delay = Delay::new(dur);
|
||||
|
||||
DelayFuture { future, delay }
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future> Future for DelayFuture<F> {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
|
||||
match this.delay.poll(cx) {
|
||||
Poll::Pending => Poll::Pending,
|
||||
Poll::Ready(_) => match this.future.poll(cx) {
|
||||
Poll::Ready(v) => Poll::Ready(v),
|
||||
Poll::Pending => Poll::Pending,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,52 @@
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::future::IntoFuture;
|
||||
use crate::task::{ready, Context, Poll};
|
||||
|
||||
#[doc(hidden)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct FlattenFuture<Fut1, Fut2> {
|
||||
state: State<Fut1, Fut2>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum State<Fut1, Fut2> {
|
||||
First(Fut1),
|
||||
Second(Fut2),
|
||||
Empty,
|
||||
}
|
||||
|
||||
impl<Fut1, Fut2> FlattenFuture<Fut1, Fut2> {
|
||||
pub(crate) fn new(future: Fut1) -> FlattenFuture<Fut1, Fut2> {
|
||||
FlattenFuture {
|
||||
state: State::First(future),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Fut1> Future for FlattenFuture<Fut1, <Fut1::Output as IntoFuture>::Future>
|
||||
where
|
||||
Fut1: Future,
|
||||
Fut1::Output: IntoFuture,
|
||||
{
|
||||
type Output = <Fut1::Output as IntoFuture>::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let Self { state } = unsafe { self.get_unchecked_mut() };
|
||||
loop {
|
||||
match state {
|
||||
State::First(fut1) => {
|
||||
let fut2 = ready!(unsafe { Pin::new_unchecked(fut1) }.poll(cx)).into_future();
|
||||
*state = State::Second(fut2);
|
||||
}
|
||||
State::Second(fut2) => {
|
||||
let v = ready!(unsafe { Pin::new_unchecked(fut2) }.poll(cx));
|
||||
*state = State::Empty;
|
||||
return Poll::Ready(v);
|
||||
}
|
||||
State::Empty => panic!("polled a completed future"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use async_macros::MaybeDone;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
use std::future::Future;
|
||||
|
||||
pin_project! {
|
||||
#[allow(missing_docs)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct Join<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>
|
||||
{
|
||||
#[pin] left: MaybeDone<L>,
|
||||
#[pin] right: MaybeDone<R>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R> Join<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
pub(crate) fn new(left: L, right: R) -> Self {
|
||||
Self {
|
||||
left: MaybeDone::new(left),
|
||||
right: MaybeDone::new(right),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R> Future for Join<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
type Output = (L::Output, R::Output);
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
|
||||
let mut left = this.left;
|
||||
let mut right = this.right;
|
||||
|
||||
if Future::poll(Pin::new(&mut left), cx).is_ready() {
|
||||
if right.as_ref().output().is_some() {
|
||||
return Poll::Ready((left.take().unwrap(), right.take().unwrap()));
|
||||
}
|
||||
}
|
||||
|
||||
if Future::poll(Pin::new(&mut right), cx).is_ready() {
|
||||
if left.as_ref().output().is_some() {
|
||||
return Poll::Ready((left.take().unwrap(), right.take().unwrap()));
|
||||
}
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,395 @@
|
||||
cfg_unstable! {
|
||||
mod delay;
|
||||
mod flatten;
|
||||
mod race;
|
||||
mod try_race;
|
||||
mod join;
|
||||
mod try_join;
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use delay::DelayFuture;
|
||||
use flatten::FlattenFuture;
|
||||
use crate::future::IntoFuture;
|
||||
use race::Race;
|
||||
use try_race::TryRace;
|
||||
use join::Join;
|
||||
use try_join::TryJoin;
|
||||
}
|
||||
|
||||
extension_trait! {
|
||||
use std::pin::Pin;
|
||||
use std::ops::{Deref, DerefMut};
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
#[doc = r#"
|
||||
A future represents an asynchronous computation.
|
||||
|
||||
A future is a value that may not have finished computing yet. This kind of
|
||||
"asynchronous value" makes it possible for a thread to continue doing useful
|
||||
work while it waits for the value to become available.
|
||||
|
||||
The [provided methods] do not really exist in the trait itself, but they become
|
||||
available when [`FutureExt`] from the [prelude] is imported:
|
||||
|
||||
```
|
||||
# #[allow(unused_imports)]
|
||||
use async_std::prelude::*;
|
||||
```
|
||||
|
||||
# The `poll` method
|
||||
|
||||
The core method of future, `poll`, *attempts* to resolve the future into a
|
||||
final value. This method does not block if the value is not ready. Instead,
|
||||
the current task is scheduled to be woken up when it's possible to make
|
||||
further progress by `poll`ing again. The `context` passed to the `poll`
|
||||
method can provide a [`Waker`], which is a handle for waking up the current
|
||||
task.
|
||||
|
||||
When using a future, you generally won't call `poll` directly, but instead
|
||||
`.await` the value.
|
||||
|
||||
[`Waker`]: ../task/struct.Waker.html
|
||||
[provided methods]: #provided-methods
|
||||
[`FutureExt`]: ../prelude/trait.FutureExt.html
|
||||
[prelude]: ../prelude/index.html
|
||||
"#]
|
||||
pub trait Future {
|
||||
#[doc = r#"
|
||||
The type of value produced on completion.
|
||||
"#]
|
||||
type Output;
|
||||
|
||||
#[doc = r#"
|
||||
Attempt to resolve the future to a final value, registering
|
||||
the current task for wakeup if the value is not yet available.
|
||||
|
||||
# Return value
|
||||
|
||||
This function returns:
|
||||
|
||||
- [`Poll::Pending`] if the future is not ready yet
|
||||
- [`Poll::Ready(val)`] with the result `val` of this future if it
|
||||
finished successfully.
|
||||
|
||||
Once a future has finished, clients should not `poll` it again.
|
||||
|
||||
When a future is not ready yet, `poll` returns `Poll::Pending` and
|
||||
stores a clone of the [`Waker`] copied from the current [`Context`].
|
||||
This [`Waker`] is then woken once the future can make progress.
|
||||
For example, a future waiting for a socket to become
|
||||
readable would call `.clone()` on the [`Waker`] and store it.
|
||||
When a signal arrives elsewhere indicating that the socket is readable,
|
||||
[`Waker::wake`] is called and the socket future's task is awoken.
|
||||
Once a task has been woken up, it should attempt to `poll` the future
|
||||
again, which may or may not produce a final value.
|
||||
|
||||
Note that on multiple calls to `poll`, only the [`Waker`] from the
|
||||
[`Context`] passed to the most recent call should be scheduled to
|
||||
receive a wakeup.
|
||||
|
||||
# Runtime characteristics
|
||||
|
||||
Futures alone are *inert*; they must be *actively* `poll`ed to make
|
||||
progress, meaning that each time the current task is woken up, it should
|
||||
actively re-`poll` pending futures that it still has an interest in.
|
||||
|
||||
The `poll` function is not called repeatedly in a tight loop -- instead,
|
||||
it should only be called when the future indicates that it is ready to
|
||||
make progress (by calling `wake()`). If you're familiar with the
|
||||
`poll(2)` or `select(2)` syscalls on Unix it's worth noting that futures
|
||||
typically do *not* suffer the same problems of "all wakeups must poll
|
||||
all events"; they are more like `epoll(4)`.
|
||||
|
||||
An implementation of `poll` should strive to return quickly, and should
|
||||
not block. Returning quickly prevents unnecessarily clogging up
|
||||
threads or event loops. If it is known ahead of time that a call to
|
||||
`poll` may end up taking awhile, the work should be offloaded to a
|
||||
thread pool (or something similar) to ensure that `poll` can return
|
||||
quickly.
|
||||
|
||||
# Panics
|
||||
|
||||
Once a future has completed (returned `Ready` from `poll`), calling its
|
||||
`poll` method again may panic, block forever, or cause other kinds of
|
||||
problems; the `Future` trait places no requirements on the effects of
|
||||
such a call. However, as the `poll` method is not marked `unsafe`,
|
||||
Rust's usual rules apply: calls must never cause undefined behavior
|
||||
(memory corruption, incorrect use of `unsafe` functions, or the like),
|
||||
regardless of the future's state.
|
||||
|
||||
[`Poll::Pending`]: ../task/enum.Poll.html#variant.Pending
|
||||
[`Poll::Ready(val)`]: ../task/enum.Poll.html#variant.Ready
|
||||
[`Context`]: ../task/struct.Context.html
|
||||
[`Waker`]: ../task/struct.Waker.html
|
||||
[`Waker::wake`]: ../task/struct.Waker.html#method.wake
|
||||
"#]
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output>;
|
||||
}
|
||||
|
||||
#[doc = r#"
|
||||
Extension methods for [`Future`].
|
||||
|
||||
[`Future`]: ../future/trait.Future.html
|
||||
"#]
|
||||
pub trait FutureExt: std::future::Future {
|
||||
/// Returns a Future that delays execution for a specified time.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// # async_std::task::block_on(async {
|
||||
/// use async_std::prelude::*;
|
||||
/// use async_std::future;
|
||||
/// use std::time::Duration;
|
||||
///
|
||||
/// let a = future::ready(1).delay(Duration::from_millis(2000));
|
||||
/// dbg!(a.await);
|
||||
/// # })
|
||||
/// ```
|
||||
#[cfg(all(feature = "default", feature = "unstable"))]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
fn delay(self, dur: Duration) -> impl Future<Output = Self::Output> [DelayFuture<Self>]
|
||||
where
|
||||
Self: Sized,
|
||||
{
|
||||
DelayFuture::new(self, dur)
|
||||
}
|
||||
|
||||
/// Flatten out the execution of this future when the result itself
|
||||
/// can be converted into another future.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// # async_std::task::block_on(async {
|
||||
/// use async_std::prelude::*;
|
||||
///
|
||||
/// let nested_future = async { async { 1 } };
|
||||
/// let future = nested_future.flatten();
|
||||
/// assert_eq!(future.await, 1);
|
||||
/// # })
|
||||
/// ```
|
||||
#[cfg(feature = "unstable")]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
fn flatten(
|
||||
self,
|
||||
) -> impl Future<Output = <Self::Output as IntoFuture>::Output>
|
||||
[FlattenFuture<Self, <Self::Output as IntoFuture>::Future>]
|
||||
where
|
||||
Self: Sized,
|
||||
<Self as Future>::Output: IntoFuture,
|
||||
{
|
||||
FlattenFuture::new(self)
|
||||
}
|
||||
|
||||
#[doc = r#"
|
||||
Waits for one of two similarly-typed futures to complete.
|
||||
|
||||
Awaits multiple futures simultaneously, returning the output of the
|
||||
first future that completes.
|
||||
|
||||
This function will return a new future which awaits for either one of both
|
||||
futures to complete. If multiple futures are completed at the same time,
|
||||
resolution will occur in the order that they have been passed.
|
||||
|
||||
Note that this function consumes all futures passed, and once a future is
|
||||
completed, all other futures are dropped.
|
||||
|
||||
# Examples
|
||||
|
||||
```
|
||||
# async_std::task::block_on(async {
|
||||
use async_std::prelude::*;
|
||||
use async_std::future;
|
||||
|
||||
let a = future::pending();
|
||||
let b = future::ready(1u8);
|
||||
let c = future::ready(2u8);
|
||||
|
||||
let f = a.race(b).race(c);
|
||||
assert_eq!(f.await, 1u8);
|
||||
# });
|
||||
```
|
||||
"#]
|
||||
#[cfg(feature = "unstable")]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
fn race<F>(
|
||||
self,
|
||||
other: F,
|
||||
) -> impl Future<Output = <Self as std::future::Future>::Output> [Race<Self, F>]
|
||||
where
|
||||
Self: std::future::Future + Sized,
|
||||
F: std::future::Future<Output = <Self as std::future::Future>::Output>,
|
||||
{
|
||||
Race::new(self, other)
|
||||
}
|
||||
|
||||
#[doc = r#"
|
||||
Waits for one of two similarly-typed fallible futures to complete.
|
||||
|
||||
Awaits multiple futures simultaneously, returning all results once complete.
|
||||
|
||||
`try_race` is similar to [`race`], but keeps going if a future
|
||||
resolved to an error until all futures have been resolved. In which case
|
||||
an error is returned.
|
||||
|
||||
The ordering of which value is yielded when two futures resolve
|
||||
simultaneously is intentionally left unspecified.
|
||||
|
||||
[`race`]: #method.race
|
||||
|
||||
# Examples
|
||||
|
||||
```
|
||||
# fn main() -> std::io::Result<()> { async_std::task::block_on(async {
|
||||
#
|
||||
use async_std::prelude::*;
|
||||
use async_std::future;
|
||||
use std::io::{Error, ErrorKind};
|
||||
|
||||
let a = future::pending::<Result<_, Error>>();
|
||||
let b = future::ready(Err(Error::from(ErrorKind::Other)));
|
||||
let c = future::ready(Ok(1u8));
|
||||
|
||||
let f = a.try_race(b).try_race(c);
|
||||
assert_eq!(f.await?, 1u8);
|
||||
#
|
||||
# Ok(()) }) }
|
||||
```
|
||||
"#]
|
||||
#[cfg(feature = "unstable")]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
fn try_race<F, T, E>(
|
||||
self,
|
||||
other: F
|
||||
) -> impl Future<Output = <Self as std::future::Future>::Output> [TryRace<Self, F>]
|
||||
where
|
||||
Self: std::future::Future<Output = Result<T, E>> + Sized,
|
||||
F: std::future::Future<Output = <Self as std::future::Future>::Output>,
|
||||
{
|
||||
TryRace::new(self, other)
|
||||
}
|
||||
|
||||
#[doc = r#"
|
||||
Waits for two similarly-typed futures to complete.
|
||||
|
||||
Awaits multiple futures simultaneously, returning the output of the
|
||||
futures once both complete.
|
||||
|
||||
This function returns a new future which polls both futures
|
||||
concurrently.
|
||||
|
||||
# Examples
|
||||
|
||||
```
|
||||
# async_std::task::block_on(async {
|
||||
use async_std::prelude::*;
|
||||
use async_std::future;
|
||||
|
||||
let a = future::ready(1u8);
|
||||
let b = future::ready(2u8);
|
||||
|
||||
let f = a.join(b);
|
||||
assert_eq!(f.await, (1u8, 2u8));
|
||||
# });
|
||||
```
|
||||
"#]
|
||||
#[cfg(any(feature = "unstable", feature = "docs"))]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
fn join<F>(
|
||||
self,
|
||||
other: F
|
||||
) -> impl Future<Output = (<Self as std::future::Future>::Output, <F as std::future::Future>::Output)> [Join<Self, F>]
|
||||
where
|
||||
Self: std::future::Future + Sized,
|
||||
F: std::future::Future<Output = <Self as std::future::Future>::Output>,
|
||||
{
|
||||
Join::new(self, other)
|
||||
}
|
||||
|
||||
#[doc = r#"
|
||||
Waits for two similarly-typed fallible futures to complete.
|
||||
|
||||
Awaits multiple futures simultaneously, returning all results once
|
||||
complete.
|
||||
|
||||
`try_join` is similar to [`join`], but returns an error immediately
|
||||
if a future resolves to an error.
|
||||
|
||||
[`join`]: #method.join
|
||||
|
||||
# Examples
|
||||
|
||||
```
|
||||
# fn main() -> std::io::Result<()> { async_std::task::block_on(async {
|
||||
#
|
||||
use async_std::prelude::*;
|
||||
use async_std::future;
|
||||
|
||||
let a = future::ready(Err("Error"));
|
||||
let b = future::ready(Ok(1u8));
|
||||
|
||||
let f = a.try_join(b);
|
||||
assert_eq!(f.await, Err("Error"));
|
||||
|
||||
let a = future::ready(Ok::<u8, String>(1u8));
|
||||
let b = future::ready(Ok::<u8, String>(2u8));
|
||||
|
||||
let f = a.try_join(b);
|
||||
assert_eq!(f.await, Ok((1u8, 2u8)));
|
||||
#
|
||||
# Ok(()) }) }
|
||||
```
|
||||
"#]
|
||||
#[cfg(any(feature = "unstable", feature = "docs"))]
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
fn try_join<F, T, E>(
|
||||
self,
|
||||
other: F
|
||||
) -> impl Future<Output = Result<(T, T), E>> [TryJoin<Self, F>]
|
||||
where
|
||||
Self: std::future::Future<Output = Result<T, E>> + Sized,
|
||||
F: std::future::Future<Output = <Self as std::future::Future>::Output>,
|
||||
{
|
||||
TryJoin::new(self, other)
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future + Unpin + ?Sized> Future for Box<F> {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future + Unpin + ?Sized> Future for &mut F {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
|
||||
impl<P> Future for Pin<P>
|
||||
where
|
||||
P: DerefMut + Unpin,
|
||||
<P as Deref>::Target: Future,
|
||||
{
|
||||
type Output = <<P as Deref>::Target as Future>::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future> Future for std::panic::AssertUnwindSafe<F> {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
unreachable!("this impl only appears in the rendered docs")
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,57 @@
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
|
||||
use async_macros::MaybeDone;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
#[allow(missing_docs)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct Race<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>
|
||||
{
|
||||
#[pin] left: MaybeDone<L>,
|
||||
#[pin] right: MaybeDone<R>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R> Race<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
pub(crate) fn new(left: L, right: R) -> Self {
|
||||
Self {
|
||||
left: MaybeDone::new(left),
|
||||
right: MaybeDone::new(right),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R> Future for Race<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
type Output = L::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
|
||||
let mut left = this.left;
|
||||
if Future::poll(Pin::new(&mut left), cx).is_ready() {
|
||||
return Poll::Ready(left.take().unwrap());
|
||||
}
|
||||
|
||||
let mut right = this.right;
|
||||
if Future::poll(Pin::new(&mut right), cx).is_ready() {
|
||||
return Poll::Ready(right.take().unwrap());
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,72 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use async_macros::MaybeDone;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
use std::future::Future;
|
||||
|
||||
pin_project! {
|
||||
#[allow(missing_docs)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct TryJoin<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>
|
||||
{
|
||||
#[pin] left: MaybeDone<L>,
|
||||
#[pin] right: MaybeDone<R>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R> TryJoin<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
pub(crate) fn new(left: L, right: R) -> Self {
|
||||
Self {
|
||||
left: MaybeDone::new(left),
|
||||
right: MaybeDone::new(right),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R, T, E> Future for TryJoin<L, R>
|
||||
where
|
||||
L: Future<Output = Result<T, E>>,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
type Output = Result<(T, T), E>;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
|
||||
let mut left = this.left;
|
||||
let mut right = this.right;
|
||||
|
||||
if Future::poll(Pin::new(&mut left), cx).is_ready() {
|
||||
if left.as_ref().output().unwrap().is_err() {
|
||||
return Poll::Ready(Err(left.take().unwrap().err().unwrap()));
|
||||
} else if right.as_ref().output().is_some() {
|
||||
return Poll::Ready(Ok((
|
||||
left.take().unwrap().ok().unwrap(),
|
||||
right.take().unwrap().ok().unwrap(),
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
if Future::poll(Pin::new(&mut right), cx).is_ready() {
|
||||
if right.as_ref().output().unwrap().is_err() {
|
||||
return Poll::Ready(Err(right.take().unwrap().err().unwrap()));
|
||||
} else if left.as_ref().output().is_some() {
|
||||
return Poll::Ready(Ok((
|
||||
left.take().unwrap().ok().unwrap(),
|
||||
right.take().unwrap().ok().unwrap(),
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,66 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use async_macros::MaybeDone;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::task::{Context, Poll};
|
||||
use std::future::Future;
|
||||
|
||||
pin_project! {
|
||||
#[allow(missing_docs)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct TryRace<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>
|
||||
{
|
||||
#[pin] left: MaybeDone<L>,
|
||||
#[pin] right: MaybeDone<R>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R> TryRace<L, R>
|
||||
where
|
||||
L: Future,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
pub(crate) fn new(left: L, right: R) -> Self {
|
||||
Self {
|
||||
left: MaybeDone::new(left),
|
||||
right: MaybeDone::new(right),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, R, T, E> Future for TryRace<L, R>
|
||||
where
|
||||
L: Future<Output = Result<T, E>>,
|
||||
R: Future<Output = L::Output>,
|
||||
{
|
||||
type Output = L::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
let mut left_errored = false;
|
||||
|
||||
// Check if the left future is ready & successful. Continue if not.
|
||||
let mut left = this.left;
|
||||
if Future::poll(Pin::new(&mut left), cx).is_ready() {
|
||||
if left.as_ref().output().unwrap().is_ok() {
|
||||
return Poll::Ready(left.take().unwrap());
|
||||
} else {
|
||||
left_errored = true;
|
||||
}
|
||||
}
|
||||
|
||||
// Check if the right future is ready & successful. Return err if left
|
||||
// future also resolved to err. Continue if not.
|
||||
let mut right = this.right;
|
||||
let is_ready = Future::poll(Pin::new(&mut right), cx).is_ready();
|
||||
if is_ready && (right.as_ref().output().unwrap().is_ok() || left_errored) {
|
||||
return Poll::Ready(right.take().unwrap());
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
@ -1,46 +1,51 @@
|
||||
use std::mem;
|
||||
use std::pin::Pin;
|
||||
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use super::read_until_internal;
|
||||
use crate::io::{self, BufRead};
|
||||
use crate::stream::Stream;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
|
||||
///
|
||||
/// This stream is created by the [`split`] method on types that implement [`BufRead`].
|
||||
///
|
||||
/// This type is an async version of [`std::io::Split`].
|
||||
///
|
||||
/// [`split`]: trait.BufRead.html#method.lines
|
||||
/// [`BufRead`]: trait.BufRead.html
|
||||
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
|
||||
#[derive(Debug)]
|
||||
pub struct Split<R> {
|
||||
pub(crate) reader: R,
|
||||
pub(crate) buf: Vec<u8>,
|
||||
pub(crate) read: usize,
|
||||
pub(crate) delim: u8,
|
||||
pin_project! {
|
||||
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
|
||||
///
|
||||
/// This stream is created by the [`split`] method on types that implement [`BufRead`].
|
||||
///
|
||||
/// This type is an async version of [`std::io::Split`].
|
||||
///
|
||||
/// [`split`]: trait.BufRead.html#method.lines
|
||||
/// [`BufRead`]: trait.BufRead.html
|
||||
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
|
||||
#[derive(Debug)]
|
||||
pub struct Split<R> {
|
||||
#[pin]
|
||||
pub(crate) reader: R,
|
||||
pub(crate) buf: Vec<u8>,
|
||||
pub(crate) read: usize,
|
||||
pub(crate) delim: u8,
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: BufRead> Stream for Split<R> {
|
||||
type Item = io::Result<Vec<u8>>;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let Self {
|
||||
reader,
|
||||
buf,
|
||||
read,
|
||||
delim,
|
||||
} = unsafe { self.get_unchecked_mut() };
|
||||
let reader = unsafe { Pin::new_unchecked(reader) };
|
||||
let n = futures_core::ready!(read_until_internal(reader, cx, *delim, buf, read))?;
|
||||
if n == 0 && buf.is_empty() {
|
||||
let this = self.project();
|
||||
let n = futures_core::ready!(read_until_internal(
|
||||
this.reader,
|
||||
cx,
|
||||
*this.delim,
|
||||
this.buf,
|
||||
this.read
|
||||
))?;
|
||||
if n == 0 && this.buf.is_empty() {
|
||||
return Poll::Ready(None);
|
||||
}
|
||||
if buf[buf.len() - 1] == *delim {
|
||||
buf.pop();
|
||||
if this.buf[this.buf.len() - 1] == *this.delim {
|
||||
this.buf.pop();
|
||||
}
|
||||
Poll::Ready(Some(Ok(mem::replace(buf, vec![]))))
|
||||
Poll::Ready(Some(Ok(mem::replace(this.buf, vec![]))))
|
||||
}
|
||||
}
|
||||
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in New Issue