Repository: alexcrichton/futures-rs
Pull request #570 (2017-09-04 23:01:36)

Implements `sync::mpsc::tip` abstraction
This is a stream that only remembers last value and also receives
updates.

The common use case for such thing is when you have a coroutine that
refreshes a value (e.g. polls DNS server for an address), and a coroutine
that uses that value in possibly blocking fashion (e.g. fetches some
data from the address). Using unbounded stream for this case means we
can have a memory leak when a consumer is too slow. And using `channel(1)`
preserves the first value while what we really need is the most recent
value generated by a producer.

The name "tip" is in the sense of end or peak, not in the sense of gratitude :) But suggestions of the better name are appreciated.

What do you think?

hawkw: > But suggestions of the better name are appreciated.

How about `Single`?
davidblewett: This is conceptually similar to log compaction in Kafka ( https://kafka.apache.org/documentation/#compaction ), so maybe look there for names?
tailhook: Okay, next take:
1. Named it `sync::slot::channel`. (Maybe `Head` from @davidblewett's link?)
2. `swap` instead of `set` with both ways of value recovery
3. Eliminated race-condition in `Drop`, but checking in receiver (so waking up receiver on each sender drop, I think it shouldn't be a problem)
4. `close()` now works by dropping Weak ref (we need to drop it in destructor anyway to avoid race conditions)

Overall looks like all suggestions are fixed. Also rebased on top of the new master.


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/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..f0e2813e
--- /dev/null
+++ b/src/sync/slot.rs
@@ -0,0 +1,170 @@
+//! An unbounded channel that only stores last value sent
+
+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: Option<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
+    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 = {
+            let strong = self.inner.as_ref()
+                .expect("sending to a closed slot");
+            if let Some(ref lock) = strong.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 = {
+            if let Some(weak) = self.inner.take() {
+                if let Some(ref lock) = weak.upgrade() {
+                    drop(weak);
+                    let mut inner = lock.lock().unwrap();
+                    inner.task.take()
+                } else {
+                    None
+                }
+            } 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: Some(Arc::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/slot.rs b/tests/slot.rs
new file mode 100644
index 00000000..6951d299
--- /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 receivesl 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,
+    }
+}