Repository: alexcrichton/futures-rs
Pull request #504 (2017-06-07 03:01:01)

Don't force allocation of the entire `Stream::buffered` buffer up front
This is problematic for large buffer sizes, in particular `Stream::buffered(usize::max_value())` does not do what one might expect it to do.  `VecDeque` is largely a drop in replacement here.

Given how much marketing of futures-rs emphasizes the "zero allocation" bit (which, without language features like `impl Trait`, I think is oversold) I'm not sure how you'll feel about this.  We can add a separate method if necessary, I suppose.

CC @rocallahan

khuey: There's also another big problem with having large amounts of buffering: every `Buffered::poll` polls every future it holds, not just the one that's ready.  The solution to that seems to be in flux (`with_unpark_event` is deprecated) so I don't propose to solve that here.
alexcrichton: Thanks for the PR! If you're working with large numbers of futures in `Buffered` though then yeah I'd definitely recommend working through the with_unpark_event-like solutions. I think that using `stream::FuturesUnordered` (the pseudo-replacement of `with_unpark_event`) may actually be pretty easy here, would you be interested in doing that? That should solve the allocation problem in addition to the "this always polls everything" problem.
khuey: I do actually want the results in order though.
alexcrichton: Right yes of course, my thinking is that:

* When a future is pushed into the queue, you push it into the internal `FuturesUnordered` queue
* Each future pushed in also gets a ticket assigned to it
* When futures pop out you check to see if the ticket is the next up
  * If so, you return the future
  * If not, you buffer the result in something like a priority queue

That way you'd get all the perf benefits and should retain minimal memory footprint I think.
khuey: Hmm.  I'll look into that tomorrow.
rocallahan: Maybe `FuturesUnordered` should not describe itself as a queue, since its elements are not ordered from the user's point of view. It's really a set.
Stebalien: In general, it's best to avoid working with a potentially unbounded (or very large) number of futures. This is a recipe for a dos-by-oom vulnerability.

Additionally, futures are zero-allocation in a steady-state so, personally, I'd *expect* it to do all of its allocation up-front. This avoids repeated re-allocations under heavy load (precisely when you don't want such allocations). 

Finally, if you correctly tune the buffer size, you'll end up allocating the full buffer at some point anyways (and tokio will never call `shrink_to_fit` on the buffer to release memory) so you might as well do this up front as it makes your program's memory profile more predictable.
rocallahan: It's useful to be able to put an upper bound on memory usage without requiring all that memory to be allocated in cases where it won't be needed.

For complex applications with highly variable workloads it's not always possible to "correctly tune" a fixed buffer size.
rocallahan: In other words, for some applications it makes sense to have "predictable memory usage" by making it always use the worst-case amount of memory. But for other applications it doesn't.
alexcrichton: > Maybe FuturesUnordered should not describe itself as a queue, since its elements are not ordered from the user's point of view. It's really a set.

[Excellent point](https://github.com/alexcrichton/futures-rs/issues/505)!

> Finally, if you correctly tune the buffer size, you'll end up allocating the full buffer at some point anyways

@Stebalien We also ran into https://github.com/tokio-rs/tokio-core/issues/124 which was fixed by https://github.com/tokio-rs/tokio-core/pull/126, so it's not 100% always the best strategy to preallocate up front from what I can tell (you may not know whether you actually reach the limit or not.

> Hmm. I'll look into that tomorrow.

@khuey ok! Let me know if there's any difficulties though, I can also help look into it and whip up an implementation.

Patch:
diff --git a/.gitignore b/.gitignore
index a9d37c56..f0ff2599 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,3 @@
 target
 Cargo.lock
+*~
diff --git a/src/sink/map_err.rs b/src/sink/map_err.rs
index cccad399..3e9aa2a3 100644
--- a/src/sink/map_err.rs
+++ b/src/sink/map_err.rs
@@ -1,6 +1,6 @@
 use sink::Sink;
 
-use {Poll, StartSend, Stream};
+use {Poll, StartSend};
 
 /// Sink for the `Sink::sink_map_err` combinator.
 #[derive(Debug)]
@@ -53,12 +53,3 @@ impl<S, F, E> Sink for SinkMapErr<S, F>
         self.sink.close().map_err(|e| self.f.take().expect("cannot use MapErr after an error")(e))
     }
 }
