Merge branch 'master' into new-scheduler
k-nasa
5 years ago
commit
84e5c5f351
@ -0,0 +1,266 @@
|
|||||||
|
# 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).
|
||||||
@ -0,0 +1,79 @@
|
|||||||
|
//! 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,24 +0,0 @@
|
|||||||
use crate::stream::Stream;
|
|
||||||
|
|
||||||
use std::pin::Pin;
|
|
||||||
use std::task::{Context, Poll};
|
|
||||||
|
|
||||||
/// A stream able to yield elements from both ends.
|
|
||||||
///
|
|
||||||
/// Something that implements `DoubleEndedStream` has one extra capability
|
|
||||||
/// over something that implements [`Stream`]: the ability to also take
|
|
||||||
/// `Item`s from the back, as well as the front.
|
|
||||||
///
|
|
||||||
/// [`Stream`]: trait.Stream.html
|
|
||||||
#[cfg(feature = "unstable")]
|
|
||||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
|
||||||
pub trait DoubleEndedStream: Stream {
|
|
||||||
/// Removes and returns an element from the end of the stream.
|
|
||||||
///
|
|
||||||
/// Returns `None` when there are no more elements.
|
|
||||||
///
|
|
||||||
/// The [trait-level] docs contain more details.
|
|
||||||
///
|
|
||||||
/// [trait-level]: trait.DoubleEndedStream.html
|
|
||||||
fn poll_next_back(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
|
|
||||||
}
|
|
||||||
@ -0,0 +1,246 @@
|
|||||||
|
use crate::stream::Stream;
|
||||||
|
|
||||||
|
use std::pin::Pin;
|
||||||
|
use std::task::{Context, Poll};
|
||||||
|
|
||||||
|
mod next_back;
|
||||||
|
mod nth_back;
|
||||||
|
mod rfind;
|
||||||
|
mod rfold;
|
||||||
|
mod try_rfold;
|
||||||
|
|
||||||
|
use next_back::NextBackFuture;
|
||||||
|
use nth_back::NthBackFuture;
|
||||||
|
use rfind::RFindFuture;
|
||||||
|
use rfold::RFoldFuture;
|
||||||
|
use try_rfold::TryRFoldFuture;
|
||||||
|
|
||||||
|
/// A stream able to yield elements from both ends.
|
||||||
|
///
|
||||||
|
/// Something that implements `DoubleEndedStream` has one extra capability
|
||||||
|
/// over something that implements [`Stream`]: the ability to also take
|
||||||
|
/// `Item`s from the back, as well as the front.
|
||||||
|
///
|
||||||
|
/// [`Stream`]: trait.Stream.html
|
||||||
|
#[cfg(feature = "unstable")]
|
||||||
|
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||||
|
pub trait DoubleEndedStream: Stream {
|
||||||
|
#[doc = r#"
|
||||||
|
Attempts to receive the next item from the back of the stream.
|
||||||
|
|
||||||
|
There are several possible return values:
|
||||||
|
|
||||||
|
* `Poll::Pending` means this stream's next_back value is not ready yet.
|
||||||
|
* `Poll::Ready(None)` means this stream has been exhausted.
|
||||||
|
* `Poll::Ready(Some(item))` means `item` was received out of the stream.
|
||||||
|
|
||||||
|
# Examples
|
||||||
|
|
||||||
|
```
|
||||||
|
# fn main() { async_std::task::block_on(async {
|
||||||
|
#
|
||||||
|
use std::pin::Pin;
|
||||||
|
|
||||||
|
use async_std::prelude::*;
|
||||||
|
use async_std::stream;
|
||||||
|
use async_std::task::{Context, Poll};
|
||||||
|
|
||||||
|
fn increment(
|
||||||
|
s: impl DoubleEndedStream<Item = i32> + Unpin,
|
||||||
|
) -> impl DoubleEndedStream<Item = i32> + Unpin {
|
||||||
|
struct Increment<S>(S);
|
||||||
|
|
||||||
|
impl<S: DoubleEndedStream<Item = i32> + Unpin> Stream for Increment<S> {
|
||||||
|
type Item = S::Item;
|
||||||
|
|
||||||
|
fn poll_next(
|
||||||
|
mut self: Pin<&mut Self>,
|
||||||
|
cx: &mut Context<'_>,
|
||||||
|
) -> Poll<Option<Self::Item>> {
|
||||||
|
match Pin::new(&mut self.0).poll_next(cx) {
|
||||||
|
Poll::Pending => Poll::Pending,
|
||||||
|
Poll::Ready(None) => Poll::Ready(None),
|
||||||
|
Poll::Ready(Some(item)) => Poll::Ready(Some(item + 1)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<S: DoubleEndedStream<Item = i32> + Unpin> DoubleEndedStream for Increment<S> {
|
||||||
|
fn poll_next_back(
|
||||||
|
mut self: Pin<&mut Self>,
|
||||||
|
cx: &mut Context<'_>,
|
||||||
|
) -> Poll<Option<Self::Item>> {
|
||||||
|
match Pin::new(&mut self.0).poll_next_back(cx) {
|
||||||
|
Poll::Pending => Poll::Pending,
|
||||||
|
Poll::Ready(None) => Poll::Ready(None),
|
||||||
|
Poll::Ready(Some(item)) => Poll::Ready(Some(item + 1)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Increment(s)
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut s = increment(stream::once(7));
|
||||||
|
|
||||||
|
assert_eq!(s.next_back().await, Some(8));
|
||||||
|
assert_eq!(s.next_back().await, None);
|
||||||
|
#
|
||||||
|
# }) }
|
||||||
|
```
|
||||||
|
"#]
|
||||||
|
fn poll_next_back(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
|
||||||
|
|
||||||
|
#[doc = r#"
|
||||||
|
Advances the stream and returns the next value.
|
||||||
|
|
||||||
|
Returns [`None`] when iteration is finished. Individual stream implementations may
|
||||||
|
choose to resume iteration, and so calling `next()` again may or may not eventually
|
||||||
|
start returning more values.
|
||||||
|
|
||||||
|
[`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None
|
||||||
|
|
||||||
|
# Examples
|
||||||
|
|
||||||
|
```
|
||||||
|
# fn main() { async_std::task::block_on(async {
|
||||||
|
#
|
||||||
|
use async_std::prelude::*;
|
||||||
|
use async_std::stream;
|
||||||
|
|
||||||
|
let mut s = stream::from_iter(vec![7u8]);
|
||||||
|
|
||||||
|
assert_eq!(s.next_back().await, Some(7));
|
||||||
|
assert_eq!(s.next_back().await, None);
|
||||||
|
#
|
||||||
|
# }) }
|
||||||
|
```
|
||||||
|
"#]
|
||||||
|
fn next_back(&mut self) -> NextBackFuture<'_, Self>
|
||||||
|
where
|
||||||
|
Self: Unpin,
|
||||||
|
{
|
||||||
|
NextBackFuture { stream: self }
|
||||||
|
}
|
||||||
|
|
||||||
|
#[doc = r#"
|
||||||
|
Returns the nth element from the back of the stream.
|
||||||
|
|
||||||
|
# Examples
|
||||||
|
|
||||||
|
Basic usage:
|
||||||
|
|
||||||
|
```
|
||||||
|
# fn main() { async_std::task::block_on(async {
|
||||||
|
#
|
||||||
|
use async_std::prelude::*;
|
||||||
|
use async_std::stream;
|
||||||
|
|
||||||
|
let mut s = stream::from_iter(vec![1u8, 2, 3, 4, 5]);
|
||||||
|
|
||||||
|
let second = s.nth_back(1).await;
|
||||||
|
assert_eq!(second, Some(4));
|
||||||
|
#
|
||||||
|
# }) }
|
||||||
|
```
|
||||||
|
"#]
|
||||||
|
fn nth_back(&mut self, n: usize) -> NthBackFuture<'_, Self>
|
||||||
|
where
|
||||||
|
Self: Unpin + Sized,
|
||||||
|
{
|
||||||
|
NthBackFuture::new(self, n)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[doc = r#"
|
||||||
|
Returns the the frist element from the right that matches the predicate.
|
||||||
|
|
||||||
|
# Examples
|
||||||
|
|
||||||
|
Basic usage:
|
||||||
|
|
||||||
|
```
|
||||||
|
# fn main() { async_std::task::block_on(async {
|
||||||
|
#
|
||||||
|
use async_std::prelude::*;
|
||||||
|
use async_std::stream;
|
||||||
|
|
||||||
|
let mut s = stream::from_iter(vec![1u8, 2, 3, 4, 5]);
|
||||||
|
|
||||||
|
let second = s.rfind(|v| v % 2 == 0).await;
|
||||||
|
assert_eq!(second, Some(4));
|
||||||
|
#
|
||||||
|
# }) }
|
||||||
|
```
|
||||||
|
"#]
|
||||||
|
fn rfind<P>(&mut self, p: P) -> RFindFuture<'_, Self, P>
|
||||||
|
where
|
||||||
|
Self: Unpin + Sized,
|
||||||
|
P: FnMut(&Self::Item) -> bool,
|
||||||
|
{
|
||||||
|
RFindFuture::new(self, p)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[doc = r#"
|
||||||
|
# Examples
|
||||||
|
|
||||||
|
Basic usage:
|
||||||
|
|
||||||
|
```
|
||||||
|
# fn main() { async_std::task::block_on(async {
|
||||||
|
#
|
||||||
|
use async_std::prelude::*;
|
||||||
|
use async_std::stream;
|
||||||
|
|
||||||
|
let s = stream::from_iter(vec![1u8, 2, 3, 4, 5]);
|
||||||
|
|
||||||
|
let second = s.rfold(0, |acc, v| v + acc).await;
|
||||||
|
|
||||||
|
assert_eq!(second, 15);
|
||||||
|
#
|
||||||
|
# }) }
|
||||||
|
```
|
||||||
|
"#]
|
||||||
|
fn rfold<B, F>(self, accum: B, f: F) -> RFoldFuture<Self, F, B>
|
||||||
|
where
|
||||||
|
Self: Sized,
|
||||||
|
F: FnMut(B, Self::Item) -> B,
|
||||||
|
{
|
||||||
|
RFoldFuture::new(self, accum, f)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[doc = r#"
|
||||||
|
A combinator that applies a function as long as it returns successfully, producing a single, final value.
|
||||||
|
Immediately returns the error when the function returns unsuccessfully.
|
||||||
|
|
||||||
|
# Examples
|
||||||
|
|
||||||
|
Basic usage:
|
||||||
|
|
||||||
|
```
|
||||||
|
# fn main() { async_std::task::block_on(async {
|
||||||
|
#
|
||||||
|
use async_std::prelude::*;
|
||||||
|
use async_std::stream;
|
||||||
|
|
||||||
|
let s = stream::from_iter(vec![1u8, 2, 3, 4, 5]);
|
||||||
|
let sum = s.try_rfold(0, |acc, v| {
|
||||||
|
if (acc+v) % 2 == 1 {
|
||||||
|
Ok(v+3)
|
||||||
|
} else {
|
||||||
|
Err("fail")
|
||||||
|
}
|
||||||
|
}).await;
|
||||||
|
|
||||||
|
assert_eq!(sum, Err("fail"));
|
||||||
|
#
|
||||||
|
# }) }
|
||||||
|
```
|
||||||
|
"#]
|
||||||
|
fn try_rfold<B, F, E>(self, accum: B, f: F) -> TryRFoldFuture<Self, F, B>
|
||||||
|
where
|
||||||
|
Self: Sized,
|
||||||
|
F: FnMut(B, Self::Item) -> Result<B, E>,
|
||||||
|
{
|
||||||
|
TryRFoldFuture::new(self, accum, f)
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,19 @@
|
|||||||
|
use core::pin::Pin;
|
||||||
|
use core::future::Future;
|
||||||
|
|
||||||
|
use crate::stream::DoubleEndedStream;
|
||||||
|
use crate::task::{Context, Poll};
|
||||||
|
|
||||||
|
#[doc(hidden)]
|
||||||
|
#[allow(missing_debug_implementations)]
|
||||||
|
pub struct NextBackFuture<'a, T: Unpin + ?Sized> {
|
||||||
|
pub(crate) stream: &'a mut T,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T: DoubleEndedStream + Unpin + ?Sized> Future for NextBackFuture<'_, T> {
|
||||||
|
type Output = Option<T::Item>;
|
||||||
|
|
||||||
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
|
Pin::new(&mut *self.stream).poll_next_back(cx)
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,41 @@
|
|||||||
|
use core::future::Future;
|
||||||
|
use core::pin::Pin;
|
||||||
|
use core::task::{Context, Poll};
|
||||||
|
|
||||||
|
use crate::stream::DoubleEndedStream;
|
||||||
|
|
||||||
|
#[doc(hidden)]
|
||||||
|
#[allow(missing_debug_implementations)]
|
||||||
|
pub struct NthBackFuture<'a, S> {
|
||||||
|
stream: &'a mut S,
|
||||||
|
n: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'a, S> NthBackFuture<'a, S> {
|
||||||
|
pub(crate) fn new(stream: &'a mut S, n: usize) -> Self {
|
||||||
|
NthBackFuture { stream, n }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'a, S> Future for NthBackFuture<'a, S>
|
||||||
|
where
|
||||||
|
S: DoubleEndedStream + Sized + Unpin,
|
||||||
|
{
|
||||||
|
type Output = Option<S::Item>;
|
||||||
|
|
||||||
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
|
let next = futures_core::ready!(Pin::new(&mut *self.stream).poll_next_back(cx));
|
||||||
|
match next {
|
||||||
|
Some(v) => match self.n {
|
||||||
|
0 => Poll::Ready(Some(v)),
|
||||||
|
_ => {
|
||||||
|
self.n -= 1;
|
||||||
|
cx.waker().wake_by_ref();
|
||||||
|
Poll::Pending
|
||||||
|
}
|
||||||
|
},
|
||||||
|
None => Poll::Ready(None),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@ -0,0 +1,41 @@
|
|||||||
|
use core::task::{Context, Poll};
|
||||||
|
use core::future::Future;
|
||||||
|
use core::pin::Pin;
|
||||||
|
|
||||||
|
use crate::stream::DoubleEndedStream;
|
||||||
|
|
||||||
|
#[doc(hidden)]
|
||||||
|
#[allow(missing_debug_implementations)]
|
||||||
|
pub struct RFindFuture<'a, S, P> {
|
||||||
|
stream: &'a mut S,
|
||||||
|
p: P,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'a, S, P> RFindFuture<'a, S, P> {
|
||||||
|
pub(super) fn new(stream: &'a mut S, p: P) -> Self {
|
||||||
|
RFindFuture { stream, p }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<S: Unpin, P> Unpin for RFindFuture<'_, S, P> {}
|
||||||
|
|
||||||
|
impl<'a, S, P> Future for RFindFuture<'a, S, P>
|
||||||
|
where
|
||||||
|
S: DoubleEndedStream + Unpin + Sized,
|
||||||
|
P: FnMut(&S::Item) -> bool,
|
||||||
|
{
|
||||||
|
type Output = Option<S::Item>;
|
||||||
|
|
||||||
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
|
let item = futures_core::ready!(Pin::new(&mut *self.stream).poll_next_back(cx));
|
||||||
|
|
||||||
|
match item {
|
||||||
|
Some(v) if (&mut self.p)(&v) => Poll::Ready(Some(v)),
|
||||||
|
Some(_) => {
|
||||||
|
cx.waker().wake_by_ref();
|
||||||
|
Poll::Pending
|
||||||
|
}
|
||||||
|
None => Poll::Ready(None),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,52 @@
|
|||||||
|
use core::future::Future;
|
||||||
|
use core::pin::Pin;
|
||||||
|
use core::task::{Context, Poll};
|
||||||
|
|
||||||
|
use pin_project_lite::pin_project;
|
||||||
|
|
||||||
|
use crate::stream::DoubleEndedStream;
|
||||||
|
|
||||||
|
pin_project! {
|
||||||
|
#[doc(hidden)]
|
||||||
|
#[allow(missing_debug_implementations)]
|
||||||
|
pub struct RFoldFuture<S, F, B> {
|
||||||
|
#[pin]
|
||||||
|
stream: S,
|
||||||
|
f: F,
|
||||||
|
acc: Option<B>,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<S, F, B> RFoldFuture<S, F, B> {
|
||||||
|
pub(super) fn new(stream: S, init: B, f: F) -> Self {
|
||||||
|
RFoldFuture {
|
||||||
|
stream,
|
||||||
|
f,
|
||||||
|
acc: Some(init),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<S, F, B> Future for RFoldFuture<S, F, B>
|
||||||
|
where
|
||||||
|
S: DoubleEndedStream + Sized,
|
||||||
|
F: FnMut(B, S::Item) -> B,
|
||||||
|
{
|
||||||
|
type Output = B;
|
||||||
|
|
||||||
|
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
|
let mut this = self.project();
|
||||||
|
loop {
|
||||||
|
let next = futures_core::ready!(this.stream.as_mut().poll_next_back(cx));
|
||||||
|
|
||||||
|
match next {
|
||||||
|
Some(v) => {
|
||||||
|
let old = this.acc.take().unwrap();
|
||||||
|
let new = (this.f)(old, v);
|
||||||
|
*this.acc = Some(new);
|
||||||
|
}
|
||||||
|
None => return Poll::Ready(this.acc.take().unwrap()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,56 @@
|
|||||||
|
use crate::future::Future;
|
||||||
|
use core::pin::Pin;
|
||||||
|
use crate::task::{Context, Poll};
|
||||||
|
|
||||||
|
use pin_project_lite::pin_project;
|
||||||
|
|
||||||
|
use crate::stream::DoubleEndedStream;
|
||||||
|
|
||||||
|
pin_project! {
|
||||||
|
#[doc(hidden)]
|
||||||
|
#[allow(missing_debug_implementations)]
|
||||||
|
pub struct TryRFoldFuture<S, F, T> {
|
||||||
|
#[pin]
|
||||||
|
stream: S,
|
||||||
|
f: F,
|
||||||
|
acc: Option<T>,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<S, F, T> TryRFoldFuture<S, F, T> {
|
||||||
|
pub(super) fn new(stream: S, init: T, f: F) -> Self {
|
||||||
|
TryRFoldFuture {
|
||||||
|
stream,
|
||||||
|
f,
|
||||||
|
acc: Some(init),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<S, F, T, E> Future for TryRFoldFuture<S, F, T>
|
||||||
|
where
|
||||||
|
S: DoubleEndedStream + Unpin,
|
||||||
|
F: FnMut(T, S::Item) -> Result<T, E>,
|
||||||
|
{
|
||||||
|
type Output = Result<T, E>;
|
||||||
|
|
||||||
|
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
|
let mut this = self.project();
|
||||||
|
loop {
|
||||||
|
let next = futures_core::ready!(this.stream.as_mut().poll_next_back(cx));
|
||||||
|
|
||||||
|
match next {
|
||||||
|
Some(v) => {
|
||||||
|
let old = this.acc.take().unwrap();
|
||||||
|
let new = (this.f)(old, v);
|
||||||
|
|
||||||
|
match new {
|
||||||
|
Ok(o) => *this.acc = Some(o),
|
||||||
|
Err(e) => return Poll::Ready(Err(e)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => return Poll::Ready(Ok(this.acc.take().unwrap())),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,68 @@
|
|||||||
|
use core::marker::PhantomData;
|
||||||
|
use core::pin::Pin;
|
||||||
|
use core::task::{Context, Poll};
|
||||||
|
|
||||||
|
use crate::stream::{DoubleEndedStream, ExactSizeStream, FusedStream, Stream};
|
||||||
|
|
||||||
|
/// A stream that never returns any items.
|
||||||
|
///
|
||||||
|
/// This stream is created by the [`pending`] function. See its
|
||||||
|
/// documentation for more.
|
||||||
|
///
|
||||||
|
/// [`pending`]: fn.pending.html
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub struct Pending<T> {
|
||||||
|
_marker: PhantomData<T>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Creates a stream that never returns any items.
|
||||||
|
///
|
||||||
|
/// The returned stream will always return `Pending` when polled.
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// ```
|
||||||
|
/// # async_std::task::block_on(async {
|
||||||
|
/// #
|
||||||
|
/// use std::time::Duration;
|
||||||
|
///
|
||||||
|
/// use async_std::prelude::*;
|
||||||
|
/// use async_std::stream;
|
||||||
|
///
|
||||||
|
/// let dur = Duration::from_millis(100);
|
||||||
|
/// let mut s = stream::pending::<()>().timeout(dur);
|
||||||
|
///
|
||||||
|
/// let item = s.next().await;
|
||||||
|
///
|
||||||
|
/// assert!(item.is_some());
|
||||||
|
/// assert!(item.unwrap().is_err());
|
||||||
|
///
|
||||||
|
/// #
|
||||||
|
/// # })
|
||||||
|
/// ```
|
||||||
|
pub fn pending<T>() -> Pending<T> {
|
||||||
|
Pending {
|
||||||
|
_marker: PhantomData,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> Stream for Pending<T> {
|
||||||
|
type Item = T;
|
||||||
|
|
||||||
|
fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> {
|
||||||
|
Poll::Pending
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> DoubleEndedStream for Pending<T> {
|
||||||
|
fn poll_next_back(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> {
|
||||||
|
Poll::Pending
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> FusedStream for Pending<T> {}
|
||||||
|
|
||||||
|
impl<T> ExactSizeStream for Pending<T> {
|
||||||
|
fn len(&self) -> usize {
|
||||||
|
0
|
||||||
|
}
|
||||||
|
}
|
||||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in New Issue