---

yaml
---
r: 146527
b: refs/heads/try2
c: 28219fc
h: refs/heads/master
i:
  146525: 5963565
  146523: 180fd20
  146519: 0e2f51a
  146511: 0f2414e
  146495: f8a1a19
v: v3

Author: alexcrichton

[refs]

@@ -5,7 +5,7 @@ refs/heads/snap-stage3: 78a7676898d9f80ab540c6df5d4c9ce35bb50463
 refs/heads/try: 519addf6277dbafccbb4159db4b710c37eaa2ec5
 refs/tags/release-0.1: 1f5c5126e96c79d22cb7862f75304136e204f105
 refs/heads/ndm: f3868061cd7988080c30d6d5bf352a5a5fe2460b
-refs/heads/try2: 9286d5113d843e65fb13ff0cf142c1bfb10124f7
+refs/heads/try2: 28219fc679e6c2f747ad3e49eb746a383797ef9b
 refs/heads/dist-snap: ba4081a5a8573875fed17545846f6f6902c8ba8d
 refs/tags/release-0.2: c870d2dffb391e14efb05aa27898f1f6333a9596
 refs/tags/release-0.3: b5f0d0f648d9a6153664837026ba1be43d3e2503

branches/try2/src/librustuv/async.rs

@@ -8,51 +8,155 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use std::cast;
 use std::libc::c_int;
+use std::rt::rtio::{Callback, RemoteCallback};
+use std::unstable::sync::Exclusive;
 
 use uvll;
-use super::{Watcher, Loop, NativeHandle, AsyncCallback, WatcherInterop};
-use super::status_to_maybe_uv_error;
+use super::{Loop, UvHandle};
 
-pub struct AsyncWatcher(*uvll::uv_async_t);
-impl Watcher for AsyncWatcher { }
+// The entire point of async is to call into a loop from other threads so it
+// does not need to home.
+pub struct AsyncWatcher {
+    handle: *uvll::uv_async_t,
+
+    // A flag to tell the callback to exit, set from the dtor. This is
+    // almost never contested - only in rare races with the dtor.
+    exit_flag: Exclusive<bool>
+}
+
+struct Payload {
+    callback: ~Callback,
+    exit_flag: Exclusive<bool>,
+}
 
 impl AsyncWatcher {
-    pub fn new(loop_: &mut Loop, cb: AsyncCallback) -> AsyncWatcher {
+    pub fn new(loop_: &mut Loop, cb: ~Callback) -> AsyncWatcher {
+        let handle = UvHandle::alloc(None::<AsyncWatcher>, uvll::UV_ASYNC);
+        assert_eq!(unsafe {
+            uvll::async_init(loop_.native_handle(), handle, async_cb)
+        }, 0);
+        let flag = Exclusive::new(false);
+        let payload = ~Payload { callback: cb, exit_flag: flag.clone() };
         unsafe {
-            let handle = uvll::malloc_handle(uvll::UV_ASYNC);
-            assert!(handle.is_not_null());
-            let mut watcher: AsyncWatcher = NativeHandle::from_native_handle(handle);
-            watcher.install_watcher_data();
-            let data = watcher.get_watcher_data();
-            data.async_cb = Some(cb);
-            assert_eq!(0, uvll::uv_async_init(loop_.native_handle(), handle, async_cb));
-            return watcher;
+            let payload: *u8 = cast::transmute(payload);
+            uvll::set_data_for_uv_handle(handle, payload);
         }
+        return AsyncWatcher { handle: handle, exit_flag: flag, };
+    }
+}
 
-        extern fn async_cb(handle: *uvll::uv_async_t, status: c_int) {
-            let mut watcher: AsyncWatcher = NativeHandle::from_native_handle(handle);
-            let status = status_to_maybe_uv_error(status);
-            let data = watcher.get_watcher_data();
-            let cb = data.async_cb.get_ref();
-            (*cb)(watcher, status);
-        }
+impl UvHandle<uvll::uv_async_t> for AsyncWatcher {
+    fn uv_handle(&self) -> *uvll::uv_async_t { self.handle }
+    unsafe fn from_uv_handle<'a>(h: &'a *T) -> &'a mut AsyncWatcher {
+        fail!("async watchers can't be built from their handles");
     }