-
-impl<S: Stream, F> Stream for SinkMapErr<S, F> {
-    type Item = S::Item;
-    type Error = S::Error;
-
-    fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
-        self.sink.poll()
-    }
-}
diff --git a/src/stream/buffer_unordered.rs b/src/stream/buffer_unordered.rs
deleted file mode 100644
index cdece4e7..00000000
--- a/src/stream/buffer_unordered.rs
+++ /dev/null
@@ -1,135 +0,0 @@
-use std::prelude::v1::*;
-use std::fmt;
-
-use {Async, IntoFuture, Poll};
-use stream::{Stream, Fuse, FuturesUnordered};
-
-/// 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>,
-
-    // Handles "in-flight" futures
-    queue: FuturesUnordered<<S::Item as IntoFuture>::Future>,
-
-    // Max number of futures
-    max: usize,
-}
-
-impl<S> fmt::Debug for BufferUnordered<S>
-    where S: Stream + fmt::Debug,
-          S::Item: IntoFuture,
-          <<S as Stream>::Item as IntoFuture>::Future: fmt::Debug,
-{
-    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
-        fmt.debug_struct("BufferUnordered")
-            .field("stream", &self.stream)
-            .field("queue", &self.queue)
-            .field("max", &self.max)
-            .finish()
-    }
-}
-
-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),
-        queue: FuturesUnordered::new(),
-        max: amt,
-    }
-}
-
-impl<S> BufferUnordered<S>
-    where S: Stream,
-          S::Item: IntoFuture<Error=<S as Stream>::Error>,
-{
-    /// Acquires a reference to the underlying stream that this combinator is
-    /// pulling from.
-    pub fn get_ref(&self) -> &S {
-        self.stream.get_ref()
-    }
-
-    /// Acquires a mutable reference to the underlying stream that this
-    /// combinator is pulling from.
-    ///
-    /// Note that care must be taken to avoid tampering with the state of the
-    /// stream which may otherwise confuse this combinator.
-    pub fn get_mut(&mut self) -> &mut S {
-        self.stream.get_mut()
-    }
-
-    /// Consumes this combinator, returning the underlying stream.
-    ///
-    /// Note that this may discard intermediate state of this combinator, so
-    /// care should be taken to avoid losing resources when this is called.
-    pub fn into_inner(self) -> S {
-        self.stream.into_inner()
-    }
-}
-
-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 up, try to spawn off as many futures as possible by filling up
-        // our slab of futures.
-        while self.queue.len() < self.max {
-            let future = match try!(self.stream.poll()) {
-                Async::Ready(Some(s)) => s.into_future(),
-                Async::Ready(None) |
-                Async::NotReady => break,
-            };
-
-            self.queue.push(future);
-        }
-
-        // Try polling a new future
-        if let Some(val) = try_ready!(self.queue.poll()) {
-            return Ok(Async::Ready(Some(val)));
-        }
-
-        // If we've gotten this far, then there are no events for us to process
-        // and nothing was ready, so figure out if we're not done yet  or if
-        // we've reached the end.
-        if self.stream.is_done() {
-            Ok(Async::Ready(None))
-        } else {
-            Ok(Async::NotReady)
-        }
-    }
-}
-
-// Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for BufferUnordered<S>
-    where S: ::sink::Sink + Stream,
-          S::Item: IntoFuture,
-{
-    type SinkItem = S::SinkItem;
-    type SinkError = S::SinkError;
-
-    fn start_send(&mut self, item: S::SinkItem) -> ::StartSend<S::SinkItem, S::SinkError> {
-        self.stream.start_send(item)
-    }
-
-    fn poll_complete(&mut self) -> Poll<(), S::SinkError> {
-        self.stream.poll_complete()
-    }
-
-    fn close(&mut self) -> Poll<(), S::SinkError> {
-        self.stream.close()
-    }
-}
diff --git a/src/stream/buffered.rs b/src/stream/buffered.rs
index 0bcf6dbf..1817302e 100644
--- a/src/stream/buffered.rs
+++ b/src/stream/buffered.rs
@@ -1,64 +1,63 @@
-use std::prelude::v1::*;
-
-use std::fmt;
-use std::mem;
-
-use {Async, IntoFuture, Poll, Future};
-use stream::{Stream, Fuse};
+use {Async, IntoFuture, Poll};
+use stream::{Stream, Fuse, FuturesSet};
 
 /// An adaptor for a stream of futures to execute the futures concurrently, if
 /// possible.
 ///
 /// This adaptor will buffer up a list of pending futures, and then return their
-/// results in the order that they were pulled out of the original stream. This
-/// is created by the `Stream::buffered` method.
+/// results in the order that they are returned by `Q`. This is created by the
+/// `Stream::buffered` or the `Stream::buffer_unordered` method.
 #[must_use = "streams do nothing unless polled"]
