Repository: alexcrichton/futures-rs
Pull request #410 (2017-03-10 06:55:09)

typo fix


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 58f70d02..5fcc93fa 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -19,6 +19,7 @@ travis-ci = { repository = "alexcrichton/futures-rs" }
 appveyor = { repository = "alexcrichton/futures-rs" }
 
 [dependencies]
+log = { version = "0.3", default-features = false }
 
 [features]
 use_std = []
diff --git a/futures-cpupool/src/lib.rs b/futures-cpupool/src/lib.rs
index 946b3770..d4a803d8 100644
--- a/futures-cpupool/src/lib.rs
+++ b/futures-cpupool/src/lib.rs
@@ -283,10 +283,7 @@ impl<F: Future> Future for MySender<F, Result<F::Item, F::Error>> {
             Ok(Async::NotReady) => return Ok(Async::NotReady),
             Err(e) => Err(e),
         };
-
-        // if the receiving end has gone away then that's ok, we just ignore the
-        // send error here.
-        drop(self.tx.take().unwrap().send(res));
+        self.tx.take().unwrap().complete(res);
         Ok(Async::Ready(()))
     }
 }
diff --git a/src/future/select2.rs b/src/future/select2.rs
index e16e2e03..e18cf81a 100644
--- a/src/future/select2.rs
+++ b/src/future/select2.rs
@@ -6,7 +6,6 @@ use future::Either;
 ///
 /// This is created by the `Future::merge` method.
 #[must_use = "futures do nothing unless polled"]
-#[derive(Debug)]
 pub struct Select2<A, B> {
     inner: Option<(A, B)>,
 }
diff --git a/src/lib.rs b/src/lib.rs
index ba78fd3b..083931d2 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -155,13 +155,16 @@
 //! [README]: https://github.com/alexcrichton/futures-rs#futures-rs
 
 #![no_std]
-#![deny(missing_docs, missing_debug_implementations)]
+#![deny(missing_docs)]
 #![doc(html_root_url = "https://docs.rs/futures/0.1")]
 
 #[macro_use]
 #[cfg(feature = "use_std")]
 extern crate std;
 
