Repository: alexcrichton/futures-rs
Pull request #618 (2017-10-22 03:40:46)

Implement FuturesUnordered::iter_mut
Implementing `FuturesUnordered::iter_mut` as suggested in #605. Generally it should be said that I don't have a lot of experience with unsafe rust yet, so I would appreciate a thorough review!

I did not go for the other suggested ones (`into_iter`/`drain`), since I lack the insight into how to implement this safely without leaking memory.

srijs: @carllerche Thanks for the review, addressed all feedback!
alexcrichton: Inside this module there's a public `futures_unordered` function which if we make the module public will also make that function reachable (but it's not intentional to have that function reachable. I think I'm ok making the module public for the iterator name (albeit it's a bit of a wonky setup) but can we ensure that the free function isn't exposed?
carllerche: Which release of rust included `pub(crate)`? I would guess that we should be able to use it soon if not now.
alexcrichton: Unsure, but the function can also just be moved
srijs: According to [this](https://github.com/rust-lang/rust/issues/32409#issuecomment-297080960), `pub(crate)` was stabilised in 1.18.

@alexcrichton the `futures_unordered` function is exported, according to the docs: https://docs.rs/futures/0.1.16/futures/stream/fn.futures_unordered.html


alexcrichton: @srijs right yeah but as is this would also export `futures::stream::futures_unordered::futures_unordered` (that is, a function in the module). Ideally only `futures::stream::futures_unordered` would be the exported function.
srijs: @alexcrichton Now that we have `FromIterator` for `FuturesUnordered`, could it be an option to deprecate the free-standing function, or do still see a use-case for it which is not covered by `from_iter`?
alexcrichton: Ah yeah so for now I personally like the free function for symmetry with the rest of the crate, but eventual deprecation is of course possible!
srijs: @alexcrichton Feedback addressed, let me know if you'd like me to change anything else!

Patch:
diff --git a/.travis.yml b/.travis.yml
index 9c6f16b6..f3bf51c6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,7 +12,7 @@ matrix:
       after_success:
         - travis-cargo doc-upload
     - os: linux
-      rust: 1.13.0
+      rust: 1.15.0
       script: cargo test
 sudo: false
 script:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 15b7d116..28ab54a0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,12 +1,3 @@
-# 0.1.17 - 2017-10-31
-
-* Add a `close` method on `sink::Wait`
-* Undeprecate `stream::iter` as `stream::iter_result`
-* Improve performance of wait-related methods
-* Tweak buffered sinks with a 0 capacity to forward directly to the underlying
-  sink.
-* Add `FromIterator` implementation for `FuturesOrdered` and `FuturesUnordered`.
-
 # 0.1.16 - 2017-09-15
 
 * A `prelude` module has been added to glob import from and pick up a whole
diff --git a/Cargo.toml b/Cargo.toml
index 2e702a99..1c12d106 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "futures"
-version = "0.1.17"
+version = "0.1.16"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT/Apache-2.0"
 readme = "README.md"
diff --git a/README.md b/README.md
index f5a5b0c8..43cf2d5f 100644
--- a/README.md
+++ b/README.md
@@ -44,17 +44,8 @@ futures = { version = "0.1.14", default-features = false }
 
 # License
 
-This project is licensed under either of
+`futures-rs` is primarily distributed under the terms of both the MIT license and
+the Apache License (Version 2.0), with portions covered by various BSD-like
+licenses.
 
- * Apache License, Version 2.0, ([LICENSE-APACHE](LICENSE-APACHE) or
-   http://www.apache.org/licenses/LICENSE-2.0)
- * MIT license ([LICENSE-MIT](LICENSE-MIT) or
-   http://opensource.org/licenses/MIT)
-
-at your option.
-
-### Contribution
-
-Unless you explicitly state otherwise, any contribution intentionally submitted
-for inclusion in Serde by you, as defined in the Apache-2.0 license, shall be
-dual licensed as above, without any additional terms or conditions.
+See LICENSE-APACHE, and LICENSE-MIT for details.
diff --git a/benches/thread_notify.rs b/benches/thread_notify.rs
deleted file mode 100644
index 92932353..00000000
--- a/benches/thread_notify.rs
+++ /dev/null
@@ -1,114 +0,0 @@
-#![feature(test)]
-
-extern crate futures;
-extern crate test;
-
-use futures::{Future, Poll, Async};
-use futures::task::{self, Task};
-
-use test::Bencher;
-
-#[bench]
-fn thread_yield_single_thread_one_wait(b: &mut Bencher) {
-    const NUM: usize = 10_000;
-
-    struct Yield {
-        rem: usize,
-    }
-
-    impl Future for Yield {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<(), ()> {
-            if self.rem == 0 {
-                Ok(Async::Ready(()))
-            } else {
-                self.rem -= 1;
-                task::current().notify();
-                Ok(Async::NotReady)
-            }
-        }
-    }
-
-    b.iter(|| {
-        let y = Yield { rem: NUM };
-        y.wait().unwrap();
-    });
-}
-
-#[bench]
-fn thread_yield_single_thread_many_wait(b: &mut Bencher) {
-    const NUM: usize = 10_000;
-
-    struct Yield {
-        rem: usize,
-    }
-
-    impl Future for Yield {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<(), ()> {
-            if self.rem == 0 {
-                Ok(Async::Ready(()))
-            } else {
-                self.rem -= 1;
-                task::current().notify();
-                Ok(Async::NotReady)
-            }
-        }
-    }
-
-    b.iter(|| {
-        for _ in 0..NUM {
-            let y = Yield { rem: 1 };
-            y.wait().unwrap();
-        }
-    });
-}
-
-#[bench]
-fn thread_yield_multi_thread(b: &mut Bencher) {
-    use std::sync::mpsc;
-    use std::thread;
-
-    const NUM: usize = 1_000;
-
-    let (tx, rx) = mpsc::sync_channel::<Task>(10_000);
-
-    struct Yield {
-        rem: usize,
-        tx: mpsc::SyncSender<Task>,
-    }
-
-    impl Future for Yield {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<(), ()> {
-            if self.rem == 0 {
-                Ok(Async::Ready(()))
-            } else {
-                self.rem -= 1;
-                self.tx.send(task::current()).unwrap();
-                Ok(Async::NotReady)
-            }
-        }
-    }
-
-    thread::spawn(move || {
-        while let Ok(task) = rx.recv() {
-            task.notify();
-        }
-    });
-
-    b.iter(move || {
-        let y = Yield {
-            rem: NUM,
-            tx: tx.clone(),
-        };
-
-        y.wait().unwrap();
-    });
-}
diff --git a/src/stream/futures_unordered.rs b/src/stream/futures_unordered.rs
index 632bd831..2940fd34 100644
--- a/src/stream/futures_unordered.rs
+++ b/src/stream/futures_unordered.rs
@@ -1,3 +1,5 @@
+//! An unbounded set of futures.
+
 use std::cell::UnsafeCell;
 use std::fmt::{self, Debug};
 use std::iter::FromIterator;
@@ -9,7 +11,7 @@ use std::sync::atomic::{AtomicPtr, AtomicBool};
 use std::sync::{Arc, Weak};
 use std::usize;
 
-use {task, Stream, Future, Poll, Async, IntoFuture};
+use {task, Stream, Future, Poll, Async};
 use executor::{Notify, UnsafeNotify, NotifyHandle};
 use task_impl::{self, AtomicTask};
 
@@ -51,29 +53,6 @@ pub struct FuturesUnordered<F> {
 unsafe impl<T: Send> Send for FuturesUnordered<T> {}
 unsafe impl<T: Sync> Sync for FuturesUnordered<T> {}
 
-/// 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 they become
-/// 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 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 set = FuturesUnordered::new();
-
-    for future in futures {
-        set.push(future.into_future());
-    }
-
-    return set
-}
-
 // FuturesUnordered is implemented using two linked lists. One which links all
 // futures managed by a `FuturesUnordered` and one that tracks futures that have
 // been scheduled for polling. The first linked list is not thread safe and is
@@ -207,6 +186,15 @@ impl<T> FuturesUnordered<T> {
         self.inner.enqueue(ptr);
     }
 
+    /// Returns an iterator that allows modifying each future in the set.
+    pub fn iter_mut(&mut self) -> IterMut<T> {
+        IterMut {
+            node: self.head_all,
+            len: self.len,
+            _marker: PhantomData
+        }
+    }
+
     fn release_node(&mut self, node: Arc<Node<T>>) {
         // The future is done, try to reset the queued flag. This will prevent
         // `notify` from doing any work in the future
@@ -440,6 +428,37 @@ impl<F: Future> FromIterator<F> for FuturesUnordered<F> {
     }
 }
 
+#[derive(Debug)]
+/// Mutable iterator over all futures in the unordered set.
+pub struct IterMut<'a, F: 'a> {
+    node: *const Node<F>,
+    len: usize,
+    _marker: PhantomData<&'a mut FuturesUnordered<F>>
+}
+
+impl<'a, F> Iterator for IterMut<'a, F> {
+    type Item = &'a mut F;
+
+    fn next(&mut self) -> Option<&'a mut F> {
+        if self.node.is_null() {
+            return None;
+        }
+        unsafe {
+            let future = (*(*self.node).future.get()).as_mut().unwrap();
+            let next = *(*self.node).next_all.get();
+            self.node = next;
+            self.len -= 1;
+            return Some(future);
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.len, Some(self.len))
+    }
+}
+
+impl<'a, F> ExactSizeIterator for IterMut<'a, F> {}
+
 impl<T> Inner<T> {
     /// The enqueue function from the 1024cores intrusive MPSC queue algorithm.
     fn enqueue(&self, node: *const Node<T>) {
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 53e866a3..e3ea2e3f 100644
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -103,7 +103,7 @@ if_std! {
     mod wait;
     mod channel;
     mod split;
-    mod futures_unordered;
+    pub mod futures_unordered;
     mod futures_ordered;
     pub use self::buffered::Buffered;
     pub use self::buffer_unordered::BufferUnordered;
@@ -112,7 +112,7 @@ if_std! {
     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_unordered::FuturesUnordered;
     pub use self::futures_ordered::{futures_ordered, FuturesOrdered};
 
     #[doc(hidden)]
@@ -1102,3 +1102,27 @@ impl<'a, S: ?Sized + Stream> Stream for &'a mut S {
         (**self).poll()
     }
 }
+
+/// 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 they become
+/// 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 set can also be used to dynamically push more
+/// futures into the set as they become available.
+#[cfg(feature = "use_std")]
+pub fn futures_unordered<I>(futures: I) -> FuturesUnordered<<I::Item as IntoFuture>::Future>
+    where I: IntoIterator,
+        I::Item: IntoFuture
+{
+    let mut set = FuturesUnordered::new();
+
+    for future in futures {
+        set.push(future.into_future());
+    }
+
+    return set
+}
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
index 0a46e9af..b5ea8d1a 100644
--- a/src/sync/mod.rs
+++ b/src/sync/mod.rs
@@ -12,6 +12,7 @@
 
 pub mod oneshot;
 pub mod mpsc;
+pub mod slot;
 mod bilock;
 
 pub use self::bilock::{BiLock, BiLockGuard, BiLockAcquire, BiLockAcquired};
diff --git a/src/sync/slot.rs b/src/sync/slot.rs
new file mode 100644
index 00000000..790b66d0
--- /dev/null
+++ b/src/sync/slot.rs
@@ -0,0 +1,183 @@
+//! An unbounded channel that only stores last value sent
+
+use std::mem;
+use std::sync::{Arc, Weak, Mutex};
+
+use task::{self, Task};
+use {Sink, Stream, AsyncSink, Async, Poll, StartSend};
+
+/// Slot is very similar to unbounded channel but only stores last value sent
+///
+/// I.e. if you want to send some value between from producer to a consumer
+/// and if consumer is slow it should skip old values, the slot is
+/// a structure for the task.
+
+/// The transmission end of a channel which is used to send values
+///
+/// If the receiver is not fast enough only the last value is preserved and
+/// other ones are discarded.
+#[derive(Debug)]
+pub struct Sender<T> {
+    inner: Weak<Mutex<Inner<T>>>,
+}
+
+/// The receiving end of a channel which preserves only the last value
+#[derive(Debug)]
+pub struct Receiver<T> {
+    inner: Arc<Mutex<Inner<T>>>,
+}
+
+/// Error type for sending, used when the receiving end of a channel is
+/// dropped
+#[derive(Clone, PartialEq, Eq, Debug)]
+pub struct SendError<T>(T);
+
+#[derive(Debug)]
+struct Inner<T> {
+    value: Option<T>,
+    task: Option<Task>,
+}
+
+trait AssertKindsSender: Send + Sync + Clone {}
+impl AssertKindsSender for Sender<u32> {}
+
+trait AssertKindsReceiver: Send + Sync {}
+impl AssertKindsReceiver for Receiver<u32> {}
+
+impl<T> Sender<T> {
+    /// Sets the new new value of the stream and notifies the consumer if any
+    ///
+    /// This function will store the `value` provided as the current value for
+    /// this channel, replacing any previous value that may have been there. If
+    /// the receiver may still be able to receive this message, then `Ok` is
+    /// returned with the previous value that was in this channel.
+    ///
+    /// If `Ok(Some)` is returned then this value overwrote a previous value,
+    /// and the value was never received by the receiver. If `Ok(None)` is
+    /// returned, then no previous value was found and the `value` is queued up
+    /// to be received by the receiver.
+    ///
+    /// # Errors
+    ///
+    /// This function will return an `Err` if the receiver has gone away and
+    /// it's impossible to send this value to the receiver. The error returned
+    /// retains ownership of the `value` provided and can be extracted, if
+    /// necessary.
+    pub fn swap(&self, value: T) -> Result<Option<T>, SendError<T>> {
+        let result;
+        // Do this step first so that the lock is dropped when
+        // `unpark` is called
+        let task = {
+            if let Some(ref lock) = self.inner.upgrade() {
+                let mut inner = lock.lock().unwrap();
+                result = inner.value.take();
+                inner.value = Some(value);
+                inner.task.take()
+            } else {
+                return Err(SendError(value));
+            }
+        };
+        if let Some(task) = task {
+            task.notify();
+        }
+        return Ok(result);
+    }
+}
+
+impl<T> Sink for Sender<T> {
+    type SinkItem = T;
+    type SinkError = SendError<T>;
+    fn start_send(&mut self, item: T) -> StartSend<T, SendError<T>> {
+        self.swap(item)?;
+        Ok(AsyncSink::Ready)
+    }
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(Async::Ready(()))
+    }
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        // Do this step first so that the lock is dropped *and*
+        // weakref is dropped when `unpark` is called
+        let task = {
+            let weak = mem::replace(&mut self.inner, Weak::new());
+            if let Some(ref lock) = weak.upgrade() {
+                drop(weak);
+                let mut inner = lock.lock().unwrap();
+                inner.task.take()
+            } else {
+                None
+            }
+        };
+        // notify on any drop of a sender, so eventually receiver wakes up
+        // when there are no senders and closes the stream
+        if let Some(task) = task {
+            task.notify();
+        }
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        self.close().ok();
+    }
+}
+
+impl<T> Stream for Receiver<T> {
+    type Item = T;
+    type Error = ();  // actually void
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        let result = {
+            let mut inner = self.inner.lock().unwrap();
+            if inner.value.is_none() {
+                if Arc::weak_count(&self.inner) == 0 {
+                    // no senders, terminate the stream
+                    return Ok(Async::Ready(None));
+                } else {
+                    inner.task = Some(task::current());
+                }
+            }
+            inner.value.take()
+        };
+        match result {
+            Some(value) => Ok(Async::Ready(Some(value))),
+            None => Ok(Async::NotReady),
+        }
+    }
+}
+
+/// Creates an in-memory Stream which only preserves last value
+///
+/// This method is somewhat similar to `channel(1)` but instead of preserving
+/// first value sent (and erroring on sender side) it replaces value if
+/// consumer is not fast enough and preserves last values sent on any
+/// poll of a stream.
+///
+/// # Example
+///
+/// ```
+/// use std::thread;
+/// use futures::prelude::*;
+/// use futures::stream::iter_ok;
+/// use futures::sync::slot;
+///
+/// let (tx, rx) = slot::channel::<i32>();
+///
+/// tx.send_all(iter_ok(vec![1, 2, 3])).wait();
+///
+/// let received = rx.collect().wait().unwrap();
+/// assert_eq!(received, vec![3]);
+/// ```
+pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
+    let inner = Arc::new(Mutex::new(Inner {
+        value: None,
+        task: None,
+    }));
+    return (Sender { inner: Arc::downgrade(&inner) },
+            Receiver { inner: inner });
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Sender<T> {
+        Sender { inner: self.inner.clone() }
+    }
+}
diff --git a/src/task_impl/std/mod.rs b/src/task_impl/std/mod.rs
index 108c4e57..af9fee10 100644
--- a/src/task_impl/std/mod.rs
+++ b/src/task_impl/std/mod.rs
@@ -5,8 +5,9 @@ use std::fmt;
 use std::marker::PhantomData;
 use std::mem;
 use std::ptr;
-use std::sync::{Arc, Mutex, Condvar, Once, ONCE_INIT};
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Once, ONCE_INIT};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread;
 
 use {Future, Stream, Sink, Poll, Async, StartSend, AsyncSink};
 use super::core;
