native: Introduce libnative

This commit introduces a new crate called "native" which will be the crate that
implements the 1:1 runtime of rust. This currently entails having an
implementation of std::rt::Runtime inside of libnative as well as moving all of
the native I/O implementations to libnative.

The current snag is that the start lang item must currently be defined in
libnative in order to start running, but this will change in the future.

Cool fact about this crate, there are no extra features that are enabled.

Note that this commit does not include any makefile support necessary for
building libnative, that's all coming in a later commit.

src/libnative/io/file.rs

@@ -10,28 +10,21 @@
 
 //! Blocking posix-based file I/O
 
-#[allow(non_camel_case_types)];
+use std::c_str::CString;
+use std::io::IoError;
+use std::io;
+use std::libc::c_int;
+use std::libc;
+use std::os;
+use std::rt::rtio;
+use std::unstable::intrinsics;
+use std::vec;
 
-use c_str::CString;
-use io::IoError;
-use io;
-use libc::c_int;
-use libc;
-use ops::Drop;
-use option::{Some, None, Option};
-use os;
-use path::{Path, GenericPath};
-use ptr::RawPtr;
-use result::{Result, Ok, Err};
-use rt::rtio;
 use super::IoResult;
-use unstable::intrinsics;
-use vec::ImmutableVector;
-use vec;
 
