Repository: alexcrichton/futures-rs
Pull request #134 (2016-09-08 16:07:34)

[RFC] Add an unordered buffer stream adapter.
Unlike the current buffered stream, this one does not preserve order. This is primarily useful when handling a stream of requests. Example use case:

```rust
my_requests.map(|(client, req)| {
    log_request_begin(req);
    do_some_work.then(move |reply| {
        client.send_response(reply)
    })
}).unordered(connection_pool_size).for_each(|result| {
    log_request_end(result);
})
```

Stebalien: This is in response to the removal of `forget` and the resulting conversation (https://github.com/alexcrichton/futures-rs/issues/33#issuecomment-244386180). Basically, I need a way to drive outstanding requests without caring about order (long running requests should not hold up quick requests). Initially, I thought I wanted an unbounded buffer but quickly realized that that would be a very bad idea :tm:. Instead, hung requests should just timeout.

The PR as written appears to work but:

1. I haven't tested it thoroughly.
2. "unorderd" is not a good name for this.
Stebalien: /cc @aturon 
alexcrichton: Nice! This looks good to me, the only question I'd have is around the name. To mirror `buffered` perhaps we could get the word "buffer" in the name somehow as well? Maybe `buffer_unordered`?
Stebalien: Done. I think the name could still be better (dispatch? drive?) but at least it (mostly) parallels buffered now.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 327a98b7..bc995ae3 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -13,7 +13,6 @@ sudo: false
 before_script:
   - pip install 'travis-cargo<0.2' --user && export PATH=$HOME/.local/bin:$PATH
 script:
-  - export CARGO_TARGET_DIR=`pwd`/target
   - cargo build
   - cargo build --no-default-features
   - cargo test
diff --git a/FAQ.md b/FAQ.md
index ad17ef69..994caf4b 100644
--- a/FAQ.md
+++ b/FAQ.md
@@ -36,9 +36,11 @@ example, to mesh in with the rest of a future's computation.
 
 ### How do I call `poll`?
 
-In general it's not recommended to call `poll` unless you're implementing
-another `poll` function. If you need to poll a future, however, you can use
-`task::spawn` followed by the `poll_future` method on `Spawn<T>`.
+Right now, call `.forget()`. That method will drive a future to completion and
+drop all associated resources as soon as it's completed.
+
+Eventually more flavorful methods of configuring a `Task` will be available, but
+right now `.forget()` is all we have.
 
 ### How do I return a future?
 
diff --git a/README.md b/README.md
index 8d192a57..e0429f09 100644
--- a/README.md
+++ b/README.md
@@ -4,8 +4,9 @@ This library is an implementation of **zero cost futures** in Rust.
 
 [![Build Status](https://travis-ci.org/alexcrichton/futures-rs.svg?branch=master)](https://travis-ci.org/alexcrichton/futures-rs)
 [![Build status](https://ci.appveyor.com/api/projects/status/yl5w3ittk4kggfsh?svg=true)](https://ci.appveyor.com/project/alexcrichton/futures-rs)
+[![Coverage Status](https://coveralls.io/repos/github/alexcrichton/futures-rs/badge.svg?branch=master)](https://coveralls.io/github/alexcrichton/futures-rs?branch=master)
 
-[Documentation](https://alexcrichton.github.io/futures-rs)
+[Documentation](http://alexcrichton.com/futures-rs)
 
 [Tutorial][tutorial]
 
@@ -99,14 +100,18 @@ More information can be found in [the tutorial][tutorial-future-trait]
 ## I/O with futures
 
 With the power of zero-allocation futures we can take futures all the way down
-the stack to the I/O layer. The [Tokio] project is a one-stop shop for async I/O
-in Rust, starting with the [`tokio-core` crate][tokio-core] at the very bottom
-layer all the way up to the [`tokio-proto` crate][tokio-proto] to enable easily
-building new protocols.
+the stack to the I/O layer. The [`tokio-core` crate][tokio-core] provides an
+abstraction for I/O objects as a stream of readiness notifications plus `Read`
+and `Write` traits along with a number of combinators you'd expect when working
+with `std::io`.
+
+These abstractions can be implemented by the with [`mio`][mio] to power I/O.
+Finally we can then use these abstractions to build a [`tokio-tls`
+crate][tokio-tls] crate to provide TLS/SSL streams over arbitrary read/write
+streams. Finally [`tokio-proto`] can be layered on top for easily writing highly
+efficient protocol implementations.
 
-[Tokio]: https://github.com/tokio-rs/tokio
 [tokio-core]: https://tokio-rs.github.io/tokio-core
-[tokio-proto]: https://tokio-rs.github.io/tokio-proto
 [tokio-tls]: https://tokio-rs.github.io/tokio-tls
 [mio]: https://github.com/carllerche/mio
 
diff --git a/TUTORIAL.md b/TUTORIAL.md
index f147bf45..b423d21a 100644
--- a/TUTORIAL.md
+++ b/TUTORIAL.md
@@ -29,6 +29,7 @@ This document is split up into a few sections:
 * [Returning futures][returning-futures]
 * [`Task` and `Future`][task-and-future]
 * [Task local data][task-local-data]
+* [Event loop data][event-loop-data]
 
 If you'd like to help contribute to this document you can find it on
 [GitHub][online-doc].
@@ -65,16 +66,15 @@ Here we're adding a dependency on three crates:
 * [`futures`] - the definition and core implementation of [`Future`] and
   [`Stream`].
 * [`tokio-core`] - bindings to the [`mio`] crate providing concrete
-  implementations of [`Future`] and [`Stream`] for TCP and UDP as well as the
-  heart of the [Tokio] project
-* [`tokio-tls`] - an SSL/TLS implementation for [Tokio]
+  implementations of [`Future`] and [`Stream`] for TCP and UDP.
+* [`tokio-tls`] - an SSL/TLS implementation built on top of futures
 
 [`mio`]: https://crates.io/crates/mio
 [`tokio-core`]: https://github.com/tokio-rs/tokio-core
 [`tokio-tls`]: https://github.com/tokio-rs/tokio-tls
 [`Future`]: http://alexcrichton.com/futures-rs/futures/trait.Future.html
 [`Stream`]: http://alexcrichton.com/futures-rs/futures/stream/trait.Stream.html
-[Tokio]: https://github.com/tokio-rs/tokio
+[Tokio]: https://github.com/tokio-rs
 
 The [`futures`] crate itself is a low-level implementation of futures which does
 not assume any particular runtime or I/O layer. For the examples below we'll be
@@ -93,15 +93,14 @@ extern crate tokio_tls;
 use std::net::ToSocketAddrs;
 
 use futures::Future;
-use tokio_core::reactor::Core;
-use tokio_core::net::TcpStream;
+use tokio_core::Loop;
 use tokio_tls::ClientContext;
 
 fn main() {
-    let mut core = Core::new().unwrap();
+    let mut lp = Loop::new().unwrap();
     let addr = "www.rust-lang.org:443".to_socket_addrs().unwrap().next().unwrap();
 
-    let socket = TcpStream::connect(&addr, &core.handle());
+    let socket = lp.handle().tcp_connect(&addr);
 
     let tls_handshake = socket.and_then(|socket| {
         let cx = ClientContext::new().unwrap();
@@ -118,7 +117,7 @@ fn main() {
         tokio_core::io::read_to_end(socket, Vec::new())
     });
 
-    let (_, data) = core.run(response).unwrap();
+    let (_, data) = lp.run(response).unwrap();
     println!("{}", String::from_utf8_lossy(&data));
 }
 ```
@@ -130,30 +129,30 @@ There's a lot to digest here, though, so let's walk through it
 line-by-line. First up in `main()`:
 
 ```rust
-let mut cre = Core::new().unwrap();
+let mut lp = Loop::new().unwrap();
 let addr = "www.rust-lang.org:443".to_socket_addrs().unwrap().next().unwrap();
 ```
 
-Here we [create an event loop][core-new] on which we will perform all our
+Here we [create an event loop][loop-new] on which we will perform all our
 I/O. Then we resolve the "www.rust-lang.org" host name by using
 the standard library's [`to_socket_addrs`] method.
 
-[core-new]: https://tokio-rs.github.io/tokio-core/tokio_core/reactor/struct.Core.html#method.new
+[loop-new]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.Loop.html#method.new
 [`to_socket_addrs`]: https://doc.rust-lang.org/std/net/trait.ToSocketAddrs.html
 
 Next up:
 
 ```rust
-let socket = TcpStream::connect(&addr, &core.handle());
+let socket = lp.handle().tcp_connect(&addr);
 ```
 
 We [get a handle] to our event loop and connect to the host with
-[`TcpStream::connect`]. Note, though, that [`TcpStream::connect`] returns a
-future! This means that we don't actually have the socket yet, but rather it
-will be fully connected at some later point in time.
+[`tcp_connect`]. Note, though, that [`tcp_connect`] returns a future! This
+means that we don't actually have the socket yet, but rather it will
+be fully connected at some later point in time.
 
-[get a handle]: https://tokio-rs.github.io/tokio-core/tokio_core/reactor/struct.Core.html#method.handle
-[`TcpStream::connect`]: https://tokio-rs.github.io/tokio-core/tokio_core/net/struct.TcpStream.html#method.connect
+[get a handle]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.Loop.html#method.handle
+[`tcp_connect`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopHandle.html#method.tcp_connect
 
 Once our socket is available we need to perform three tasks to download the
 rust-lang.org home page:
@@ -174,14 +173,14 @@ let tls_handshake = socket.and_then(|socket| {
 });
 ```
 
-Here we use the [`and_then`] method on the [`Future`] trait to continue
-building on the future returned by [`TcpStream::connect`]. The [`and_then`] method
-takes a closure which receives the resolved value of this previous future. In
-this case `socket` will have type [`TcpStream`]. The [`and_then`] closure,
-however, will not run if [`TcpStream::connect`] returned an error.
+Here we use the [`and_then`] method on the [`Future`] trait to continue building
+on the future returned by [`tcp_connect`]. The [`and_then`] method takes a
+closure which receives the resolved value of this previous future. In this case
+`socket` will have type [`TcpStream`]. The [`and_then`] closure, however, will
+not run if [`tcp_connect`] returned an error.
 
 [`and_then`]: http://alexcrichton.com/futures-rs/futures/trait.Future.html#method.and_then
-[`TcpStream`]: https://tokio-rs.github.io/tokio-core/tokio_core/net/struct.TcpStream.html
+[`TcpStream`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.TcpStream.html
 
 Once we have our `socket`, we create a client TLS context via
 [`ClientContext::new`]. This type from the [`tokio-tls`] crate
@@ -193,7 +192,7 @@ as the second.
 [`ClientContext::new`]: https://tokio-rs.github.io/tokio-tls/tokio_tls/struct.ClientContext.html#method.new
 [`handshake`]: https://tokio-rs.github.io/tokio-tls/tokio_tls/struct.ClientContext.html#method.handshake
 
-Like with [`TcpStream::connect`] from before, the [`handshake`] method
+Like with [`tcp_connect`] from before, the [`handshake`] method
 returns a future. The actual TLS handshake may take some time as the
 client and server need to perform some I/O, agree on certificates,
 etc. Once resolved, however, the future will become a [`TlsStream`],
@@ -260,25 +259,25 @@ To actually execute our future and drive it to completion we'll need to run the
 event loop:
 
 ```rust
-let (_, data) = core.run(response).unwrap();
+let (_, data) = lp.run(response).unwrap();
 println!("{}", String::from_utf8_lossy(&data));
 ```
 
 Here we pass our `response` future, our entire HTTP request, and we pass it to
-the event loop, [asking it to resolve the future][`core_run`]. The event loop will
+the event loop, [asking it to resolve the future][`loop_run`]. The event loop will
 then run until the future has been resolved, returning the result of the future
 which in this case is `io::Result<(TcpStream, Vec<u8>)>`.
 
-[`core_run`]: https://tokio-rs.github.io/tokio-core/tokio_core/reactor/struct.Core.html#method.run
+[`loop_run`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.Loop.html#method.run
 
-Note that this `core.run(..)` call will block the calling thread until the
-future can itself be resolved. This means that `data` here has type `Vec<u8>`.
-We then print it out to stdout as usual.
+Note that this `lp.run(..)` call will block the calling thread until the future
+can itself be resolved. This means that `data` here has type `Vec<u8>`. We then
+print it out to stdout as usual.
 
-Phew! At this point we've seen futures [initiate a TCP
-connection][`TcpStream::connect`] [create a chain of computation][`and_then`],
-and [read data from a socket][`read_to_end`]. But this is only a hint of what
-futures can do, so let's dive more into the traits themselves!
+Phew! At this point we've seen futures [initiate a TCP connection][`tcp_connect`]
+[create a chain of computation][`and_then`], and [read data from a
+socket][`read_to_end`]. But this is only a hint of what futures can
+do, so let's dive more into the traits themselves!
 
 ---
 
@@ -338,7 +337,7 @@ Or when taking a future you might write:
 ```rust
 fn foo<F>(future: F)
     where F: Future<Error = io::Error>,
-          F::Item: Clone,
+          F::Item: Future,
 {
     // ...
 }
@@ -370,29 +369,22 @@ conveys a number of restrictions and abilities:
 * During a `poll`, futures can mutate their own state.
 * When `poll`'d, futures are owned by another entity.
 
-Next we see that [`Poll`][poll-type] is actually a type alias:
+Next we see the [`Poll`][poll-type] type being returned, which looks like:
 
-[poll-type]: http://alexcrichton.com/futures-rs/futures/type.Poll.html
+[`Task`]: http://alexcrichton.com/futures-rs/futures/struct.Task.html
+[poll-type]: http://alexcrichton.com/futures-rs/futures/enum.Poll.html
 
 ```rust
-type Poll<T, E> = Result<Async<T>, E>;
-```
-
-Let's follow through and see what [`Async`] is as well:
-
-```rust
-pub enum Async<T> {
-    Ready(T),
+enum Poll<T, E> {
     NotReady,
+    Ok(T),
+    Err(E),
 }
 ```
 
-[`Async`]: http://alexcrichton.com/futures-rs/futures/enum.Async.html
-
 Through this `enum` futures can communicate whether the future's value is ready
-to go. If an error ever happens, then `Err` is returned immediately. Otherwise,
-the [`Async`] type indicates whether the futures is ready with a successful
-payload or not ready.
+to go via `Poll::Ok` or `Poll::Err`. If the value isn't ready yet then
+`Poll::NotReady` is returned.
 
 The [`Future`] trait, like `Iterator`, doesn't specify what happens after
 [`poll`] is called if the future has already resolved. Many implementations will
@@ -400,7 +392,7 @@ panic, some may never resolve again, etc. This means that implementors of the
 [`Future`] trait don't need to maintain state to check if [`poll`] has already
 returned successfully.
 
-If a call to [`poll`] returns `NotReady`, then futures still need to know
+If a call to [`poll`] returns `Poll::NotReady`, then futures still need to know
 how to figure out when to get poll'd later! To accomplish this, a future must
 ensure that when `NotReady` is returned the *current task* is arranged to
 receive a notification when the value may be available. We'll [talk more about
@@ -408,14 +400,14 @@ tasks later][task-and-future], and it's a sufficient mental model to understand
 that `NotReady` means not only the item isn't ready, but you're now registered
 to receive a notification when it is ready.
 
-To actually deliver notifications, the [`park`] method is the primary
-entry point. This function returns a threadsafe [`Task`] handle with one
-primary method, [`unpark`]. The `unpark` method, when called, indicates that a
-future can make progress, and may be able to resolve to a value.
+To actually deliver notifications, the `task::park` method is the primary entry
+point. This function returns a [`TaskHandle`] which implements `Send + 'static`
+and has one primary method, [`unpark`]. The `unpark` method, when called,
+indicates that a future can make progress, and may be able to resolve to a
+value.
 
-[`Task`]: http://alexcrichton.com/futures-rs/futures/task/struct.Task.html
-[`unpark`]: http://alexcrichton.com/futures-rs/futures/task/struct.Task.html#method.unpark
-[`park`]: http://alexcrichton.com/futures-rs/futures/task/fn.park.html
+[`TaskHandle`]: http://alexcrichton.com/futures-rs/futures/struct.TaskHandle.html
+[`unpark`]: http://alexcrichton.com/futures-rs/futures/struct.TaskHandle.html#method.unpark
 
 More detailed documentation can be found on the [`poll`] methods itself.
 
@@ -513,7 +505,7 @@ returns `Option<Self::Item>` instead of `Self::Item`.
 [stream-poll]: http://alexcrichton.com/futures-rs/futures/stream/trait.Stream.html#tymethod.poll
 
 A [`Stream`] produces optionally many values over time, signaling termination of
-the stream by returning `Ready(None)`. At its heart a [`Stream`] represents
+the stream by returning `Poll::Ok(None)`. At its heart a [`Stream`] represents
 an asynchronous stream of values being produced in order.
 
 A [`Stream`] is actually just a special instance of a [`Future`], and can be
@@ -544,50 +536,65 @@ take a look at an example of streams now, the [`incoming`] implementation of
 [`Stream`] on [`TcpListener`]. The simple server below will accept connections,
 write out the word "Hello!" to them, and then close the socket:
 
-[`incoming`]: https://tokio-rs.github.io/tokio-core/tokio_core/net/struct.TcpListener.html#method.incoming
-[`TcpListener`]: https://tokio-rs.github.io/tokio-core/tokio_core/net/struct.TcpListener.html
+[`incoming`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.TcpListener.html#method.incoming
+[`TcpListener`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.TcpListener.html
 
 ```rust
 extern crate futures;
 extern crate tokio_core;
 
+use futures::Future;
 use futures::stream::Stream;
-use tokio_core::reactor::Core;
-use tokio_core::net::TcpListener;
+use tokio_core::Loop;
 
 fn main() {
-    let mut core = Core::new().unwrap();
+    let mut lp = Loop::new().unwrap();
     let address = "127.0.0.1:8080".parse().unwrap();
-    let listener = TcpListener::bind(&address, &core.handle()).unwrap();
-
-    let addr = listener.local_addr().unwrap();
-    println!("Listening for connections on {}", addr);
-
-    let clients = listener.incoming();
-    let welcomes = clients.and_then(|(socket, _peer_addr)| {
-        tokio_core::io::write_all(socket, b"Hello!\n")
-    });
-    let server = welcomes.for_each(|(_socket, _welcome)| {
-        Ok(())
+    let listener = lp.handle().tcp_listen(&address);
+
+    let server = listener.and_then(|listener| {
+        let addr = listener.local_addr().unwrap();
+        println!("Listening for connections on {}", addr);
+
+        let clients = listener.incoming();
+        let welcomes = clients.and_then(|(socket, _peer_addr)| {
+            tokio_core::io::write_all(socket, b"Hello!\n")
+        });
+        welcomes.for_each(|(_socket, _welcome)| {
+            Ok(())
+        })
     });
 
-    core.run(server).unwrap();
+    lp.run(server).unwrap();
 }
 ```
 
 Like before, let's walk through this line-by-line:
 
 ```rust
-let mut core = Core::new().unwrap();
+let mut lp = Loop::new().unwrap();
 let address = "127.0.0.1:8080".parse().unwrap();
-let listener = TcpListener::bind(&address, &core.handle()).unwrap();
+let listener = lp.handle().tcp_listen(&address);
 ```
 
 Here we initialize our event loop, like before, and then we use the
-[`TcpListener::bind`] function to create a TCP listener which will accept
-sockets.
+[`tcp_listen`] method on [`LoopHandle`] to create a TCP listener which will
+accept sockets.
+
+[`tcp_listen`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopHandle.html#method.tcp_listen
+
+Next up we see:
+
+```rust
+let server = listener.and_then(|listener| {
+    // ...
+});
+```
 
-[`TcpListener::bind`]: https://tokio-rs.github.io/tokio-core/tokio_core/net/struct.TcpListener.html#method.bind
+Here we see that [`tcp_listen`], like [`tcp_connect`] from before, did not
+return a [`TcpListener`] but rather a future which will resolve to a TCP
+listener. We then employ the [`and_then`] method on [`Future`] to define what
+happens once the TCP listener is available.
 
 Now that we've got the TCP listener we can inspect its state:
 
@@ -662,30 +669,23 @@ these therefore has the effect of taking each socket from the stream and
 processing all chained operations on it before taking the next socket.
 
 If, instead, we want to handle all clients concurrently, we can use the
-[`spawn`] method on [`Handle`]\:
-
-[`spawn`]: https://tokio-rs.github.io/tokio-core/tokio_core/reactor/struct.Handle.html#method.spawn
+[`forget`] method:
 
 ```rust
 let clients = listener.incoming();
 let welcomes = clients.map(|(socket, _peer_addr)| {
-    tokio_core::io::write_all(socket, b"hello!\n")
+    tokio_core::io::write_all(socket, b"Hello!\n")
 });
-let handle = core.handle();
-let server = welcomes.for_each(|future| {
-    handle.spawn(future.then(|_| Ok(())));
+welcomes.for_each(|future| {
+    future.forget();
     Ok(())
-});
+})
 ```
 
 Instead of [`and_then`][stream-and-then] we're using [`map`][stream-map] here
 which changes our stream of clients to a stream of futures. We then change our
-[`for_each`] closure to *[`spawn`]* the future, which allows the future to
-execute concurrently on the event loop. Note that [`spawn`] requires the future
-to have the item/error types as `()`.
-
-We'll talk a bit more about spawning futures in the [tasks and
-futures][task-and-future] section.
+[`for_each`] closure to *[`forget`]* the future, which allows the future to
+execute concurrently.
 
 ---
 
@@ -694,10 +694,10 @@ futures][task-and-future] section.
 
 [Back to top][top]
 
-Alright! At this point we've got a good understanding of the [`Future`] and
-[`Stream`] traits, both how they're implemented as well as how they're composed
-together.  But where do all these futures originally come from? Let's take a
-look at a few concrete implementations of futures and streams.
+Alright! At this point we've got a good understanding of the [`Future`] and [`Stream`]
+traits, both how they're implemented as well as how they're composed together.
+But where do all these futures originally come from? Let's take a look at a few
+concrete implementations of futures and streams.
 
 First, any value already available is trivially a future that is immediately
 ready. For this, the [`done`], [`failed`], [`finished`] functions suffice. The
@@ -953,35 +953,41 @@ come from? Additionally, where'd [`poll`] get called from in the first place?
 
 Enter, a [`Task`]!
 
-In the [`futures`] crate a task drives a computation represented by
-futures. Tasks can be thought of as being similar to green threads or normal OS
-threads: they're spawned to represent concurrent computations (which in this
-case are futures). Any particular instance of a `Future` may be short-lived,
-only a part of a larger computation. For example, in our ["hello
-world"][hello-world] example we had a number of futures, but only one actually
-ran at a time. For the entire program, we had one task that followed the
-logical "thread of execution" as each future resolved and the overall
-computation progressed.
+In the [`futures`] crate the [`Task`] struct drives a computation represented by
+futures. Any particular instance of a `Future` may be short-lived, only a part
+of a larger computation. For example, in our ["hello world"][hello-world]
+example we had a number of futures, but only one actually ran at a time. For the
+entire program, we had one [`Task`] that followed the logical "thread of
+execution" as each future resolved and the overall computation progressed.
 
-When a future is [`spawn`]ed it is fused with a task, and then this structure
-can be polled for completion. Precisely how and when a `poll` happens is up to
-the function which spawned the future. Normally you won't call [`spawn`] but
-rather one of [`CpuPool::spawn`] with a thread pool or [`Handle::spawn`] with
-an event loop. These internally use [`spawn`] and handle managing calls to
-`poll` for you.
+In short, a `Task` is the entity that actually orchestrates the top-level calls
+to `poll`. Its main method, [`run`], does exactly this.  Internally, `Task` has
+synchronization such that if [`unpark`] is called on multiple threads then the
+resulting calls to [`poll`] are properly coordinated.
 
-[`CpuPool::spawn`]: http://alexcrichton.com/futures-rs/futures_cpupool/struct.CpuPool.html#method.spawn
-[`Handle::spawn`]: https://tokio-rs.github.io/tokio-core/tokio_core/reactor/struct.Handle.html#method.spawn
+[`run`]: http://alexcrichton.com/futures-rs/futures/struct.Task.html#method.run
 
+Tasks themselves are not typically created manually but rather are manufactured
+through use of the [`forget`] function. This function on the [`Future`] trait
+creates a new [`Task`] and then asks it to run the future, resolving the entire
+chain of composed futures in sequence.
 
 _The clever implementation of `Task` is the key to the `futures` crate's
-efficiency: when a future is spawned, each of the `Future`s in the chain of
+efficiency: when a `Task` is created, each of the `Future`s in the chain of
 computations is combined into a single state machine structure and moved
 together from the stack into the heap_. This action is the only allocation
 imposed by the futures library. In effect, the `Task` behaves as if you had
 written an efficient state machine by hand while allowing you to express that
 state machine as straight-line sequence of computations.
 
+[`forget`]: http://alexcrichton.com/futures-rs/futures/trait.Future.html#method.forget
+
+Conceptually a [`Task`] is somewhat similar to an OS thread's stack. Where an OS
+thread runs functions that all have access to the stack which is available
+across blocking I/O calls, an asynchronous computation runs individual futures
+over time which all have access to a [`Task`] that is persisted throughout the
+lifetime of the computation.
+
 ---
 
 ## Task-local data
@@ -994,8 +1000,8 @@ one piece of a larger asynchronous computation. This means that futures come
 and go, but there could also be data that lives for the entire span of a
 computation that many futures need access to.
 
-Futures themselves are often required to be `'static`, so we have two choices
-to share data between futures:
+Futures themselves require `'static`, so we have two choices to share data
+between futures:
 
 * If the data is only ever used by one future at a time we can thread through
   ownership of the data between each future.
@@ -1010,20 +1016,73 @@ In the [`Task` and `Future`][task-and-future] section we saw how an
 asynchronous computation has access to a [`Task`] for its entire lifetime, and
 from the signature of [`poll`] we also see that it has mutable access to
 this task. The [`Task`] API leverages these facts and allows you to store
-data inside a `Task`. Data associated with a `Task` can be created with two
-methods:
-
-* The [`task_local!`] macro behaves similarly to the `thread_local!` macro in
-  the standard library. Data initialized this way is lazily initialized on
-  first access for a task, and then it's destroyed when a task itself is
-  destroyed.
-* The [`TaskRc`] structure provides the ability to create a reference-counted
-  piece of data that can only be accessed on an appropriate task. It can be
-  cloned, however, like an `Rc`.
-
-[`task_local!`]: http://alexcrichton.com/futures-rs/futures/macro.task_local.html
-[`TaskRc`]: http://alexcrichton.com/futures-rs/futures/task/struct.TaskRc.html
-
-Note that both of these methods will fuse data with the currently running task,
-which may not always be desired. These two should likely be used sparingly in
-general.
+data inside a `Task`.
+
+Data associated with a `Task` with [`TaskData::new`] to return a [`TaskData`]
+handle. This handle can then be cloned regardless of the underlying data. To
+access the data at a later time you can use the [`with`] methods.
+
+[`TaskData`]: http://alexcrichton.com/futures-rs/futures/task/struct.TaskData.html
+[`TaskData::new`]: http://alexcrichton.com/futures-rs/futures/task/struct.TaskData.html#method.new
+[`with`]: http://alexcrichton.com/futures-rs/futures/task/struct.TaskData.html#method.with
+
+---
+
+## Event loop data
+[event-loop-data]: #event-loop-data
+
+[Back to top][top]
+
+We've now seen that we can store data into a [`Task`] with [`TaskData`], but
+data is sometimes not `Send` or otherwise needs to be persisted yet not tied to
+a [`Task`]. For this purpose the [`tokio-core`] crate provides a similar
+abstraction, [`LoopData`].
+
+[`LoopData`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopData.html
+
+The [`LoopData`] is similar to [`TaskData`] where it's a handle to data
+conceptually owned by the event loop. The key property of [`LoopData`], however,
+is that it implements the `Send` trait regardless of the underlying data.
+
+One key method on the [`Future`] trait, [`forget`], requires that the future is
+`Send`. This is not always possible for a future itself, but a `LoopData<F:
+Future>` is indeed `Send` and the `Future` trait is implemented directly for
+`LoopData<F>`. This means that if a future is not `Send`, it can trivially be
+made `Send` by "pinning" it to an event loop through `LoopData`.
+
+A [`LoopData`] handle is a bit easier to access than a [`TaskData`], as you
+don't need to get the data from a task. Instead you can simply attempt to access
+the data with [`get`][loop-get] or [`get_mut`][loop-get-mut]. Both of these
+methods return `Option` where `None` is returned if you're not on the event
+loop.
+
+[loop-get]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopData.html#method.get
+[loop-get-mut]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopData.html#method.get_mut
+
+In the case that `None` is returned, a future may have to return to the event
+loop in order to make progress. In order to guarantee this the [`executor`]
+associated with the data can be acquired and passed to [`Task::poll_on`]. This
+will request that the task poll itself on the specified executor, which in this
+case will run the poll request on the event loop where the data can be accessed.
+
+[`executor`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopData.html#method.executor
+[`Task::poll_on`]: http://alexcrichton.com/futures-rs/futures/struct.Task.html#method.poll_on
+
+A [`LoopData`] can be created through one of two methods:
+
+* If you've got a handle to the [event loop itself][`Loop`] then you can call
+  the [`Loop::add_loop_data`] method. This allows inserting data directly and a
+  handle is immediately returned.
+* If all you have is a [`LoopHandle`] then you can call the
+  [`LoopHandle::add_loop_data`] method. This, unlike [`Loop::add_loop_data`],
+  requires a `Send` closure which will be used to create the relevant data. Also
+  unlike the [`Loop`] method, this function will return a future that resolves
+  to the [`LoopData`] value.
+
+[`Loop`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.Loop.html
+[`Loop::add_loop_data`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.Loop.html#method.add_loop_data
+[`LoopHandle`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopHandle.html
+[`LoopHandle::add_loop_data`]: https://tokio-rs.github.io/tokio-core/tokio_core/struct.LoopHandle.html#method.add_loop_data
+
+Task-local data and event-loop data provide the ability for futures to easily
+share sendable and non-sendable data amongst many futures.
diff --git a/src/lib.rs b/src/lib.rs
index 207acd33..eec24c87 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,9 +1,8 @@
-//! Zero-cost Futures in Rust
+//! Futures at zero cost
 //!
 //! This library is an implementation of futures in Rust which aims to provide
-//! a robust implementation of handling asynchronous computations, ergonomic
-//! composition and usage, and zero cost abstractions over what would otherwise
-//! be written by hand.
+//! zero cost abstractions, `Iterator`-like ergonomics, and easy composability
+//! between `Future`-related APIs.
 //!
 //! Futures are a concept for an object which is a proxy for another value that
 //! may not be ready yet. For example issuing an HTTP request may return a
@@ -251,21 +250,10 @@ mod chain;
 ///
 /// # The `poll` method
 ///
-/// The core method of future, `poll`, is used to attempt to generate the value
-/// of a `Future`. This method *does not block* but is allowed to inform the
-/// caller that the value is not ready yet. Implementations of `poll` may
-/// themselves do work to generate the value, but it's guaranteed that this will
-/// never block the calling thread.
-///
-/// A key aspect of this method is that if the value is not yet available the
-/// current task is scheduled to receive a notification when it's later ready to
-/// be made available. This follows what's typically known as a "readiness" or
-/// "pull" model where values are pulled out of futures on demand, and
-/// otherwise a task is notified when a value might be ready to get pulled out.
-///
-/// The `poll` method is not intended to be called in general, but rather is
-/// typically called in the context of a "task" which drives a future to
-/// completion. For more information on this see the `task` module.
+/// The core method of future, `poll`, is not intended to be called in general.
+/// The `poll` method is typically called in the context of a "task" which
+/// drives a future to completion. For more information on this see the `task`
+/// module.
 ///
 /// # Combinators
 ///
@@ -282,12 +270,16 @@ mod chain;
 /// otherwise have to write down.
 // TODO: expand this
 pub trait Future {
+
     /// The type of value that this future will resolved with if it is
     /// successful.
     type Item;
 
     /// The type of error that this future will resolve with if it fails in a
     /// normal fashion.
+    ///
+    /// Futures may also fail due to panics or cancellation, but that is
+    /// expressed through the `PollError` type, not this type.
     type Error;
 
     /// Query this future to see if its value has become available, registering
@@ -308,17 +300,17 @@ pub trait Future {
     /// This function, `poll`, is the primary method for 'making progress'
     /// within a tree of futures. For example this method will be called
     /// repeatedly as the internal state machine makes its various transitions.
-    /// Executors are responsible for ensuring that this function is called in
-    /// the right location (e.g. always on an I/O thread or not). Unless it is
-    /// otherwise arranged to be so, it should be ensured that **implementations
-    /// of this function finish very quickly**.
-    ///
-    /// Returning quickly prevents unnecessarily clogging up threads and/or
-    /// event loops while a `poll` function call, for example, takes up compute
-    /// resources to perform some expensive computation. 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.
+    /// Additionally, this function may not necessarily have many guarantees
+    /// about *where* it's run (e.g. always on an I/O thread or not). Unless it
+    /// is otherwise arranged to be so, it should be ensured that
+    /// **implementations of this function finish very quickly**.
+    ///
+    /// This prevents unnecessarily clogging up threads and/or event loops while
+    /// a `poll` function call, for example, takes up compute resources to
+    /// perform some expensive computation. 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.
     ///
     /// # Return value
     ///
@@ -326,7 +318,7 @@ pub trait Future {
     /// `Err` if the future is finished but resolved to an error, or
     /// `Async::Ready` with the result of this future if it's finished
     /// successfully. Once a future has finished it is considered a contract
-    /// error to continue polling the future.
+    /// error to continue polling it.
     ///
     /// If `NotReady` is returned, then the future will internally register
     /// interest in the value being produced for the current task. In other
diff --git a/src/stream/buffer_unordered.rs b/src/stream/buffer_unordered.rs
new file mode 100644
index 00000000..6a58c9df
--- /dev/null
+++ b/src/stream/buffer_unordered.rs
@@ -0,0 +1,85 @@
+use std::prelude::v1::*;
+
+use {Async, IntoFuture, Poll, Future};
+use stream::{Stream, Fuse};
+
+/// An adaptor for a stream of futures to execute the futures concurrently, if
+/// possible, delivering results as they become available.
+///
+/// This adaptor will buffer up a list of pending futures, and then return their
+/// results in the order that they complete. This is created by the
+/// `Stream::buffer_unordered` method.
+#[must_use = "streams do nothing unless polled"]
+pub struct BufferUnordered<S>
+    where S: Stream,
+          S::Item: IntoFuture,
+{
+    stream: Fuse<S>,
+    futures: Vec<Option<<S::Item as IntoFuture>::Future>>,
+    cur: usize,
+}
+
+pub fn new<S>(s: S, amt: usize) -> BufferUnordered<S>
+    where S: Stream,
+          S::Item: IntoFuture<Error=<S as Stream>::Error>,
+{
+    BufferUnordered {
+        stream: super::fuse::new(s),
+        futures: (0..amt).map(|_| None).collect(),
+        cur: 0,
+    }
+}
+
+impl<S> Stream for BufferUnordered<S>
+    where S: Stream,
+          S::Item: IntoFuture<Error=<S as Stream>::Error>,
+{
+    type Item = <S::Item as IntoFuture>::Item;
+    type Error = <S as Stream>::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        // First, try to fill in all the futures
+        for future in &mut self.futures {
+            if future.is_none() {
+                match try!(self.stream.poll()) {
+                    Async::Ready(Some(s)) => {
+                        *future = Some(s.into_future());
+                    }
+                    Async::Ready(None) => break,
+                    Async::NotReady => break,
+                }
+            }
+        }
+
+        // Next, try and step futures forward until we find a ready one.
+        // Always start at `cur` for fairness.
+        let mut waiting = false;
+        for i in 0..self.futures.len() {
+            let mut idx = self.cur + i;
+            if idx >= self.futures.len() {
+                idx -= self.futures.len();
+            }
+            let future = &mut self.futures[idx];
+            let result = match *future {
+                Some(ref mut s) => match s.poll() {
+                    Ok(Async::NotReady) => {
+                        waiting = true;
+                        continue
+                    },
+                    Ok(Async::Ready(e)) => Ok(Async::Ready(Some(e))),
+                    Err(e) => Err(e),
+                },
+                None => continue,
+            };
+            self.cur = i + 1;
+            *future = None;
+            return result;
+        }
+
+        Ok(if waiting || !self.stream.is_done() {
+            Async::NotReady
+        } else {
+            Async::Ready(None)
+        })
+    }
+}
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 4c17b59a..115413c2 100755
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -57,10 +57,12 @@ pub use self::peek::Peekable;
 
 if_std! {
     mod buffered;
+    mod buffer_unordered;
     mod channel;
     mod collect;
     mod wait;
     pub use self::buffered::Buffered;
+    pub use self::buffer_unordered::BufferUnordered;
     pub use self::channel::{channel, Sender, Receiver, FutureSender};
     pub use self::collect::Collect;
     pub use self::wait::Wait;
@@ -650,6 +652,24 @@ pub trait Stream {
         buffered::new(self, amt)
     }
 
+    /// An adaptor for creating a buffered list of pending futures (unordered).
+    ///
+    /// If this stream's item can be converted into a future, then this adaptor
+    /// will buffer up to `amt` futures and then return results in the order
+    /// in which they complete. No more than `amt` futures will be buffered at
+    /// any point in time, and less than `amt` may also be buffered depending on
+    /// the state of each future.
+    ///
+    /// The returned stream will be a stream of each future's result, with
+    /// errors passed through whenever they occur.
+    #[cfg(feature = "use_std")]
+    fn buffer_unordered(self, amt: usize) -> BufferUnordered<Self>
+        where Self::Item: IntoFuture<Error = <Self as Stream>::Error>,
+              Self: Sized
+    {
+        buffer_unordered::new(self, amt)
+    }
+
     /// An adapter for merging the output of two streams.
     ///
     /// The merged stream produces items from one or both of the underlying
diff --git a/src/task/mod.rs b/src/task/mod.rs
index 9dd7ba18..1529ec02 100644
--- a/src/task/mod.rs
+++ b/src/task/mod.rs
@@ -209,7 +209,7 @@ impl<F: Future> Spawn<F> {
     /// `task::park()` will return a handle that contains the `unpark`
     /// specified.
     ///
-    /// If this function returns `NotReady`, then the `unpark` should have been
+    /// If this function returns `NotReady`, then the `unpark` should be
     /// scheduled to receive a notification when poll can be called again.
     /// Otherwise if `Ready` or `Err` is returned, the `Spawn` task can be
     /// safely destroyed.


Test patch:
diff --git a/tests/stream.rs b/tests/stream.rs
index bf2c2432..264e1097 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -155,6 +155,45 @@ fn buffered() {
     assert_eq!(rx.next(), None);
 }
 
+#[test]
+fn unordered() {
+    let (tx, rx) = channel::<_, u32>();
+    let (a, b) = oneshot::<u32>();
+    let (c, d) = oneshot::<u32>();
+
+    tx.send(Ok(b.map_err(|_| 2).boxed()))
+      .and_then(|tx| tx.send(Ok(d.map_err(|_| 4).boxed())))
+      .forget();
+
+    let mut rx = rx.buffer_unordered(2);
+    sassert_empty(&mut rx);
+    let mut rx = rx.wait();
+    c.complete(3);
+    assert_eq!(rx.next(), Some(Ok(3)));
+    a.complete(5);
+    assert_eq!(rx.next(), Some(Ok(5)));
+    assert_eq!(rx.next(), None);
+
+    let (tx, rx) = channel::<_, u32>();
+    let (a, b) = oneshot::<u32>();
+    let (c, d) = oneshot::<u32>();
+
+    tx.send(Ok(b.map_err(|_| 2).boxed()))
+      .and_then(|tx| tx.send(Ok(d.map_err(|_| 4).boxed())))
+      .forget();
+
+    // We don't even get to see `c` until `a` completes.
+    let mut rx = rx.buffer_unordered(1);
+    sassert_empty(&mut rx);
+    c.complete(3);
+    sassert_empty(&mut rx);
+    a.complete(5);
+    let mut rx = rx.wait();
+    assert_eq!(rx.next(), Some(Ok(5)));
+    assert_eq!(rx.next(), Some(Ok(3)));
+    assert_eq!(rx.next(), None);
+}
+
 #[test]
 fn zip() {
     assert_done(|| list().zip(list()).collect(),