@@ -237,15 +238,14 @@ impl<F: Future> Spawn<F> {
     /// to complete. When a future cannot make progress it will use
     /// `thread::park` to block the current thread.
     pub fn wait_future(&mut self) -> Result<F::Item, F::Error> {
-        ThreadNotify::with_current(|notify| {
+        let unpark = Arc::new(ThreadNotify::new(thread::current()));
 
-            loop {
-                match self.poll_future_notify(notify, 0)? {
-                    Async::NotReady => notify.park(),
-                    Async::Ready(e) => return Ok(e),
-                }
+        loop {
+            match self.poll_future_notify(&unpark, 0)? {
+                Async::NotReady => unpark.park(),
+                Async::Ready(e) => return Ok(e),
             }
-        })
+        }
     }
 
     /// A specialized function to request running a future to completion on the
@@ -296,17 +296,15 @@ impl<S: Stream> Spawn<S> {
     /// Like `wait_future`, except only waits for the next element to arrive on
     /// the underlying stream.
     pub fn wait_stream(&mut self) -> Option<Result<S::Item, S::Error>> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                match self.poll_stream_notify(notify, 0) {
-                    Ok(Async::NotReady) => notify.park(),
-                    Ok(Async::Ready(Some(e))) => return Some(Ok(e)),
-                    Ok(Async::Ready(None)) => return None,
-                    Err(e) => return Some(Err(e)),
-                }
+        let unpark = Arc::new(ThreadNotify::new(thread::current()));
+        loop {
+            match self.poll_stream_notify(&unpark, 0) {
+                Ok(Async::NotReady) => unpark.park(),
+                Ok(Async::Ready(Some(e))) => return Some(Ok(e)),
+                Ok(Async::Ready(None)) => return None,
+                Err(e) => return Some(Err(e)),
             }
-        })
+        }
     }
 }
 
