Repository: alexcrichton/futures-rs (rust)

Issue:
there is a bug in `SpawnHandle`
the underlying future will be canceled when I call `SpawnHandle::forget`,
```rust
extern crate futures;
extern crate tokio_core;

use std::thread;
use std::time::Duration;

use futures::{Future, Stream};
use futures::future::{Executor, ExecuteError, empty};
use futures::sync::mpsc;
use futures::sync::oneshot::{self, Execute};
use tokio_core::reactor::{Core, Remote};

struct MyExecutor {
    handle: Remote,
}

impl MyExecutor {
    fn new(handle: Remote) -> Self {
        MyExecutor { handle: handle }
    }
}

impl<F> Executor<Execute<F>> for MyExecutor
where
    F: Future + Send + 'static,
    F::Item: Send,
    F::Error: Send,
{
    fn execute(&self, future: Execute<F>) -> Result<(), ExecuteError<Execute<F>>> {
        self.handle.spawn(move |_| future);
        Ok(())
    }
}

fn main() {
    let mut core = Core::new().unwrap();
    let my_executor = MyExecutor::new(core.remote());
    thread::spawn(move || {
        let (tx, rx) = mpsc::unbounded();
        let f = rx.for_each(|t| Ok(println!("{}", t)));
        let handle = oneshot::spawn(f, &my_executor);
        handle.forget();
        for i in 1..10 {
            tx.unbounded_send(i).unwrap(); // panic here
            thread::sleep(Duration::from_secs(1));
        }
    });
    core.run(empty::<(), ()>()).unwrap();
}

```

this code will panic due to the drop of the `UnboundedSender`. 

