Add a test for CpuPool shutting down

Fix an off-by-one in the count to send shutdown messages.

Closes rust-lang#65

futures-cpupool/src/lib.rs

@@ -177,7 +177,7 @@ impl Clone for CpuPool {
 
 impl Drop for CpuPool {
     fn drop(&mut self) {
-        if self.inner.cnt.fetch_sub(1, Ordering::Relaxed) != 0 {
+        if self.inner.cnt.fetch_sub(1, Ordering::Relaxed) > 1 {
             return
         }
         for _ in 0..self.inner.size {

futures-cpupool/tests/smoke.rs

@@ -1,7 +1,10 @@
 extern crate futures;
 extern crate futures_cpupool;
 
+use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
 use std::sync::mpsc::channel;
+use std::thread;
+use std::time::Duration;
 
 use futures::Future;
 use futures_cpupool::CpuPool;
@@ -40,3 +43,32 @@ fn select() {
     assert!(item1 != item2);
     assert!((item1 == 1 && item2 == 2) || (item1 == 2 && item2 == 1));
 }
+
+#[test]
+fn threads_go_away() {
+    static CNT: AtomicUsize = ATOMIC_USIZE_INIT;
+
+    struct A;
+
+    impl Drop for A {
+        fn drop(&mut self) {
+            CNT.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    thread_local!(static FOO: A = A);
+
+    let pool = CpuPool::new(2);
+    get(pool.execute(|| {
+        FOO.with(|_| ())
+    })).unwrap();
+    drop(pool);
+
+    for _ in 0..100 {
+        if CNT.load(Ordering::SeqCst) == 1 {
+            return
+        }
+        thread::sleep(Duration::from_millis(10));
+    }
+    panic!("thread didn't exit");
+}