Consider having polling an error represent the final value. In other words, a poll that returns an error means that poll should never be called again. In the case of a Stream, this means that an error indicates the stream has terminated.

This issue is a placeholder for the associated discussion.

cc @aturon

Currently Future::poll seems to be expected to call task::park which then fetches the current task from TLS and panics if there is no task in TLS.

This results in an unintuitive API (it's not clear at first glance that poll()'s expected interface/implementation is related to tasks) and a potential run-time failure that could be checked at compile time.

So my suggestion is to instead pass the task driving the Future explicitly to Future::poll as an additional function argument, either as a Task reference, a closure calling task::park() (if that's enough), or a similar mechanism, instead of storing it in the TLS variable CURRENT_TASK.

Also, "poll" is a confusing name, since it creates the expectation that it is a function that anyone can call to get the value of the future if it has already completed, but it is in fact an "internal" function that drives future execution instead and currently even panics if called outside a task.

Something like "drive", "execute", "run", "run_next_step" or similar would be a better name.

Updated description

The primary change made in this PR is to restructure memory management and notifications throughout the "task system" in the futures crate. It is intended that this will have very little impact, if any, on consumers of futures. Implementations of runtimes of futures (e.g. crates like tokio-core) are the target for this series of changes, enabling a suite of possible optimizations that were not previously feasible. One major use case that is now enabled is usage of the task and executor modules in the no_std ecosystem. This means that bare-metal applications of futures should be able to use the same task system that the std-based futures ecosystem uses.

One of the largest changes being made to support this is an update to the memory management of objects behind Task handles. Previously it was required that Arc<Unpark> instances were passed into the various Spawn::poll_* functions, but new functions Spawn::poll_*_notify were added which operate with a NotifyHandle instead. A NotifyHandle is conceptually very similar to an Arc<Unpark> instance, but it works through an unsafe trait UnsafeNotify to manage memory instead of requiring Arc is used. You can still use Arc safely, however, if you'd like.

In addition to supporting more forms of memory management, the poll_*_notify functions also take a new id parameter. This parameter is intended to be an opaque bag-of-bits to the futures crate itself but runtimes can use this to identify the future being notified. This is intended to enable situations where the same instance of a NotifyHandle can be used for all futures executed by using the id field to distinguish which future is ready when it gets notified.

API Additions

• A FuturesUnordered::push method was added and the FuturesUnordered type itself was completely rewritten to efficiently track a large number of futures.
• A Task::will_notify_current method was added with a slightly different implementation than Task::is_current but with stronger guarantees and documentation wording about its purpose.

Compatibility Notes

As with all 0.1.x releases this PR is intended to be 100% backwards compatible. All code that previously compiled should continue to do so with these changes. As with other changes, though, there are also some updates to be aware of:

• The task::park function has been renamed to task::current.
• The Task::unpark function has been renamed to Task::notify, and in general terminology around "unpark" has shifted to terminology around "notify"
• The Unpark trait has been deprecated in favor of the Notify trait mentioned above.
• The UnparkEvent structure has been deprecated. It currently should perform the same as it used to, but it's planned that in a future 0.1.x release the performance will regress for crates that have not transitioned away. The primary primitive to replace this is the addition of a push function on the FuturesUnordered type. If this does not help implement your use case though, please let us know!
• The Task::is_current method is now deprecated, and you likely want to use Task::will_notify_current instead, but let us know if this doesn't suffice!

Original description

This PR is a work in progress

I'm submitting this PR early to hopefully try to illustrate my thoughts and get some early feedback.

Checklist

• [x] Land #442
• [x] Update tokio-core
• [x] Run sscache tests using new task system
• [x] Switch Arc<Unpark> to UnparkHandle #432
• [x] Decide on #312 (leaning towards yes)
• [x] Allow executors to customize wait behavior #360 (deferring this until later)
• [x] Fix Task::is_current
• [x] Remove Notify::is_current, I don't think this is needed anymore.
• [x] Consider GetNotifyHandle https://github.com/alexcrichton/futures-rs/issues/129
• [x] Should ref_inc -> ref_dec be moved to UnsafeNotify (@alexcrichton says no).
• [x] Consider getting rid of poll_*_notify on Stream and Sink. Also, maybe name it poll_notify if it is only for Future.
• [x] Merge https://github.com/carllerche/futures-rs/pull/4
• [x] u64 vs. usize

Overview

The previous implementation of the task system required a number of allocations per task instance. Each task required a dedicated Arc<Unpark> handle which means that executors require at least two allocations per task.

Things get worse when using with_unpark_event as nested calls to with_unpark_event result in Vec allocation and cloning during each call to task::park.

This commit provides an overhaul to the task system to work around these problems. The Unpark trait is changed so that only one instance is required per executor. In order to identify which task is being unparked, Unpark::unpark takes an unpark_id: u64 argument.

with_unpark_event is removed in favor of UnparkContext which satisfies a similar end goal, but requires only a single allocation per lifetime of the UnparkContext.

The new Unpark trait

In general, tasks are driven to completion by executors and executors are able to handle large number of tasks. As such, the Unpark trait has been tweaked to require a single allocation per executor instead of one per task. The idea is that the executor creates one Arc<Unpark> handle and uses the unpark_id: u64 to identify which task is unparked.

In the case of tokio-core, each task is stored in a slab and the the unpark_id is the slab index. Now, given that an Arc is no longer used, it can be that a slab slot is released and repurposed for a different task while there are still outstanding Task handles referencing the now released task.

There are two potential ways to deal with this.

a) Not care. Futures need to be able to handle spurious wake ups already. Spurious wake ups can be reduced by splitting the u64 into 28 bits for the slab offset and use the rest of the u64 as a slot usage counter.