-pub struct Buffered<S>
+pub struct Buffered<Q, S>
     where S: Stream,
           S::Item: IntoFuture,
+          Q: FuturesSet<<<S as Stream>::Item as IntoFuture>::Future>,
 {
     stream: Fuse<S>,
-    futures: Vec<State<<S::Item as IntoFuture>::Future>>,
-    cur: usize,
+
+    // Handles "in-flight" futures
+    queue: Q,
+
+    // Max number of futures
+    max: usize,
 }
 
-impl<S> fmt::Debug for Buffered<S>
+if_std! {
+use std::fmt;
+
+impl<Q, S> fmt::Debug for Buffered<Q, S>
     where S: Stream + fmt::Debug,
           S::Item: IntoFuture,
           <<S as Stream>::Item as IntoFuture>::Future: fmt::Debug,
-          <<S as Stream>::Item as IntoFuture>::Item: fmt::Debug,
-          <<S as Stream>::Item as IntoFuture>::Error: fmt::Debug,
+          Q: FuturesSet<<<S as Stream>::Item as IntoFuture>::Future> + fmt::Debug,
 {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
-        fmt.debug_struct("Stream")
+        fmt.debug_struct("Buffered")
             .field("stream", &self.stream)
-            .field("futures", &self.futures)
-            .field("cur", &self.cur)
+            .field("queue", &self.queue)
+            .field("max", &self.max)
             .finish()
     }
 }
 
-#[derive(Debug)]
-enum State<S: Future> {
-    Empty,
-    Running(S),
-    Finished(Result<S::Item, S::Error>),
 }
 
-pub fn new<S>(s: S, amt: usize) -> Buffered<S>
+pub fn new<Q, S>(s: S, amt: usize) -> Buffered<Q, S>
     where S: Stream,
           S::Item: IntoFuture<Error=<S as Stream>::Error>,
+          Q: FuturesSet<<<S as Stream>::Item as IntoFuture>::Future>,
 {
     Buffered {
         stream: super::fuse::new(s),
-        futures: (0..amt).map(|_| State::Empty).collect(),
-        cur: 0,
+        queue: Q::new(),
+        max: amt,
     }
 }
 
-impl<S> Buffered<S>
+impl<Q, S> Buffered<Q, S>
     where S: Stream,
-          S::Item: IntoFuture,
+          S::Item: IntoFuture<Error=<S as Stream>::Error>,
+          Q: FuturesSet<<<S as Stream>::Item as IntoFuture>::Future>,
 {
     /// Acquires a reference to the underlying stream that this combinator is
     /// pulling from.
@@ -84,10 +83,48 @@ impl<S> Buffered<S>
     }
 }
 
+impl<Q, S> Stream for Buffered<Q, S>
+    where S: Stream,
+          S::Item: IntoFuture<Error=<S as Stream>::Error>,
+          Q: FuturesSet<<<S as Stream>::Item as IntoFuture>::Future>,
+{
+    type Item = <S::Item as IntoFuture>::Item;
+    type Error = <S as Stream>::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        // First up, try to spawn off as many futures as possible by filling up
+        // our slab of futures.
+        while self.queue.len() < self.max {
+            let future = match try!(self.stream.poll()) {
+                Async::Ready(Some(s)) => s.into_future(),
+                Async::Ready(None) |
+                Async::NotReady => break,
+            };
+
+            self.queue.push(future);
+        }
+
+        // Try polling a new future
+        if let Some(val) = try_ready!(self.queue.poll()) {
+            return Ok(Async::Ready(Some(val)));
+        }
+
+        // If we've gotten this far, then there are no events for us to process
+        // and nothing was ready, so figure out if we're not done yet  or if
+        // we've reached the end.
+        if self.stream.is_done() {
+            Ok(Async::Ready(None))
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Buffered<S>
+impl<Q, S> ::sink::Sink for Buffered<Q, S>
     where S: ::sink::Sink + Stream,
           S::Item: IntoFuture,
+          Q: FuturesSet<<<S as Stream>::Item as IntoFuture>::Future>,
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -104,67 +141,3 @@ impl<S> ::sink::Sink for Buffered<S>
         self.stream.close()
     }
 }