+}
+
+extern fn async_cb(handle: *uvll::uv_async_t, status: c_int) {
+    assert!(status == 0);
+    let payload: &mut Payload = unsafe {
+        cast::transmute(uvll::get_data_for_uv_handle(handle))
+    };
+
+    // 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 {
+        payload.exit_flag.with_imm(|&should_exit| should_exit)
+    };
+
+    payload.callback.call();
+
+    if should_exit {
+        unsafe { uvll::close(handle, close_cb) }
+    }
+}
 
-    pub fn send(&mut self) {
+extern fn close_cb(handle: *uvll::uv_handle_t) {
+    // drop the payload
+    let _payload: ~Payload = unsafe {
+        cast::transmute(uvll::get_data_for_uv_handle(handle))
+    };
+    // and then free the handle
+    unsafe { uvll::free_handle(handle) }
+}
+
+impl RemoteCallback for AsyncWatcher {
+    fn fire(&mut self) {
+        unsafe { uvll::async_send(self.handle) }
+    }
+}
+
+impl Drop for AsyncWatcher {
+    fn drop(&mut self) {
         unsafe {
-            let handle = self.native_handle();
-            uvll::uv_async_send(handle);
+            do self.exit_flag.with |should_exit| {
+                // NB: These two things need to happen atomically. Otherwise
+                // the event handler could wake up due to a *previous*
+                // signal and see the exit flag, destroying the handle
+                // before the final send.
+                *should_exit = true;
+                uvll::async_send(self.handle)
+            }
         }
     }
 }
 
-impl NativeHandle<*uvll::uv_async_t> for AsyncWatcher {
-    fn from_native_handle(handle: *uvll::uv_async_t) -> AsyncWatcher {
-        AsyncWatcher(handle)
-    }
-    fn native_handle(&self) -> *uvll::uv_async_t {
-        match self { &AsyncWatcher(ptr) => ptr }
+#[cfg(test)]
+mod test_remote {
+    use std::cell::Cell;
+    use std::rt::test::*;
+    use std::rt::thread::Thread;
+    use std::rt::tube::Tube;
+    use std::rt::rtio::EventLoop;
+    use std::rt::local::Local;
+    use std::rt::sched::Scheduler;
+
+    #[test]
+    fn test_uv_remote() {
+        do run_in_mt_newsched_task {
+            let mut tube = Tube::new();
+            let tube_clone = tube.clone();
+            let remote_cell = Cell::new_empty();
+            do Local::borrow |sched: &mut Scheduler| {
+                let tube_clone = tube_clone.clone();
+                let tube_clone_cell = Cell::new(tube_clone);
+                let remote = do sched.event_loop.remote_callback {
+                    // This could be called multiple times
+                    if !tube_clone_cell.is_empty() {
+                        tube_clone_cell.take().send(1);
+                    }
+                };
+                remote_cell.put_back(remote);
+            }
+            let thread = do Thread::start {
+                remote_cell.take().fire();
+            };
+
+            assert!(tube.recv() == 1);
+            thread.join();
+        }
     }
 }
 

branches/try2/src/librustuv/idle.rs

@@ -8,70 +8,98 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use std::libc::c_int;
+use std::cast;
+use std::libc::{c_int, c_void};
 
 use uvll;
-use super::{Watcher, Loop, NativeHandle, IdleCallback, status_to_maybe_uv_error};
-
-pub struct IdleWatcher(*uvll::uv_idle_t);
-impl Watcher for IdleWatcher { }
+use super::{Loop, UvHandle};
+use std::rt::rtio::{Callback, PausibleIdleCallback};
+
+pub struct IdleWatcher {
+    handle: *uvll::uv_idle_t,
+    idle_flag: bool,
+    closed: bool,
+    callback: Option<~Callback>,
+}
 
 impl IdleWatcher {
-    pub fn new(loop_: &mut Loop) -> IdleWatcher {
+    pub fn new(loop_: &mut Loop) -> ~IdleWatcher {
+        let handle = UvHandle::alloc(None::<IdleWatcher>, uvll::UV_IDLE);
+        assert_eq!(unsafe {
+            uvll::idle_init(loop_.native_handle(), handle)
+        }, 0);
+        let me = ~IdleWatcher {
+            handle: handle,
+            idle_flag: false,
+            closed: false,
+            callback: None,
+        };
+        return me.install();
+    }
+
+    pub fn onetime(loop_: &mut Loop, f: proc()) {
+        let handle = UvHandle::alloc(None::<IdleWatcher>, uvll::UV_IDLE);
         unsafe {
-            let handle = uvll::malloc_handle(uvll::UV_IDLE);
-            assert!(handle.is_not_null());
-            assert_eq!(uvll::uv_idle_init(loop_.native_handle(), handle), 0);
-            let mut watcher: IdleWatcher = NativeHandle::from_native_handle(handle);
-            watcher.install_watcher_data();
-            return watcher
+            assert_eq!(uvll::idle_init(loop_.native_handle(), handle), 0);
+            let data: *c_void = cast::transmute(~f);
+            uvll::set_data_for_uv_handle(handle, data);
+            assert_eq!(uvll::idle_start(handle, onetime_cb), 0)
         }
-    }
 
-    pub fn start(&mut self, cb: IdleCallback) {
-        {
-            let data = self.get_watcher_data();
-            data.idle_cb = Some(cb);
+        extern fn onetime_cb(handle: *uvll::uv_idle_t, status: c_int) {
+            assert_eq!(status, 0);
+            unsafe {
+                let data = uvll::get_data_for_uv_handle(handle);
+                let f: ~proc() = cast::transmute(data);
+                (*f)();
+                uvll::idle_stop(handle);
+                uvll::close(handle, close_cb);
+            }
         }
 
-        unsafe {
-            assert_eq!(uvll::uv_idle_start(self.native_handle(), idle_cb), 0)
+        extern fn close_cb(handle: *uvll::uv_handle_t) {
+            unsafe { uvll::free_handle(handle) }
         }
     }
+}
 
-    pub fn restart(&mut self) {
-        unsafe {
-            assert!(self.get_watcher_data().idle_cb.is_some());
-            assert_eq!(uvll::uv_idle_start(self.native_handle(), idle_cb), 0)
+impl PausibleIdleCallback for IdleWatcher {
+    fn start(&mut self, cb: ~Callback) {
+        assert!(self.callback.is_none());
+        self.callback = Some(cb);
+        assert_eq!(unsafe { uvll::idle_start(self.handle, idle_cb) }, 0)
+        self.idle_flag = true;
+    }
+    fn pause(&mut self) {
+        if self.idle_flag == true {
+            assert_eq!(unsafe {uvll::idle_stop(self.handle) }, 0);
+            self.idle_flag = false;
         }
     }
-
-    pub fn stop(&mut self) {
-        // NB: Not resetting the Rust idle_cb to None here because `stop` is
-        // likely called from *within* the idle callback, causing a use after
-        // free
-
-        unsafe {
-            assert_eq!(uvll::uv_idle_stop(self.native_handle()), 0);
+    fn resume(&mut self) {
+        if self.idle_flag == false {
+            assert_eq!(unsafe { uvll::idle_start(self.handle, idle_cb) }, 0)
+            self.idle_flag = true;
+        }
+    }
+    fn close(&mut self) {
+        self.pause();
+        if !self.closed {
+            self.closed = true;
+            self.close_async_();
         }
     }
 }
 
-impl NativeHandle<*uvll::uv_idle_t> for IdleWatcher {
-    fn from_native_handle(handle: *uvll::uv_idle_t) -> IdleWatcher {
-        IdleWatcher(handle)
-    }
-    fn native_handle(&self) -> *uvll::uv_idle_t {
-        match self { &IdleWatcher(ptr) => ptr }
-    }
+impl UvHandle<uvll::uv_idle_t> for IdleWatcher {
+    fn uv_handle(&self) -> *uvll::uv_idle_t { self.handle }
 }
 
 extern fn idle_cb(handle: *uvll::uv_idle_t, status: c_int) {
-    let mut idle_watcher: IdleWatcher = NativeHandle::from_native_handle(handle);
-    let data = idle_watcher.get_watcher_data();
-    let cb: &IdleCallback = data.idle_cb.get_ref();
-    let status = status_to_maybe_uv_error(status);
-    (*cb)(idle_watcher, status);
+    assert_eq!(status, 0);
+    let idle: &mut IdleWatcher = unsafe { UvHandle::from_uv_handle(&handle) };
+    assert!(idle.callback.is_some());
+    idle.callback.get_mut_ref().call();
 }
 
 #[cfg(test)]