@@ -342,16 +340,14 @@ impl<S: Sink> Spawn<S> {
     /// be blocked until it's able to send the value.
     pub fn wait_send(&mut self, mut value: S::SinkItem)
                      -> Result<(), S::SinkError> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                value = match self.start_send_notify(value, notify, 0)? {
-                    AsyncSink::NotReady(v) => v,
-                    AsyncSink::Ready => return Ok(()),
-                };
-                notify.park();
-            }
-        })
+        let notify = Arc::new(ThreadNotify::new(thread::current()));
+        loop {
+            value = match self.start_send_notify(value, &notify, 0)? {
+                AsyncSink::NotReady(v) => v,
+                AsyncSink::Ready => return Ok(()),
+            };
+            notify.park();
+        }
     }
 
     /// Blocks the current thread until it's able to flush this sink.
@@ -363,15 +359,13 @@ impl<S: Sink> Spawn<S> {
     /// The thread will be blocked until `poll_complete` returns that it's
     /// ready.
     pub fn wait_flush(&mut self) -> Result<(), S::SinkError> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                if self.poll_flush_notify(notify, 0)?.is_ready() {
-                    return Ok(())
-                }
-                notify.park();
+        let notify = Arc::new(ThreadNotify::new(thread::current()));
+        loop {
+            if self.poll_flush_notify(&notify, 0)?.is_ready() {
+                return Ok(())
             }
-        })
+            notify.park();
+        }
     }
 
     /// Blocks the current thread until it's able to close this sink.
