Implement native TCP I/O

5 files changed, 467 insertions, 22 deletions

diff --git a/src/libnative/io/file.rs b/src/libnative/io/file.rs
index 543132cce15..6197bd70c76 100644
--- a/src/libnative/io/file.rs
+++ b/src/libnative/io/file.rs
@@ -26,7 +26,7 @@ use super::IoResult;
 #[cfg(windows)] use std::ptr;
 #[cfg(windows)] use std::str;
 
-fn keep_going(data: &[u8], f: |*u8, uint| -> i64) -> i64 {
+pub fn keep_going(data: &[u8], f: |*u8, uint| -> i64) -> i64 {
     #[cfg(windows)] static eintr: int = 0; // doesn't matter
     #[cfg(not(windows))] static eintr: int = libc::EINTR as int;
 
@@ -92,7 +92,7 @@ impl FileDesc {
             Ok(ret as uint)
         }
     }
-    fn inner_write(&mut self, buf: &[u8]) -> Result<(), IoError> {
+    pub fn inner_write(&mut self, buf: &[u8]) -> Result<(), IoError> {
         #[cfg(windows)] type wlen = libc::c_uint;
         #[cfg(not(windows))] type wlen = libc::size_t;
         let ret = keep_going(buf, |buf, len| {
@@ -106,6 +106,8 @@ impl FileDesc {
             Ok(())
         }
     }
+
+    pub fn fd(&self) -> fd_t { self.fd }
 }
 
 impl io::Reader for FileDesc {
@@ -902,7 +904,7 @@ pub fn utime(p: &CString, atime: u64, mtime: u64) -> IoResult<()> {
 
 #[cfg(test)]
 mod tests {
-    use super::{CFile, FileDesc, CloseFd};
+    use super::{CFile, FileDesc};
     use std::io;
     use std::libc;
     use std::os;
diff --git a/src/libnative/io/mod.rs b/src/libnative/io/mod.rs
index 56096142349..9e76dea5ebf 100644
--- a/src/libnative/io/mod.rs
+++ b/src/libnative/io/mod.rs
@@ -44,6 +44,7 @@ pub use self::process::Process;
 // Native I/O implementations
 pub mod file;
 pub mod process;
+pub mod net;
 
 type IoResult<T> = Result<T, IoError>;
 
@@ -60,7 +61,20 @@ fn translate_error(errno: i32, detail: bool) -> IoError {
     fn get_err(errno: i32) -> (io::IoErrorKind, &'static str) {
         match errno {
             libc::EOF => (io::EndOfFile, "end of file"),
-            _ => (io::OtherIoError, "unknown error"),
+            libc::WSAECONNREFUSED => (io::ConnectionRefused, "connection refused"),
+            libc::WSAECONNRESET => (io::ConnectionReset, "connection reset"),
+            libc::WSAEACCES => (io::PermissionDenied, "permission denied"),
+            libc::WSAEWOULDBLOCK =>
+                (io::ResourceUnavailable, "resource temporarily unavailable"),
+            libc::WSAENOTCONN => (io::NotConnected, "not connected"),
+            libc::WSAECONNABORTED => (io::ConnectionAborted, "connection aborted"),
+            libc::WSAEADDRNOTAVAIL => (io::ConnectionRefused, "address not available"),
+            libc::WSAEADDRINUSE => (io::ConnectionRefused, "address in use"),
+
+            x => {
+                debug!("ignoring {}: {}", x, os::last_os_error());
+                (io::OtherIoError, "unknown error")
+            }
         }
     }
 
@@ -69,13 +83,25 @@ fn translate_error(errno: i32, detail: bool) -> IoError {
         // XXX: this should probably be a bit more descriptive...
         match errno {
             libc::EOF => (io::EndOfFile, "end of file"),
+            libc::ECONNREFUSED => (io::ConnectionRefused, "connection refused"),
+            libc::ECONNRESET => (io::ConnectionReset, "connection reset"),
+            libc::EPERM | libc::EACCES =>
+                (io::PermissionDenied, "permission denied"),
+            libc::EPIPE => (io::BrokenPipe, "broken pipe"),
+            libc::ENOTCONN => (io::NotConnected, "not connected"),
+            libc::ECONNABORTED => (io::ConnectionAborted, "connection aborted"),
+            libc::EADDRNOTAVAIL => (io::ConnectionRefused, "address not available"),
+            libc::EADDRINUSE => (io::ConnectionRefused, "address in use"),
 
             // These two constants can have the same value on some systems, but
             // different values on others, so we can't use a match clause
             x if x == libc::EAGAIN || x == libc::EWOULDBLOCK =>
                 (io::ResourceUnavailable, "resource temporarily unavailable"),
 
-            _ => (io::OtherIoError, "unknown error"),
+            x => {
+                debug!("ignoring {}: {}", x, os::last_os_error());
+                (io::OtherIoError, "unknown error")
+            }
         }
     }
 
@@ -121,15 +147,24 @@ fn retry(f: || -> libc::c_int) -> IoResult<libc::c_int> {
 
 /// Implementation of rt::rtio's IoFactory trait to generate handles to the
 /// native I/O functionality.
-pub struct IoFactory;
+pub struct IoFactory {
+    priv cannot_construct_outside_of_this_module: ()
+}
+
+impl IoFactory {
+    pub fn new() -> IoFactory {
+        net::init();
+        IoFactory { cannot_construct_outside_of_this_module: () }
+    }
+}
 
 impl rtio::IoFactory for IoFactory {
     // networking
-    fn tcp_connect(&mut self, _addr: SocketAddr) -> IoResult<~RtioTcpStream> {
-        Err(unimpl())
+    fn tcp_connect(&mut self, addr: SocketAddr) -> IoResult<~RtioTcpStream> {
+        net::TcpStream::connect(addr).map(|s| ~s as ~RtioTcpStream)
     }
-    fn tcp_bind(&mut self, _addr: SocketAddr) -> IoResult<~RtioTcpListener> {
-        Err(unimpl())
+    fn tcp_bind(&mut self, addr: SocketAddr) -> IoResult<~RtioTcpListener> {
+        net::TcpListener::bind(addr).map(|s| ~s as ~RtioTcpListener)
     }
     fn udp_bind(&mut self, _addr: SocketAddr) -> IoResult<~RtioUdpSocket> {
         Err(unimpl())
@@ -217,9 +252,7 @@ impl rtio::IoFactory for IoFactory {
     }
     fn tty_open(&mut self, fd: c_int, _readable: bool) -> IoResult<~RtioTTY> {
         if unsafe { libc::isatty(fd) } != 0 {
-            // Don't ever close the stdio file descriptors, nothing good really
-            // comes of that.
-            Ok(~file::FileDesc::new(fd, fd > libc::STDERR_FILENO) as ~RtioTTY)
+            Ok(~file::FileDesc::new(fd, true) as ~RtioTTY)
         } else {
             Err(IoError {
                 kind: io::MismatchedFileTypeForOperation,
diff --git a/src/libnative/io/net.rs b/src/libnative/io/net.rs
new file mode 100644
index 00000000000..aaa95ce0cfb
--- /dev/null
+++ b/src/libnative/io/net.rs
@@ -0,0 +1,412 @@
+// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use std::cast;
+use std::io::net::ip;
+use std::io;
+use std::libc;
+use std::mem;
+use std::rt::rtio;
+use std::unstable::intrinsics;
+
+use super::IoResult;
+use super::file::keep_going;
+
+#[cfg(windows)] pub type sock_t = libc::SOCKET;
+#[cfg(unix)]    pub type sock_t = super::file::fd_t;
+
+pub struct TcpStream {
+    priv fd: sock_t,
+}
+
+#[cfg(target_endian = "big")] pub fn htons(x: u16) -> u16 { x }
+#[cfg(target_endian = "big")] pub fn ntohs(x: u16) -> u16 { x }
+#[cfg(target_endian = "little")]
+pub fn htons(u: u16) -> u16 {
+    unsafe { intrinsics::bswap16(u as i16) as u16 }
+}
+#[cfg(target_endian = "little")]
+pub fn ntohs(u: u16) -> u16 {
+    unsafe { intrinsics::bswap16(u as i16) as u16 }
+}
+
+fn addr_to_sockaddr(addr: ip::SocketAddr) -> (libc::sockaddr_storage, uint) {
+    unsafe {
+        let storage: libc::sockaddr_storage = intrinsics::init();
+        let len = match addr.ip {
+            ip::Ipv4Addr(a, b, c, d) => {
+                let storage: *mut libc::sockaddr_in = cast::transmute(&storage);
+                (*storage).sin_family = libc::AF_INET as libc::sa_family_t;
+                (*storage).sin_port = htons(addr.port);
+                (*storage).sin_addr.s_addr = (d as u32 << 24) |
+                                             (c as u32 << 16) |
+                                             (b as u32 <<  8) |
+                                             (a as u32 <<  0);
+                mem::size_of::<libc::sockaddr_in>()
+            }
+            ip::Ipv6Addr(a, b, c, d, e, f, g, h) => {
+                let storage: *mut libc::sockaddr_in6 = cast::transmute(&storage);
+                (*storage).sin6_family = libc::AF_INET6 as libc::sa_family_t;
+                (*storage).sin6_port = htons(addr.port);
+                (*storage).sin6_addr.s6_addr[0] = htons(a);
+                (*storage).sin6_addr.s6_addr[1] = htons(b);
+                (*storage).sin6_addr.s6_addr[2] = htons(c);
+                (*storage).sin6_addr.s6_addr[3] = htons(d);
+                (*storage).sin6_addr.s6_addr[4] = htons(e);
+                (*storage).sin6_addr.s6_addr[5] = htons(f);
+                (*storage).sin6_addr.s6_addr[6] = htons(g);
+                (*storage).sin6_addr.s6_addr[7] = htons(h);
+                mem::size_of::<libc::sockaddr_in6>()
+            }
+        };
+        return (storage, len);
+    }
+}
+
+fn socket(addr: ip::SocketAddr) -> IoResult<sock_t> {
+    unsafe {
+        let fam = match addr.ip {
+            ip::Ipv4Addr(..) => libc::AF_INET,
+            ip::Ipv6Addr(..) => libc::AF_INET6,
+        };
+        match libc::socket(fam, libc::SOCK_STREAM, 0) {
+            -1 => Err(super::last_error()),
+            fd => Ok(fd),
+        }
+    }
+}
+
+fn sockname(fd: sock_t,
+            f: extern "system" unsafe fn(sock_t, *mut libc::sockaddr,
+                                         *mut libc::socklen_t) -> libc::c_int)
+    -> IoResult<ip::SocketAddr>
+{
+    let mut storage: libc::sockaddr_storage = unsafe { intrinsics::init() };
+    let mut len = mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
+    unsafe {
+        let storage = &mut storage as *mut libc::sockaddr_storage;
+        let ret = f(fd,
+                    storage as *mut libc::sockaddr,
+                    &mut len as *mut libc::socklen_t);
+        if ret != 0 {
+            return Err(super::last_error())
+        }
+    }
+    match storage.ss_family as libc::c_int {
+        libc::AF_INET => {
+            assert!(len as uint >= mem::size_of::<libc::sockaddr_in>());
+            let storage: &mut libc::sockaddr_in = unsafe {
+                cast::transmute(&mut storage)
+            };
+            let addr = storage.sin_addr.s_addr as u32;
+            let a = (addr >>  0) as u8;
+            let b = (addr >>  8) as u8;
+            let c = (addr >> 16) as u8;
+            let d = (addr >> 24) as u8;
+            Ok(ip::SocketAddr {
+                ip: ip::Ipv4Addr(a, b, c, d),
+                port: ntohs(storage.sin_port),
+            })
+        }
+        libc::AF_INET6 => {
+            assert!(len as uint >= mem::size_of::<libc::sockaddr_in6>());
+            let storage: &mut libc::sockaddr_in6 = unsafe {
+                cast::transmute(&mut storage)
+            };
+            let a = ntohs(storage.sin6_addr.s6_addr[0]);
+            let b = ntohs(storage.sin6_addr.s6_addr[1]);
+            let c = ntohs(storage.sin6_addr.s6_addr[2]);
+            let d = ntohs(storage.sin6_addr.s6_addr[3]);
+            let e = ntohs(storage.sin6_addr.s6_addr[4]);
+            let f = ntohs(storage.sin6_addr.s6_addr[5]);
+            let g = ntohs(storage.sin6_addr.s6_addr[6]);
+            let h = ntohs(storage.sin6_addr.s6_addr[7]);
+            Ok(ip::SocketAddr {
+                ip: ip::Ipv6Addr(a, b, c, d, e, f, g, h),
+                port: ntohs(storage.sin6_port),
+            })
+        }
+        _ => {
+            Err(io::standard_error(io::OtherIoError))
+        }
+    }
+}
+
+#[cfg(unix)]
+pub fn init() {}
+
+#[cfg(windows)]
+pub fn init() {
+    static WSADESCRIPTION_LEN: uint = 256;
+    static WSASYS_STATUS_LEN: uint = 128;
+    struct WSADATA {
+        wVersion: libc::WORD,
+        wHighVersion: libc::WORD,
+        szDescription: [u8, ..WSADESCRIPTION_LEN + 1],
+        szSystemStatus: [u8, ..WSASYS_STATUS_LEN + 1],
+        iMaxSockets: u16,
+        iMaxUdpDg: u16,
+        lpVendorInfo: *u8,
+    }
+    type LPWSADATA = *mut WSADATA;
+
+    #[link(name = "ws2_32")]
+    extern "system" {
+        fn WSAStartup(wVersionRequested: libc::WORD,
+                       lpWSAData: LPWSADATA) -> libc::c_int;
+    }
+
+    unsafe {
+        use std::unstable::mutex::{Mutex, MUTEX_INIT};
+        static mut LOCK: Mutex = MUTEX_INIT;
+        static mut INITIALIZED: bool = false;
+        if INITIALIZED { return }
+        LOCK.lock();
+        if !INITIALIZED {
+            let mut data: WSADATA = intrinsics::init();
+            let ret = WSAStartup(0x202,      // version 2.2
+                                 &mut data);
+            assert_eq!(ret, 0);
+            INITIALIZED = true;
+        }
+        LOCK.unlock();
+    }
+}
+
+impl TcpStream {
+    pub fn connect(addr: ip::SocketAddr) -> IoResult<TcpStream> {
+        unsafe {
+            socket(addr).and_then(|fd| {
+                let (addr, len) = addr_to_sockaddr(addr);
+                let addrp = &addr as *libc::sockaddr_storage;
+                let ret = TcpStream { fd: fd };
+                match libc::connect(fd, addrp as *libc::sockaddr,
+                                    len as libc::socklen_t) {
+                    -1 => Err(super::last_error()),
+                    _ => Ok(ret),
+                }
+            })
+        }
+    }
+
+    pub fn fd(&self) -> sock_t { self.fd }
+
+    fn set_nodelay(&mut self, nodelay: bool) -> IoResult<()> {
+        unsafe {
+            let on = nodelay as libc::c_int;
+            let on = &on as *libc::c_int;
+            super::mkerr_libc(libc::setsockopt(self.fd,
+                                               libc::IPPROTO_TCP,
+                                               libc::TCP_NODELAY,
+                                               on as *libc::c_void,
+                                               mem::size_of::<libc::c_void>()
+                                                    as libc::socklen_t))
+        }
+    }
+
+    fn set_keepalive(&mut self, seconds: Option<uint>) -> IoResult<()> {
+        unsafe {
+            let on = seconds.is_some() as libc::c_int;
+            let on = &on as *libc::c_int;
+            let ret = libc::setsockopt(self.fd,
+                                       libc::SOL_SOCKET,
+                                       libc::SO_KEEPALIVE,
+                                       on as *libc::c_void,
+                                       mem::size_of::<libc::c_void>()
+                                            as libc::socklen_t);
+            if ret != 0 { return Err(super::last_error()) }
+
+            match seconds {
+                Some(n) => self.set_tcp_keepalive(n),
+                None => Ok(())
+            }
+        }
+    }
+
+    #[cfg(target_os = "macos")]
+    unsafe fn set_tcp_keepalive(&mut self, seconds: uint) -> IoResult<()> {
+        let delay = seconds as libc::c_uint;
+        let delay = &delay as *libc::c_uint;
+        let ret = libc::setsockopt(self.fd,
+                                   libc::IPPROTO_TCP,
+                                   libc::TCP_KEEPALIVE,
+                                   delay as *libc::c_void,
+                                   mem::size_of::<libc::c_uint>()
+                                        as libc::socklen_t);
+        super::mkerr_libc(ret)
+    }
+    #[cfg(target_os = "freebsd")]
+    unsafe fn set_tcp_keepalive(&mut self, seconds: uint) -> IoResult<()> {
+        let delay = seconds as libc::c_uint;
+        let delay = &delay as *libc::c_uint;
+        let ret = libc::setsockopt(self.fd,
+                                   libc::IPPROTO_TCP,
+                                   libc::TCP_KEEPIDLE,
+                                   delay as *libc::c_void,
+                                   mem::size_of::<libc::c_uint>()
+                                        as libc::socklen_t);
+        super::mkerr_libc(ret)
+    }
+    #[cfg(not(target_os = "macos"), not(target_os = "freebsd"))]
+    unsafe fn set_tcp_keepalive(&mut self, _seconds: uint) -> IoResult<()> {
+        Ok(())
+    }
+}
+
+#[cfg(windows)] type wrlen = libc::c_int;
+#[cfg(not(windows))] type wrlen = libc::size_t;
+
+impl rtio::RtioTcpStream for TcpStream {
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
+        let ret = keep_going(buf, |buf, len| {
+            unsafe {
+                libc::recv(self.fd,
+                           buf as *mut libc::c_void,
+                           len as wrlen,
+                           0) as i64
+            }
+        });
+        if ret == 0 {
+            Err(io::standard_error(io::EndOfFile))
+        } else if ret < 0 {
+            Err(super::last_error())
+        } else {
+            Ok(ret as uint)
+        }
+    }
+    fn write(&mut self, buf: &[u8]) -> IoResult<()> {
+        let ret = keep_going(buf, |buf, len| {
+            unsafe {
+                libc::send(self.fd,
+                           buf as *mut libc::c_void,
+                           len as wrlen,
+                           0) as i64
+            }
+        });
+        if ret < 0 {
+            Err(super::last_error())
+        } else {
+            Ok(())
+        }
+    }
+    fn peer_name(&mut self) -> IoResult<ip::SocketAddr> {
+        sockname(self.fd, libc::getpeername)
+    }
+    fn control_congestion(&mut self) -> IoResult<()> {
+        self.set_nodelay(false)
+    }
+    fn nodelay(&mut self) -> IoResult<()> {
+        self.set_nodelay(true)
+    }
+    fn keepalive(&mut self, delay_in_seconds: uint) -> IoResult<()> {
+        self.set_keepalive(Some(delay_in_seconds))
+    }
+    fn letdie(&mut self) -> IoResult<()> {
+        self.set_keepalive(None)
+    }
+}
+
+impl rtio::RtioSocket for TcpStream {
+    fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
+        sockname(self.fd, libc::getsockname)
+    }
+}
+
+impl Drop for TcpStream {
+    #[cfg(unix)]
+    fn drop(&mut self) {
+        unsafe { libc::close(self.fd); }
+    }
+
+    #[cfg(windows)]
+    fn drop(&mut self) {
+        unsafe { libc::closesocket(self.fd); }
+    }
+}
+
+pub struct TcpListener {
+    priv fd: sock_t,
+}
+
+impl TcpListener {
+    pub fn bind(addr: ip::SocketAddr) -> IoResult<TcpListener> {
+        unsafe {
+            socket(addr).and_then(|fd| {
+                let (addr, len) = addr_to_sockaddr(addr);
+                let addrp = &addr as *libc::sockaddr_storage;
+                let ret = TcpListener { fd: fd };
+                match libc::bind(fd, addrp as *libc::sockaddr,
+                                 len as libc::socklen_t) {
+                    -1 => Err(super::last_error()),
+                    _ => Ok(ret),
+                }
+            })
+        }
+    }
+
+    pub fn fd(&self) -> sock_t { self.fd }
+
+    pub fn native_listen(self, backlog: int) -> IoResult<TcpAcceptor> {
+        match unsafe { libc::listen(self.fd, backlog as libc::c_int) } {
+            -1 => Err(super::last_error()),
+            _ => Ok(TcpAcceptor { fd: self.fd })
+        }
+    }
+}
+
+impl rtio::RtioTcpListener for TcpListener {
+    fn listen(~self) -> IoResult<~rtio::RtioTcpAcceptor> {
+        self.native_listen(128).map(|a| ~a as ~rtio::RtioTcpAcceptor)
+    }
+}
+
+impl rtio::RtioSocket for TcpListener {
+    fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
+        sockname(self.fd, libc::getsockname)
+    }
+}
+
+pub struct TcpAcceptor {
+    priv fd: sock_t,
+}
+
+impl TcpAcceptor {
+    pub fn fd(&self) -> sock_t { self.fd }
+
+    pub fn native_accept(&mut self) -> IoResult<TcpStream> {
+        unsafe {
+            let mut storage: libc::sockaddr_storage = intrinsics::init();
+            let storagep = &mut storage as *mut libc::sockaddr_storage;
+            let size = mem::size_of::<libc::sockaddr_storage>();
+            let mut size = size as libc::socklen_t;
+            match libc::accept(self.fd,
+                               storagep as *mut libc::sockaddr,
+                               &mut size as *mut libc::socklen_t) {
+                -1 => Err(super::last_error()),
+                fd => Ok(TcpStream { fd: fd })
+            }
+        }
+    }
+}
+
+impl rtio::RtioSocket for TcpAcceptor {
+    fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
+        sockname(self.fd, libc::getsockname)
+    }
+}
+
+impl rtio::RtioTcpAcceptor for TcpAcceptor {
+    fn accept(&mut self) -> IoResult<~rtio::RtioTcpStream> {
+        self.native_accept().map(|s| ~s as ~rtio::RtioTcpStream)
+    }
+
+    fn accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
+    fn dont_accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
+}
diff --git a/src/libnative/io/process.rs b/src/libnative/io/process.rs
index 33abb27f16b..3fda4486921 100644
--- a/src/libnative/io/process.rs
+++ b/src/libnative/io/process.rs
@@ -407,8 +407,8 @@ fn spawn_process_os(prog: &str, args: &[~str],
     }
 
     let pipe = os::pipe();
-    let mut input = file::FileDesc::new(pipe.input, file::CloseFd);
-    let mut output = file::FileDesc::new(pipe.out, file::CloseFd);
+    let mut input = file::FileDesc::new(pipe.input, true);
+    let mut output = file::FileDesc::new(pipe.out, true);
 
     unsafe { set_cloexec(output.fd()) };
 
diff --git a/src/libnative/task.rs b/src/libnative/task.rs
index 12e361d8041..8f2dff42404 100644
--- a/src/libnative/task.rs
+++ b/src/libnative/task.rs
@@ -34,6 +34,7 @@ pub fn new() -> ~Task {
     task.put_runtime(~Ops {
         lock: unsafe { Mutex::new() },
         awoken: false,
+        io: io::IoFactory::new(),
     } as ~rt::Runtime);
     return task;
 }
@@ -86,8 +87,9 @@ pub fn spawn_opts(opts: TaskOpts, f: proc()) {
 // This structure is the glue between channels and the 1:1 scheduling mode. This
 // structure is allocated once per task.
 struct Ops {
-    lock: Mutex,  // native synchronization
-    awoken: bool, // used to prevent spurious wakeups
+    lock: Mutex,       // native synchronization
+    awoken: bool,      // used to prevent spurious wakeups
+    io: io::IoFactory, // local I/O factory
 }
 
 impl rt::Runtime for Ops {
@@ -217,11 +219,7 @@ impl rt::Runtime for Ops {
     }
 
     fn local_io<'a>(&'a mut self) -> Option<rtio::LocalIo<'a>> {
-        static mut io: io::IoFactory = io::IoFactory;
-        // Unsafety is from accessing `io`, which is guaranteed to be safe
-        // because you can't do anything usable with this statically initialized
-        // unit struct.
-        Some(unsafe { rtio::LocalIo::new(&mut io as &mut rtio::IoFactory) })
+        Some(rtio::LocalIo::new(&mut self.io as &mut rtio::IoFactory))
     }
 }