Merge branch 'reunite' of https://github.com/spinda/futures-rs

Author: alexcrichton

.travis.yml

@@ -12,7 +12,7 @@ matrix:
       after_success:
         - travis-cargo doc-upload
     - os: linux
-      rust: 1.10.0
+      rust: 1.17.0
       script: cargo test
 sudo: false
 script:

src/stream/split.rs

@@ -1,10 +1,21 @@
 use {StartSend, Sink, Stream, Poll, Async, AsyncSink};
+use std::error::Error;
+use std::fmt;
 use sync::BiLock;
 
 /// A `Stream` part of the split pair
 #[derive(Debug)]
 pub struct SplitStream<S>(BiLock<S>);
 
+impl<S> SplitStream<S> {
+    /// Attempts to put the two "halves" of a split `Stream + Sink` back
+    /// together. Succeeds only if the `SplitStream<S>` and `SplitSink<S>` are
+    /// a matching pair originating from the same call to `Stream::split`.
+    pub fn reunite(self, other: SplitSink<S>) -> Result<S, ReuniteError<S>> {
+        other.reunite(self)
+    }
+}
+
 impl<S: Stream> Stream for SplitStream<S> {
     type Item = S::Item;
     type Error = S::Error;
@@ -21,6 +32,17 @@ impl<S: Stream> Stream for SplitStream<S> {
 #[derive(Debug)]
 pub struct SplitSink<S>(BiLock<S>);
 
+impl<S> SplitSink<S> {
+    /// Attempts to put the two "halves" of a split `Stream + Sink` back
+    /// together. Succeeds only if the `SplitStream<S>` and `SplitSink<S>` are
+    /// a matching pair originating from the same call to `Stream::split`.
+    pub fn reunite(self, other: SplitStream<S>) -> Result<S, ReuniteError<S>> {
+        self.0.reunite(other.0).map_err(|err| {
+            ReuniteError(SplitSink(err.0), SplitStream(err.1))
+        })
+    }
+}
+
 impl<S: Sink> Sink for SplitSink<S> {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -55,3 +77,27 @@ pub fn split<S: Stream + Sink>(s: S) -> (SplitSink<S>, SplitStream<S>) {
     let write = SplitSink(b);
     (write, read)
 }
+
+/// Error indicating a `SplitSink<S>` and `SplitStream<S>` were not two halves
+/// of a `Stream + Split`, and thus could not be `reunite`d.
+pub struct ReuniteError<T>(pub SplitSink<T>, pub SplitStream<T>);
+
+impl<T> fmt::Debug for ReuniteError<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_tuple("ReuniteError")
+            .field(&"...")
+            .finish()
+    }
+}
+
+impl<T> fmt::Display for ReuniteError<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "tried to reunite a SplitStream and SplitSink that don't form a pair")
+    }
+}
+
+impl<T> Error for ReuniteError<T> {
+    fn description(&self) -> &str {
+        "tried to reunite a SplitStream and SplitSink that don't form a pair"
+    }
+}

src/sync/bilock.rs

@@ -1,5 +1,7 @@
 use std::boxed::Box;
 use std::cell::UnsafeCell;
+use std::error::Error;
+use std::fmt;
 use std::mem;
 use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
@@ -35,7 +37,7 @@ pub struct BiLock<T> {
 #[derive(Debug)]
 struct Inner<T> {
     state: AtomicUsize,
-    inner: UnsafeCell<T>,
+    inner: Option<UnsafeCell<T>>,
 }
 
 unsafe impl<T: Send> Send for Inner<T> {}
@@ -50,7 +52,7 @@ impl<T> BiLock<T> {
     pub fn new(t: T) -> (BiLock<T>, BiLock<T>) {
         let inner = Arc::new(Inner {
             state: AtomicUsize::new(0),
-            inner: UnsafeCell::new(t),
+            inner: Some(UnsafeCell::new(t)),
         });
 
         (BiLock { inner: inner.clone() }, BiLock { inner: inner })
@@ -131,6 +133,21 @@ impl<T> BiLock<T> {
         }
     }
 
+    /// Attempts to put the two "halves" of a `BiLock<T>` back together and
+    /// recover the original value. Succeeds only if the two `BiLock<T>`s
+    /// originated from the same call to `BiLock::new`.
+    pub fn reunite(self, other: Self) -> Result<T, ReuniteError<T>> {
+        if Arc::ptr_eq(&self.inner, &other.inner) {
+            drop(other);
+            let inner = Arc::try_unwrap(self.inner)
+                .ok()
+                .expect("futures: try_unwrap failed in BiLock<T>::reunite");
+            Ok(unsafe { inner.into_inner() })
+        } else {
+            Err(ReuniteError(self, other))
+        }
+    }
+
     fn unlock(&self) {
         match self.inner.state.swap(0, SeqCst) {
             // we've locked the lock, shouldn't be possible for us to see an
@@ -149,12 +166,42 @@ impl<T> BiLock<T> {
     }
 }
 
+impl<T> Inner<T> {
+    unsafe fn into_inner(mut self) -> T {
+        mem::replace(&mut self.inner, None).unwrap().into_inner()
+    }
+}
+
 impl<T> Drop for Inner<T> {
     fn drop(&mut self) {
         assert_eq!(self.state.load(SeqCst), 0);
     }
 }
 
+/// Error indicating two `BiLock<T>`s were not two halves of a whole, and
+/// thus could not be `reunite`d.
+pub struct ReuniteError<T>(pub BiLock<T>, pub BiLock<T>);
+
+impl<T> fmt::Debug for ReuniteError<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_tuple("ReuniteError")
+            .field(&"...")
+            .finish()
+    }
+}
+
+impl<T> fmt::Display for ReuniteError<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "tried to reunite two BiLocks that don't form a pair")
+    }
+}
+
+impl<T> Error for ReuniteError<T> {
+    fn description(&self) -> &str {
+        "tried to reunite two BiLocks that don't form a pair"
+    }
+}
+
 /// Returned RAII guard from the `poll_lock` method.
 ///
 /// This structure acts as a sentinel to the data in the `BiLock<T>` itself,
