forked from mirror/async-std
Compare commits
1 Commits
master
...
yoshuawuyt
| Author | SHA1 | Date |
|---|---|---|
Yoshua Wuyts
|
b7eeed7785 | 5 years ago |
@ -1,3 +0,0 @@
|
|||||||
Our contribution policy can be found at [async.rs/contribute][policy].
|
|
||||||
|
|
||||||
[policy]: https://async.rs/contribute/
|
|
||||||
@ -1,40 +0,0 @@
|
|||||||
#![feature(test)]
|
|
||||||
|
|
||||||
extern crate test;
|
|
||||||
|
|
||||||
use async_std::sync::{Arc, 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;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,11 +0,0 @@
|
|||||||
#![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 {}));
|
|
||||||
}
|
|
||||||
@ -1,266 +0,0 @@
|
|||||||
# Production-Ready Accept Loop
|
|
||||||
|
|
||||||
A production-ready accept loop needs the following things:
|
|
||||||
1. Handling errors
|
|
||||||
2. Limiting the number of simultanteous connections to avoid deny-of-service
|
|
||||||
(DoS) attacks
|
|
||||||
|
|
||||||
|
|
||||||
## Handling errors
|
|
||||||
|
|
||||||
There are two kinds of errors in an accept loop:
|
|
||||||
1. Per-connection errors. The system uses them to notify that there was a
|
|
||||||
connection in the queue and it's dropped by the peer. Subsequent connections
|
|
||||||
can be already queued so next connection must be accepted immediately.
|
|
||||||
2. Resource shortages. When these are encountered it doesn't make sense to
|
|
||||||
accept the next socket immediately. But the listener stays active, so you server
|
|
||||||
should try to accept socket later.
|
|
||||||
|
|
||||||
Here is the example of a per-connection error (printed in normal and debug mode):
|
|
||||||
```
|
|
||||||
Error: Connection reset by peer (os error 104)
|
|
||||||
Error: Os { code: 104, kind: ConnectionReset, message: "Connection reset by peer" }
|
|
||||||
```
|
|
||||||
|
|
||||||
And the following is the most common example of a resource shortage error:
|
|
||||||
```
|
|
||||||
Error: Too many open files (os error 24)
|
|
||||||
Error: Os { code: 24, kind: Other, message: "Too many open files" }
|
|
||||||
```
|
|
||||||
|
|
||||||
### Testing Application
|
|
||||||
|
|
||||||
To test your application for these errors try the following (this works
|
|
||||||
on unixes only).
|
|
||||||
|
|
||||||
Lower limits and start the application:
|
|
||||||
```
|
|
||||||
$ ulimit -n 100
|
|
||||||
$ cargo run --example your_app
|
|
||||||
Compiling your_app v0.1.0 (/work)
|
|
||||||
Finished dev [unoptimized + debuginfo] target(s) in 5.47s
|
|
||||||
Running `target/debug/examples/your_app`
|
|
||||||
Server is listening on: http://127.0.0.1:1234
|
|
||||||
```
|
|
||||||
Then in another console run the [`wrk`] benchmark tool:
|
|
||||||
```
|
|
||||||
$ wrk -c 1000 http://127.0.0.1:1234
|
|
||||||
Running 10s test @ http://localhost:8080/
|
|
||||||
2 threads and 1000 connections
|
|
||||||
$ telnet localhost 1234
|
|
||||||
Trying ::1...
|
|
||||||
Connected to localhost.
|
|
||||||
```
|
|
||||||
|
|
||||||
Important is to check the following things:
|
|
||||||
|
|
||||||
1. The application doesn't crash on error (but may log errors, see below)
|
|
||||||
2. It's possible to connect to the application again once load is stopped
|
|
||||||
(few seconds after `wrk`). This is what `telnet` does in example above,
|
|
||||||
make sure it prints `Connected to <hostname>`.
|
|
||||||
3. The `Too many open files` error is logged in the appropriate log. This
|
|
||||||
requires to set "maximum number of simultaneous connections" parameter (see
|
|
||||||
below) of your application to a value greater then `100` for this example.
|
|
||||||
4. Check CPU usage of the app while doing a test. It should not occupy 100%
|
|
||||||
of a single CPU core (it's unlikely that you can exhaust CPU by 1000
|
|
||||||
connections in Rust, so this means error handling is not right).
|
|
||||||
|
|
||||||
#### Testing non-HTTP applications
|
|
||||||
|
|
||||||
If it's possible, use the appropriate benchmark tool and set the appropriate
|
|
||||||
number of connections. For example `redis-benchmark` has a `-c` parameter for
|
|
||||||
that, if you implement redis protocol.
|
|
||||||
|
|
||||||
Alternatively, can still use `wrk`, just make sure that connection is not
|
|
||||||
immediately closed. If it is, put a temporary timeout before handing
|
|
||||||
the connection to the protocol handler, like this:
|
|
||||||
|
|
||||||
```rust,edition2018
|
|
||||||
# extern crate async_std;
|
|
||||||
# use std::time::Duration;
|
|
||||||
# use async_std::{
|
|
||||||
# net::{TcpListener, ToSocketAddrs},
|
|
||||||
# prelude::*,
|
|
||||||
# };
|
|
||||||
#
|
|
||||||
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
|
|
||||||
#
|
|
||||||
#async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
|
|
||||||
# let listener = TcpListener::bind(addr).await?;
|
|
||||||
# let mut incoming = listener.incoming();
|
|
||||||
while let Some(stream) = incoming.next().await {
|
|
||||||
task::spawn(async {
|
|
||||||
task::sleep(Duration::from_secs(10)).await; // 1
|
|
||||||
connection_loop(stream).await;
|
|
||||||
});
|
|
||||||
}
|
|
||||||
# Ok(())
|
|
||||||
# }
|
|
||||||
```
|
|
||||||
|
|
||||||
1. Make sure the sleep coroutine is inside the spawned task, not in the loop.
|
|
||||||
|
|
||||||
[`wrk`]: https://github.com/wg/wrk
|
|
||||||
|
|
||||||
|
|
||||||
### Handling Errors Manually
|
|
||||||
|
|
||||||
Here is how basic accept loop could look like:
|
|
||||||
|
|
||||||
```rust,edition2018
|
|
||||||
# extern crate async_std;
|
|
||||||
# use std::time::Duration;
|
|
||||||
# use async_std::{
|
|
||||||
# net::{TcpListener, ToSocketAddrs},
|
|
||||||
# prelude::*,
|
|
||||||
# };
|
|
||||||
#
|
|
||||||
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
|
|
||||||
#
|
|
||||||
async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
|
|
||||||
let listener = TcpListener::bind(addr).await?;
|
|
||||||
let mut incoming = listener.incoming();
|
|
||||||
while let Some(result) = incoming.next().await {
|
|
||||||
let stream = match stream {
|
|
||||||
Err(ref e) if is_connection_error(e) => continue, // 1
|
|
||||||
Err(e) => {
|
|
||||||
eprintln!("Error: {}. Pausing for 500ms."); // 3
|
|
||||||
task::sleep(Duration::from_millis(500)).await; // 2
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
Ok(s) => s,
|
|
||||||
};
|
|
||||||
// body
|
|
||||||
}
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
1. Ignore per-connection errors.
|
|
||||||
2. Sleep and continue on resource shortage.
|
|
||||||
3. It's important to log the message, because these errors commonly mean the
|
|
||||||
misconfiguration of the system and are helpful for operations people running
|
|
||||||
the application.
|
|
||||||
|
|
||||||
Be sure to [test your application](#testing-application).
|
|
||||||
|
|
||||||
|
|
||||||
### External Crates
|
|
||||||
|
|
||||||
The crate [`async-listen`] has a helper to achieve this task:
|
|
||||||
```rust,edition2018
|
|
||||||
# extern crate async_std;
|
|
||||||
# extern crate async_listen;
|
|
||||||
# use std::time::Duration;
|
|
||||||
# use async_std::{
|
|
||||||
# net::{TcpListener, ToSocketAddrs},
|
|
||||||
# prelude::*,
|
|
||||||
# };
|
|
||||||
#
|
|
||||||
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
|
|
||||||
#
|
|
||||||
use async_listen::{ListenExt, error_hint};
|
|
||||||
|
|
||||||
async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
|
|
||||||
|
|
||||||
let listener = TcpListener::bind(addr).await?;
|
|
||||||
let mut incoming = listener
|
|
||||||
.incoming()
|
|
||||||
.log_warnings(log_accept_error) // 1
|
|
||||||
.handle_errors(Duration::from_millis(500));
|
|
||||||
while let Some(socket) = incoming.next().await { // 2
|
|
||||||
// body
|
|
||||||
}
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
fn log_accept_error(e: &io::Error) {
|
|
||||||
eprintln!("Error: {}. Listener paused for 0.5s. {}", e, error_hint(e)) // 3
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
1. Logs resource shortages (`async-listen` calls them warnings). If you use
|
|
||||||
`log` crate or any other in your app this should go to the log.
|
|
||||||
2. Stream yields sockets without `Result` wrapper after `handle_errors` because
|
|
||||||
all errors are already handled.
|
|
||||||
3. Together with the error we print a hint, which explains some errors for end
|
|
||||||
users. For example, it recommends increasing open file limit and gives
|
|
||||||
a link.
|
|
||||||
|
|
||||||
[`async-listen`]: https://crates.io/crates/async-listen/
|
|
||||||
|
|
||||||
Be sure to [test your application](#testing-application).
|
|
||||||
|
|
||||||
|
|
||||||
## Connections Limit
|
|
||||||
|
|
||||||
Even if you've applied everything described in
|
|
||||||
[Handling Errors](#handling-errors) section, there is still a problem.
|
|
||||||
|
|
||||||
Let's imagine you have a server that needs to open a file to process
|
|
||||||
client request. At some point, you might encounter the following situation:
|
|
||||||
|
|
||||||
1. There are as many client connection as max file descriptors allowed for
|
|
||||||
the application.
|
|
||||||
2. Listener gets `Too many open files` error so it sleeps.
|
|
||||||
3. Some client sends a request via the previously open connection.
|
|
||||||
4. Opening a file to serve request fails, because of the same
|
|
||||||
`Too many open files` error, until some other client drops a connection.
|
|
||||||
|
|
||||||
There are many more possible situations, this is just a small illustation that
|
|
||||||
limiting number of connections is very useful. Generally, it's one of the ways
|
|
||||||
to control resources used by a server and avoiding some kinds of deny of
|
|
||||||
service (DoS) attacks.
|
|
||||||
|
|
||||||
### `async-listen` crate
|
|
||||||
|
|
||||||
Limiting maximum number of simultaneous connections with [`async-listen`]
|
|
||||||
looks like the following:
|
|
||||||
|
|
||||||
```rust,edition2018
|
|
||||||
# extern crate async_std;
|
|
||||||
# extern crate async_listen;
|
|
||||||
# use std::time::Duration;
|
|
||||||
# use async_std::{
|
|
||||||
# net::{TcpListener, TcpStream, ToSocketAddrs},
|
|
||||||
# prelude::*,
|
|
||||||
# };
|
|
||||||
#
|
|
||||||
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
|
|
||||||
#
|
|
||||||
use async_listen::{ListenExt, Token, error_hint};
|
|
||||||
|
|
||||||
async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
|
|
||||||
|
|
||||||
let listener = TcpListener::bind(addr).await?;
|
|
||||||
let mut incoming = listener
|
|
||||||
.incoming()
|
|
||||||
.log_warnings(log_accept_error)
|
|
||||||
.handle_errors(Duration::from_millis(500)) // 1
|
|
||||||
.backpressure(100);
|
|
||||||
while let Some((token, socket)) = incoming.next().await { // 2
|
|
||||||
task::spawn(async move {
|
|
||||||
connection_loop(&token, stream).await; // 3
|
|
||||||
});
|
|
||||||
}
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
async fn connection_loop(_token: &Token, stream: TcpStream) { // 4
|
|
||||||
// ...
|
|
||||||
}
|
|
||||||
# fn log_accept_error(e: &io::Error) {
|
|
||||||
# eprintln!("Error: {}. Listener paused for 0.5s. {}", e, error_hint(e));
|
|
||||||
# }
|
|
||||||
```
|
|
||||||
|
|
||||||
1. We need to handle errors first, because [`backpressure`] helper expects
|
|
||||||
stream of `TcpStream` rather than `Result`.
|
|
||||||
2. The token yielded by a new stream is what is counted by backpressure helper.
|
|
||||||
I.e. if you drop a token, new connection can be established.
|
|
||||||
3. We give the connection loop a reference to token to bind token's lifetime to
|
|
||||||
the lifetime of the connection.
|
|
||||||
4. The token itsellf in the function can be ignored, hence `_token`
|
|
||||||
|
|
||||||
[`backpressure`]: https://docs.rs/async-listen/0.1.2/async_listen/trait.ListenExt.html#method.backpressure
|
|
||||||
|
|
||||||
Be sure to [test this behavior](#testing-application).
|
|
||||||
@ -1,14 +1,11 @@
|
|||||||
# Tutorial: Writing a chat
|
# Tutorial: Writing a chat
|
||||||
|
|
||||||
Nothing is simpler than creating a chat server, right?
|
Nothing is as simple as a chat server, right? Not quite, chat servers
|
||||||
Not quite, chat servers expose you to all the fun of asynchronous programming:
|
already expose you to all the fun of asynchronous programming: how
|
||||||
|
do you handle clients connecting concurrently. How do you handle them disconnecting?
|
||||||
|
|
||||||
How will the server handle clients connecting concurrently?
|
How do you distribute the messages?
|
||||||
|
|
||||||
How will it handle them disconnecting?
|
In this tutorial, we will show you how to write one in `async-std`.
|
||||||
|
|
||||||
How will it distribute the messages?
|
|
||||||
|
|
||||||
This tutorial explains how to write a chat server in `async-std`.
|
|
||||||
|
|
||||||
You can also find the tutorial in [our repository](https://github.com/async-rs/async-std/blob/master/examples/a-chat).
|
You can also find the tutorial in [our repository](https://github.com/async-rs/async-std/blob/master/examples/a-chat).
|
||||||
|
|||||||
@ -1,45 +0,0 @@
|
|||||||
//! 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 mut reader = stream.clone();
|
|
||||||
let mut writer = stream;
|
|
||||||
io::copy(&mut reader, &mut 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(())
|
|
||||||
})
|
|
||||||
}
|
|
||||||
@ -0,0 +1,169 @@
|
|||||||
|
cfg_unstable! {
|
||||||
|
mod delay;
|
||||||
|
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use delay::DelayFuture;
|
||||||
|
}
|
||||||
|
|
||||||
|
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 {
|
||||||
|
/// 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_attr(feature = "docs", doc(cfg(unstable)))]
|
||||||
|
#[cfg(any(feature = "unstable", feature = "docs"))]
|
||||||
|
fn delay(self, dur: Duration) -> impl Future<Output = Self::Output> [DelayFuture<Self>]
|
||||||
|
where
|
||||||
|
Self: Future + Sized
|
||||||
|
{
|
||||||
|
DelayFuture::new(self, dur)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -1,52 +0,0 @@
|
|||||||
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"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,60 +0,0 @@
|
|||||||
use std::pin::Pin;
|
|
||||||
|
|
||||||
use crate::future::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,
|
|
||||||
{
|
|
||||||
#[pin] left: MaybeDone<L>,
|
|
||||||
#[pin] right: MaybeDone<R>,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<L, R> Join<L, R>
|
|
||||||
where
|
|
||||||
L: Future,
|
|
||||||
R: Future,
|
|
||||||
{
|
|
||||||
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,
|
|
||||||
{
|
|
||||||
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;
|
|
||||||
|
|
||||||
let is_left_ready = Future::poll(Pin::new(&mut left), cx).is_ready();
|
|
||||||
if is_left_ready && right.as_ref().output().is_some() {
|
|
||||||
return Poll::Ready((left.take().unwrap(), right.take().unwrap()));
|
|
||||||
}
|
|
||||||
|
|
||||||
let is_right_ready = Future::poll(Pin::new(&mut right), cx).is_ready();
|
|
||||||
if is_right_ready && left.as_ref().output().is_some() {
|
|
||||||
return Poll::Ready((left.take().unwrap(), right.take().unwrap()));
|
|
||||||
}
|
|
||||||
|
|
||||||
Poll::Pending
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,432 +0,0 @@
|
|||||||
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;
|
|
||||||
}
|
|
||||||
|
|
||||||
cfg_unstable_default! {
|
|
||||||
use crate::future::timeout::TimeoutFuture;
|
|
||||||
}
|
|
||||||
|
|
||||||
extension_trait! {
|
|
||||||
use core::pin::Pin;
|
|
||||||
use core::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: core::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(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(2u16);
|
|
||||||
|
|
||||||
let f = a.join(b);
|
|
||||||
assert_eq!(f.await, (1u8, 2u16));
|
|
||||||
# });
|
|
||||||
```
|
|
||||||
"#]
|
|
||||||
#[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,
|
|
||||||
{
|
|
||||||
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::<u8, &str>("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::<u16, String>(2u16));
|
|
||||||
|
|
||||||
let f = a.try_join(b);
|
|
||||||
assert_eq!(f.await, Ok((1u8, 2u16)));
|
|
||||||
#
|
|
||||||
# Ok(()) }) }
|
|
||||||
```
|
|
||||||
"#]
|
|
||||||
#[cfg(any(feature = "unstable", feature = "docs"))]
|
|
||||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
|
||||||
fn try_join<F, A, B, E>(
|
|
||||||
self,
|
|
||||||
other: F
|
|
||||||
) -> impl Future<Output = Result<(A, B), E>> [TryJoin<Self, F>]
|
|
||||||
where
|
|
||||||
Self: std::future::Future<Output = Result<A, E>> + Sized,
|
|
||||||
F: std::future::Future<Output = Result<B, E>>,
|
|
||||||
{
|
|
||||||
TryJoin::new(self, other)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[doc = r#"
|
|
||||||
Waits for both the future and a timeout, if the timeout completes before
|
|
||||||
the future, it returns an TimeoutError.
|
|
||||||
|
|
||||||
# Example
|
|
||||||
```
|
|
||||||
# async_std::task::block_on(async {
|
|
||||||
#
|
|
||||||
use std::time::Duration;
|
|
||||||
|
|
||||||
use async_std::prelude::*;
|
|
||||||
use async_std::future;
|
|
||||||
|
|
||||||
let fut = future::ready(0);
|
|
||||||
let dur = Duration::from_millis(100);
|
|
||||||
let res = fut.timeout(dur).await;
|
|
||||||
assert!(res.is_ok());
|
|
||||||
|
|
||||||
let fut = future::pending::<()>();
|
|
||||||
let dur = Duration::from_millis(100);
|
|
||||||
let res = fut.timeout(dur).await;
|
|
||||||
assert!(res.is_err())
|
|
||||||
#
|
|
||||||
# });
|
|
||||||
```
|
|
||||||
"#]
|
|
||||||
#[cfg(any(all(feature = "default", feature = "unstable"), feature = "docs"))]
|
|
||||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
|
||||||
fn timeout(self, dur: Duration) -> impl Future<Output = Self::Output> [TimeoutFuture<Self>]
|
|
||||||
where Self: Sized
|
|
||||||
{
|
|
||||||
TimeoutFuture::new(self, dur)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
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")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,57 +0,0 @@
|
|||||||
use std::future::Future;
|
|
||||||
use std::pin::Pin;
|
|
||||||
|
|
||||||
use crate::future::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
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,72 +0,0 @@
|
|||||||
use std::pin::Pin;
|
|
||||||
|
|
||||||
use crate::future::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,
|
|
||||||
{
|
|
||||||
#[pin] left: MaybeDone<L>,
|
|
||||||
#[pin] right: MaybeDone<R>,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<L, R> TryJoin<L, R>
|
|
||||||
where
|
|
||||||
L: Future,
|
|
||||||
R: Future,
|
|
||||||
{
|
|
||||||
pub(crate) fn new(left: L, right: R) -> Self {
|
|
||||||
Self {
|
|
||||||
left: MaybeDone::new(left),
|
|
||||||
right: MaybeDone::new(right),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<L, R, A, B, E> Future for TryJoin<L, R>
|
|
||||||
where
|
|
||||||
L: Future<Output = Result<A, E>>,
|
|
||||||
R: Future<Output = Result<B, E>>,
|
|
||||||
{
|
|
||||||
type Output = Result<(A, B), 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
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,66 +0,0 @@
|
|||||||
use std::pin::Pin;
|
|
||||||
|
|
||||||
use crate::future::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,79 +0,0 @@
|
|||||||
//! A type that wraps a future to keep track of its completion status.
|
|
||||||
//!
|
|
||||||
//! This implementation was taken from the original `macro_rules` `join/try_join`
|
|
||||||
//! macros in the `futures-preview` crate.
|
|
||||||
|
|
||||||
use std::future::Future;
|
|
||||||
use std::mem;
|
|
||||||
use std::pin::Pin;
|
|
||||||
use std::task::{Context, Poll};
|
|
||||||
|
|
||||||
use futures_core::ready;
|
|
||||||
|
|
||||||
/// A future that may have completed.
|
|
||||||
#[derive(Debug)]
|
|
||||||
pub(crate) enum MaybeDone<Fut: Future> {
|
|
||||||
/// A not-yet-completed future
|
|
||||||
Future(Fut),
|
|
||||||
|
|
||||||
/// The output of the completed future
|
|
||||||
Done(Fut::Output),
|
|
||||||
|
|
||||||
/// The empty variant after the result of a [`MaybeDone`] has been
|
|
||||||
/// taken using the [`take`](MaybeDone::take) method.
|
|
||||||
Gone,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<Fut: Future> MaybeDone<Fut> {
|
|
||||||
/// Create a new instance of `MaybeDone`.
|
|
||||||
pub(crate) fn new(future: Fut) -> MaybeDone<Fut> {
|
|
||||||
Self::Future(future)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Returns an [`Option`] containing a reference to the output of the future.
|
|
||||||
/// The output of this method will be [`Some`] if and only if the inner
|
|
||||||
/// future has been completed and [`take`](MaybeDone::take)
|
|
||||||
/// has not yet been called.
|
|
||||||
#[inline]
|
|
||||||
pub(crate) fn output(self: Pin<&Self>) -> Option<&Fut::Output> {
|
|
||||||
let this = self.get_ref();
|
|
||||||
match this {
|
|
||||||
MaybeDone::Done(res) => Some(res),
|
|
||||||
_ => None,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Attempt to take the output of a `MaybeDone` without driving it
|
|
||||||
/// towards completion.
|
|
||||||
#[inline]
|
|
||||||
pub(crate) fn take(self: Pin<&mut Self>) -> Option<Fut::Output> {
|
|
||||||
unsafe {
|
|
||||||
let this = self.get_unchecked_mut();
|
|
||||||
match this {
|
|
||||||
MaybeDone::Done(_) => {}
|
|
||||||
MaybeDone::Future(_) | MaybeDone::Gone => return None,
|
|
||||||
};
|
|
||||||
if let MaybeDone::Done(output) = mem::replace(this, MaybeDone::Gone) {
|
|
||||||
Some(output)
|
|
||||||
} else {
|
|
||||||
unreachable!()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<Fut: Future> Future for MaybeDone<Fut> {
|
|
||||||
type Output = ();
|
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
|
||||||
let res = unsafe {
|
|
||||||
match Pin::as_mut(&mut self).get_unchecked_mut() {
|
|
||||||
MaybeDone::Future(a) => ready!(Pin::new_unchecked(a).poll(cx)),
|
|
||||||
MaybeDone::Done(_) => return Poll::Ready(()),
|
|
||||||
MaybeDone::Gone => panic!("MaybeDone polled after value taken"),
|
|
||||||
}
|
|
||||||
};
|
|
||||||
self.set(MaybeDone::Done(res));
|
|
||||||
Poll::Ready(())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,42 +0,0 @@
|
|||||||
use crate::utils::Context;
|
|
||||||
|
|
||||||
use std::{error::Error as StdError, fmt, io};
|
|
||||||
|
|
||||||
/// Wrap `std::io::Error` with additional message
|
|
||||||
///
|
|
||||||
/// Keeps the original error kind and stores the original I/O error as `source`.
|
|
||||||
impl<T> Context for Result<T, std::io::Error> {
|
|
||||||
fn context(self, message: impl Fn() -> String) -> Self {
|
|
||||||
self.map_err(|e| VerboseError::wrap(e, message()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug)]
|
|
||||||
pub(crate) struct VerboseError {
|
|
||||||
source: io::Error,
|
|
||||||
message: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl VerboseError {
|
|
||||||
pub(crate) fn wrap(source: io::Error, message: impl Into<String>) -> io::Error {
|
|
||||||
io::Error::new(
|
|
||||||
source.kind(),
|
|
||||||
VerboseError {
|
|
||||||
source,
|
|
||||||
message: message.into(),
|
|
||||||
},
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl fmt::Display for VerboseError {
|
|
||||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
||||||
write!(f, "{}", self.message)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl StdError for VerboseError {
|
|
||||||
fn source(&self) -> Option<&(dyn StdError + 'static)> {
|
|
||||||
Some(&self.source)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in New Issue