rustuv: Implement clone/close_accept

This commits implements {Tcp,Unix}Acceptor::{clone,close_accept} methods for
all of librustuv.

This implementation rewrites much of Access, AccessTimeout, and AcceptTimeout to
have type parameter for shared state that all acceptors share (a shared queue of
sockets). The incoming/outgoing channels have been removed as all timeouts and
such are now managed on the event loop rather than concurrently.

Author: alexcrichton

src/librustuv/access.rs

@@ -22,38 +22,40 @@ use std::cell::UnsafeCell;
 
 use homing::HomingMissile;
 
-pub struct Access {
-    inner: Arc<UnsafeCell<Inner>>,
+pub struct Access<T> {
+    inner: Arc<UnsafeCell<Inner<T>>>,
 }
 
-pub struct Guard<'a> {
-    access: &'a mut Access,
+pub struct Guard<'a, T> {
+    access: &'a mut Access<T>,
     missile: Option<HomingMissile>,
 }
 
-struct Inner {
+struct Inner<T> {
     queue: Vec<(BlockedTask, uint)>,
     held: bool,
     closed: bool,
+    data: T,
 }
 
-impl Access {
-    pub fn new() -> Access {
+impl<T: Send> Access<T> {
+    pub fn new(data: T) -> Access<T> {
         Access {
             inner: Arc::new(UnsafeCell::new(Inner {
                 queue: vec![],
                 held: false,
                 closed: false,
+                data: data,
             }))
         }
     }
 
     pub fn grant<'a>(&'a mut self, token: uint,
-                     missile: HomingMissile) -> Guard<'a> {
+                     missile: HomingMissile) -> Guard<'a, T> {
         // This unsafety is actually OK because the homing missile argument
         // guarantees that we're on the same event loop as all the other objects
         // attempting to get access granted.
-        let inner: &mut Inner = unsafe { &mut *self.inner.get() };
+        let inner = unsafe { &mut *self.inner.get() };
 
         if inner.held {
             let t: Box<Task> = Local::take();
@@ -69,6 +71,15 @@ impl Access {
         Guard { access: self, missile: Some(missile) }
     }
 
+    pub fn unsafe_get(&self) -> *mut T {
+        unsafe { &mut (*self.inner.get()).data  as *mut _ }
+    }
+
+    // Safe version which requires proof that you are on the home scheduler.
+    pub fn get_mut<'a>(&'a mut self, _missile: &HomingMissile) -> &'a mut T {
+        unsafe { &mut *self.unsafe_get() }
+    }
+
     pub fn close(&self, _missile: &HomingMissile) {
         // This unsafety is OK because with a homing missile we're guaranteed to
         // be the only task looking at the `closed` flag (and are therefore
@@ -82,35 +93,55 @@ impl Access {
     // is only safe to invoke while on the home event loop, and there is no
     // guarantee that this i being invoked on the home event loop.
     pub unsafe fn dequeue(&mut self, token: uint) -> Option<BlockedTask> {
-        let inner: &mut Inner = &mut *self.inner.get();
+        let inner = &mut *self.inner.get();
         match inner.queue.iter().position(|&(_, t)| t == token) {
             Some(i) => Some(inner.queue.remove(i).unwrap().val0()),
             None => None,
         }
     }
+
+    /// Test whether this access is closed, using a homing missile to prove
+    /// that it's safe
+    pub fn is_closed(&self, _missile: &HomingMissile) -> bool {
+        unsafe { (*self.inner.get()).closed }
+    }
 }
 
-impl Clone for Access {
-    fn clone(&self) -> Access {
+impl<T: Send> Clone for Access<T> {
+    fn clone(&self) -> Access<T> {
         Access { inner: self.inner.clone() }
     }
 }
 
-impl<'a> Guard<'a> {
+impl<'a, T: Send> Guard<'a, T> {
     pub fn is_closed(&self) -> bool {
         // See above for why this unsafety is ok, it just applies to the read
         // instead of the write.
         unsafe { (*self.access.inner.get()).closed }
     }
 }
 
+impl<'a, T: Send> Deref<T> for Guard<'a, T> {
+    fn deref<'a>(&'a self) -> &'a T {
+        // A guard represents exclusive access to a piece of data, so it's safe
+        // to hand out shared and mutable references
+        unsafe { &(*self.access.inner.get()).data }
+    }
+}
+
+impl<'a, T: Send> DerefMut<T> for Guard<'a, T> {
+    fn deref_mut<'a>(&'a mut self) -> &'a mut T {
+        unsafe { &mut (*self.access.inner.get()).data }
+    }
+}
+
 #[unsafe_destructor]
-impl<'a> Drop for Guard<'a> {
+impl<'a, T> Drop for Guard<'a, T> {
     fn drop(&mut self) {
         // This guard's homing missile is still armed, so we're guaranteed to be
         // on the same I/O event loop, so this unsafety should be ok.
         assert!(self.missile.is_some());
-        let inner: &mut Inner = unsafe {
+        let inner: &mut Inner<T> = unsafe {
             mem::transmute(self.access.inner.get())
         };
 
@@ -133,7 +164,8 @@ impl<'a> Drop for Guard<'a> {
     }
 }
 
-impl Drop for Inner {
+#[unsafe_destructor]
+impl<T> Drop for Inner<T> {
     fn drop(&mut self) {
         assert!(!self.held);
         assert_eq!(self.queue.len(), 0);

src/librustuv/net.rs

@@ -22,7 +22,7 @@ use stream::StreamWatcher;
 use super::{Loop, Request, UvError, Buf, status_to_io_result,
             uv_error_to_io_error, UvHandle, slice_to_uv_buf,
             wait_until_woken_after, wakeup};
-use timeout::{AccessTimeout, AcceptTimeout, ConnectCtx};
+use timeout::{AccessTimeout, ConnectCtx, AcceptTimeout};
 use uvio::UvIoFactory;
 use uvll;
 
@@ -158,20 +158,20 @@ pub struct TcpWatcher {
     // stream object, so we use these access guards in order to arbitrate among
     // multiple concurrent reads and writes. Note that libuv *can* read and
     // write simultaneously, it just can't read and read simultaneously.
-    read_access: AccessTimeout,
-    write_access: AccessTimeout,
+    read_access: AccessTimeout<()>,
+    write_access: AccessTimeout<()>,
 }
 
 pub struct TcpListener {
     home: HomeHandle,
-    handle: *mut uvll::uv_pipe_t,
-    outgoing: Sender<Result<Box<rtio::RtioTcpStream + Send>, IoError>>,
-    incoming: Receiver<Result<Box<rtio::RtioTcpStream + Send>, IoError>>,
+    handle: *mut uvll::uv_tcp_t,
 }
 
 pub struct TcpAcceptor {
-    listener: Box<TcpListener>,
-    timeout: AcceptTimeout,
+    home: HomeHandle,
+    handle: *mut uvll::uv_tcp_t,
+    access: AcceptTimeout<Box<rtio::RtioTcpStream + Send>>,
+    refcount: Refcount,
 }
 
 // TCP watchers (clients/streams)
@@ -192,8 +192,8 @@ impl TcpWatcher {
             handle: handle,
             stream: StreamWatcher::new(handle, true),
             refcount: Refcount::new(),
-            read_access: AccessTimeout::new(),
-            write_access: AccessTimeout::new(),
+            read_access: AccessTimeout::new(()),
+            write_access: AccessTimeout::new(()),
         }
     }
 
@@ -291,7 +291,7 @@ impl rtio::RtioTcpStream for TcpWatcher {
         let task = {
             let m = self.fire_homing_missile();
             self.read_access.access.close(&m);
-    self.stream.cancel_read(uvll::EOF as libc::ssize_t)
+            self.stream.cancel_read(uvll::EOF as libc::ssize_t)
         };
         let _ = task.map(|t| t.reawaken());
         Ok(())
@@ -354,12 +354,9 @@ impl TcpListener {
         assert_eq!(unsafe {
             uvll::uv_tcp_init(io.uv_loop(), handle)
         }, 0);
-        let (tx, rx) = channel();
         let l = box TcpListener {
             home: io.make_handle(),
             handle: handle,
-            outgoing: tx,
-            incoming: rx,
         };
         let mut storage = unsafe { mem::zeroed() };
         let _len = addr_to_sockaddr(address, &mut storage);
@@ -392,15 +389,19 @@ impl rtio::RtioSocket for TcpListener {
 impl rtio::RtioTcpListener for TcpListener {
     fn listen(self: Box<TcpListener>)
               -> Result<Box<rtio::RtioTcpAcceptor + Send>, IoError> {
+        let _m = self.fire_homing_missile();
+
         // create the acceptor object from ourselves
-        let mut acceptor = box TcpAcceptor {
-            listener: self,
-            timeout: AcceptTimeout::new(),
-        };
+        let acceptor = (box TcpAcceptor {
+            handle: self.handle,
+            home: self.home.clone(),
+            access: AcceptTimeout::new(),
+            refcount: Refcount::new(),
+        }).install();
+        self.handle = 0 as *mut _;
 
-        let _m = acceptor.fire_homing_missile();
         // FIXME: the 128 backlog should be configurable
-        match unsafe { uvll::uv_listen(acceptor.listener.handle, 128, listen_cb) } {
+        match unsafe { uvll::uv_listen(acceptor.handle, 128, listen_cb) } {
             0 => Ok(acceptor as Box<rtio::RtioTcpAcceptor + Send>),
             n => Err(uv_error_to_io_error(UvError(n))),
         }
@@ -409,7 +410,7 @@ impl rtio::RtioTcpListener for TcpListener {
 
 extern fn listen_cb(server: *mut uvll::uv_stream_t, status: c_int) {
     assert!(status != uvll::ECANCELED);
-    let tcp: &mut TcpListener = unsafe { UvHandle::from_uv_handle(&server) };
+    let tcp: &mut TcpAcceptor = unsafe { UvHandle::from_uv_handle(&server) };
     let msg = match status {
         0 => {
             let loop_ = Loop::wrap(unsafe {
@@ -421,11 +422,15 @@ extern fn listen_cb(server: *mut uvll::uv_stream_t, status: c_int) {
         }
         n => Err(uv_error_to_io_error(UvError(n)))
     };
-    tcp.outgoing.send(msg);
+
+    // If we're running then we have exclusive access, so the unsafe_get() is ok
+    unsafe { tcp.access.push(msg); }
 }
 
 impl Drop for TcpListener {
     fn drop(&mut self) {
+        if self.handle.is_null() { return }
+
         let _m = self.fire_homing_missile();
         self.close();
     }
@@ -434,40 +439,68 @@ impl Drop for TcpListener {
 // TCP acceptors (bound servers)
 
 impl HomingIO for TcpAcceptor {
-    fn home<'r>(&'r mut self) -> &'r mut HomeHandle { self.listener.home() }
+    fn home<'r>(&'r mut self) -> &'r mut HomeHandle { &mut self.home }
 }
 
 impl rtio::RtioSocket for TcpAcceptor {
     fn socket_name(&mut self) -> Result<rtio::SocketAddr, IoError> {
         let _m = self.fire_homing_missile();
-        socket_name(Tcp, self.listener.handle)
+        socket_name(Tcp, self.handle)
     }
 }
 
+impl UvHandle<uvll::uv_tcp_t> for TcpAcceptor {
+    fn uv_handle(&self) -> *mut uvll::uv_tcp_t { self.handle }
+}
+
 impl rtio::RtioTcpAcceptor for TcpAcceptor {
     fn accept(&mut self) -> Result<Box<rtio::RtioTcpStream + Send>, IoError> {
-        self.timeout.accept(&self.listener.incoming)
+        let m = self.fire_homing_missile();
+        let loop_ = self.uv_loop();
+        self.access.accept(m, &loop_)
     }
 
     fn accept_simultaneously(&mut self) -> Result<(), IoError> {
         let _m = self.fire_homing_missile();
         status_to_io_result(unsafe {
-            uvll::uv_tcp_simultaneous_accepts(self.listener.handle, 1)
+            uvll::uv_tcp_simultaneous_accepts(self.handle, 1)
         })
     }
 
     fn dont_accept_simultaneously(&mut self) -> Result<(), IoError> {
         let _m = self.fire_homing_missile();
         status_to_io_result(unsafe {
-            uvll::uv_tcp_simultaneous_accepts(self.listener.handle, 0)
+            uvll::uv_tcp_simultaneous_accepts(self.handle, 0)
         })
     }
 
     fn set_timeout(&mut self, ms: Option<u64>) {
         let _m = self.fire_homing_missile();
-        match ms {
-            None => self.timeout.clear(),
-            Some(ms) => self.timeout.set_timeout(ms, &mut *self.listener),
+        let loop_ = self.uv_loop();
+        self.access.set_timeout(ms, &loop_, &self.home);
+    }
+
+    fn clone(&self) -> Box<rtio::RtioTcpAcceptor + Send> {
+        box TcpAcceptor {
+            refcount: self.refcount.clone(),
+            home: self.home.clone(),
+            handle: self.handle,
+            access: self.access.clone(),
+        } as Box<rtio::RtioTcpAcceptor + Send>
+    }
+
+    fn close_accept(&mut self) -> Result<(), IoError> {
+        let m = self.fire_homing_missile();
+        self.access.close(m);
+        Ok(())
+    }
+}
+
+impl Drop for TcpAcceptor {
+    fn drop(&mut self) {
+        let _m = self.fire_homing_missile();
+        if self.refcount.decrement() {
+            self.close();
         }
     }
 }
@@ -482,8 +515,8 @@ pub struct UdpWatcher {
 
     // See above for what these fields are
     refcount: Refcount,
-    read_access: AccessTimeout,
-    write_access: AccessTimeout,
+    read_access: AccessTimeout<()>,
+    write_access: AccessTimeout<()>,
 
     blocked_sender: Option<BlockedTask>,
 }
@@ -507,8 +540,8 @@ impl UdpWatcher {
             handle: unsafe { uvll::malloc_handle(uvll::UV_UDP) },
             home: io.make_handle(),
             refcount: Refcount::new(),
-            read_access: AccessTimeout::new(),
-            write_access: AccessTimeout::new(),
+            read_access: AccessTimeout::new(()),
+            write_access: AccessTimeout::new(()),
             blocked_sender: None,
         };
         assert_eq!(unsafe {