@@ -380,15 +374,13 @@ impl<S: Sink> Spawn<S> {
     /// is not ready to be close yet, then the current thread will be blocked
     /// until it's closed.
     pub fn wait_close(&mut self) -> Result<(), S::SinkError> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                if self.close_notify(notify, 0)?.is_ready() {
-                    return Ok(())
-                }
-                notify.park();
+        let notify = Arc::new(ThreadNotify::new(thread::current()));
+        loop {
+            if self.close_notify(&notify, 0)?.is_ready() {
+                return Ok(())
             }
-        })
+            notify.park();
+        }
     }
 }
 
@@ -485,88 +477,29 @@ impl Unpark for RunInner {
 // ===== ThreadNotify =====
 
 struct ThreadNotify {
-    state: AtomicUsize,
-    mutex: Mutex<()>,
-    condvar: Condvar,
-}
-
-const IDLE: usize = 0;
-const NOTIFY: usize = 1;
-const SLEEP: usize = 2;
-
-thread_local! {
-    static CURRENT_THREAD_NOTIFY: Arc<ThreadNotify> = Arc::new(ThreadNotify {
-        state: AtomicUsize::new(IDLE),
-        mutex: Mutex::new(()),
-        condvar: Condvar::new(),
-    });
+    thread: thread::Thread,
+    ready: AtomicBool,
 }
 
 impl ThreadNotify {
-    fn with_current<F, R>(f: F) -> R
-        where F: FnOnce(&Arc<ThreadNotify>) -> R,
-    {
-        CURRENT_THREAD_NOTIFY.with(|notify| f(notify))
+    fn new(thread: thread::Thread) -> ThreadNotify {
+        ThreadNotify {
+            thread: thread,
+            ready: AtomicBool::new(false),
+        }
     }
 
     fn park(&self) {
-        // If currently notified, then we skip sleeping. This is checked outside
-        // of the lock to avoid acquiring a mutex if not necessary.
-        match self.state.compare_and_swap(NOTIFY, IDLE, Ordering::SeqCst) {
-            NOTIFY => return,
-            IDLE => {},
-            _ => unreachable!(),
-        }
-
-        // The state is currently idle, so obtain the lock and then try to
-        // transition to a sleeping state.
-        let mut m = self.mutex.lock().unwrap();
-
-        // Transition to sleeping
-        match self.state.compare_and_swap(IDLE, SLEEP, Ordering::SeqCst) {
-            NOTIFY => {
-                // Notified before we could sleep, consume the notification and
-                // exit
-                self.state.store(IDLE, Ordering::SeqCst);
-                return;
-            }
-            IDLE => {},
-            _ => unreachable!(),
-        }
-
-        // Loop until we've been notified
-        loop {
-            m = self.condvar.wait(m).unwrap();
-
-            // Transition back to idle, loop otherwise
-            if NOTIFY == self.state.compare_and_swap(NOTIFY, IDLE, Ordering::SeqCst) {
-                return;
-            }
+        if !self.ready.swap(false, Ordering::SeqCst) {
+            thread::park();
         }
     }
 }
 
 impl Notify for ThreadNotify {
     fn notify(&self, _unpark_id: usize) {
-        // First, try transitioning from IDLE -> NOTIFY, this does not require a
-        // lock.
-        match self.state.compare_and_swap(IDLE, NOTIFY, Ordering::SeqCst) {
-            IDLE | NOTIFY => return,
-            SLEEP => {}
-            _ => unreachable!(),
-        }
-
-        // The other half is sleeping, this requires a lock
-        let _m = self.mutex.lock().unwrap();
-
-        // Transition from SLEEP -> NOTIFY
-        match self.state.compare_and_swap(SLEEP, NOTIFY, Ordering::SeqCst) {
-            SLEEP => {}
-            _ => return,
-        }
-
-        // Wakeup the sleeper
-        self.condvar.notify_one();
+        self.ready.store(true, Ordering::SeqCst);
+        self.thread.unpark()
     }
 }
 
diff --git a/src/unsync/mod.rs b/src/unsync/mod.rs
index aaa5a707..407b9725 100644
--- a/src/unsync/mod.rs
+++ b/src/unsync/mod.rs
@@ -5,3 +5,4 @@
 
 pub mod mpsc;
 pub mod oneshot;
+pub mod slot;
diff --git a/src/unsync/slot.rs b/src/unsync/slot.rs
new file mode 100644
index 00000000..4dff1fb4
--- /dev/null
+++ b/src/unsync/slot.rs
@@ -0,0 +1,170 @@
+//! An unbounded channel that only stores last value sent
+
+use std::rc::{Rc, Weak};
+use std::cell::RefCell;
+
+use task::{self, Task};
+use {Sink, Stream, AsyncSink, Async, Poll, StartSend};
+
+/// Slot is very similar to unbounded channel but only stores last value sent
+///
+/// I.e. if you want to send some value between from producer to a consumer
+/// and if consumer is slow it should skip old values, the slot is
+/// a structure for the task.
+
+/// The transmission end of a channel which is used to send values
+///
+/// If the receiver is not fast enough only the last value is preserved and
+/// other ones are discarded.
+#[derive(Debug)]
+pub struct Sender<T> {
+    inner: Weak<RefCell<Inner<T>>>,
+}
+
+/// The receiving end of a channel which preserves only the last value
+#[derive(Debug)]
+pub struct Receiver<T> {
+    inner: Rc<RefCell<Inner<T>>>,
+}
+
+/// Error type for sending, used when the receiving end of a channel is
+/// dropped
+#[derive(Clone, PartialEq, Eq, Debug)]
+pub struct SendError<T>(T);
+
+#[derive(Debug)]
+struct Inner<T> {
+    value: Option<T>,
+    task: Option<Task>,
+}
+
+impl<T> Sender<T> {
+    /// Sets the new new value of the stream and notifies the consumer if any.
+    ///
+    /// This function will store the `value` provided as the current value for
+    /// this channel, replacing any previous value that may have been there. If
+    /// the receiver may still be able to receive this message, then `Ok` is
+    /// returned with the previous value that was in this channel.
+    ///
+    /// If `Ok(Some)` is returned then this value overwrote a previous value,
+    /// and the value was never received by the receiver. If `Ok(None)` is
+    /// returned, then no previous value was found and the `value` is queued up
+    /// to be received by the receiver.
+    ///
+    /// # Errors
+    ///
+    /// This function will return an `Err` if the receiver has gone away and
+    /// it's impossible to send this value to the receiver. The error returned
+    /// retains ownership of the `value` provided and can be extracted, if
+    /// necessary.
+    pub fn swap(&self, value: T) -> Result<Option<T>, SendError<T>> {
+        let result;
+        // Do this step first so that the cell is dropped when
+        // `unpark` is called
+        let task = {
+            if let Some(ref cell) = self.inner.upgrade() {
+                let mut inner = cell.borrow_mut();
+                result = inner.value.take();
+                inner.value = Some(value);
+                inner.task.take()
+            } else {
+                return Err(SendError(value));
+            }
+        };
+        if let Some(task) = task {
+            task.notify();
+        }
+        return Ok(result);
+    }
+}
+
+impl<T> Sink for Sender<T> {
+    type SinkItem = T;
+    type SinkError = SendError<T>;
+    fn start_send(&mut self, item: T) -> StartSend<T, SendError<T>> {
+        self.swap(item)?;
+        Ok(AsyncSink::Ready)
+    }
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(Async::Ready(()))
+    }
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        // Do this step first so that the cell is dropped *and*
+        // weakref is dropped when `unpark` is called
+        let task = self.inner.upgrade()
+            .and_then(|inner| inner.borrow_mut().task.take());
+        self.inner = Weak::new();
+        // notify on any drop of a sender, so eventually receiver wakes up
+        // when there are no senders and closes the stream
+        if let Some(task) = task {
+            task.notify();
+        }
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        self.close().ok();
+    }
+}
+
+impl<T> Stream for Receiver<T> {
+    type Item = T;
+    type Error = ();  // actually void
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        let result = {
+            let mut inner = self.inner.borrow_mut();
+            if inner.value.is_none() {
+                if Rc::weak_count(&self.inner) == 0 {
+                    // no senders, terminate the stream
+                    return Ok(Async::Ready(None));
+                } else {
+                    inner.task = Some(task::current());
+                }
+            }
+            inner.value.take()
+        };
+        match result {
+            Some(value) => Ok(Async::Ready(Some(value))),
+            None => Ok(Async::NotReady),
+        }
+    }
+}
+
+/// Creates an in-memory Stream which only preserves last value
+///
+/// This method is somewhat similar to `channel(1)` but instead of preserving
+/// first value sent (and erroring on sender side) it replaces value if
+/// consumer is not fast enough and preserves last values sent on any
+/// poll of a stream.
+///
+/// # Example
+///
+/// ```
+/// use std::thread;
+/// use futures::prelude::*;
+/// use futures::stream::iter_ok;
+/// use futures::unsync::slot;
+///
+/// let (tx, rx) = slot::channel::<i32>();
+///
+/// tx.send_all(iter_ok(vec![1, 2, 3])).wait();
+///
+/// let received = rx.collect().wait().unwrap();
+/// assert_eq!(received, vec![3]);
+/// ```
+pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
+    let inner = Rc::new(RefCell::new(Inner {
+        value: None,
+        task: None,
+    }));
+    return (Sender { inner: Rc::downgrade(&inner) },
+            Receiver { inner: inner });
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Sender<T> {
+        Sender { inner: self.inner.clone() }
+    }
+}