b) Use Unpark::ref_inc and Unpark::ref_dec to allow the executor implementation to handle its own reference counting.

Option b) would allow an executor implementation to store a pointer as the unpark_id and the ref_inc and ref_dec allow for atomic reference counting. This could be used in cases where using a slab is not an option.

UnparkContext

This is quite similar in spirit to with_unpark_event except it requires significantly less allocations when used in a nested situation.

It does have some different behavior. Given the following:

// Currently in task A
let my_context = UnparkContext::new(my_unpark);

my_context.with(0, || {
    let task = task::park();

    thread::spawn(move || {
        thread::sleep(a_few_secs);
        task.unpark();
    });
});

my_executor.spawn(move || {
    // Currently in task B

    my_context.with(0, || {
        // some work here
    });
});

Task B will be the root task that is notified.

Since futures 0.2 is considered just a snapshot that libraries shouldn't expose, most of the ecosystem is still using futures 0.1. It is not expected that 0.2 will see any more development, nor that the ecosystem will move to it. Instead, work is ongoing to try to get futures into libstd.

However, the version on crates.io is 0.2, and the version that is shown on docs.rs is also 0.2. This leads to a lot of confusion when new users try to get started in the ecosystem, since they don't understand why futures returned by libraries don't implement futures::Future (from v2) (example).

Could it just be yanked/deprecated/etc, with a 0.1.22 published to get the docs.rs and crates.io listings to suggest 0.1 until the new version is actually ready?

When the crossbeam-channel RFC was introduced a few months ago, I found it very hard to understand what select meant. I have the same issue in future-rs, where I find the name equally as opaque as in crossbeam. Luckily I have since been pointed in the direction of some Unix history which explains the design and naming behind select, but I feel like we can still do better than requiring a history lesson for a function name to make sense.

Therefore I would like to propose that select is slightly renamed to make it more clear what it is actually doing, to e.g. select_any or select_any_ready maybe (or any other name that is deemed better). Although the presence of select_ok already makes the naming tough, and not having a pure select breaks somewhat with the precedent for this functionality (since it is named as such in Unix and Go for example), but I think there are many Rust users who are not familiar with these precedents and will hence be confused by the name (myself included). At the same time, by keeping the word "select" in the name, it is still easy to search for, for those that do know the precedent (and if "select" is kept as the first part of the name, anyone just looking to type select in their IDE will also be helped by autocomplete).