Pull request description:
Fix {sync,unsync}::oneshot::spawn (fixes #669)
- `keep_running` was set to inverted values.
- also add unit tests

Patch:
diff --git a/src/sync/oneshot.rs b/src/sync/oneshot.rs
index 9e79442b..d95883d3 100644
--- a/src/sync/oneshot.rs
+++ b/src/sync/oneshot.rs
@@ -477,7 +477,7 @@ pub fn spawn<F, E>(future: F, executor: &E) -> SpawnHandle<F::Item, F::Error>
 {
     let data = Arc::new(ExecuteInner {
         inner: Inner::new(),
-        keep_running: AtomicBool::new(true),
+        keep_running: AtomicBool::new(false),
     });
     executor.execute(Execute {
         future: future,
@@ -506,7 +506,7 @@ impl<T, E> SpawnHandle<T, E> {
     /// well if the future hasn't already resolved. This function can be used
     /// when to drop this future but keep executing the underlying future.
     pub fn forget(self) {
-        self.rx.keep_running.store(false, SeqCst);
+        self.rx.keep_running.store(true, SeqCst);
     }
 }
 
diff --git a/src/unsync/oneshot.rs b/src/unsync/oneshot.rs
index c895e68a..7ae2890f 100644
--- a/src/unsync/oneshot.rs
+++ b/src/unsync/oneshot.rs
@@ -260,7 +260,7 @@ pub fn spawn<F, E>(future: F, executor: &E) -> SpawnHandle<F::Item, F::Error>
     where F: Future,
           E: Executor<Execute<F>>,
 {
-    let flag = Rc::new(Cell::new(true));
+    let flag = Rc::new(Cell::new(false));
     let (tx, rx) = channel();
     executor.execute(Execute {
         future: future,
@@ -293,7 +293,7 @@ impl<T, E> SpawnHandle<T, E> {
     /// well if the future hasn't already resolved. This function can be used
     /// when to drop this future but keep executing the underlying future.
     pub fn forget(self) {
-        self.keep_running.set(false);
+        self.keep_running.set(true);
     }
 }
 


Test patch:
diff --git a/tests/oneshot.rs b/tests/oneshot.rs
index 71d2d97c..45c19968 100644
--- a/tests/oneshot.rs
+++ b/tests/oneshot.rs
@@ -120,3 +120,134 @@ fn cancel_sends() {
     drop(tx);
     t.join().unwrap();
 }
+
+#[test]
+fn spawn_sends_items() {
+    let core = local_executor::Core::new();
+    let future = ok::<_, ()>(1);
+    let rx = spawn(future, &core);
+    assert_eq!(core.run(rx).unwrap(), 1);
+}
+
+#[test]
+fn spawn_kill_dead_stream() {
+    use std::thread;
+    use std::time::Duration;
+    use futures::future::Either;
+    use futures::sync::oneshot;
+
+    // a future which never returns anything (forever accepting incoming
+    // connections), but dropping it leads to observable side effects
+    // (like closing listening sockets, releasing limited resources,
+    // ...)
+    #[derive(Debug)]
+    struct Dead {
+        // when dropped you should get Err(oneshot::Canceled) on the
+        // receiving end
+        done: oneshot::Sender<()>,
+    }
+    impl Future for Dead {
+        type Item = ();
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            Ok(Async::NotReady)
+        }
+    }
+
+    // need to implement a timeout for the test, as it would hang
+    // forever right now
+    let (timeout_tx, timeout_rx) = oneshot::channel();
+    thread::spawn(move || {
+        thread::sleep(Duration::from_millis(1000));
+        let _ = timeout_tx.send(());
+    });
+
+    let core = local_executor::Core::new();
+    let (done_tx, done_rx) = oneshot::channel();
+    let future = Dead{done: done_tx};
+    let rx = spawn(future, &core);
+    let res = core.run(
+        Ok::<_, ()>(())
+        .into_future()
+        .then(move |_| {
+            // now drop the spawned future: maybe some timeout exceeded,
+            // or some connection on this end was closed by the remote
+            // end.
+            drop(rx);
+            // and wait for the spawned future to release its resources
+            done_rx
+        })
+        .select2(timeout_rx)
+    );
+    match res {
+        Err(Either::A((oneshot::Canceled, _))) => (),
+        Ok(Either::B(((), _))) => {
+            panic!("dead future wasn't canceled (timeout)");
+        },
+        _ => {
+            panic!("dead future wasn't canceled (unexpected result)");
+        },
+    }
+}
+
+#[test]
+fn spawn_dont_kill_forgot_dead_stream() {
+    use std::thread;
+    use std::time::Duration;
+    use futures::future::Either;
+    use futures::sync::oneshot;
+
+    // a future which never returns anything (forever accepting incoming
+    // connections), but dropping it leads to observable side effects
+    // (like closing listening sockets, releasing limited resources,
+    // ...)
+    #[derive(Debug)]
+    struct Dead {
+        // when dropped you should get Err(oneshot::Canceled) on the
+        // receiving end
+        done: oneshot::Sender<()>,
+    }
+    impl Future for Dead {
+        type Item = ();
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            Ok(Async::NotReady)
+        }
+    }
+
+    // need to implement a timeout for the test, as it would hang
+    // forever right now
+    let (timeout_tx, timeout_rx) = oneshot::channel();
+    thread::spawn(move || {
+        thread::sleep(Duration::from_millis(1000));
+        let _ = timeout_tx.send(());
+    });
+
+    let core = local_executor::Core::new();
+    let (done_tx, done_rx) = oneshot::channel();
+    let future = Dead{done: done_tx};
+    let rx = spawn(future, &core);
+    let res = core.run(
+        Ok::<_, ()>(())
+        .into_future()
+        .then(move |_| {
+            // forget the spawned future: should keep running, i.e. hit
+            // the timeout below.
+            rx.forget();
+            // and wait for the spawned future to release its resources
+            done_rx
+        })
+        .select2(timeout_rx)
+    );
+    match res {
+        Err(Either::A((oneshot::Canceled, _))) => {
+            panic!("forgotten dead future was canceled");
+        },
+        Ok(Either::B(((), _))) => (), // reached timeout
+        _ => {
+            panic!("forgotten dead future was canceled (unexpected result)");
+        },
+    }
+}
diff --git a/tests/unsync-oneshot.rs b/tests/unsync-oneshot.rs
index 7f19d830..55b0ca5a 100644
--- a/tests/unsync-oneshot.rs
+++ b/tests/unsync-oneshot.rs
@@ -2,7 +2,10 @@ extern crate futures;
 
 use futures::prelude::*;
 use futures::future;
-use futures::unsync::oneshot::{channel, Canceled};
+use futures::unsync::oneshot::{channel, Canceled, spawn};
+
+mod support;
+use support::local_executor;
 
 #[test]
 fn smoke() {
@@ -53,3 +56,134 @@ fn is_canceled() {
     drop(rx);
     assert!(tx.is_canceled());
 }
+
+#[test]
+fn spawn_sends_items() {
+    let core = local_executor::Core::new();
+    let future = future::ok::<_, ()>(1);
+    let rx = spawn(future, &core);
+    assert_eq!(core.run(rx).unwrap(), 1);
+}
+
+#[test]
+fn spawn_kill_dead_stream() {
+    use std::thread;
+    use std::time::Duration;
+    use futures::future::Either;
+    use futures::sync::oneshot;
+
+    // a future which never returns anything (forever accepting incoming
+    // connections), but dropping it leads to observable side effects
+    // (like closing listening sockets, releasing limited resources,
+    // ...)
+    #[derive(Debug)]
+    struct Dead {
+        // when dropped you should get Err(oneshot::Canceled) on the
+        // receiving end
+        done: oneshot::Sender<()>,
+    }
+    impl Future for Dead {
+        type Item = ();
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            Ok(Async::NotReady)
+        }
+    }
+
+    // need to implement a timeout for the test, as it would hang
+    // forever right now
+    let (timeout_tx, timeout_rx) = oneshot::channel();
+    thread::spawn(move || {
+        thread::sleep(Duration::from_millis(1000));
+        let _ = timeout_tx.send(());
+    });
+
+    let core = local_executor::Core::new();
+    let (done_tx, done_rx) = oneshot::channel();
+    let future = Dead{done: done_tx};
+    let rx = spawn(future, &core);
+    let res = core.run(
+        Ok::<_, ()>(())
+        .into_future()
+        .then(move |_| {
+            // now drop the spawned future: maybe some timeout exceeded,
+            // or some connection on this end was closed by the remote
+            // end.
+            drop(rx);
+            // and wait for the spawned future to release its resources
+            done_rx
+        })
+        .select2(timeout_rx)
+    );
+    match res {
+        Err(Either::A((oneshot::Canceled, _))) => (),
+        Ok(Either::B(((), _))) => {
+            panic!("dead future wasn't canceled (timeout)");
+        },
+        _ => {
+            panic!("dead future wasn't canceled (unexpected result)");
+        },
+    }
+}
+
+#[test]
+fn spawn_dont_kill_forgot_dead_stream() {
+    use std::thread;
+    use std::time::Duration;
+    use futures::future::Either;
+    use futures::sync::oneshot;
+
+    // a future which never returns anything (forever accepting incoming
+    // connections), but dropping it leads to observable side effects
+    // (like closing listening sockets, releasing limited resources,
+    // ...)
+    #[derive(Debug)]
+    struct Dead {
+        // when dropped you should get Err(oneshot::Canceled) on the
+        // receiving end
+        done: oneshot::Sender<()>,
+    }
+    impl Future for Dead {
+        type Item = ();
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            Ok(Async::NotReady)
+        }
+    }
+
+    // need to implement a timeout for the test, as it would hang
+    // forever right now
+    let (timeout_tx, timeout_rx) = oneshot::channel();
+    thread::spawn(move || {
+        thread::sleep(Duration::from_millis(1000));
+        let _ = timeout_tx.send(());
+    });
+
+    let core = local_executor::Core::new();
+    let (done_tx, done_rx) = oneshot::channel();
+    let future = Dead{done: done_tx};
+    let rx = spawn(future, &core);
+    let res = core.run(
+        Ok::<_, ()>(())
+        .into_future()
+        .then(move |_| {
+            // forget the spawned future: should keep running, i.e. hit
+            // the timeout below.
+            rx.forget();
+            // and wait for the spawned future to release its resources
+            done_rx
+        })
+        .select2(timeout_rx)
+    );
+    match res {
+        Err(Either::A((oneshot::Canceled, _))) => {
+            panic!("forgotten dead future was canceled");
+        },
+        Ok(Either::B(((), _))) => (), // reached timeout
+        _ => {
+            panic!("forgotten dead future was canceled (unexpected result)");
+        },
+    }
+}