---

yaml
---
r: 154608
b: refs/heads/try2
c: cb8df7a
h: refs/heads/master
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: 110168de2a7b529a7c4839ca1e19c4c42f68be12
+refs/heads/try2: cb8df7a8e3c53f16d22f76da50e6e2e4734bdf62
 refs/heads/dist-snap: ba4081a5a8573875fed17545846f6f6902c8ba8d
 refs/tags/release-0.2: c870d2dffb391e14efb05aa27898f1f6333a9596
 refs/tags/release-0.3: b5f0d0f648d9a6153664837026ba1be43d3e2503

branches/try2/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);

branches/try2/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 {