I feel this would be a very rustic improvement to make; it should still be easy for veterans to use, but welcomes newcomers at the same time, by optimizing for readability over familiarity.

I have a CI pipeline that tests my project on glibc and musl and it appears that when I pull in futures-util as a dependency, this compiles just fine on glibc but does not compile on musl.

Here is the build log: https://travis-ci.org/github/jbaublitz/neli/jobs/738716295

This appears to be due to futures-macro exporting all of its procedural macros with #[proc_macro_hack] which is not allowed on musl.

I know that rust and musl have had problems in the past related to proc macros and I'm wondering if this is the right place to report this or if this should go into the rust repo.

(I'm opening this issue to continue the discussion from https://github.com/alexcrichton/futures-rs/pull/305.)

The problem is that Shared, as it is currently designed, can interact poorly with certain futures, such as the ModedFuture sketched below. We should either formalize some reason why ModedFuture is an invalid implementation of Future, or we should redesign Shared to accommodate this kind of situation.

extern crate futures;
extern crate tokio_core;

use futures::{Future, Poll};
use futures::sync::oneshot;
use std::rc::Rc;
use std::cell::RefCell;

enum Mode { Left, Right }

struct ModedFutureInner<F> where F: Future {
    mode: Mode,
    left: F,
    right: F,
    task: Option<::futures::task::Task>,
}

struct ModedFuture<F> where F: Future {
    inner: Rc<RefCell<ModedFutureInner<F>>>,
}

struct ModedFutureHandle<F> where F: Future {
    inner: Rc<RefCell<ModedFutureInner<F>>>,
}

impl <F> ModedFuture<F> where F: Future {
    pub fn new(left: F, right: F, mode: Mode) -> (ModedFutureHandle<F>, ModedFuture<F>) {
        let inner = Rc::new(RefCell::new(ModedFutureInner {
            left: left, right: right, mode: mode, task: None,
         }));
        (ModedFutureHandle { inner: inner.clone() }, ModedFuture { inner: inner })
    }
}

impl <F> ModedFutureHandle<F> where F: Future {
    pub fn switch_mode(&mut self, mode: Mode) {
        self.inner.borrow_mut().mode = mode;
        if let Some(t) = self.inner.borrow_mut().task.take() {
            // The other future may have become ready while we were ignoring it.
            t.unpark();
        }
    }
}

impl <F> Future for ModedFuture<F> where F: Future {
    type Item = F::Item;
    type Error = F::Error;
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        let ModedFutureInner { ref mut mode, ref mut left, ref mut right, ref mut task } =
            *self.inner.borrow_mut();
        *task = Some(::futures::task::park());
        match *mode {
            Mode::Left => left.poll(),
            Mode::Right => right.poll(),
        }
    }
}

pub fn main() {
    let mut core = ::tokio_core::reactor::Core::new().unwrap();
    let handle = core.handle();

    let (tx, rx) = oneshot::channel::<u32>();
    let f1 = rx.shared();
    let f2 = f1.clone();

    let (mut mfh, mf) = ModedFuture::new(
        Box::new(f1.map_err(|_| ()).map(|v| *v)) as Box<Future<Item=u32, Error=()>>,
        Box::new(::futures::future::empty()) as Box<Future<Item=u32, Error=()>>,
        Mode::Left);

    let (tx3, rx3) = oneshot::channel::<u32>();
    handle.spawn(f2.map(|x| tx3.complete(*x)).map_err(|_| ()));

    core.turn(Some(::std::time::Duration::from_millis(1)));

    handle.spawn(mf.map(|_| ()));

    core.turn(Some(::std::time::Duration::from_millis(1)));

    mfh.switch_mode(Mode::Right);

    tx.complete(11); // It seems like this ought to cause f2 and then rx3 to get resolved.

    // This hangs forever.
    core.run(rx3).unwrap();
}