+#[macro_use]
+extern crate log;
+
 macro_rules! if_std {
     ($($i:item)*) => ($(
         #[cfg(feature = "use_std")]
diff --git a/src/poll.rs b/src/poll.rs
index 75d9426f..56364494 100644
--- a/src/poll.rs
+++ b/src/poll.rs
@@ -64,7 +64,7 @@ impl<T> From<T> for Async<T> {
 #[derive(Copy, Clone, Debug, PartialEq)]
 pub enum AsyncSink<T> {
     /// The `start_send` attempt succeeded, so the sending process has
-    /// *started*; you muse use `Sink::poll_complete` to drive the send
+    /// *started*; you must use `Sink::poll_complete` to drive the send
     /// to completion.
     Ready,
 
diff --git a/src/sink/mod.rs b/src/sink/mod.rs
index b0c82448..048604c1 100644
--- a/src/sink/mod.rs
+++ b/src/sink/mod.rs
@@ -22,10 +22,8 @@ mod map_err;
 
 if_std! {
     mod buffer;
-    mod wait;
 
     pub use self::buffer::Buffer;
-    pub use self::wait::Wait;
 
     // TODO: consider expanding this via e.g. FromIterator
     impl<T> Sink for ::std::vec::Vec<T> {
@@ -148,24 +146,19 @@ pub trait Sink {
     fn start_send(&mut self, item: Self::SinkItem)
                   -> StartSend<Self::SinkItem, Self::SinkError>;
 
-    /// Flush all output from this sink, if necessary.
+    /// Make progress on all pending requests, and determine whether they have
+    /// completed.
     ///
-    /// Some sinks may buffer intermediate data as an optimization to improve
-    /// throughput. In other words, if a sink has a corresponding receiver then
-    /// a successful `start_send` above may not guarantee that the value is
-    /// actually ready to be received by the receiver. This function is intended
-    /// to be used to ensure that values do indeed make their way to the
-    /// receiver.
-    ///
-    /// This function will attempt to process any pending requests on behalf of
-    /// the sink and drive it to completion. When `Ready`
+    /// Since sinks are asynchronous, no single method completes all of their
+    /// work in one shot. Instead, you use `poll_complete` to repeatedly drive
+    /// the sink to make progress on requests (such as `start_send`). As with
+    /// `Future::poll`, if the pending requests are not able to complete during
+    /// this call, the current task is automatically scheduled to be woken up
+    /// again once more progress is possible.
     ///
     /// # Return value
     ///
-    /// Returns `Ok(Async::Ready(()))` when no buffered items remain. If this
-    /// value is returned then it is guaranteed that all previous values sent
-    /// via `start_send` will be guaranteed to available to a listening
-    /// receiver.
+    /// Returns `Ok(Async::Ready(()))` when no unprocessed requests remain.
     ///
     /// Returns `Ok(Async::NotReady)` if there is more work left to do, in which
     /// case the current task is scheduled to wake up when more progress may be
@@ -185,32 +178,8 @@ pub trait Sink {
     ///
     /// - It is called outside of the context of a task.
     /// - A previous call to `start_send` or `poll_complete` yielded an error.
-    ///
-    /// # Compatibility nodes
-    ///
-    /// The name of this method may be slightly misleading as the original
-    /// intention was to have this method be more general than just flushing
-    /// requests. Over time though it was decided to trim back the ambitions of
-    /// this method to what it's always done, just flushing.
-    ///
-    /// In the 0.2 release series of futures this method will be renamed to
-    /// `poll_flush`. For 0.1, however, the breaking change is not happening
-    /// yet.
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError>;
 
-    /// Creates a new object which will produce a synchronous sink.
-    ///
-    /// The sink returned does **not** implement the `Sink` trait, and instead
-    /// only has two methods: `send` and `flush`. These two methods correspond
-    /// to `start_send` and `poll_complete` above except are executed in a
-    /// blocking fashion.
-    #[cfg(feature = "use_std")]
-    fn wait(self) -> Wait<Self>
-        where Self: Sized
-    {
-        wait::new(self)
-    }
-
     /// Composes a function *in front of* the sink.
     ///
     /// This adapter produces a new sink that passes each value through the
diff --git a/src/sink/wait.rs b/src/sink/wait.rs
deleted file mode 100644
index ae4498c5..00000000
--- a/src/sink/wait.rs
+++ /dev/null
@@ -1,50 +0,0 @@
-use sink::Sink;
-use executor;
-
-/// A sink combinator which converts an asynchronous sink to a **blocking
-/// sink**.
-///
-/// Created by the `Sink::wait` method, this function transforms any sink into a
-/// blocking version. This is implemented by blocking the current thread when a
-/// sink is otherwise unable to make progress.
-#[must_use = "sinks do nothing unless used"]
-#[derive(Debug)]
-pub struct Wait<S> {
-    sink: executor::Spawn<S>,
-}
-
-pub fn new<S: Sink>(s: S) -> Wait<S> {
-    Wait {
-        sink: executor::spawn(s),
-    }
-}
-
-impl<S: Sink> Wait<S> {
-    /// Sends a value to this sink, blocking the current thread until it's able
-    /// to do so.
-    ///
-    /// This function will take the `value` provided and call the underlying
-    /// sink's `start_send` function until it's ready to accept the value. If
-    /// the function returns `NotReady` then the current thread is blocked
-    /// until it is otherwise ready to accept the value.
-    ///
-    /// # Return value
-    ///
-    /// If `Ok(())` is returned then the `value` provided was successfully sent
-    /// along the sink, and if `Err(e)` is returned then an error occurred
-    /// which prevented the value from being sent.
-    pub fn send(&mut self, value: S::SinkItem) -> Result<(), S::SinkError> {
-        self.sink.wait_send(value)
-    }
-
-    /// Flushes any buffered data in this sink, blocking the current thread
-    /// until it's entirely flushed.
-    ///
-    /// This function will call the underlying sink's `poll_complete` method
-    /// until it returns that it's ready to proceed. If the method returns
-    /// `NotReady` the current thread will be blocked until it's otherwise
-    /// ready to proceed.
-    pub fn flush(&mut self) -> Result<(), S::SinkError> {
-        self.sink.wait_flush()
-    }
-}
diff --git a/src/stream/concat.rs b/src/stream/concat.rs
deleted file mode 100644
index 7ee0af98..00000000
--- a/src/stream/concat.rs
+++ /dev/null
@@ -1,81 +0,0 @@
-use core::mem;
-
-use {Poll, Async};
-use future::Future;
-use stream::Stream;
-
-/// A stream combinator to concatenate the results of a stream into the first
-/// yielded item.
-///
-/// This structure is produced by the `Stream::concat` method.
-#[derive(Debug)]
-#[must_use = "streams do nothing unless polled"]
-pub struct Concat<S>
-    where S: Stream,
-{
-    stream: S,
-    extend: Inner<S::Item>,
-}
-
-pub fn new<S>(s: S) -> Concat<S>
-    where S: Stream,
-          S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator,
-{
-    Concat {
-        stream: s,
-        extend: Inner::First,
-    }
-}
-
-impl<S> Future for Concat<S>
-    where S: Stream,
-          S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator,
-
-{
-    type Item = S::Item;
-    type Error = S::Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        loop {
-            match self.stream.poll() {
-                Ok(Async::Ready(Some(i))) => {
-                    match self.extend {
-                        Inner::First => {
-                            self.extend = Inner::Extending(i);
-                        },
-                        Inner::Extending(ref mut e) => {
-                            e.extend(i);
-                        },
-                        Inner::Done => unreachable!(),
-                    }
-                },
-                Ok(Async::Ready(None)) => return Ok(Async::Ready(expect(self.extend.take()))),
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
-                Err(e) => {
-                    self.extend.take();
-                    return Err(e)
-                }
-            }
-        }
-    }
-}
-
-#[derive(Debug)]
-enum Inner<E> {
-    First,
-    Extending(E),
-    Done,
-}
-
-impl<E> Inner<E> {
-    fn take(&mut self) -> Option<E> {
-        match mem::replace(self, Inner::Done) {
-            Inner::Extending(e) => Some(e),
-            _ => None,
-        }
-    }
-}
-
-fn expect<T>(opt: Option<T>) -> T {
-    opt.expect("cannot poll Concat again")
-}
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 7e15c44b..19dee75b 100755
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -27,7 +27,6 @@ pub use self::repeat::{repeat, Repeat};
 
 mod and_then;
 mod chain;
-mod concat;
 mod empty;
 mod filter;
 mod filter_map;
@@ -53,7 +52,6 @@ mod zip;
 mod forward;
 pub use self::and_then::AndThen;
 pub use self::chain::Chain;
-pub use self::concat::Concat;
 pub use self::empty::{Empty, empty};
 pub use self::filter::Filter;
 pub use self::filter_map::FilterMap;
@@ -534,40 +532,6 @@ pub trait Stream {
         collect::new(self)
     }
 
-    /// Concatenate all results of a stream into a single extendable
-    /// destination, returning a future representing the end result.
-    ///
-    /// This combinator will extend the first item with the contents
-    /// of all the successful results of the stream. If an error
-    /// occurs, all the results will be dropped and the error will be
-    /// returned.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use std::thread;
-    ///
-    /// use futures::{Future, Sink, Stream};
-    /// use futures::sync::mpsc;
-    ///
-    /// let (mut tx, rx) = mpsc::channel(1);
-    ///
-    /// thread::spawn(move || {
-    ///     for i in (0..3).rev() {
-    ///         let n = i * 3;
-    ///         tx = tx.send(vec![n + 1, n + 2, n + 3]).wait().unwrap();
-    ///     }
-    /// });
-    /// let result = rx.concat();
-    /// assert_eq!(result.wait(), Ok(vec![7, 8, 9, 4, 5, 6, 1, 2, 3]));
-    /// ```
-    fn concat(self) -> Concat<Self>
-        where Self: Sized,
-              Self::Item: Extend<<<Self as Stream>::Item as IntoIterator>::Item> + IntoIterator,
-    {
-        concat::new(self)
-    }
-
     /// Execute an accumulating computation over a stream, collecting all the
     /// values into one final result.
     ///
diff --git a/src/stream/wait.rs b/src/stream/wait.rs
index 8f5ef11b..e3aba8b8 100644
--- a/src/stream/wait.rs
+++ b/src/stream/wait.rs
@@ -8,7 +8,6 @@ use executor;
 /// into a standard iterator. This is implemented by blocking the current thread
 /// while items on the underlying stream aren't ready yet.
 #[must_use = "iterators do nothing unless advanced"]
-#[derive(Debug)]
 pub struct Wait<S> {
     stream: executor::Spawn<S>,
 }
diff --git a/src/sync/oneshot.rs b/src/sync/oneshot.rs
index 2b44aa59..d744ee36 100644
--- a/src/sync/oneshot.rs
+++ b/src/sync/oneshot.rs
@@ -91,7 +91,7 @@ struct Inner<T> {
 ///     }).wait();
 /// });
 ///
-/// c.send(3).unwrap();
+/// c.complete(3);
 /// ```
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     let inner = Arc::new(Inner {
@@ -110,28 +110,13 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
 }
 
 impl<T> Sender<T> {
-    #[deprecated(note = "renamed to `send`", since = "0.1.11")]
-    #[doc(hidden)]
-    #[cfg(feature = "with-deprecated")]
-    pub fn complete(self, t: T) {
-        drop(self.send(t));
-    }
-
     /// Completes this oneshot with a successful result.
     ///
     /// This function will consume `self` and indicate to the other end, the
     /// `Receiver`, that the error provided is the result of the computation this
     /// represents.
-    ///
-    /// If the value is successfully enqueued for the remote end to receive,
-    /// then `Ok(())` is returned. If the receiving end was deallocated before
-    /// this function was called, however, then `Err` is returned with the value
-    /// provided.
-    pub fn send(self, t: T) -> Result<(), T> {
-        if self.inner.complete.load(SeqCst) {
-            return Err(t)
-        }
-
+    pub fn complete(self, t: T) {
+        // First up, flag that this method was called and then store the data.
         // Note that this lock acquisition should always succeed as it can only
         // interfere with `poll` in `Receiver` which is only called when the
         // `complete` flag is true, which we're setting here.
@@ -139,7 +124,6 @@ impl<T> Sender<T> {
         assert!(slot.is_none());
         *slot = Some(t);
         drop(slot);
-        Ok(())
     }
 
     /// Polls this `Sender` half to detect whether the `Receiver` this has
diff --git a/src/task_impl/data.rs b/src/task_impl/data.rs
index f7076e4b..ccd942f2 100644
--- a/src/task_impl/data.rs
+++ b/src/task_impl/data.rs
@@ -113,9 +113,9 @@ impl<T: Send + 'static> LocalKey<T> {
         where F: FnOnce(&T) -> R
     {
         let key = (self.__key)();
-        super::with(|task| {
+        super::with(|_, data| {
             let raw_pointer = {
-                let mut data = task.map.borrow_mut();
+                let mut data = data.borrow_mut();
                 let entry = data.entry(key).or_insert_with(|| {
                     Box::new((self.__init)())
                 });
diff --git a/src/task_impl/mod.rs b/src/task_impl/mod.rs
index a261a76a..e0d4dcb2 100644
--- a/src/task_impl/mod.rs
+++ b/src/task_impl/mod.rs
@@ -2,13 +2,13 @@ use std::prelude::v1::*;
 
 use std::cell::Cell;
 use std::fmt;
-use std::mem;
 use std::sync::Arc;
 use std::sync::atomic::{Ordering, AtomicBool, AtomicUsize, ATOMIC_USIZE_INIT};
 use std::thread;
 
-use {Poll, Future, Async, Stream, Sink, StartSend, AsyncSink};
+use {Poll, Future, Async};
 use future::BoxFuture;
+use stream::Stream;
 
 mod unpark_mutex;
 use self::unpark_mutex::UnparkMutex;
@@ -20,15 +20,8 @@ mod data;
 pub use self::task_rc::TaskRc;
 pub use self::data::LocalKey;
 
-struct BorrowedTask<'a> {
-    id: usize,
-    unpark: &'a Arc<Unpark>,
-    map: &'a data::LocalMap,
-    events: Events,
-}
-
-thread_local!(static CURRENT_TASK: Cell<*const BorrowedTask<'static>> = {
-    Cell::new(0 as *const _)
+thread_local!(static CURRENT_TASK: Cell<(*const Task, *const data::LocalMap)> = {
+    Cell::new((0 as *const _, 0 as *const _))
 });
 
 fn fresh_task_id() -> usize {
@@ -41,32 +34,29 @@ fn fresh_task_id() -> usize {
     id
 }
 
-fn set<'a, F, R>(task: &BorrowedTask<'a>, f: F) -> R
+fn set<F, R>(task: &Task, data: &data::LocalMap, f: F) -> R
     where F: FnOnce() -> R
 {
-    struct Reset(*const BorrowedTask<'static>);
+    struct Reset((*const Task, *const data::LocalMap));
     impl Drop for Reset {
         fn drop(&mut self) {
             CURRENT_TASK.with(|c| c.set(self.0));
         }
     }
 
-    CURRENT_TASK.with(move |c| {
+    CURRENT_TASK.with(|c| {
         let _reset = Reset(c.get());
-        let task = unsafe {
-            mem::transmute::<&BorrowedTask<'a>,
-                             *const BorrowedTask<'static>>(task)
-        };
-        c.set(task);
+        c.set((task as *const _, data as *const _));
         f()
     })
 }
 
-fn with<F: FnOnce(&BorrowedTask) -> R, R>(f: F) -> R {
-    let task = CURRENT_TASK.with(|c| c.get());
+fn with<F: FnOnce(&Task, &data::LocalMap) -> R, R>(f: F) -> R {
+    let (task, data) = CURRENT_TASK.with(|c| c.get());
     assert!(!task.is_null(), "no Task is currently running");
+    debug_assert!(!data.is_null());
     unsafe {
-        f(&*task)
+        f(&*task, &*data)
     }
 }
 
@@ -117,13 +107,7 @@ fn _assert_kinds() {
 /// is, this method can be dangerous to call outside of an implementation of
 /// `poll`.
 pub fn park() -> Task {
-    with(|task| {
-        Task {
-            id: task.id,
-            events: task.events.clone(),
-            unpark: task.unpark.clone(),
-        }
-    })
+    with(|task, _| task.clone())
 }
 
 impl Task {
@@ -144,7 +128,7 @@ impl Task {
     /// other words, the task is currently running on the thread calling the
     /// function.
     pub fn is_current(&self) -> bool {
-        with(|current| current.id == self.id)
+        with(|current, _| current.id == self.id)
     }
 }
 
@@ -181,14 +165,13 @@ impl fmt::Debug for Task {
 pub fn with_unpark_event<F, R>(event: UnparkEvent, f: F) -> R
     where F: FnOnce() -> R
 {
-    with(|task| {
-        let new_task = BorrowedTask {
+    with(|task, data| {
+        let new_task = Task {
             id: task.id,
-            unpark: task.unpark,
+            unpark: task.unpark.clone(),
             events: task.events.with_event(event),
-            map: task.map,
         };
-        set(&new_task, f)
+        set(&new_task, data, f)
     })
 }
 
@@ -215,15 +198,6 @@ impl UnparkEvent {
     }
 }
 
-impl fmt::Debug for UnparkEvent {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("UnparkEvent")
-         .field("set", &"...")
-         .field("item", &self.item)
-         .finish()
-    }
-}
-
 /// A concurrent set which allows for the insertion of `usize` values.
 ///
 /// `EventSet`s are used to communicate precise information about the event(s)
@@ -334,7 +308,7 @@ impl<F: Future> Spawn<F> {
     /// Otherwise if `Ready` or `Err` is returned, the `Spawn` task can be
     /// safely destroyed.
     pub fn poll_future(&mut self, unpark: Arc<Unpark>) -> Poll<F::Item, F::Error> {
-        self.enter(&unpark, |f| f.poll())
+        self.enter(unpark, |f| f.poll())
     }
 
     /// Waits for the internal future to complete, blocking this thread's
@@ -393,7 +367,7 @@ impl<S: Stream> Spawn<S> {
     /// Like `poll_future`, except polls the underlying stream.
     pub fn poll_stream(&mut self, unpark: Arc<Unpark>)
                        -> Poll<Option<S::Item>, S::Error> {
-        self.enter(&unpark, |stream| stream.poll())
+        self.enter(unpark, |stream| stream.poll())
     }
 
     /// Like `wait_future`, except only waits for the next element to arrive on
@@ -411,86 +385,17 @@ impl<S: Stream> Spawn<S> {
     }
 }
 
-impl<S: Sink> Spawn<S> {
-    /// Invokes the underlying `start_send` method with this task in place.
-    ///
-    /// If the underlying operation returns `NotReady` then the `unpark` value
-    /// passed in will receive a notification when the operation is ready to be
-    /// attempted again.
-    pub fn start_send(&mut self, value: S::SinkItem, unpark: &Arc<Unpark>)
-                       -> StartSend<S::SinkItem, S::SinkError> {
-        self.enter(unpark, |sink| sink.start_send(value))
-    }
-
-    /// Invokes the underlying `poll_complete` method with this task in place.
-    ///
-    /// If the underlying operation returns `NotReady` then the `unpark` value
-    /// passed in will receive a notification when the operation is ready to be
-    /// attempted again.
-    pub fn poll_flush(&mut self, unpark: &Arc<Unpark>)
-                       -> Poll<(), S::SinkError> {
-        self.enter(unpark, |sink| sink.poll_complete())
-    }
-
-    /// Blocks the current thread until it's able to send `value` on this sink.
-    ///
-    /// This function will send the `value` on the sink that this task wraps. If
-    /// the sink is not ready to send the value yet then the current thread will
-    /// be blocked until it's able to send the value.
-    pub fn wait_send(&mut self, mut value: S::SinkItem)
-                     -> Result<(), S::SinkError> {
-        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
-        let unpark2 = unpark.clone() as Arc<Unpark>;
-        loop {
-            value = match try!(self.start_send(value, &unpark2)) {
-                AsyncSink::NotReady(v) => v,
-                AsyncSink::Ready => return Ok(()),
-            };
-            unpark.park();
-        }
-    }
-
-    /// Blocks the current thread until it's able to flush this sink.
-    ///
-    /// This function will call the underlying sink's `poll_complete` method
-    /// until it returns that it's ready, proxying out errors upwards to the
-    /// caller if one occurs.
-    ///
-    /// The thread will be blocked until `poll_complete` returns that it's
-    /// ready.
-    pub fn wait_flush(&mut self) -> Result<(), S::SinkError> {
-        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
-        let unpark2 = unpark.clone() as Arc<Unpark>;
-        loop {
-            if try!(self.poll_flush(&unpark2)).is_ready() {
-                return Ok(())
-            }
-            unpark.park();
-        }
-    }
-}
-
 impl<T> Spawn<T> {
-    fn enter<F, R>(&mut self, unpark: &Arc<Unpark>, f: F) -> R
+    fn enter<F, R>(&mut self, unpark: Arc<Unpark>, f: F) -> R
         where F: FnOnce(&mut T) -> R
     {
-        let task = BorrowedTask {
+        let task = Task {
             id: self.id,
             unpark: unpark,
             events: Events::new(),
-            map: &self.data,
         };
         let obj = &mut self.obj;
-        set(&task, || f(obj))
-    }
-}
-
-impl<T: fmt::Debug> fmt::Debug for Spawn<T> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Spawn")
-         .field("obj", &self.obj)
-         .field("id", &self.id)
-         .finish()
+        set(&task, &self.data, || f(obj))
     }
 }
 
@@ -586,14 +491,6 @@ impl Run {
     }
 }
 
-impl fmt::Debug for Run {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Run")
-         .field("contents", &"...")
-         .finish()
-    }
-}
-
 impl Unpark for Inner {
     fn unpark(&self) {
         match self.mutex.notify() {
diff --git a/src/task_impl/task_rc.rs b/src/task_impl/task_rc.rs
index b942f26a..e246b5d7 100644
--- a/src/task_impl/task_rc.rs
+++ b/src/task_impl/task_rc.rs
@@ -89,7 +89,7 @@ impl<A> TaskRc<A> {
     ///
     /// This function will panic if a task is not currently running.
     pub fn new(a: A) -> TaskRc<A> {
-        super::with(|task| {
+        super::with(|task, _| {
             TaskRc {
                 task_id: task.id,
                 ptr: Arc::new(UnsafeCell::new(a)),
@@ -113,7 +113,7 @@ impl<A> TaskRc<A> {
         where F: FnOnce(&A) -> R
     {
         // for safety here, see docs at the top of this module
-        super::with(|task| {
+        super::with(|task, _| {
             assert!(self.task_id == task.id,
                     "TaskRc being accessed on task it does not belong to");
             f(unsafe { &*self.ptr.get() })
diff --git a/src/unsync/oneshot.rs b/src/unsync/oneshot.rs
index 6008efa6..555a6b6e 100644
--- a/src/unsync/oneshot.rs
+++ b/src/unsync/oneshot.rs
@@ -74,17 +74,9 @@ impl<T> Sender<T> {
     /// This function will consume `self` and indicate to the other end, the
     /// `Receiver`, that the error provided is the result of the computation this
     /// represents.
-    ///
-    /// If the value is successfully enqueued for the remote end to receive,
-    /// then `Ok(())` is returned. If the receiving end was deallocated before
-    /// this function was called, however, then `Err` is returned with the value
-    /// provided.
-    pub fn send(self, val: T) -> Result<(), T> {
+    pub fn complete(self, val: T) {
         if let Some(inner) = self.inner.upgrade() {
             inner.borrow_mut().value = Some(val);
-            Ok(())
-        } else {
-            Err(val)
         }
     }
 


Test patch:
diff --git a/tests/all.rs b/tests/all.rs
index bc872bb3..3bf201d8 100644
--- a/tests/all.rs
+++ b/tests/all.rs
@@ -110,7 +110,7 @@ fn flatten() {
 fn smoke_oneshot() {
     assert_done(|| {
         let (c, p) = oneshot::channel();
-        c.send(1).unwrap();
+        c.complete(1);
         p
     }, Ok(1));
     assert_done(|| {
@@ -149,7 +149,7 @@ fn select_cancels() {
     // assert!(f.poll(&mut Task::new()).is_not_ready());
     assert!(brx.try_recv().is_err());
     assert!(drx.try_recv().is_err());
-    a.send(1).unwrap();
+    a.complete(1);
     let res = executor::spawn(f).poll_future(unpark_panic());
     assert!(res.ok().unwrap().is_ready());
     assert_eq!(brx.recv().unwrap(), 1);
@@ -164,7 +164,7 @@ fn select_cancels() {
     let mut f = executor::spawn(b.select(d).then(unselect));
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
-    a.send(1).unwrap();
+    a.complete(1);
     assert!(f.poll_future(unpark_panic()).ok().unwrap().is_ready());
     drop((c, f));
     assert!(drx.recv().is_err());
@@ -210,7 +210,7 @@ fn join_incomplete() {
     let mut f = executor::spawn(ok(1).join(b).map(move |r| tx.send(r).unwrap()));
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
-    a.send(2).unwrap();
+    a.complete(2);
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_ready());
     assert_eq!(rx.recv().unwrap(), (1, 2));
 
@@ -219,7 +219,7 @@ fn join_incomplete() {
     let mut f = executor::spawn(b.join(Ok(2)).map(move |r| tx.send(r).unwrap()));
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
-    a.send(1).unwrap();
+    a.complete(1);
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_ready());
     assert_eq!(rx.recv().unwrap(), (1, 2));
 
@@ -278,7 +278,7 @@ fn select2() {
         let f = b.select(d);
         let (tx, rx) = channel();
         f.map(move |r| tx.send(r).unwrap()).forget();
-        a.send(1).unwrap();
+        a.complete(1);
         let (val, next) = rx.recv().unwrap();
         assert_eq!(val, 1);
         let (tx, rx) = channel();
@@ -300,7 +300,7 @@ fn select2() {
         let (tx, rx) = channel();
         next.map(move |r| tx.send(r).unwrap()).forget();
         assert_eq!(rx.try_recv().err().unwrap(), TryRecvError::Empty);
-        a.send(2).unwrap();
+        a.complete(2);
         assert_eq!(rx.recv().unwrap(), 2);
     }
 
diff --git a/tests/buffer_unordered.rs b/tests/buffer_unordered.rs
index e47a0d8d..8874c291 100644
--- a/tests/buffer_unordered.rs
+++ b/tests/buffer_unordered.rs
@@ -42,14 +42,14 @@ fn works() {
     let o4 = rx3.recv().unwrap();
     assert!(rx4.try_recv().is_err());
 
-    o1.send(1).unwrap();
+    o1.complete(1);
     assert_eq!(rx4.recv(), Ok(1));
-    o3.send(3).unwrap();
+    o3.complete(3);
     assert_eq!(rx4.recv(), Ok(3));
     tx2.send(()).unwrap();
-    o2.send(2).unwrap();
+    o2.complete(2);
     assert_eq!(rx4.recv(), Ok(2));
-    o4.send(4).unwrap();
+    o4.complete(4);
     assert_eq!(rx4.recv(), Ok(4));
 
     let o5 = rx3.recv().unwrap();
@@ -58,15 +58,15 @@ fn works() {
     let o8 = rx3.recv().unwrap();
     let o9 = rx3.recv().unwrap();
 
-    o5.send(5).unwrap();
+    o5.complete(5);
     assert_eq!(rx4.recv(), Ok(5));
-    o8.send(8).unwrap();
+    o8.complete(8);
     assert_eq!(rx4.recv(), Ok(8));
-    o9.send(9).unwrap();
+    o9.complete(9);
     assert_eq!(rx4.recv(), Ok(9));
-    o7.send(7).unwrap();
+    o7.complete(7);
     assert_eq!(rx4.recv(), Ok(7));
-    o6.send(6).unwrap();
+    o6.complete(6);
     assert_eq!(rx4.recv(), Ok(6));
 
     t1.join().unwrap();
diff --git a/tests/eager_drop.rs b/tests/eager_drop.rs
index 94f30d2b..507a9fdd 100644
--- a/tests/eager_drop.rs
+++ b/tests/eager_drop.rs
@@ -60,7 +60,7 @@ fn and_then_drops_eagerly() {
         ok(1)
     }).forget();
     assert!(rx2.try_recv().is_err());
-    c.send(()).unwrap();
+    c.complete(());
     rx2.recv().unwrap();
 }
 
diff --git a/tests/eventual.rs b/tests/eventual.rs
index 5bf3680d..feb4606b 100644
--- a/tests/eventual.rs
+++ b/tests/eventual.rs
@@ -47,7 +47,7 @@ fn and_then2() {
 #[test]
 fn oneshot1() {
     let (c, p) = oneshot::channel::<i32>();
-    let t = thread::spawn(|| c.send(1).unwrap());
+    let t = thread::spawn(|| c.complete(1));
 
     let (tx, rx) = mpsc::channel();
     p.map(move |e| tx.send(e).unwrap()).forget();
@@ -58,7 +58,7 @@ fn oneshot1() {
 #[test]
 fn oneshot2() {
     let (c, p) = oneshot::channel::<i32>();
-    let t = thread::spawn(|| c.send(1).unwrap());
+    let t = thread::spawn(|| c.complete(1));
     t.join().unwrap();
 
     let (tx, rx) = mpsc::channel();
@@ -72,7 +72,7 @@ fn oneshot3() {
     let (tx, rx) = mpsc::channel();
     p.map(move |e| tx.send(e).unwrap()).forget();
 
-    let t = thread::spawn(|| c.send(1).unwrap());
+    let t = thread::spawn(|| c.complete(1));
     t.join().unwrap();
 
     assert_eq!(rx.recv(), Ok(1));
@@ -102,7 +102,7 @@ fn oneshot5() {
 fn oneshot6() {
     let (c, p) = oneshot::channel::<i32>();
     drop(p);
-    c.send(2).unwrap_err();
+    c.complete(2);
 }
 
 #[test]
@@ -145,7 +145,7 @@ fn or_else1() {
 
     assert!(rx.try_recv().is_err());
     drop(c1);
-    c2.send(3).unwrap();
+    c2.complete(3);
     p1.or_else(|_| p2).map(move |v| tx.send(v).unwrap()).forget();
 
     assert_eq!(rx.recv(), Ok(2));
@@ -165,7 +165,7 @@ fn or_else2() {
         ok::<i32, i32>(1)
     }).forget();
 
-    c1.send(2).unwrap();
+    c1.complete(2);
     assert!(rx.recv().is_err());
 }
 
@@ -186,9 +186,9 @@ fn join2() {
     let (tx, rx) = mpsc::channel();
     p1.join(p2).map(move |v| tx.send(v).unwrap()).forget();
     assert!(rx.try_recv().is_err());
-    c1.send(1).unwrap();
+    c1.complete(1);
     assert!(rx.try_recv().is_err());
-    c2.send(2).unwrap();
+    c2.complete(2);
     assert_eq!(rx.recv(), Ok((1, 2)));
     assert!(rx.recv().is_err());
 }
@@ -227,11 +227,11 @@ fn join5() {
     let (tx, rx) = mpsc::channel();
     p1.join(p2).join(p3).map(move |v| tx.send(v).unwrap()).forget();
     assert!(rx.try_recv().is_err());
-    c1.send(1).unwrap();
+    c1.complete(1);
     assert!(rx.try_recv().is_err());
-    c2.send(2).unwrap();
+    c2.complete(2);
     assert!(rx.try_recv().is_err());
-    c3.send(3).unwrap();
+    c3.complete(3);
     assert_eq!(rx.recv(), Ok(((1, 2), 3)));
     assert!(rx.recv().is_err());
 }
@@ -243,14 +243,14 @@ fn select1() {
     let (tx, rx) = mpsc::channel();
     p1.select(p2).map(move |v| tx.send(v).unwrap()).forget();
     assert!(rx.try_recv().is_err());
-    c1.send(1).unwrap();
+    c1.complete(1);
     let (v, p2) = rx.recv().unwrap();
     assert_eq!(v, 1);
     assert!(rx.recv().is_err());
 
     let (tx, rx) = mpsc::channel();
     p2.map(move |v| tx.send(v).unwrap()).forget();
-    c2.send(2).unwrap();
+    c2.complete(2);
     assert_eq!(rx.recv(), Ok(2));
     assert!(rx.recv().is_err());
 }
@@ -269,7 +269,7 @@ fn select2() {
 
     let (tx, rx) = mpsc::channel();
     p2.map(move |v| tx.send(v).unwrap()).forget();
-    c2.send(2).unwrap();
+    c2.complete(2);
     assert_eq!(rx.recv(), Ok(2));
     assert!(rx.recv().is_err());
 }
@@ -299,7 +299,7 @@ fn select4() {
 
     let t = thread::spawn(move || {
         for c in rx {
-            c.send(1).unwrap();
+            c.complete(1);
         }
     });
 
diff --git a/tests/futures_unordered.rs b/tests/futures_unordered.rs
index d877f03b..8b9b29f8 100644
--- a/tests/futures_unordered.rs
+++ b/tests/futures_unordered.rs
@@ -17,11 +17,11 @@ fn works_1() {
     let stream = futures_unordered(vec![a_rx, b_rx, c_rx]);
 
     let mut spawn = futures::executor::spawn(stream);
-    b_tx.send(99).unwrap();
+    b_tx.complete(99);
     assert_eq!(Some(Ok(99)), spawn.wait_stream());
 
-    a_tx.send(33).unwrap();
-    c_tx.send(33).unwrap();
+    a_tx.complete(33);
+    c_tx.complete(33);
     assert_eq!(Some(Ok(33)), spawn.wait_stream());
     assert_eq!(Some(Ok(33)), spawn.wait_stream());
     assert_eq!(None, spawn.wait_stream());
@@ -36,10 +36,10 @@ fn works_2() {
     let stream = futures_unordered(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();
+    a_tx.complete(33);
+    b_tx.complete(33);
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_ready());
-    c_tx.send(33).unwrap();
+    c_tx.complete(33);
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_ready());
 }
 
@@ -59,10 +59,10 @@ fn finished_future_ok() {
         assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
     }
 
-    b_tx.send(Box::new(())).unwrap();
+    b_tx.complete(Box::new(()));
     let next = spawn.poll_stream(support::unpark_noop()).unwrap();
     assert!(next.is_ready());
-    c_tx.send(Box::new(())).unwrap();
+    c_tx.complete(Box::new(()));
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
 }
diff --git a/tests/shared.rs b/tests/shared.rs
index f74f5a94..bacb9ad5 100644
--- a/tests/shared.rs
+++ b/tests/shared.rs
@@ -19,7 +19,7 @@ fn send_shared_oneshot_and_wait_on_multiple_threads(threads_number: u32) {
             assert!(*cloned_future.wait().unwrap() == 6);
         })
     }).collect::<Vec<_>>();
-    tx.send(6).unwrap();
+    tx.complete(6);
     assert!(*f.wait().unwrap() == 6);
     for f in threads {
         f.join().unwrap();
@@ -57,13 +57,13 @@ fn drop_on_one_task_ok() {
     let (tx3, rx3) = oneshot::channel::<u32>();
 
     let t2 = thread::spawn(|| {
-        drop(f2.map(|x| tx3.send(*x).unwrap()).map_err(|_| ()).wait());
+        drop(f2.map(|x| tx3.complete(*x)).map_err(|_| ()).wait());
     });
 
-    tx2.send(11).unwrap(); // cancel `f1`
+    tx2.complete(11); // cancel `f1`
     t1.join().unwrap();
 
-    tx.send(42).unwrap(); // Should cause `f2` and then `rx3` to get resolved.
+    tx.complete(42); // Should cause `f2` and then `rx3` to get resolved.
     let result = rx3.wait().unwrap();
     assert_eq!(result, 42);
     t2.join().unwrap();
@@ -97,7 +97,7 @@ fn peek() {
     }
 
     // Completing the underlying future has no effect, because the value has not been `poll`ed in.
-    tx0.send(42).unwrap();
+    tx0.complete(42);
     for _ in 0..2 {
         assert!(f1.peek().is_none());
         assert!(f2.peek().is_none());
@@ -123,20 +123,20 @@ fn polled_then_ignored() {
     let (tx2, rx2) = oneshot::channel::<u32>();
     let (tx3, rx3) = oneshot::channel::<u32>();
 
-    core.spawn(f1.map(|n| tx3.send(*n).unwrap()).map_err(|_|()));
+    core.spawn(f1.map(|n| tx3.complete(*n)).map_err(|_|()));
 
     core.run(future::ok::<(),()>(())).unwrap(); // Allow f1 to be polled.
 
     core.spawn(f2.map_err(|_| ()).map(|x| *x).select(rx2.map_err(|_| ())).map_err(|_| ())
-        .and_then(|(_, f2)| rx3.map_err(|_| ()).map(move |n| {drop(f2); tx1.send(n).unwrap()})));
+        .and_then(|(_, f2)| rx3.map_err(|_| ()).map(move |n| {drop(f2); tx1.complete(n)})));
 
     core.run(future::ok::<(),()>(())).unwrap();  // Allow f2 to be polled.
 
-    tx2.send(11).unwrap(); // Resolve rx2, causing f2 to no longer get polled.
+    tx2.complete(11); // Resolve rx2, causing f2 to no longer get polled.
 
-    core.run(future::ok::<(),()>(())).unwrap(); // Let the send() propagate.
+    core.run(future::ok::<(),()>(())).unwrap(); // Let the complete() propagate.
 
-    tx0.send(42).unwrap(); // Should cause f1, then rx3, and then rx1 to resolve.
+    tx0.complete(42); // Should cause f1, then rx3, and then rx1 to resolve.
 
     assert_eq!(core.run(rx1).unwrap(), 42);
 }
@@ -166,7 +166,7 @@ fn recursive_poll() {
     // deadlock or panic due to a recursive lock() on a mutex.
     core.run(future::ok::<(),()>(())).unwrap();
 
-    tx1.send(()).unwrap(); // Break the cycle.
+    tx1.complete(()); // Break the cycle.
     drop(tx0);
     core.run(f3).unwrap();
 }
@@ -198,7 +198,7 @@ fn recursive_poll_with_unpark() {
     // deadlock or panic due to a recursive lock() on a mutex.
     core.run(future::ok::<(),()>(())).unwrap();
 
-    tx1.send(()).unwrap(); // Break the cycle.
+    tx1.complete(()); // Break the cycle.
     drop(tx0);
     core.run(f3).unwrap();
 }
diff --git a/tests/sink.rs b/tests/sink.rs
index 97283a79..ebd0c431 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -144,7 +144,7 @@ fn with_flush() {
     let flag = Flag::new();
     let mut task = executor::spawn(sink.flush());
     assert!(task.poll_future(flag.clone()).unwrap().is_not_ready());
-    tx.send(()).unwrap();
+    tx.complete(());
     assert!(flag.get());
 
     let sink = match task.poll_future(flag.clone()).unwrap() {
diff --git a/tests/stream.rs b/tests/stream.rs
index 71ecb866..6b1bfae6 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -165,9 +165,9 @@ fn buffered() {
 
     let mut rx = rx.buffered(2);
     sassert_empty(&mut rx);
-    c.send(3).unwrap();
+    c.complete(3);
     sassert_empty(&mut rx);
-    a.send(5).unwrap();
+    a.complete(5);
     let mut rx = rx.wait();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
@@ -183,9 +183,9 @@ fn buffered() {
 
     let mut rx = rx.buffered(1);
     sassert_empty(&mut rx);
-    c.send(3).unwrap();
+    c.complete(3);
     sassert_empty(&mut rx);
-    a.send(5).unwrap();
+    a.complete(5);
     let mut rx = rx.wait();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
@@ -205,9 +205,9 @@ fn unordered() {
     let mut rx = rx.buffer_unordered(2);
     sassert_empty(&mut rx);
     let mut rx = rx.wait();
-    c.send(3).unwrap();
+    c.complete(3);
     assert_eq!(rx.next(), Some(Ok(3)));
-    a.send(5).unwrap();
+    a.complete(5);
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), None);
 
@@ -222,9 +222,9 @@ fn unordered() {
     // We don't even get to see `c` until `a` completes.
     let mut rx = rx.buffer_unordered(1);
     sassert_empty(&mut rx);
-    c.send(3).unwrap();
+    c.complete(3);
     sassert_empty(&mut rx);
-    a.send(5).unwrap();
+    a.complete(5);
     let mut rx = rx.wait();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
@@ -321,12 +321,3 @@ fn forward() {
     assert_done(move || iter(vec![Ok(4), Ok(5)]).forward(v).map(|(_, s)| s),
                 Ok::<_, ()>(vec![0, 1, 2, 3, 4, 5]));
 }
-
-#[test]
-fn concat() {
-    let a = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Ok(vec![4, 5, 6]), Ok(vec![7, 8, 9])]);
-    assert_done(move || a.concat(), Ok(vec![1, 2, 3, 4, 5, 6, 7, 8, 9]));
-
-    let b = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Err(()), Ok(vec![7, 8, 9])]);
-    assert_done(move || b.concat(), Err(()));
-}
diff --git a/tests/unsync-oneshot.rs b/tests/unsync-oneshot.rs
index 565c3953..38c6d97e 100644
--- a/tests/unsync-oneshot.rs
+++ b/tests/unsync-oneshot.rs
@@ -7,7 +7,7 @@ use futures::unsync::oneshot::{channel, Canceled};
 #[test]
 fn smoke() {
     let (tx, rx) = channel();
-    tx.send(33).unwrap();
+    tx.complete(33);
     assert_eq!(rx.wait().unwrap(), 33);
 }
 
@@ -28,7 +28,7 @@ fn tx_complete_rx_unparked() {
     let (tx, rx) = channel();
 
     let res = rx.join(future::lazy(move || {
-        tx.send(55).unwrap();
+        tx.complete(55);
         Ok(11)
     }));
     assert_eq!(res.wait().unwrap(), (55, 11));