-#[cfg(windows)] use os::win32::{as_utf16_p, fill_utf16_buf_and_decode};
+#[cfg(windows)] use std::os::win32::{as_utf16_p, fill_utf16_buf_and_decode};
-#[cfg(windows)] use ptr;
+#[cfg(windows)] use std::ptr;
-#[cfg(windows)] use str;
+#[cfg(windows)] use std::str;
 
 fn keep_going(data: &[u8], f: |*u8, uint| -> i64) -> i64 {
     #[cfg(windows)] static eintr: int = 0; // doesn't matter
@@ -490,8 +483,8 @@ fn prune(root: &CString, dirs: ~[Path]) -> ~[Path] {
     unsafe {
         #[cfg(not(windows))]
         unsafe fn get_list(p: &CString) -> IoResult<~[Path]> {
-            use libc::{dirent_t};
+            use std::libc::{dirent_t};
-            use libc::{opendir, readdir, closedir};
+            use std::libc::{opendir, readdir, closedir};
             extern {
                 fn rust_list_dir_val(ptr: *dirent_t) -> *libc::c_char;
             }
@@ -517,14 +510,14 @@ unsafe fn get_list(p: &CString) -> IoResult<~[Path]> {
 
         #[cfg(windows)]
         unsafe fn get_list(p: &CString) -> IoResult<~[Path]> {
-            use libc::consts::os::extra::INVALID_HANDLE_VALUE;
+            use std::libc::consts::os::extra::INVALID_HANDLE_VALUE;
-            use libc::{wcslen, free};
+            use std::libc::{wcslen, free};
-            use libc::funcs::extra::kernel32::{
+            use std::libc::funcs::extra::kernel32::{
                 FindFirstFileW,
                 FindNextFileW,
                 FindClose,
             };
-            use libc::types::os::arch::extra::HANDLE;
+            use std::libc::types::os::arch::extra::HANDLE;
             use os::win32::{
                 as_utf16_p
             };
@@ -906,12 +899,12 @@ fn os_utime(p: &CString, atime: u64, mtime: u64) -> c_int {
 
 #[cfg(test)]
 mod tests {
-    use io::native::file::{CFile, FileDesc};
+    use std::io::native::file::{CFile, FileDesc};
-    use io;
+    use std::io::fs;
-    use libc;
+    use std::io;
-    use os;
+    use std::libc;
-    use result::Ok;
+    use std::os;
-    use rt::rtio::RtioFileStream;
+    use std::rt::rtio::RtioFileStream;
 
     #[ignore(cfg(target_os = "freebsd"))] // hmm, maybe pipes have a tiny buffer
     #[test]

src/libnative/io/mod.rs

@@ -21,24 +21,21 @@
 //! play. The only dependencies of these modules are the normal system libraries
 //! that you would find on the respective platform.
 
-use c_str::CString;
+use std::c_str::CString;
-use comm::SharedChan;
+use std::comm::SharedChan;
-use libc::c_int;
+use std::libc::c_int;
-use libc;
+use std::libc;
-use option::{Option, None, Some};
+use std::os;
-use os;
+use std::rt::rtio;
-use path::Path;
+use std::rt::rtio::{RtioTcpStream, RtioTcpListener, RtioUdpSocket,
-use result::{Result, Ok, Err};
+                    RtioUnixListener, RtioPipe, RtioFileStream, RtioProcess,
-use rt::rtio;
+                    RtioSignal, RtioTTY, CloseBehavior, RtioTimer};
-use rt::rtio::{RtioTcpStream, RtioTcpListener, RtioUdpSocket, RtioUnixListener,
+use std::io;
-               RtioPipe, RtioFileStream, RtioProcess, RtioSignal, RtioTTY,
+use std::io::IoError;
-               CloseBehavior, RtioTimer};
+use std::io::net::ip::SocketAddr;
-use io;
+use std::io::process::ProcessConfig;
-use io::IoError;
+use std::io::signal::Signum;
-use io::net::ip::SocketAddr;
+use ai = std::io::net::addrinfo;
-use io::process::ProcessConfig;
-use io::signal::Signum;
-use ai = io::net::addrinfo;
 
 // Local re-exports
 pub use self::file::FileDesc;
@@ -114,6 +111,9 @@ fn mkerr_winbool(ret: libc::c_int) -> IoResult<()> {
 pub struct IoFactory;
 
 impl rtio::IoFactory for IoFactory {
+    // all native io factories are the same
+    fn id(&self) -> uint { 0 }
+
     // networking
     fn tcp_connect(&mut self, _addr: SocketAddr) -> IoResult<~RtioTcpStream> {
         Err(unimpl())
@@ -223,6 +223,3 @@ fn signal(&mut self, _signal: Signum, _channel: SharedChan<Signum>)
         Err(unimpl())
     }
 }
-
-pub static mut NATIVE_IO_FACTORY: IoFactory = IoFactory;
-

src/libnative/io/process.rs

@@ -8,18 +8,16 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use io;
+use std::cast;
-use libc::{pid_t, c_void, c_int};
+use std::io;
-use libc;
+use std::libc::{pid_t, c_void, c_int};
-use os;
+use std::libc;
-use prelude::*;
+use std::os;
-use ptr;
+use std::ptr;
-use rt::rtio;
+use std::rt::rtio;
-use super::file;
+use p = std::io::process;
-#[cfg(windows)]
-use cast;
 
-use p = io::process;
+use super::file;
 
 /**
  * A value representing a child process.
@@ -179,22 +177,22 @@ fn spawn_process_os(prog: &str, args: &[~str],
                     env: Option<~[(~str, ~str)]>,
                     dir: Option<&Path>,
                     in_fd: c_int, out_fd: c_int, err_fd: c_int) -> SpawnProcessResult {
-    use libc::types::os::arch::extra::{DWORD, HANDLE, STARTUPINFO};
+    use std::libc::types::os::arch::extra::{DWORD, HANDLE, STARTUPINFO};
-    use libc::consts::os::extra::{
+    use std::libc::consts::os::extra::{
         TRUE, FALSE,
         STARTF_USESTDHANDLES,
         INVALID_HANDLE_VALUE,
         DUPLICATE_SAME_ACCESS
     };
-    use libc::funcs::extra::kernel32::{
+    use std::libc::funcs::extra::kernel32::{
         GetCurrentProcess,
         DuplicateHandle,
         CloseHandle,
         CreateProcessA
     };
-    use libc::funcs::extra::msvcrt::get_osfhandle;
+    use std::libc::funcs::extra::msvcrt::get_osfhandle;
 
-    use mem;
+    use std::mem;
 
     unsafe {
 
@@ -256,10 +254,10 @@ fn spawn_process_os(prog: &str, args: &[~str],
             fail!("failure in CreateProcess: {}", *msg);
         }
 
-        // We close the thread handle because we don't care about keeping the
+        // We close the thread handle because std::we don't care about keeping the
         // thread id valid, and we aren't keeping the thread handle around to be
         // able to close it later. We don't close the process handle however
-        // because we want the process id to stay valid at least until the
+        // because std::we want the process id to stay valid at least until the
         // calling code closes the process handle.
         CloseHandle(pi.hThread);
 
@@ -362,8 +360,8 @@ fn spawn_process_os(prog: &str, args: &[~str],
                     env: Option<~[(~str, ~str)]>,
                     dir: Option<&Path>,
                     in_fd: c_int, out_fd: c_int, err_fd: c_int) -> SpawnProcessResult {
-    use libc::funcs::posix88::unistd::{fork, dup2, close, chdir, execvp};
+    use std::libc::funcs::posix88::unistd::{fork, dup2, close, chdir, execvp};
-    use libc::funcs::bsd44::getdtablesize;
+    use std::libc::funcs::bsd44::getdtablesize;
 
     mod rustrt {
         extern {
@@ -433,7 +431,7 @@ unsafe fn set_environ(envp: *c_void) {
 
 #[cfg(unix)]
 fn with_argv<T>(prog: &str, args: &[~str], cb: |**libc::c_char| -> T) -> T {
-    use vec;
+    use std::vec;
 
     // We can't directly convert `str`s into `*char`s, as someone needs to hold
     // a reference to the intermediary byte buffers. So first build an array to
@@ -459,7 +457,7 @@ fn with_argv<T>(prog: &str, args: &[~str], cb: |**libc::c_char| -> T) -> T {
 
 #[cfg(unix)]
 fn with_envp<T>(env: Option<~[(~str, ~str)]>, cb: |*c_void| -> T) -> T {
-    use vec;
+    use std::vec;
 
     // On posixy systems we can pass a char** for envp, which is a
     // null-terminated array of "k=v\n" strings. Like `with_argv`, we have to
@@ -540,8 +538,8 @@ fn waitpid(pid: pid_t) -> int {
 
     #[cfg(windows)]
     fn waitpid_os(pid: pid_t) -> int {
-        use libc::types::os::arch::extra::DWORD;
+        use std::libc::types::os::arch::extra::DWORD;
-        use libc::consts::os::extra::{
+        use std::libc::consts::os::extra::{
             SYNCHRONIZE,
             PROCESS_QUERY_INFORMATION,
             FALSE,
@@ -549,7 +547,7 @@ fn waitpid_os(pid: pid_t) -> int {
             INFINITE,
             WAIT_FAILED
         };
-        use libc::funcs::extra::kernel32::{
+        use std::libc::funcs::extra::kernel32::{
             OpenProcess,
             GetExitCodeProcess,
             CloseHandle,
@@ -585,7 +583,7 @@ fn waitpid_os(pid: pid_t) -> int {
 
     #[cfg(unix)]
     fn waitpid_os(pid: pid_t) -> int {
-        use libc::funcs::posix01::wait::*;
+        use std::libc::funcs::posix01::wait;
 
         #[cfg(target_os = "linux")]
         #[cfg(target_os = "android")]
@@ -612,7 +610,7 @@ fn WEXITSTATUS(status: i32) -> i32 {
         }
 
         let mut status = 0 as c_int;
-        if unsafe { waitpid(pid, &mut status, 0) } == -1 {
+        if unsafe { wait::waitpid(pid, &mut status, 0) } == -1 {
             fail!("failure in waitpid: {}", os::last_os_error());
         }
 

src/libnative/lib.rs

@@ -0,0 +1,61 @@
+// 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.
+
+//! The native runtime crate
+//!
+//! This crate contains an implementation of 1:1 scheduling for a "native"
+//! runtime. In addition, all I/O provided by this crate is the thread blocking
+//! version of I/O.
+
+#[link(name = "native",
+       package_id = "native",
+       vers = "0.9-pre",
+       uuid = "535344a7-890f-5a23-e1f3-e0d118805141",
+       url = "https://github.com/mozilla/rust/tree/master/src/native")];
+
+#[license = "MIT/ASL2"];
+#[crate_type = "rlib"];
+#[crate_type = "dylib"];
+
+// NB this crate explicitly does *not* allow glob imports, please seriously
+//    consider whether they're needed before adding that feature here.
+
+use std::cast;
+use std::os;
+use std::rt;
+use std::task::try;
+
+pub mod io;
+pub mod task;
+
+// XXX: this should not exist here
+#[cfg(stage0)]
+#[lang = "start"]
+pub fn start(main: *u8, argc: int, argv: **u8) -> int {
+    rt::init(argc, argv);
+
+    // Bootstrap ourselves by installing a local Task and then immediately
+    // spawning a thread to run 'main'. Always spawn a new thread for main so
+    // the stack size of 'main' is known (and the bounds can be set
+    // appropriately).
+    //
+    // Once the main task has completed, then we wait for everyone else to exit.
+    task::run(task::new(), proc() {
+        let main: extern "Rust" fn() = unsafe { cast::transmute(main) };
+        match do try { main() } {
+            Ok(()) => { os::set_exit_status(0); }
+            Err(..) => { os::set_exit_status(rt::DEFAULT_ERROR_CODE); }
+        }
+    });
+    task::wait_for_completion();
+
+    unsafe { rt::cleanup(); }
+    os::get_exit_status()
+}

src/libnative/task.rs

@@ -0,0 +1,257 @@
+// 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.
+
+//! Tasks implemented on top of OS threads
+//!
+//! This module contains the implementation of the 1:1 threading module required
+//! by rust tasks. This implements the necessary API traits laid out by std::rt
+//! in order to spawn new tasks and deschedule the current task.
+
+use std::cast;
+use std::rt::env;
+use std::rt::local::Local;
+use std::rt::rtio;
+use std::rt::task::{Task, BlockedTask};
+use std::rt::thread::Thread;
+use std::rt;
+use std::sync::atomics::{AtomicUint, SeqCst, INIT_ATOMIC_UINT};
+use std::task::TaskOpts;
+use std::unstable::mutex::{Mutex, MUTEX_INIT};
+use std::unstable::stack;
+
+use io;
+use task;
+
+static mut THREAD_CNT: AtomicUint = INIT_ATOMIC_UINT;
+static mut LOCK: Mutex = MUTEX_INIT;
+
+/// Waits for all spawned threads to finish completion. This should only be used
+/// by the main task in order to wait for all other tasks to terminate.
+///
+/// This mirrors the same semantics as the green scheduling model.
+pub fn wait_for_completion() {
+    static mut someone_waited: bool = false;
+
+    unsafe {
+        LOCK.lock();
+        assert!(!someone_waited);
+        someone_waited = true;
+        while THREAD_CNT.load(SeqCst) > 0 {
+            LOCK.wait();
+        }
+        LOCK.unlock();
+        LOCK.destroy();
+    }
+
+}
+
+// Signal that a thread has finished execution, possibly waking up a blocker
+// waiting for all threads to have finished.
+fn signal_done() {
+    unsafe {
+        LOCK.lock();
+        if THREAD_CNT.fetch_sub(1, SeqCst) == 1 {
+            LOCK.signal();
+        }
+        LOCK.unlock();
+    }
+}
+
+/// Creates a new Task which is ready to execute as a 1:1 task.
+pub fn new() -> ~Task {
+    let mut task = ~Task::new();
+    task.put_runtime(~Ops {
+        lock: unsafe { Mutex::new() },
+    } as ~rt::Runtime);
+    return task;
+}
+
+/// Spawns a new task given the configuration options and a procedure to run
+/// inside the task.
+pub fn spawn(opts: TaskOpts, f: proc()) {
+    // must happen before the spawn, no need to synchronize with a lock.
+    unsafe { THREAD_CNT.fetch_add(1, SeqCst); }
+
+    let TaskOpts {
+        watched: _watched,
+        notify_chan, name, stack_size
+    } = opts;
+
+    let mut task = new();
+    task.name = name;
+    match notify_chan {
+        Some(chan) => {
+            let on_exit = proc(task_result) { chan.send(task_result) };
+            task.death.on_exit = Some(on_exit);
+        }
+        None => {}
+    }
+
+    let stack = stack_size.unwrap_or(env::min_stack());
+    let task = task;
+
+    // Spawning a new OS thread guarantees that __morestack will never get
+    // triggered, but we must manually set up the actual stack bounds once this
+    // function starts executing. This raises the lower limit by a bit because
+    // by the time that this function is executing we've already consumed at
+    // least a little bit of stack (we don't know the exact byte address at
+    // which our stack started).
+    Thread::spawn_stack(stack, proc() {
+        let something_around_the_top_of_the_stack = 1;
+        let addr = &something_around_the_top_of_the_stack as *int;
+        unsafe {
+            let my_stack = addr as uint;
+            stack::record_stack_bounds(my_stack - stack + 1024, my_stack);
+        }
+
+        run(task, f);
+        signal_done();
+    })
+}
+
+/// Runs a task once, consuming the task. The given procedure is run inside of
+/// the task.
+pub fn run(t: ~Task, f: proc()) {
+    let mut f = Some(f);
+    t.run(|| { f.take_unwrap()(); });
+}
+
+// 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
+}
+
+impl rt::Runtime for Ops {
+    fn yield_now(~self, mut cur_task: ~Task) {
+        // put the task back in TLS and then invoke the OS thread yield
+        cur_task.put_runtime(self as ~rt::Runtime);
+        Local::put(cur_task);
+        Thread::yield_now();
+    }
+
+    fn maybe_yield(~self, mut cur_task: ~Task) {
+        // just put the task back in TLS, on OS threads we never need to
+        // opportunistically yield b/c the OS will do that for us (preemption)
+        cur_task.put_runtime(self as ~rt::Runtime);
+        Local::put(cur_task);
+    }
+
+    fn wrap(~self) -> ~Any {
+        self as ~Any
+    }
+
+    // This function gets a little interesting. There are a few safety and
+    // ownership violations going on here, but this is all done in the name of
+    // shared state. Additionally, all of the violations are protected with a
+    // mutex, so in theory there are no races.
+    //
+    // The first thing we need to do is to get a pointer to the task's internal
+    // mutex. This address will not be changing (because the task is allocated
+    // on the heap). We must have this handle separately because the task will
+    // have its ownership transferred to the given closure. We're guaranteed,
+    // however, that this memory will remain valid because *this* is the current
+    // task's execution thread.
+    //
+    // The next weird part is where ownership of the task actually goes. We
+    // relinquish it to the `f` blocking function, but upon returning this
+    // function needs to replace the task back in TLS. There is no communication
+    // from the wakeup thread back to this thread about the task pointer, and
+    // there's really no need to. In order to get around this, we cast the task
+    // to a `uint` which is then used at the end of this function to cast back
+    // to a `~Task` object. Naturally, this looks like it violates ownership
+    // semantics in that there may be two `~Task` objects.
+    //
+    // The fun part is that the wakeup half of this implementation knows to
+    // "forget" the task on the other end. This means that the awakening half of
+    // things silently relinquishes ownership back to this thread, but not in a
+    // way that the compiler can understand. The task's memory is always valid
+    // for both tasks because these operations are all done inside of a mutex.
+    //
+    // You'll also find that if blocking fails (the `f` function hands the
+    // BlockedTask back to us), we will `cast::forget` the handles. The
+    // reasoning for this is the same logic as above in that the task silently
+    // transfers ownership via the `uint`, not through normal compiler
+    // semantics.
+    fn deschedule(mut ~self, times: uint, mut cur_task: ~Task,
+                  f: |BlockedTask| -> Result<(), BlockedTask>) {
+        let my_lock: *mut Mutex = &mut self.lock as *mut Mutex;
+        cur_task.put_runtime(self as ~rt::Runtime);
+
+        unsafe {
+            let cur_task_dupe = *cast::transmute::<&~Task, &uint>(&cur_task);
+            let task = BlockedTask::block(cur_task);
+
+            if times == 1 {
+                (*my_lock).lock();
+                match f(task) {
+                    Ok(()) => (*my_lock).wait(),
+                    Err(task) => { cast::forget(task.wake()); }
+                }
+                (*my_lock).unlock();
+            } else {
+                let mut iter = task.make_selectable(times);
+                (*my_lock).lock();
+                let success = iter.all(|task| {
+                    match f(task) {
+                        Ok(()) => true,
+                        Err(task) => {
+                            cast::forget(task.wake());
+                            false
+                        }
+                    }
+                });
+                if success {
+                    (*my_lock).wait();
+                }
+                (*my_lock).unlock();
+            }
+            // re-acquire ownership of the task
+            cur_task = cast::transmute::<uint, ~Task>(cur_task_dupe);
+        }
+
+        // put the task back in TLS, and everything is as it once was.
+        Local::put(cur_task);
+    }
+
+    // See the comments on `deschedule` for why the task is forgotten here, and
+    // why it's valid to do so.
+    fn reawaken(mut ~self, mut to_wake: ~Task, _can_resched: bool) {
+        unsafe {
+            let lock: *mut Mutex = &mut self.lock as *mut Mutex;
+            to_wake.put_runtime(self as ~rt::Runtime);
+            cast::forget(to_wake);
+            (*lock).lock();
+            (*lock).signal();
+            (*lock).unlock();
+        }
+    }
+
+    fn spawn_sibling(~self, mut cur_task: ~Task, opts: TaskOpts, f: proc()) {
+        cur_task.put_runtime(self as ~rt::Runtime);
+        Local::put(cur_task);
+
+        task::spawn(opts, f);
+    }
+
+    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) })
+    }
+}
+
+impl Drop for Ops {
+    fn drop(&mut self) {
+        unsafe { self.lock.destroy() }
+    }
+}

src/libstd/io/mod.rs

@@ -164,9 +164,6 @@
 performing an asynchronous request; the task is only resumed once the
 asynchronous request completes.
 
-For blocking (but possibly more efficient) implementations, look
-in the `io::native` module.
-
 # Error Handling
 
 I/O is an area where nearly every operation can result in unexpected
@@ -349,8 +346,6 @@
 /// Buffered I/O wrappers
 pub mod buffered;
 
-pub mod native;
-
 /// Signal handling
 pub mod signal;