Skip to content

init FutureExt #308

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
4 commits merged into from
Oct 15, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/fs/dir_builder.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
use std::future::Future;

use cfg_if::cfg_if;

use crate::future::Future;
use crate::io;
use crate::path::Path;
use crate::task::blocking;
Expand Down
3 changes: 2 additions & 1 deletion src/fs/open_options.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
use std::future::Future;

use cfg_if::cfg_if;

use crate::fs::File;
use crate::future::Future;
use crate::io;
use crate::path::Path;
use crate::task::blocking;
Expand Down
145 changes: 145 additions & 0 deletions src/future/future.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
use crate::utils::extension_trait;

cfg_if::cfg_if! {
if #[cfg(feature = "docs")] {
use std::pin::Pin;
use std::ops::{Deref, DerefMut};

use crate::task::{Context, Poll};
}
}

extension_trait! {
#[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")
}
}
}
5 changes: 2 additions & 3 deletions src/future/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,6 @@
//! | `future::select` | `Result<T, E>` | Return on first value
//! | `future::try_select` | `Result<T, E>` | Return on first `Ok`, reject on last Err

#[doc(inline)]
pub use std::future::Future;

#[doc(inline)]
pub use async_macros::{join, try_join};

Expand All @@ -53,10 +50,12 @@ pub use async_macros::{select, try_select};

use cfg_if::cfg_if;

pub use future::Future;
pub use pending::pending;
pub use poll_fn::poll_fn;
pub use ready::ready;

pub(crate) mod future;
mod pending;
mod poll_fn;
mod ready;
Expand Down
4 changes: 3 additions & 1 deletion src/io/buf_read/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ extension_trait! {
https://docs.rs/futures-preview/0.3.0-alpha.17/futures/io/trait.AsyncBufRead.html
[provided methods]: #provided-methods
"#]
pub trait BufRead [BufReadExt: futures_io::AsyncBufRead] {
pub trait BufRead {
#[doc = r#"
Returns the contents of the internal buffer, filling it with more data from the
inner reader if it is empty.
Expand All @@ -67,7 +67,9 @@ extension_trait! {
should no longer be returned in calls to `read`.
"#]
fn consume(self: Pin<&mut Self>, amt: usize);
}

pub trait BufReadExt: futures_io::AsyncBufRead {
#[doc = r#"
Reads all bytes into `buf` until the delimiter `byte` or EOF is reached.

Expand Down
4 changes: 3 additions & 1 deletion src/io/read/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ extension_trait! {
[`poll_read`]: #tymethod.poll_read
[`poll_read_vectored`]: #method.poll_read_vectored
"#]
pub trait Read [ReadExt: futures_io::AsyncRead] {
pub trait Read {
#[doc = r#"
Attempt to read from the `AsyncRead` into `buf`.
"#]
Expand All @@ -70,7 +70,9 @@ extension_trait! {
) -> Poll<io::Result<usize>> {
unreachable!("this impl only appears in the rendered docs")
}
}

pub trait ReadExt: futures_io::AsyncRead {
#[doc = r#"
Reads some bytes from the byte stream.

Expand Down
6 changes: 4 additions & 2 deletions src/io/seek.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ extension_trait! {
https://docs.rs/futures-preview/0.3.0-alpha.17/futures/io/trait.AsyncSeek.html
[provided methods]: #provided-methods
"#]
pub trait Seek [SeekExt: futures_io::AsyncSeek] {
pub trait Seek {
#[doc = r#"
Attempt to seek to an offset, in bytes, in a stream.
"#]
Expand All @@ -42,7 +42,9 @@ extension_trait! {
cx: &mut Context<'_>,
pos: SeekFrom,
) -> Poll<io::Result<u64>>;
}

pub trait SeekExt: futures_io::AsyncSeek {
#[doc = r#"
Seeks to a new position in a byte stream.

Expand Down Expand Up @@ -70,7 +72,7 @@ extension_trait! {
fn seek(
&mut self,
pos: SeekFrom,
) -> impl Future<Output = io::Result<u64>> [SeekFuture<'_, Self>]
) -> impl Future<Output = io::Result<u64>> + '_ [SeekFuture<'_, Self>]
where
Self: Unpin,
{
Expand Down
4 changes: 3 additions & 1 deletion src/io/write/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ extension_trait! {
[`poll_flush`]: #tymethod.poll_flush
[`poll_close`]: #tymethod.poll_close
"#]
pub trait Write [WriteExt: futures_io::AsyncWrite] {
pub trait Write {
#[doc = r#"
Attempt to write bytes from `buf` into the object.
"#]
Expand Down Expand Up @@ -80,7 +80,9 @@ extension_trait! {
Attempt to close the object.
"#]
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
}

pub trait WriteExt: futures_io::AsyncWrite {
#[doc = r#"
Writes some bytes into the byte stream.

Expand Down
2 changes: 2 additions & 0 deletions src/prelude.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ pub use crate::stream::Stream;
#[doc(no_inline)]
pub use crate::task_local;

#[doc(hidden)]
pub use crate::future::future::FutureExt as _;
#[doc(hidden)]
pub use crate::io::buf_read::BufReadExt as _;
#[doc(hidden)]
Expand Down
4 changes: 3 additions & 1 deletion src/stream/stream/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ extension_trait! {
https://docs.rs/futures-preview/0.3.0-alpha.17/futures/stream/trait.Stream.html
[provided methods]: #provided-methods
"#]
pub trait Stream [StreamExt: futures_core::stream::Stream] {
pub trait Stream {
#[doc = r#"
The type of items yielded by this stream.
"#]
Expand Down Expand Up @@ -180,7 +180,9 @@ extension_trait! {
```
"#]
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}

pub trait StreamExt: futures_core::stream::Stream {
#[doc = r#"
Advances the stream and returns the next value.

Expand Down
2 changes: 1 addition & 1 deletion src/task/blocking.rs
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
//! A thread pool for running blocking functions asynchronously.

use std::future::Future;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;
use std::time::Duration;

use crossbeam_channel::{bounded, Receiver, Sender};
use lazy_static::lazy_static;

use crate::future::Future;
use crate::task::task::{JoinHandle, Tag};
use crate::utils::abort_on_panic;

Expand Down
2 changes: 1 addition & 1 deletion src/task/task.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use std::fmt;
use std::future::Future;
use std::i64;
use std::mem;
use std::num::NonZeroU64;
Expand All @@ -7,7 +8,6 @@ use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;

use super::task_local;
use crate::future::Future;
use crate::task::{Context, Poll};

/// A handle to a task.
Expand Down
2 changes: 1 addition & 1 deletion src/task/task_local.rs
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
use std::cell::UnsafeCell;
use std::error::Error;
use std::fmt;
use std::future::Future;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;

use lazy_static::lazy_static;

use super::worker;
use crate::future::Future;
use crate::utils::abort_on_panic;

/// Declares task-local values.
Expand Down
Loading