std::rt: Fix a race in UvRemoteCallback's dtor that misses callbacks

Full description in comments.

2 files changed, 74 insertions, 7 deletions

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index c0481d06493..b07591183b7 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -440,6 +440,7 @@ impl Scheduler {
 //                return Some(this);
             }
             Some(Shutdown) => {
+                rtdebug!("shutting down");
 //                this.event_loop.callback(Scheduler::run_sched_once);
                 if this.sleepy {
                     // There may be an outstanding handle on the
@@ -1166,6 +1167,51 @@ mod test {
         }
     }
 
+    // A regression test that the final message is always handled.
+    // Used to deadlock because Shutdown was never recvd.
+    #[test]
+    fn no_missed_messages() {
+        use rt::work_queue::WorkQueue;
+        use rt::sleeper_list::SleeperList;
+        use rt::stack::StackPool;
+        use rt::uv::uvio::UvEventLoop;
+        use rt::sched::{Shutdown, TaskFromFriend};
+        use util;
+
+        do run_in_bare_thread {
+            do stress_factor().times {
+                let sleepers = SleeperList::new();
+                let queue = WorkQueue::new();
+                let queues = ~[queue.clone()];
+
+                let mut sched = ~Scheduler::new(
+                    ~UvEventLoop::new(),
+                    queue,
+                    queues.clone(),
+                    sleepers.clone());
+
+                let mut handle = sched.make_handle();
+
+                let sched = Cell::new(sched);
+
+                let thread = do Thread::start {
+                    let mut sched = sched.take();
+                    let bootstrap_task = ~Task::new_root(&mut sched.stack_pool, None, ||());
+                    sched.bootstrap(bootstrap_task);
+                };
+
+                let mut stack_pool = StackPool::new();
+                let task = ~Task::new_root(&mut stack_pool, None, ||());
+                handle.send(TaskFromFriend(task));
+
+                handle.send(Shutdown);
+                util::ignore(handle);
+
+                thread.join();
+            }
+        }
+    }
+
     #[test]
     fn multithreading() {
         use rt::comm::*;
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index a26b8a3ad59..1ad295641a7 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -162,14 +162,32 @@ impl UvRemoteCallback {
         let exit_flag_clone = exit_flag.clone();
         let async = do AsyncWatcher::new(loop_) |watcher, status| {
             assert!(status.is_none());
+
+            // The synchronization logic here is subtle. To review, the uv async handle
+            // type promises that, after it is triggered the remote callback is definitely
+            // called at least once. UvRemoteCallback needs to maintain those semantics
+            // while also shutting down cleanly from the dtor. In our case that means that,
+            // when the UvRemoteCallback dtor calls `async.send()`, here `f` is always called
+            // later.
+
+            // In the dtor both the exit flag is set and the async callback fired under a lock.
+            // Here, before calling `f`, we take the lock and check the flag. Because we are
+            // checking the flag before calling `f`, and the flag is set under the same lock
+            // as the send, then if the flag is set then we're guaranteed to call `f` after
+            // the final send.
+
+            // If the check was done after `f()` then there would be a period between that call
+            // and the check where the dtor could be called in the other thread, missing the
+            // final callback while still destroying the handle.
+
+            let should_exit = unsafe { exit_flag_clone.with_imm(|&should_exit| should_exit) };
+
             f();
-            unsafe {
-                do exit_flag_clone.with_imm |&should_exit| {
-                    if should_exit {
-                        watcher.close(||());
-                    }
-                }
+
+            if should_exit {
+                watcher.close(||());
             }
+
         };
         UvRemoteCallback {
             async: async,
@@ -218,7 +236,10 @@ mod test_remote {
                 let tube_clone = tube_clone.clone();
                 let tube_clone_cell = Cell::new(tube_clone);
                 let remote = do sched.event_loop.remote_callback {
-                    tube_clone_cell.take().send(1);
+                    // This could be called multiple times
+                    if !tube_clone_cell.is_empty() {
+                        tube_clone_cell.take().send(1);
+                    }
                 };
                 remote_cell.put_back(remote);
             }