@@ -168,13 +215,13 @@ pub struct BiLockGuard<'a, T: 'a> {
 impl<'a, T> Deref for BiLockGuard<'a, T> {
     type Target = T;
     fn deref(&self) -> &T {
-        unsafe { &*self.inner.inner.inner.get() }
+        unsafe { &*self.inner.inner.inner.as_ref().unwrap().get() }
     }
 }
 
 impl<'a, T> DerefMut for BiLockGuard<'a, T> {
     fn deref_mut(&mut self) -> &mut T {
-        unsafe { &mut *self.inner.inner.inner.get() }
+        unsafe { &mut *self.inner.inner.inner.as_ref().unwrap().get() }
     }
 }
 
@@ -231,13 +278,13 @@ impl<T> BiLockAcquired<T> {
 impl<T> Deref for BiLockAcquired<T> {
     type Target = T;
     fn deref(&self) -> &T {
-        unsafe { &*self.inner.as_ref().unwrap().inner.inner.get() }
+        unsafe { &*self.inner.as_ref().unwrap().inner.inner.as_ref().unwrap().get() }
     }
 }
 
 impl<T> DerefMut for BiLockAcquired<T> {
     fn deref_mut(&mut self) -> &mut T {
-        unsafe { &mut *self.inner.as_mut().unwrap().inner.inner.get() }
+        unsafe { &mut *self.inner.as_mut().unwrap().inner.inner.as_ref().unwrap().get() }
     }
 }
 

tests/bilock.rs

@@ -15,25 +15,31 @@ use support::*;
 fn smoke() {
     let future = future::lazy(|| {
         let (a, b) = BiLock::new(1);
-        let mut lock = match a.poll_lock() {
-            Async::Ready(l) => l,
-            Async::NotReady => panic!("poll not ready"),
-        };
-        assert_eq!(*lock, 1);
-        *lock = 2;
 
-        assert!(b.poll_lock().is_not_ready());
-        assert!(a.poll_lock().is_not_ready());
-        drop(lock);
+        {
+            let mut lock = match a.poll_lock() {
+                Async::Ready(l) => l,
+                Async::NotReady => panic!("poll not ready"),
+            };
+            assert_eq!(*lock, 1);
+            *lock = 2;
+
+            assert!(b.poll_lock().is_not_ready());
+            assert!(a.poll_lock().is_not_ready());
+        }
 
         assert!(b.poll_lock().is_ready());
         assert!(a.poll_lock().is_ready());
 
-        let lock = match b.poll_lock() {
-            Async::Ready(l) => l,
-            Async::NotReady => panic!("poll not ready"),
-        };
-        assert_eq!(*lock, 2);
+        {
+            let lock = match b.poll_lock() {
+                Async::Ready(l) => l,
+                Async::NotReady => panic!("poll not ready"),
+            };
+            assert_eq!(*lock, 2);
+        }
+
+        assert_eq!(a.reunite(b).expect("bilock/smoke: reunite error"), 2);
 
         Ok::<(), ()>(())
     });
@@ -73,6 +79,8 @@ fn concurrent() {
         Async::NotReady => panic!("poll not ready"),
     }
 
+    assert_eq!(a.reunite(b).expect("bilock/concurrent: reunite error"), 2 * N);
+
     struct Increment {
         remaining: usize,
         a: Option<BiLock<usize>>,

tests/split.rs

@@ -39,6 +39,8 @@ fn test_split() {
     {
         let j = Join(iter(vec![Ok(10), Ok(20), Ok(30)]), &mut dest);
         let (sink, stream) = j.split();
+        let j = sink.reunite(stream).expect("test_split: reunite error");
+        let (sink, stream) = j.split();
         sink.send_all(stream).wait().unwrap();
     }
     assert_eq!(dest, vec![10, 20, 30]);