-
-impl<S> Stream for Buffered<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 i in 0..self.futures.len() {
-            let mut idx = self.cur + i;
-            if idx >= self.futures.len() {
-                idx -= self.futures.len();
-            }
-
-            if let State::Empty = self.futures[idx] {
-                match try!(self.stream.poll()) {
-                    Async::Ready(Some(future)) => {
-                        let future = future.into_future();
-                        self.futures[idx] = State::Running(future);
-                    }
-                    Async::Ready(None) => break,
-                    Async::NotReady => break,
-                }
-            }
-        }
-
-        // Next, try and step all the futures forward
-        for future in &mut self.futures {
-            let result = match *future {
-                State::Running(ref mut s) => {
-                    match s.poll() {
-                        Ok(Async::NotReady) => continue,
-                        Ok(Async::Ready(e)) => Ok(e),
-                        Err(e) => Err(e),
-                    }
-                }
-                _ => continue,
-            };
-            *future = State::Finished(result);
-        }
-
-        // Check to see if our current future is done.
-        if let State::Finished(_) = self.futures[self.cur] {
-            let r = match mem::replace(&mut self.futures[self.cur], State::Empty) {
-                State::Finished(r) => r,
-                _ => panic!(),
-            };
-            self.cur += 1;
-            if self.cur >= self.futures.len() {
-                self.cur = 0;
-            }
-            return Ok(Async::Ready(Some(try!(r))))
-        }
-
-        if self.stream.is_done() {
-            if let State::Empty = self.futures[self.cur] {
-                return Ok(Async::Ready(None))
-            }
-        }
-        Ok(Async::NotReady)
-    }
-}
diff --git a/src/stream/futures_ordered.rs b/src/stream/futures_ordered.rs
new file mode 100644
index 00000000..11209487
--- /dev/null
+++ b/src/stream/futures_ordered.rs
@@ -0,0 +1,204 @@
+use std::cmp::{Eq, PartialEq, PartialOrd, Ord, Ordering};
+use std::collections::BinaryHeap;
+use std::fmt::{self, Debug};
+
+use {Async, Future, IntoFuture, Poll, Stream};
+use stream::{FuturesSet, FuturesUnordered};
+
+#[derive(Debug)]
+struct OrderWrapper<T> {
+    item: T,
+    index: usize,
+}
+
+impl<T> PartialEq for OrderWrapper<T> {
+    fn eq(&self, other: &Self) -> bool {
+        self.index == other.index
+    }
+}
+
+impl<T> Eq for OrderWrapper<T> {}
+
+impl<T> PartialOrd for OrderWrapper<T> {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl<T> Ord for OrderWrapper<T> {
+    fn cmp(&self, other: &Self) -> Ordering {
+        // BinaryHeap is a max heap, so compare backwards here.
+        other.index.cmp(&self.index)
+    }
+}
+
+impl<T> Future for OrderWrapper<T>
+    where T: Future
+{
+    type Item = OrderWrapper<T::Item>;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let result = try_ready!(self.item.poll());
+        Ok(Async::Ready(OrderWrapper {
+            item: result,
+            index: self.index
+        }))
+    }
+}
+
+/// An unbounded queue of futures.
+///
+/// This "combinator" is similar to `FuturesUnordered`, but it imposes an order
+/// on top of the set of futures. While futures in the set will race to
+/// completion in parallel, results will only be returned in the order their
+/// originating futures were added to the queue.
+///
+/// Futures are pushed into this queue and their realized values are yielded in
+/// order. This structure is optimized to manage a large number of futures.
+/// Futures managed by `FuturesOrdered` will only be polled when they generate
+/// notifications. This reduces the required amount of work needed to coordinate
+/// large numbers of futures.
+///
+/// When a `FuturesOrdered` is first created, it does not contain any futures.
+/// Calling `poll` in this state will result in `Ok(Async::Ready(None))` to be
+/// returned. Futures are submitted to the queue using `push`; however, the
+/// future will **not** be polled at this point. `FuturesOrdered` will only
+/// poll managed futures when `FuturesOrdered::poll` is called. As such, it
+/// is important to call `poll` after pushing new futures.
+///
+/// If `FuturesOrdered::poll` returns `Ok(Async::Ready(None))` this means that
+/// the queue is currently not managing any futures. A future may be submitted
+/// to the queue at a later time. At that point, a call to
+/// `FuturesOrdered::poll` will either return the future's resolved value
+/// **or** `Ok(Async::NotReady)` if the future has not yet completed. When
+/// multiple futures are submitted to the queue, `FuturesOrdered::poll` will
+/// return `Ok(Async::NotReady)` until the first future completes, even if
+/// some of the later futures have already completed.
+///
+/// Note that you can create a ready-made `FuturesOrdered` via the
+/// `futures_ordered` function in the `stream` module, or you can start with an
+/// empty queue with the `FuturesOrdered::new` constructor.
+#[must_use = "streams do nothing unless polled"]
+pub struct FuturesOrdered<T>
+    where T: Future
+{
+    in_progress: FuturesUnordered<OrderWrapper<T>>,
+    queued_results: BinaryHeap<OrderWrapper<T::Item>>,
+    next_incoming_index: usize,
+    next_outgoing_index: usize,
+}
+
+/// Converts a list of futures into a `Stream` of results from the futures.
+///
+/// This function will take an list of futures (e.g. a vector, an iterator,
+/// etc), and return a stream. The stream will yield items as they become
+/// available on the futures internally, in the order that their originating
+/// futures were submitted to the queue. If the futures complete out of order,
+/// items will be stored internally within `FuturesOrdered` until all preceding
+/// items have been yielded.
+///
+/// Note that the returned queue can also be used to dynamically push more
+/// futures into the queue as they become available.
+pub fn futures_ordered<I>(futures: I) -> FuturesOrdered<<I::Item as IntoFuture>::Future>
+    where I: IntoIterator,
+          I::Item: IntoFuture
+{
+    let mut queue = FuturesOrdered::new();
+
+    for future in futures {
+        queue.push(future.into_future());
+    }
+
+    return queue
+}
+
+impl<T> FuturesSet<T> for FuturesOrdered<T>
+    where T: Future
+{
+    /// Constructs a new, empty `FuturesOrdered`
+    ///
+    /// The returned `FuturesOrdered` does not contain any futures and, in this
+    /// state, `FuturesOrdered::poll` will return `Ok(Async::Ready(None))`.
+    fn new() -> FuturesOrdered<T> {
+        FuturesOrdered {
+            in_progress: FuturesUnordered::new(),
+            queued_results: BinaryHeap::new(),
+            next_incoming_index: 0,
+            next_outgoing_index: 0,
+        }
+    }
+
+    /// Returns the number of futures contained in the queue.
+    ///
+    /// This represents the total number of in-flight futures, both
+    /// those currently processing and those that have completed but
+    /// which are waiting for earlier futures to complete.
+    fn len(&self) -> usize {
+        self.in_progress.len() + self.queued_results.len()
+    }
+
+    /// Push a future into the queue.
+    ///
+    /// This function submits the given future to the internal set for managing.
+    /// This function will not call `poll` on the submitted future. The caller
+    /// must ensure that `FuturesOrdered::poll` is called in order to receive
+    /// task notifications.
+    fn push(&mut self, future: T) {
+        let wrapped = OrderWrapper {
+            item: future,
+            index: self.next_incoming_index,
+        };
+        self.next_incoming_index += 1;
+        self.in_progress.push(wrapped);
+    }
+}
+
+impl<T> FuturesOrdered<T>
+    where T: Future
+{
+    /// Returns `true` if the queue contains no futures
+    pub fn is_empty(&self) -> bool {
+        self.in_progress.is_empty() && self.queued_results.is_empty()
+    }
+}
+
+impl<T> Stream for FuturesOrdered<T>
+    where T: Future
+{
+    type Item = T::Item;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        // Get any completed futures from the unordered set.
+        loop {
+            match try!(self.in_progress.poll()) {
+                Async::Ready(Some(result)) => self.queued_results.push(result),
+                Async::Ready(None) | Async::NotReady => break,
+            }
+        }
+
+        if let Some(next_result) = self.queued_results.peek() {
+            // PeekMut::pop is not stable yet QQ
+            if next_result.index != self.next_outgoing_index {
+                return Ok(Async::NotReady);
+            }
+        } else if !self.in_progress.is_empty() {
+            return Ok(Async::NotReady);
+        } else {
+            return Ok(Async::Ready(None));
+        }
+
+        let next_result = self.queued_results.pop().unwrap();
+        self.next_outgoing_index += 1;
+        Ok(Async::Ready(Some(next_result.item)))
+    }
+}
+
+impl<T: Debug> Debug for FuturesOrdered<T>
+    where T: Future
+{
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "FuturesOrdered {{ ... }}")
+    }
+}
diff --git a/src/stream/futures_set.rs b/src/stream/futures_set.rs
new file mode 100644
index 00000000..a14270d8
--- /dev/null
+++ b/src/stream/futures_set.rs
@@ -0,0 +1,30 @@
+use {Future, Stream};
+
+/// Trait for types which manage a set of `Future`s.
+///
+/// A `FuturesSet` polls its set of `Future`s and returns produced values
+/// through its implementation of `Stream`.
+///
+/// Two implementations of `FuturesSet` are provided:
+/// `FuturesUnordered` and `FuturesOrdered`, which produce items in the order
+/// in which they are produced and the order which their corresponding futures
+/// were added to the set, respectively.
+pub trait FuturesSet<T> : Stream<Item = T::Item, Error = T::Error>
+    where T: Future
+{
+    /// Constructs a new, empty implementation of `FuturesSet`.
+    ///
+    /// The returned `FuturesSet` does not contain any futures and, in this
+    /// state, `poll` will return `Ok(Async::Ready(None))`.
+    fn new() -> Self;
+
+    /// Returns the number of futures contained in the set.
+    fn len(&self) -> usize;
+
+    /// Push a future into the set.
+    ///
+    /// This function submits the given future to the set for managing. This
+    /// function will not call `poll` on the submitted future. The caller must
+    /// ensure that `poll` is called in order to receive task notifications.
+    fn push(&mut self, future: T);
+}
diff --git a/src/stream/futures_unordered.rs b/src/stream/futures_unordered.rs
index 17743eae..2a670f4c 100644
--- a/src/stream/futures_unordered.rs
+++ b/src/stream/futures_unordered.rs
@@ -10,15 +10,16 @@ use std::usize;
 
 use {task, Stream, Future, Poll, Async, IntoFuture};
 use executor::{Notify, UnsafeNotify, NotifyHandle};