Test patch:
diff --git a/tests/futures_unordered.rs b/tests/futures_unordered.rs
index e29f3d96..9b8c08d0 100644
--- a/tests/futures_unordered.rs
+++ b/tests/futures_unordered.rs
@@ -83,3 +83,45 @@ fn finished_future_ok() {
     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());
 }
+
+#[test]
+fn iter_mut_cancel() {
+    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 mut stream = futures_unordered(vec![a_rx, b_rx, c_rx]);
+
+    for rx in stream.iter_mut() {
+        rx.close();
+    }
+
+    assert!(a_tx.is_canceled());
+    assert!(b_tx.is_canceled());
+    assert!(c_tx.is_canceled());
+
+    let mut spawn = futures::executor::spawn(stream);
+    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), spawn.wait_stream());
+    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), spawn.wait_stream());
+    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), spawn.wait_stream());
+    assert_eq!(None, spawn.wait_stream());
+}
+
+#[test]
+fn iter_mut_len() {
+    let mut stream = futures_unordered(vec![
+        futures::future::empty::<(),()>(),
+        futures::future::empty::<(),()>(),
+        futures::future::empty::<(),()>()
+    ]);
+
+    let mut iter_mut = stream.iter_mut();
+    assert_eq!(iter_mut.len(), 3);
+    assert!(iter_mut.next().is_some());
+    assert_eq!(iter_mut.len(), 2);
+    assert!(iter_mut.next().is_some());
+    assert_eq!(iter_mut.len(), 1);
+    assert!(iter_mut.next().is_some());
+    assert_eq!(iter_mut.len(), 0);
+    assert!(iter_mut.next().is_none());
+}
diff --git a/tests/slot-unsync.rs b/tests/slot-unsync.rs
new file mode 100644
index 00000000..d216a6f0
--- /dev/null
+++ b/tests/slot-unsync.rs
@@ -0,0 +1,107 @@
+#![cfg(feature = "use_std")]
+
+extern crate futures;
+
+use futures::prelude::*;
+use futures::future::{lazy, ok};
+use futures::stream::unfold;
+use futures::unsync::slot;
+use futures::unsync::mpsc;
+
+mod support;
+use support::*;
+
+#[test]
+fn send_recv() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    tx.send(1).wait().unwrap();
+
+    assert_eq!(rx.next().unwrap(), Ok(1));
+}
+
+#[test]
+fn swap() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    assert_eq!(tx.swap(1), Ok(None));
+    assert_eq!(tx.swap(2), Ok(Some(1)));
+    assert_eq!(rx.next().unwrap(), Ok(2));
+    assert_eq!(tx.swap(3), Ok(None));
+    assert_eq!(rx.next().unwrap(), Ok(3));
+}
+
+#[test]
+fn send_recv_no_buffer() {
+    let (mut tx, mut rx) = slot::channel::<i32>();
+
+    // Run on a task context
+    lazy(move || {
+        assert!(tx.poll_complete().unwrap().is_ready());
+
+        // Send first message
+
+        let res = tx.start_send(1).unwrap();
+        assert!(is_ready(&res));
+
+        // Send second message
+        let res = tx.start_send(2).unwrap();
+        assert!(is_ready(&res));
+
+        // Take the value
+        assert_eq!(rx.poll().unwrap(), Async::Ready(Some(2)));
+
+        let res = tx.start_send(3).unwrap();
+        assert!(is_ready(&res));
+
+        // Take the value
+        assert_eq!(rx.poll().unwrap(), Async::Ready(Some(3)));
+
+        Ok::<(), ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn send_shared_recv() {
+    let (tx1, rx) = slot::channel::<i32>();
+    let tx2 = tx1.clone();
+    let mut rx = rx.wait();
+
+    tx1.send(1).wait().unwrap();
+    assert_eq!(rx.next().unwrap(), Ok(1));
+
+    tx2.send(2).wait().unwrap();
+    assert_eq!(rx.next().unwrap(), Ok(2));
+}
+
+#[test]
+fn tx_close_gets_none() {
+    let (_, mut rx) = slot::channel::<i32>();
+
+    // Run on a task context
+    lazy(move || {
+        assert_eq!(rx.poll(), Ok(Async::Ready(None)));
+        assert_eq!(rx.poll(), Ok(Async::Ready(None)));
+
+        Ok::<(), ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn spawn_sends_items() {
+    let core = local_executor::Core::new();
+    let stream = unfold(0, |i| Some(ok::<_,u8>((i, i + 1))));
+    let rx = mpsc::spawn(stream, &core, 1);
+    assert_eq!(core.run(rx.take(4).collect()).unwrap(),
+               [0, 1, 2, 3]);
+}
+
+fn is_ready<T>(res: &AsyncSink<T>) -> bool {
+    match *res {
+        AsyncSink::Ready => true,
+        _ => false,
+    }
+}
+
diff --git a/tests/slot.rs b/tests/slot.rs
new file mode 100644
index 00000000..eaf10a0a
--- /dev/null
+++ b/tests/slot.rs
@@ -0,0 +1,199 @@
+#![cfg(feature = "use_std")]
+
+extern crate futures;
+
+use futures::prelude::*;
+use futures::future::{lazy, ok};
+use futures::stream::unfold;
+use futures::sync::slot;
+use futures::sync::mpsc;
+
+use std::time::Duration;
+use std::thread;
+
+mod support;
+use support::*;
+
+
+trait AssertSend: Send {}
+impl AssertSend for slot::Sender<i32> {}
+impl AssertSend for slot::Receiver<i32> {}
+
+#[test]
+fn send_recv() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    tx.send(1).wait().unwrap();
+
+    assert_eq!(rx.next().unwrap(), Ok(1));
+}
+
+#[test]
+fn swap() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    assert_eq!(tx.swap(1), Ok(None));
+    assert_eq!(tx.swap(2), Ok(Some(1)));
+    assert_eq!(rx.next().unwrap(), Ok(2));
+    assert_eq!(tx.swap(3), Ok(None));
+    assert_eq!(rx.next().unwrap(), Ok(3));
+}
+
+#[test]
+fn send_recv_no_buffer() {
+    let (mut tx, mut rx) = slot::channel::<i32>();
+
+    // Run on a task context
+    lazy(move || {
+        assert!(tx.poll_complete().unwrap().is_ready());
+
+        // Send first message
+
+        let res = tx.start_send(1).unwrap();
+        assert!(is_ready(&res));
+
+        // Send second message
+        let res = tx.start_send(2).unwrap();
+        assert!(is_ready(&res));
+
+        // Take the value
+        assert_eq!(rx.poll().unwrap(), Async::Ready(Some(2)));
+
+        let res = tx.start_send(3).unwrap();
+        assert!(is_ready(&res));
+
+        // Take the value
+        assert_eq!(rx.poll().unwrap(), Async::Ready(Some(3)));
+
+        Ok::<(), ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn send_shared_recv() {
+    let (tx1, rx) = slot::channel::<i32>();
+    let tx2 = tx1.clone();
+    let mut rx = rx.wait();
+
+    tx1.send(1).wait().unwrap();
+    assert_eq!(rx.next().unwrap(), Ok(1));
+
+    tx2.send(2).wait().unwrap();
+    assert_eq!(rx.next().unwrap(), Ok(2));
+}
+
+#[test]
+fn send_recv_threads() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    thread::spawn(move|| {
+        tx.send(1).wait().unwrap();
+    });
+
+    assert_eq!(rx.next().unwrap(), Ok(1));
+}
+
+#[test]
+fn send_recv_threads_no_capacity() {
+    let (mut tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    let t = thread::spawn(move|| {
+        tx = tx.send(1).wait().unwrap();
+        tx = tx.send(2).wait().unwrap();
+    });
+
+    thread::sleep(Duration::from_millis(100));
+    assert_eq!(rx.next().unwrap(), Ok(2));
+
+    t.join().unwrap();
+}
+
+#[test]
+fn tx_close_gets_none() {
+    let (_, mut rx) = slot::channel::<i32>();
+
+    // Run on a task context
+    lazy(move || {
+        assert_eq!(rx.poll(), Ok(Async::Ready(None)));
+        assert_eq!(rx.poll(), Ok(Async::Ready(None)));
+
+        Ok::<(), ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn spawn_sends_items() {
+    let core = local_executor::Core::new();
+    let stream = unfold(0, |i| Some(ok::<_,u8>((i, i + 1))));
+    let rx = mpsc::spawn(stream, &core, 1);
+    assert_eq!(core.run(rx.take(4).collect()).unwrap(),
+               [0, 1, 2, 3]);
+}
+
+#[test]
+fn stress_shared_bounded_hard() {
+    const AMT: u32 = 10000;
+    const NTHREADS: u32 = 8;
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    let t = thread::spawn(move|| {
+        for _ in 0..AMT * NTHREADS {
+            if let Some(value) = rx.next() {
+                assert_eq!(value, Ok(1));
+            } else {
+                // less items is okay, but it must be end of stream
+                break;
+            }
+        }
+
+        if rx.next().is_some() {
+            panic!();
+        }
+    });
+
+    for _ in 0..NTHREADS {
+        let mut tx = tx.clone();
+
+        thread::spawn(move|| {
+            for _ in 0..AMT {
+                tx = tx.send(1).wait().unwrap();
+            }
+        });
+    }
+
+    drop(tx);
+
+    t.join().ok().unwrap();
+}
+
+/// Stress test that receiver properly receives last message
+/// after sender dropped.
+#[test]
+fn stress_drop_sender() {
+    fn list() -> Box<Stream<Item=i32, Error=u32>> {
+        let (tx, rx) = slot::channel();
+        tx.send(Ok(1))
+          .and_then(|tx| tx.send(Ok(2)))
+          .and_then(|tx| tx.send(Ok(3)))
+          .forget();
+        Box::new(rx.then(|r| r.unwrap()))
+    }
+
+    for _ in 0..10000 {
+        assert_eq!(
+            list().wait().collect::<Result<Vec<_>, _>>().unwrap().last(),
+            Some(&3));
+    }
+}
+
+fn is_ready<T>(res: &AsyncSink<T>) -> bool {
+    match *res {
+        AsyncSink::Ready => true,
+        _ => false,
+    }
+}