I can't seem to make this work, perhaps I'm being dumb! In any case, I think this should be in the documentation as an example.

In my case, I have background processes loading data for a given ID. Obviously if a piece of data is already being loaded then I want to join the existing waiters rather than starting a new background process.

I've implemented this using a Map <Id, Vec <Complete>>>, then the first loader triggers the completion of subsequent loaders when it has completed. This is a lot of boilerplate.

I've tried all sorts of things to get this to work but somehow I can't get anything else to compile. Waiting for a future consumes it, so I can only do that once. I have tried to replace the future with a new one and wait on that, like I might do in JavaScript, but this also doesn't work.

If anyone can show me an example then that would be great, if not then I'll probably create an abstraction around my current method and submit this as a pull request for the library.

I am one of the developers of https://github.com/akka/akka/ and I just stumbled upon this nice library (I am a Rust lurker mostly). Futures are a very nice building block for asynchronous programs, but eventually one reaches the point where some kind of streaming abstraction is needed. The simplest streaming approach is the RX-style chained Observables, however, as we and others found out, with asynchronous call-chains backpressure becomes an issue as blocking is no longer available to throttle a producer. To solve this issue, the reactive-streams (RS) standard has been created: http://www.reactive-streams.org backed by various companies interested in the JVM landscape. This set of interoperability interfaces was designed by multiple teams together. This standard is also on its way to become part of JDK9. There is also an effort to expose its semantics as a wire-level protocol http://reactivesocket.io which nicely completements the RS standard (which mainly focuses on in-JVM asynchronous, ordered, backpressured communications).

Since I imagine that eventually the need for asynchronous streams will arise here, I think these standards can be interesting for Rust, too. While RS might not be perfect, it was a result of a long design process and now has mostly consensus about it in the JVM land, so it would be nice to see a Rust implementation that is similar enough to be easily connectable to JVMs, maybe via reactive-socket.

Sorry for the shameless plug :)

The other day I had a conversation in which we were a few mins in before we realized we were talking at cross purposes because of a sync/sink confusion.

Maybe rename futures::sync? Or "sink" -> "drain" (or something)?

Forgive me if this is not the appropriate place to be discussing things like this, but has the possibility of allowing futures to specify what sort of executors they may be spawned on been examined at all? This would allow libraries to support extra features like task thread-affinity or priority.

The way I'm thinking of doing this would be to make the following changes to Future and Context:

pub trait Future<S: Spawn + ?Sized = dyn Spawn> {
    type Output;

    fn poll(self: PinMut<Self>, cx: &mut Context<S>) -> Poll<Self::Output>;
}

pub struct Context<'a, S: Spawn + 'a + ?Sized = dyn Spawn> {
    local_waker: &'a LocalWaker,
    spawner: &'a mut S,
}
// And appropriate changes to method signatures in Context::new and Context::with_spawner

Existing futures would not need to change because of the defaulted type parameter, but futures could require additional features by changing S to something else. For instance, if a future has a non-send subfuture one could create a new trait SpawnLocal: Spawn which takes LocalFutureObj instead of FutureObj, and implement Future<dyn SpawnLocal>.

If anyone is interested, I have a sort of proof-of-concept for the changes in libcore here. (It's not an actual fork of libcore though, since I'm not quite sure how to do that.)

Hi there! So this is a really fun issue I spent some time debugging. I had written an adapter that sits on top of a Peekable<FuturesUnordered<T>>. I had some code that did something like:

// self.stream is a Peekable<FuturesUnordered<T>>

match self.stream.poll_next(ctx) {
    // do some stuff here
}
self.stream.get_mut().push(new_future);
self.stream.poll_peek(ctx);

I found that if the inner FuturesUnordered dropped to having no futures from the first poll_next, then poll_peek always returns Poll::Ready(None) after that.

This is because Peekable uses the Fuse adapter internally (source code link). The