+use stream::{FuturesSet};
 use task_impl::{self, AtomicTask};
 
-/// An unbounded queue of futures.
+/// An unbounded set of futures.
 ///
 /// This "combinator" also serves a special function in this library, providing
-/// the ability to maintain a queue of futures that and manage driving them all
+/// the ability to maintain a set of futures that and manage driving them all
 /// to completion.
 ///
-/// Futures are pushed into this queue and their realized values are yielded as
+/// Futures are pushed into this set and their realized values are yielded as
 /// they are ready. This structure is optimized to manage a large number of
 /// futures. Futures managed by `FuturesUnordered` will only be polled when they
 /// generate notifications. This reduces the required amount of work needed to
@@ -26,20 +27,20 @@ use task_impl::{self, AtomicTask};
 ///
 /// When a `FuturesUnordered` is first created, it does not contain any futures.
 /// Calling `poll` in this state will result in `Ok(Async::Ready(None))` to be
-/// returned. Futures are submitted to the queue using `push`; however, the
+/// returned. Futures are submitted to the set using `push`; however, the
 /// future will **not** be polled at this point. `FuturesUnordered` will only
 /// poll managged futures when `FuturesUnordered::poll` is called. As such, it
 /// is important to call `poll` after pushing new futures.
 ///
 /// If `FuturesUnordered::poll` returns `Ok(Async::Ready(None))` this means that
