core::rt: Begin implementing TcpStream

This ended up touching a lot of code related to error handling.

Author: brson

src/libcore/macros.rs

@@ -30,6 +30,14 @@ macro_rules! rtdebug (
     ($( $arg:expr),+) => ( $(let _ = $arg)*; )
 )
 
+macro_rules! rtassert (
+    ( $arg:expr ) => ( {
+        if !$arg {
+            abort!("assertion failed: %s", stringify!($arg));
+        }
+    } )
+)
+
 macro_rules! abort(
     ($( $msg:expr),+) => ( {
         rtdebug!($($msg),+);

src/libcore/rt/io/mod.rs

@@ -252,7 +252,9 @@ pub use self::stdio::println;
 
 pub use self::file::FileStream;
 pub use self::net::ip::IpAddr;
+#[cfg(not(stage0))]
 pub use self::net::tcp::TcpListener;
+#[cfg(not(stage0))]
 pub use self::net::tcp::TcpStream;
 pub use self::net::udp::UdpStream;
 
@@ -266,6 +268,7 @@ pub mod file;
 
 /// Synchronous, non-blocking network I/O.
 pub mod net {
+    #[cfg(not(stage0))]
     pub mod tcp;
     pub mod udp;
     pub mod ip;
@@ -326,12 +329,14 @@ pub struct IoError {
 
 #[deriving(Eq)]
 pub enum IoErrorKind {
+    PreviousIoError,
+    OtherIoError,
+    EndOfFile,
     FileNotFound,
-    FilePermission,
+    PermissionDenied,
     ConnectionFailed,
     Closed,
-    OtherIoError,
-    PreviousIoError
+    ConnectionRefused,
 }
 
 // XXX: Can't put doc comments on macros

src/libcore/rt/io/net/tcp.rs

@@ -8,63 +8,179 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use prelude::*;
-use super::super::*;
-use super::ip::IpAddr;
+use option::{Option, Some, None};
+use result::{Result, Ok, Err};
+use ops::Drop;
+use rt::sched::local_sched::unsafe_borrow_io;
+use rt::io::net::ip::IpAddr;
+use rt::io::{Reader, Writer, Listener};
+use rt::io::io_error;
+use rt::rtio;
+use rt::rtio::{IoFactory, TcpListener, Stream};
 
-pub struct TcpStream;
+pub struct TcpStream {
+    rtstream: ~rtio::StreamObject
+}
 
 impl TcpStream {
-    pub fn connect(_addr: IpAddr) -> Option<TcpStream> {
-        fail!()
+    fn new(s: ~rtio::StreamObject) -> TcpStream {
+        TcpStream {
+            rtstream: s
+        }
+    }
+
+    pub fn connect(addr: IpAddr) -> Option<TcpStream> {
+        let stream = unsafe {
+            rtdebug!("borrowing io to connect");
+            let io = unsafe_borrow_io();
+            rtdebug!("about to connect");
+            io.connect(addr)
+        };
+
+        match stream {
+            Ok(s) => {
+                Some(TcpStream::new(s))
+            }
+            Err(ioerr) => {
+                rtdebug!("failed to connect: %?", ioerr);
+                io_error::cond.raise(ioerr);
+                return None;
+            }
+        }
     }
 }
 
 impl Reader for TcpStream {
-    fn read(&mut self, _buf: &mut [u8]) -> Option<uint> { fail!() }
+    fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
+        let bytes_read = self.rtstream.read(buf);
+        match bytes_read {
+            Ok(read) => Some(read),
+            Err(_) => {
+                abort!("TODO");
+            }
+        }
+    }
 
     fn eof(&mut self) -> bool { fail!() }
 }
 
 impl Writer for TcpStream {
-    fn write(&mut self, _buf: &[u8]) { fail!() }
+    fn write(&mut self, buf: &[u8]) {
+        let res = self.rtstream.write(buf);
+        match res {
+            Ok(_) => (),
+            Err(_) => {
+                abort!("TODO");
+            }
+        }
+    }
 
     fn flush(&mut self) { fail!() }
 }
 
-pub struct TcpListener;
+impl Drop for TcpStream {
+    fn finalize(&self) {
+        self.rtstream.close();
+    }
+}
+
+pub struct TcpListener {
+    rtlistener: ~rtio::TcpListenerObject
+}
 
 impl TcpListener {
-    pub fn bind(_addr: IpAddr) -> Option<TcpListener> {
-        fail!()
+    pub fn bind(addr: IpAddr) -> Option<TcpListener> {
+        let listener = unsafe { unsafe_borrow_io().bind(addr) };
+        match listener {
+            Ok(l) => {
+                Some(TcpListener {
+                    rtlistener: l
+                })
+            }
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+                return None;
+            }
+        }
     }
 }
 
 impl Listener<TcpStream> for TcpListener {
-    fn accept(&mut self) -> Option<TcpStream> { fail!() }
+    fn accept(&mut self) -> Option<TcpStream> {
+        let rtstream = self.rtlistener.listen();
+        match rtstream {
+            Some(s) => {
+                Some(TcpStream::new(s))
+            }
+            None => {
+                abort!("TODO");
+            }
+        }
+    }
+}
+
+impl Drop for TcpListener {
+    fn finalize(&self) {
+        self.rtlistener.close();
+    }
 }
 
 #[cfg(test)]
 mod test {
+    use super::*;
+    use rt::test::*;
+    use rt::io::net::ip::Ipv4;
+    use rt::io::*;
+
+    #[test]
+    fn bind_error() {
+        do run_in_newsched_task {
+            let mut called = false;
+            do io_error::cond.trap(|e| {
+                assert!(e.kind == PermissionDenied);
+                called = true;
+            }).in {
+                let addr = Ipv4(0, 0, 0, 0, 1);
+                let listener = TcpListener::bind(addr);
+                assert!(listener.is_none());
+            }
+            assert!(called);
+        }
+    }
+
+    #[test]
+    fn connect_error() {
+        do run_in_newsched_task {
+            let mut called = false;
+            do io_error::cond.trap(|e| {
+                assert!(e.kind == ConnectionRefused);
+                called = true;
+            }).in {
+                let addr = Ipv4(0, 0, 0, 0, 1);
+                let stream = TcpStream::connect(addr);
+                assert!(stream.is_none());
+            }
+            assert!(called);
+        }
+    }
 
-    #[test] #[ignore]
+    #[test]
     fn smoke_test() {
-        /*do run_in_newsched_task {
+        do run_in_newsched_task {
             let addr = next_test_ip4();
 
-            do spawn_immediately {
-                let listener = TcpListener::bind(addr);
-                do listener.accept() {
-                    let mut buf = [0];
-                    listener.read(buf);
-                    assert!(buf[0] == 99);
-                }
+            do spawntask_immediately {
+                let mut listener = TcpListener::bind(addr);
+                let mut stream = listener.accept();
+                let mut buf = [0];
+                stream.read(buf);
+                assert!(buf[0] == 99);
             }
 
-            do spawn_immediately {
-                let stream = TcpStream::connect(addr);
+            do spawntask_immediately {
+                let mut stream = TcpStream::connect(addr);
                 stream.write([99]);
             }
-        }*/
+        }
     }
 }

src/libcore/rt/local_services.rs

@@ -177,7 +177,8 @@ pub unsafe fn unsafe_borrow_local_services() -> &mut LocalServices {
             transmute_mut_region(&mut task.local_services)
         }
         None => {
-            fail!(~"no local services for schedulers yet")
+            // Don't fail. Infinite recursion
+            abort!("no local services for schedulers yet")
         }
     }
 }

src/libcore/rt/mod.rs

@@ -8,15 +8,23 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-/*! The Rust runtime, including the scheduler and I/O interface */
+/*! The Rust runtime, including the scheduler and I/O interface
+
+# XXX
+
+* Unsafe uses of borrowed pointers should just use unsafe pointers
+* Unwinding is not wired up correctly
+
+*/
+
 
 #[doc(hidden)];
 
 use libc::c_char;
 
 #[path = "sched/mod.rs"]
 mod sched;
-mod rtio;
+pub mod rtio;
 pub mod uvll;
 mod uvio;
 #[path = "uv/mod.rs"]

src/libcore/rt/rtio.rs

@@ -11,6 +11,7 @@
 use option::*;
 use result::*;
 
+use rt::io::IoError;
 use super::io::net::ip::IpAddr;
 
 // XXX: ~object doesn't work currently so these are some placeholder
@@ -28,8 +29,8 @@ pub trait EventLoop {
 }
 
 pub trait IoFactory {
-    fn connect(&mut self, addr: IpAddr) -> Option<~StreamObject>;
-    fn bind(&mut self, addr: IpAddr) -> Option<~TcpListenerObject>;
+    fn connect(&mut self, addr: IpAddr) -> Result<~StreamObject, IoError>;
+    fn bind(&mut self, addr: IpAddr) -> Result<~TcpListenerObject, IoError>;
 }
 
 pub trait TcpListener {

src/libcore/rt/sched/local_sched.rs

@@ -13,18 +13,21 @@
 use prelude::*;
 use ptr::mut_null;
 use libc::c_void;
-use cast::transmute;
+use cast;
+use cell::Cell;
 
 use super::Scheduler;
 use super::super::rtio::IoFactoryObject;
 use tls = super::super::thread_local_storage;
+use unstable::finally::Finally;
+
 #[cfg(test)] use super::super::uvio::UvEventLoop;
 
 /// Give the Scheduler to thread-local storage
 pub fn put(sched: ~Scheduler) {
     unsafe {
         let key = tls_key();
-        let void_sched: *mut c_void = transmute::<~Scheduler, *mut c_void>(sched);
+        let void_sched: *mut c_void = cast::transmute(sched);
         tls::set(key, void_sched);
     }
 }
@@ -34,8 +37,8 @@ pub fn take() -> ~Scheduler {
     unsafe {
         let key = tls_key();
         let void_sched: *mut c_void = tls::get(key);
-        assert!(void_sched.is_not_null());
-        let sched = transmute::<*mut c_void, ~Scheduler>(void_sched);
+        rtassert!(void_sched.is_not_null());
+        let sched: ~Scheduler = cast::transmute(void_sched);
         tls::set(key, mut_null());
         return sched;
     }
@@ -55,8 +58,18 @@ pub fn exists() -> bool {
 /// While the scheduler is borrowed it is not available in TLS.
 pub fn borrow(f: &fn(&mut Scheduler)) {
     let mut sched = take();
-    f(sched);
-    put(sched);
+
+    // XXX: Need a different abstraction from 'finally' here to avoid unsafety
+    unsafe {
+        let unsafe_sched = cast::transmute_mut_region(&mut *sched);
+        let sched = Cell(sched);
+        
+        do (|| {
+            f(unsafe_sched);
+        }).finally {
+            put(sched.take());
+        }
+    }
 }
 
 /// Borrow a mutable reference to the thread-local Scheduler
@@ -68,11 +81,11 @@ pub fn borrow(f: &fn(&mut Scheduler)) {
 pub unsafe fn unsafe_borrow() -> &mut Scheduler {
     let key = tls_key();
     let mut void_sched: *mut c_void = tls::get(key);
-    assert!(void_sched.is_not_null());
+    rtassert!(void_sched.is_not_null());
     {
         let void_sched_ptr = &mut void_sched;
         let sched: &mut ~Scheduler = {
-            transmute::<&mut *mut c_void, &mut ~Scheduler>(void_sched_ptr)
+            cast::transmute::<&mut *mut c_void, &mut ~Scheduler>(void_sched_ptr)
         };
         let sched: &mut Scheduler = &mut **sched;
         return sched;
@@ -91,7 +104,7 @@ fn tls_key() -> tls::Key {
 fn maybe_tls_key() -> Option<tls::Key> {
     unsafe {
         let key: *mut c_void = rust_get_sched_tls_key();
-        let key: &mut tls::Key = transmute(key);
+        let key: &mut tls::Key = cast::transmute(key);
         let key = *key;
         // Check that the key has been initialized.