I worked around it by writing my own version of Peekable that didn't use Fuse internally. Note that FuturesUnordered already implements FusedStream so I didn't have to change anything about Peekable.

Potential solutions

I discussed this issue with a couple of folks and here are some thoughts we had:

1. Easiest, non-breaking change: add a reset_fuse method to the Fuse adapter, as well as any other adapters that use a Fuse internally. That would let me use get_mut() followed by reset_fuse() to inform the adapter that it should start polling for values again. (There would also be an advocacy component here -- if you write an adapter that uses one that has reset_fuse internally, you must also expose a reset_fuse method.)

2. Remove the Fuse adapter from Peekable and all other adapters that use Fuse, and replace it with internal logic. This makes a lot of logical sense to me, but it is a breaking change because it won't be possible to maintain the FusedStream for Peekable<St> where St: Stream impl.

   However, it would still be possible to write impl FusedStream for Peekable<St> where St: FusedStream. Also, other people might build their own adapters that use Fuse anyway, so some other workaround like reset_fuse is necessary for ecosystem cohesion.

3. Add a FusedStreamExt trait and a whole new series of methods peekable_fused etc that work on FusedStreams. This would solve the problem, since FuturesUnordered implements FusedStream, but this seems excessive.

4. Revisit the definition of FusedStream as a breaking change. FusedStream really isn't like FusedIterator at all -- this was quite surprising to me and to other people I discussed the issue with. I'm not sure what the semantics would be but requiring workarounds like reset_fuse feels like a little bit of a code smell.

This enables poll()ing to fill the buffer without producing or consuming the Stream's outputs.

Buffered and FuturesOrdered will also poll their inner futures produced by the stream, since they have an output value buffer available. BufferUnordered however will only poll the underlying stream, as it cannot produce outputs without consuming them.

~~Given that the Output of the BufferUnordered impl isn't particularly useful (because it doesn't indicate anything about the futures produced by the stream), maybe it should be type Output = Infallible, and never complete? For the other two, the completion is more meaningful as it indicates that the stream is empty and that all captured futures have produced outputs (though in the case of Buffered, it will stall while the buffer remains full).~~

~~It may also make sense to expose this as a poll_stream() function rather than impl Future (or otherwise a IntoFuture for &Self or something), for the sake of extension trait method resolution.~~

EDIT: moved to inherent poll_stream() instead, see latest commit. Final name tbd (buffered.poll_all()? poll_buffer()? poll_fill()?)

The tokio-postgres crate executes queries in the order in which they are first polled. If two queries depend on each other, then their initial polling order must be the correct order. Subsequent polls are irrelevant. Currently FuturesUnordered polls futures pushed onto the queue initially in that order. Can we add this guarantee to the FuturesUnordered documentation?

Would it be possible to support async transformations on arrays?

async fn map_async<const N: usize, A, B, Fut>(
    array: [A;N],
    fun: Fn(A) -> Fut
) -> [B;N]
    where Fut: Future<Output = B>;

I ran into a case where I'd like to allocate an array of TCP connections.

let sockets = map_async(["127.0.0.1:8000", "127.0.0.1:8001"], connect_socket);

Currently I need to write this in the form of a macro.

let sockets = map_async!(["127.0.0.1:8000", "127.0.0.1:8001"], connect_socket);

Discovered while discussing https://github.com/rust-lang/unsafe-code-guidelines/issues/380, probably related to https://github.com/rust-lang/unsafe-code-guidelines/issues/148.

This code (playground):

use std::thread;
use futures::{channel::mpsc, executor::block_on, sink::SinkExt, stream::StreamExt};

async fn send_sequence(n: u32, mut sender: mpsc::Sender<u32>) {
    for x in 0..n {
        sender.send(n - x).await.unwrap();
    }
}

fn main() {
    let (tx, rx) = mpsc::channel(1);
    let t = thread::spawn(move || block_on(send_sequence(20, tx)));
    let _list: Vec<_> = block_on(rx.collect());
    t.join().unwrap();
}

fails under Miri.