-/// the queue is currently not managing any futures. A future may be submitted
-/// to the queue at a later time. At that point, a call to
+/// the set is currently not managing any futures. A future may be submitted
+/// to the set at a later time. At that point, a call to
 /// `FuturesUnordered::poll` will either return the future's resolved value
 /// **or** `Ok(Async::NotReady)` if the future has not yet completed.
 ///
 /// Note that you can create a ready-made `FuturesUnordered` via the
-/// `futures_unordered` function in the `stream` module, or you can start with a
-/// blank queue with the `FuturesUnordered::new` constructor.
+/// `futures_unordered` function in the `stream` module, or you can start with an
+/// empty set with the `FuturesUnordered::new` constructor.
 #[must_use = "streams do nothing unless polled"]
 pub struct FuturesUnordered<F> {
     inner: Arc<Inner<F>>,
@@ -58,19 +59,19 @@ unsafe impl<T: Sync> Sync for FuturesUnordered<T> {}
 /// available. This function is similar to `buffer_unordered` in that it may
 /// return items in a different order than in the list specified.
 ///
-/// Note that the returned queue can also be used to dynamically push more
-/// futures onto the queue as they become available.
+/// Note that the returned set can also be used to dynamically push more
+/// futures into the set as they become available.
 pub fn futures_unordered<I>(futures: I) -> FuturesUnordered<<I::Item as IntoFuture>::Future>
     where I: IntoIterator,
           I::Item: IntoFuture
 {
-    let mut queue = FuturesUnordered::new();
+    let mut set = FuturesUnordered::new();
 
     for future in futures {
-        queue.push(future.into_future());
+        set.push(future.into_future());
     }
 
-    return queue
+    return set
 }
 
 // FuturesUnordered is implemented using two linked lists. One which links all
@@ -80,7 +81,7 @@ pub fn futures_unordered<I>(futures: I) -> FuturesUnordered<<I::Item as IntoFutu
 // second linked list is an implementation of the intrusive MPSC queue algorithm
 // described by 1024cores.net.
 //
-// When a future is submitted to the queue a node is allocated and inserted in
+// When a future is submitted to the set a node is allocated and inserted in
 // both linked lists. The next call to `poll` will (eventually) see this node
 // and call `poll` on the future.
 //
@@ -133,14 +134,14 @@ enum Dequeue<T> {
     Inconsistent,
 }
 
-impl<T> FuturesUnordered<T>
+impl<T> FuturesSet<T> for FuturesUnordered<T>
     where T: Future,
 {
     /// Constructs a new, empty `FuturesUnordered`
     ///
     /// The returned `FuturesUnordered` does not contain any futures and, in this
     /// state, `FuturesUnordered::poll` will return `Ok(Async::Ready(None))`.
-    pub fn new() -> FuturesUnordered<T> {
+    fn new() -> FuturesUnordered<T> {
         let stub = Arc::new(Node {
             future: UnsafeCell::new(None),
             next_all: UnsafeCell::new(ptr::null()),
@@ -163,28 +164,21 @@ impl<T> FuturesUnordered<T>
             inner: inner,
         }
     }
-}
 
-impl<T> FuturesUnordered<T> {
-    /// Returns the number of futures contained by the queue.
+    /// Returns the number of futures contained in the set.
     ///
     /// This represents the total number of in-flight futures.
-    pub fn len(&self) -> usize {
+    fn len(&self) -> usize {
         self.len
     }
 
-    /// Returns `true` if the queue contains no futures
-    pub fn is_empty(&self) -> bool {
-        self.len == 0
-    }
-
-    /// Push a future into the queue.
+    /// Push a future into the set.
     ///
-    /// This function submits the given future to the queue for managing. This
+    /// This function submits the given future to the set for managing. This
     /// function will not call `poll` on the submitted future. The caller must
     /// ensure that `FuturesUnordered::poll` is called in order to receive task
     /// notifications.
-    pub fn push(&mut self, future: T) {
+    fn push(&mut self, future: T) {
         let node = Arc::new(Node {
             future: UnsafeCell::new(Some(future)),
             next_all: UnsafeCell::new(ptr::null_mut()),
@@ -205,6 +199,13 @@ impl<T> FuturesUnordered<T> {
         // polling here.
         self.inner.enqueue(ptr);
     }
+}
+
+impl<T> FuturesUnordered<T> {
+    /// Returns `true` if the set contains no futures
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
 
     fn release_node(&mut self, node: Arc<Node<T>>) {
         // The future is done, try to reset the queued flag. This will prevent
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 8a7e466c..d6f8fcbc 100755
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -37,6 +37,7 @@ mod for_each;
 mod from_err;
 mod fuse;
 mod future;
+mod futures_set;
 mod map;
 mod map_err;
 mod merge;
@@ -64,6 +65,7 @@ pub use self::for_each::ForEach;
 pub use self::from_err::FromErr;
 pub use self::fuse::Fuse;
 pub use self::future::StreamFuture;
+pub use self::futures_set::FuturesSet;
 pub use self::map::Map;
 pub use self::map_err::MapErr;
 pub use self::merge::{Merge, MergedItem};
@@ -85,7 +87,6 @@ if_std! {
     use std;
 
     mod buffered;
-    mod buffer_unordered;
     mod catch_unwind;
     mod chunks;
     mod collect;
@@ -93,14 +94,15 @@ if_std! {
     mod channel;
     mod split;
     mod futures_unordered;
+    mod futures_ordered;
     pub use self::buffered::Buffered;
-    pub use self::buffer_unordered::BufferUnordered;
     pub use self::catch_unwind::CatchUnwind;
     pub use self::chunks::Chunks;
     pub use self::collect::Collect;
     pub use self::wait::Wait;
     pub use self::split::{SplitStream, SplitSink};
     pub use self::futures_unordered::{futures_unordered, FuturesUnordered};
+    pub use self::futures_ordered::{futures_ordered, FuturesOrdered};
 
     #[doc(hidden)]
     #[cfg(feature = "with-deprecated")]
@@ -880,7 +882,7 @@ pub trait Stream {
     /// This method is only available when the `use_std` feature of this
     /// library is activated, and it is activated by default.
     #[cfg(feature = "use_std")]
-    fn buffered(self, amt: usize) -> Buffered<Self>
+    fn buffered(self, amt: usize) -> Buffered<FuturesOrdered<<<Self as Stream>::Item as IntoFuture>::Future>, Self>
         where Self::Item: IntoFuture<Error = <Self as Stream>::Error>,
               Self: Sized
     {
@@ -901,11 +903,11 @@ pub trait Stream {
     /// This method is only available when the `use_std` feature of this
     /// library is activated, and it is activated by default.
     #[cfg(feature = "use_std")]
-    fn buffer_unordered(self, amt: usize) -> BufferUnordered<Self>
+    fn buffer_unordered(self, amt: usize) -> Buffered<FuturesUnordered<<<Self as Stream>::Item as IntoFuture>::Future>, Self>
         where Self::Item: IntoFuture<Error = <Self as Stream>::Error>,
               Self: Sized
     {
-        buffer_unordered::new(self, amt)
+        buffered::new(self, amt)
     }
 
     /// An adapter for merging the output of two streams.


Test patch:
diff --git a/tests/futures_ordered.rs b/tests/futures_ordered.rs
new file mode 100644
index 00000000..18ba0d6e
--- /dev/null
+++ b/tests/futures_ordered.rs
@@ -0,0 +1,69 @@
+extern crate futures;
+
+use std::any::Any;
+
+use futures::sync::oneshot;
+use futures::stream::futures_ordered;
+use futures::Future;
+
+mod support;
+
+#[test]
+fn works_1() {
+    let (a_tx, a_rx) = oneshot::channel::<u32>();
+    let (b_tx, b_rx) = oneshot::channel::<u32>();
+    let (c_tx, c_rx) = oneshot::channel::<u32>();
+
+    let stream = futures_ordered(vec![a_rx, b_rx, c_rx]);
+
+    let mut spawn = futures::executor::spawn(stream);
+    b_tx.send(99).unwrap();
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+
+    a_tx.send(33).unwrap();
+    c_tx.send(33).unwrap();
+    assert_eq!(Some(Ok(33)), spawn.wait_stream());
+    assert_eq!(Some(Ok(99)), spawn.wait_stream());
+    assert_eq!(Some(Ok(33)), spawn.wait_stream());
+    assert_eq!(None, spawn.wait_stream());
+}
+
+#[test]
+fn works_2() {
+    let (a_tx, a_rx) = oneshot::channel::<u32>();
+    let (b_tx, b_rx) = oneshot::channel::<u32>();
+    let (c_tx, c_rx) = oneshot::channel::<u32>();
+
+    let stream = futures_ordered(vec![a_rx.boxed(), b_rx.join(c_rx).map(|(a, b)| a + b).boxed()]);
+
+    let mut spawn = futures::executor::spawn(stream);
+    a_tx.send(33).unwrap();
+    b_tx.send(33).unwrap();
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_ready());
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+    c_tx.send(33).unwrap();
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_ready());
+}
+
+#[test]
+fn queue_never_unblocked() {
+    let (_a_tx, a_rx) = oneshot::channel::<Box<Any+Send>>();
+    let (b_tx, b_rx) = oneshot::channel::<Box<Any+Send>>();
+    let (c_tx, c_rx) = oneshot::channel::<Box<Any+Send>>();
+
+    let stream = futures_ordered(vec![
+        a_rx.boxed(),
+        b_rx.select(c_rx).then(|res| Ok(Box::new(res) as Box<Any+Send>)).boxed(),
+    ]);
+
+    let mut spawn = futures::executor::spawn(stream);
+    for _ in 0..10 {
+        assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+    }
+
+    b_tx.send(Box::new(())).unwrap();
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+    c_tx.send(Box::new(())).unwrap();
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+}
diff --git a/tests/ready_queue.rs b/tests/ready_queue.rs
index e90c7a5a..eae9b370 100644
--- a/tests/ready_queue.rs
+++ b/tests/ready_queue.rs
@@ -5,7 +5,7 @@ use std::panic::{self, AssertUnwindSafe};
 use futures::{Future, Stream, Poll};
 use futures::Async::*;
 use futures::future;
-use futures::stream::FuturesUnordered;
+use futures::stream::{FuturesSet, FuturesUnordered};
 use futures::sync::oneshot;
 
 trait AssertSendSync: Send + Sync {}