auto merge of #6046 : brson/rust/io, r=graydon

r? @pcwalton

Sorry this is so big, and sorry the first commit is just titled 'wip'.

Some interesting bits

* [LocalServices](f9069baa70) - This is the set of runtime capabilities that *all* Rust code should expect access to, including the local heap, GC, logging, unwinding.
* [impl Reader, etc. for Option](5fbb0949a5) - Constructors like `File::open` return Option<FileStream>. This lets you write I/O code without ever unwrapping an option.

This series adds a lot of [documentation](https://github.com/brson/rust/blob/io/src/libcore/rt/io/mod.rs#L11) to `core::rt::io`.

Makefile.in

@@ -238,7 +238,7 @@ $(foreach target,$(CFG_TARGET_TRIPLES),\
 
 CORELIB_CRATE := $(S)src/libcore/core.rc
 CORELIB_INPUTS := $(wildcard $(addprefix $(S)src/libcore/,        \
-                                           core.rc *.rs */*.rs */*/*rs))
+                                           core.rc *.rs */*.rs */*/*rs */*/*/*rs))
 
 ######################################################################
 # Standard library variables

mk/docs.mk

@@ -16,15 +16,8 @@ DOCS :=
 
 
 ######################################################################
-# Pandoc (reference-manual related)
+# Docs, from pandoc, rustdoc (which runs pandoc), and node
 ######################################################################
-ifeq ($(CFG_PANDOC),)
-  $(info cfg: no pandoc found, omitting doc/rust.pdf)
-else
-
-  ifeq ($(CFG_NODE),)
-    $(info cfg: no node found, omitting doc/tutorial.html)
-  else
 
 doc/rust.css: rust.css
 	@$(call E, cp: $@)
@@ -34,6 +27,18 @@ doc/manual.css: manual.css
 	@$(call E, cp: $@)
 	$(Q)cp -a $< $@ 2> /dev/null
 
+ifeq ($(CFG_PANDOC),)
+  $(info cfg: no pandoc found, omitting docs)
+  NO_DOCS = 1
+endif
+
+ifeq ($(CFG_NODE),)
+  $(info cfg: no node found, omitting docs)
+  NO_DOCS = 1
+endif
+
+ifneq ($(NO_DOCS),1)
+
 DOCS += doc/rust.html
 doc/rust.html: rust.md doc/version_info.html doc/rust.css doc/manual.css
 	@$(call E, pandoc: $@)
@@ -47,19 +52,8 @@ doc/rust.html: rust.md doc/version_info.html doc/rust.css doc/manual.css
          --css=manual.css \
 	     --include-before-body=doc/version_info.html \
          --output=$@
-  endif
 
-  ifeq ($(CFG_PDFLATEX),)
-    $(info cfg: no pdflatex found, omitting doc/rust.pdf)
-  else
-    ifeq ($(CFG_XETEX),)
-      $(info cfg: no xetex found, disabling doc/rust.pdf)
-    else
-      ifeq ($(CFG_LUATEX),)
-        $(info cfg: lacking luatex, disabling pdflatex)
-      else
-
-DOCS += doc/rust.pdf
+DOCS += doc/rust.tex
 doc/rust.tex: rust.md doc/version.md
 	@$(call E, pandoc: $@)
 	$(Q)$(CFG_NODE) $(S)doc/prep.js $< | \
@@ -70,17 +64,6 @@ doc/rust.tex: rust.md doc/version.md
          --from=markdown --to=latex \
          --output=$@
 
-doc/rust.pdf: doc/rust.tex
-	@$(call E, pdflatex: $@)
-	$(Q)$(CFG_PDFLATEX) \
-        -interaction=batchmode \
-        -output-directory=doc \
-        $<
-
-      endif
-    endif
-  endif
-
 DOCS += doc/rustpkg.html
 doc/rustpkg.html: rustpkg.md doc/version_info.html doc/rust.css doc/manual.css
 	@$(call E, pandoc: $@)
@@ -95,13 +78,6 @@ doc/rustpkg.html: rustpkg.md doc/version_info.html doc/rust.css doc/manual.css
 	     --include-before-body=doc/version_info.html \
          --output=$@
 
-######################################################################
-# Node (tutorial related)
-######################################################################
-  ifeq ($(CFG_NODE),)
-    $(info cfg: no node found, omitting doc/tutorial.html)
-  else
-
 DOCS += doc/tutorial.html
 doc/tutorial.html: tutorial.md doc/version_info.html doc/rust.css
 	@$(call E, pandoc: $@)
@@ -153,9 +129,29 @@ doc/tutorial-tasks.html: tutorial-tasks.md doc/version_info.html doc/rust.css
 	   --include-before-body=doc/version_info.html \
            --output=$@
 
-  endif
-endif
+  ifeq ($(CFG_PDFLATEX),)
+    $(info cfg: no pdflatex found, omitting doc/rust.pdf)
+  else
+    ifeq ($(CFG_XETEX),)
+      $(info cfg: no xetex found, disabling doc/rust.pdf)
+    else
+      ifeq ($(CFG_LUATEX),)
+        $(info cfg: lacking luatex, disabling pdflatex)
+      else
 
+DOCS += doc/rust.pdf
+doc/rust.pdf: doc/rust.tex
+	@$(call E, pdflatex: $@)
+	$(Q)$(CFG_PDFLATEX) \
+        -interaction=batchmode \
+        -output-directory=doc \
+        $<
+
+      endif
+    endif
+  endif
+
+endif # No pandoc / node
 
 ######################################################################
 # LLnextgen (grammar analysis from refman)

src/libcore/condition.rs

@@ -192,4 +192,27 @@ mod test {
 
         assert!(trapped);
     }
+
+    // Issue #6009
+    mod m {
+        condition! {
+            sadness: int -> int;
+        }
+
+        mod n {
+            use super::sadness;
+
+            #[test]
+            fn test_conditions_are_public() {
+                let mut trapped = false;
+                do sadness::cond.trap(|_| {
+                    trapped = true;
+                    0
+                }).in {
+                    sadness::cond.raise(0);
+                }
+                assert!(trapped);
+            }
+        }
+    }
 }

src/libcore/core.rc

@@ -125,6 +125,9 @@ pub mod linkhack {
     }
 }
 
+// Internal macros
+mod macros;
+
 /* The Prelude. */
 
 pub mod prelude;

src/libcore/macros.rs

@@ -0,0 +1,39 @@
+// Copyright 2012 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.
+
+#[macro_escape];
+
+// Some basic logging
+macro_rules! rtdebug_ (
+    ($( $arg:expr),+) => ( {
+        dumb_println(fmt!( $($arg),+ ));
+
+        fn dumb_println(s: &str) {
+            use io::WriterUtil;
+            let dbg = ::libc::STDERR_FILENO as ::io::fd_t;
+            dbg.write_str(s);
+            dbg.write_str("\n");
+        }
+
+    } )
+)
+
+// An alternate version with no output, for turning off logging
+macro_rules! rtdebug (
+    ($( $arg:expr),+) => ( $(let _ = $arg)*; )
+)
+
+macro_rules! abort(
+    ($( $msg:expr),+) => ( {
+        rtdebug!($($msg),+);
+
+        unsafe { ::libc::abort(); }
+    } )
+)

src/libcore/rt/io/file.rs

@@ -9,13 +9,9 @@
 // except according to those terms.
 
 use prelude::*;
-use super::misc::PathLike;
+use super::support::PathLike;
 use super::{Reader, Writer, Seek, Close};
-use super::{IoError, SeekStyle};
-
-/// Open a file with the default FileMode and FileAccess
-/// # XXX are there sane defaults here?
-pub fn open_file<P: PathLike>(_path: &P) -> FileStream { fail!() }
+use super::SeekStyle;
 
 /// # XXX
 /// * Ugh, this is ridiculous. What is the best way to represent these options?
@@ -46,7 +42,7 @@ impl FileStream {
     pub fn open<P: PathLike>(_path: &P,
                              _mode: FileMode,
                              _access: FileAccess
-                            ) -> Result<FileStream, IoError> {
+                            ) -> Option<FileStream> {
         fail!()
     }
 }

src/libcore/rt/io/mem.rs

@@ -17,7 +17,7 @@
 
 use prelude::*;
 use super::*;
-
+use cmp::min;
 
 /// Writes to an owned, growable byte vector
 pub struct MemWriter {
@@ -29,13 +29,15 @@ impl MemWriter {
 }
 
 impl Writer for MemWriter {
-    fn write(&mut self, _buf: &[u8]) { fail!() }
+    fn write(&mut self, buf: &[u8]) {
+        self.buf.push_all(buf)
+    }
 
     fn flush(&mut self) { /* no-op */ }
 }
 
 impl Seek for MemWriter {
-    fn tell(&self) -> u64 { fail!() }
+    fn tell(&self) -> u64 { self.buf.len() as u64 }
 
     fn seek(&mut self, _pos: i64, _style: SeekStyle) { fail!() }
 }
@@ -77,13 +79,27 @@ impl MemReader {
 }
 
 impl Reader for MemReader {
-    fn read(&mut self, _buf: &mut [u8]) -> Option<uint> { fail!() }
+    fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
+        { if self.eof() { return None; } }
 
-    fn eof(&mut self) -> bool { fail!() }
+        let write_len = min(buf.len(), self.buf.len() - self.pos);
+        {
+            let input = self.buf.slice(self.pos, self.pos + write_len);
+            let output = vec::mut_slice(buf, 0, write_len);
+            assert!(input.len() == output.len());
+            vec::bytes::copy_memory(output, input, write_len);
+        }
+        self.pos += write_len;
+        assert!(self.pos <= self.buf.len());
+
+        return Some(write_len);
+    }
+
+    fn eof(&mut self) -> bool { self.pos == self.buf.len() }
 }
 
 impl Seek for MemReader {
-    fn tell(&self) -> u64 { fail!() }
+    fn tell(&self) -> u64 { self.pos as u64 }
 
     fn seek(&mut self, _pos: i64, _style: SeekStyle) { fail!() }
 }
@@ -163,4 +179,43 @@ impl<'self> Seek for BufReader<'self> {
     fn tell(&self) -> u64 { fail!() }
 
     fn seek(&mut self, _pos: i64, _style: SeekStyle) { fail!() }
-}
+}
+
+#[cfg(test)]
+mod test {
+    use prelude::*;
+    use super::*;
+
+    #[test]
+    fn test_mem_writer() {
+        let mut writer = MemWriter::new();
+        assert!(writer.tell() == 0);
+        writer.write([0]);
+        assert!(writer.tell() == 1);
+        writer.write([1, 2, 3]);
+        writer.write([4, 5, 6, 7]);
+        assert!(writer.tell() == 8);
+        assert!(writer.inner() == ~[0, 1, 2, 3, 4, 5 , 6, 7]);
+    }
+
+    #[test]
+    fn test_mem_reader() {
+        let mut reader = MemReader::new(~[0, 1, 2, 3, 4, 5, 6, 7]);
+        let mut buf = [];
+        assert!(reader.read(buf) == Some(0));
+        assert!(reader.tell() == 0);
+        let mut buf = [0];
+        assert!(reader.read(buf) == Some(1));
+        assert!(reader.tell() == 1);
+        assert!(buf == [0]);
+        let mut buf = [0, ..4];
+        assert!(reader.read(buf) == Some(4));
+        assert!(reader.tell() == 5);
+        assert!(buf == [1, 2, 3, 4]);
+        assert!(reader.read(buf) == Some(3));
+        assert!(buf.slice(0, 3) == [5, 6, 7]);
+        assert!(reader.eof());
+        assert!(reader.read(buf) == None);
+        assert!(reader.eof());
+    }
+}

src/libcore/rt/io/mod.rs

@@ -11,7 +11,13 @@
 /*! Synchronous I/O
 
 This module defines the Rust interface for synchronous I/O.
-It supports file access,
+It models byte-oriented input and output with the Reader and Writer traits.
+Types that implement both `Reader` and `Writer` and called 'streams',
+and automatically implement trait `Stream`.
+Implementations are provided for common I/O streams like
+file, TCP, UDP, Unix domain sockets.
+Readers and Writers may be composed to add capabilities like string
+parsing, encoding, and compression.
 
 This will likely live in core::io, not core::rt::io.
 
@@ -27,44 +33,177 @@ Some examples of obvious things you might want to do
 
 * Read a complete file to a string, (converting newlines?)
 
-    let contents = open("message.txt").read_to_str(); // read_to_str??
+    let contents = File::open("message.txt").read_to_str(); // read_to_str??
 
 * Write a line to a file
 
-    let file = FileStream::open("message.txt", Create, Write);
+    let file = File::open("message.txt", Create, Write);
     file.write_line("hello, file!");
 
 * Iterate over the lines of a file
 
+    do File::open("message.txt").each_line |line| {
+        println(line)
+    }
+
 * Pull the lines of a file into a vector of strings
 
+    let lines = File::open("message.txt").line_iter().to_vec();
+
+* Make an simple HTTP request
+
+    let socket = TcpStream::open("localhost:8080");
+    socket.write_line("GET / HTTP/1.0");
+    socket.write_line("");
+    let response = socket.read_to_end();
+
 * Connect based on URL? Requires thinking about where the URL type lives
   and how to make protocol handlers extensible, e.g. the "tcp" protocol
   yields a `TcpStream`.
 
-    connect("tcp://localhost:8080").write_line("HTTP 1.0 GET /");
+    connect("tcp://localhost:8080");
 
 # Terms
 
-* reader
-* writer
-* stream
-* Blocking vs. non-blocking
-* synchrony and asynchrony
+* Reader - An I/O source, reads bytes into a buffer
+* Writer - An I/O sink, writes bytes from a buffer
+* Stream - Typical I/O sources like files and sockets are both Readers and Writers,
+  and are collectively referred to a `streams`.
+* Decorator - A Reader or Writer that composes with others to add additional capabilities
+  such as encoding or decoding
 
-I tend to call this implementation non-blocking, because performing I/O
-doesn't block the progress of other tasks. Is that how we want to present
-it, 'synchronous but non-blocking'?
+# Blocking and synchrony
+
+When discussing I/O you often hear the terms 'synchronous' and
+'asynchronous', along with 'blocking' and 'non-blocking' compared and
+contrasted. A synchronous I/O interface performs each I/O operation to
+completion before proceeding to the next. Synchronous interfaces are
+usually used in imperative style as a sequence of commands. An
+asynchronous interface allows multiple I/O requests to be issued
+simultaneously, without waiting for each to complete before proceeding
+to the next.
+
+Asynchronous interfaces are used to achieve 'non-blocking' I/O. In
+traditional single-threaded systems, performing a synchronous I/O
+operation means that the program stops all activity (it 'blocks')
+until the I/O is complete. Blocking is bad for performance when
+there are other computations that could be done.
+
+Asynchronous interfaces are most often associated with the callback
+(continuation-passing) style popularised by node.js. Such systems rely
+on all computations being run inside an event loop which maintains a
+list of all pending I/O events; when one completes the registered
+callback is run and the code that made the I/O request continiues.
+Such interfaces achieve non-blocking at the expense of being more
+difficult to reason about.
+
+Rust's I/O interface is synchronous - easy to read - and non-blocking by default.
+
+Remember that Rust tasks are 'green threads', lightweight threads that
+are multiplexed onto a single operating system thread. If that system
+thread blocks then no other task may proceed. Rust tasks are
+relatively cheap to create, so as long as other tasks are free to
+execute then non-blocking code may be written by simply creating a new
+task.
+
+When discussing blocking in regards to Rust's I/O model, we are
+concerned with whether performing I/O blocks other Rust tasks from
+proceeding. In other words, when a task calls `read`, it must then
+wait (or 'sleep', or 'block') until the call to `read` is complete.
+During this time, other tasks may or may not be executed, depending on
+how `read` is implemented.
+
+
+Rust's default I/O implementation is non-blocking; by cooperating
+directly with the task scheduler it arranges to never block progress
+of *other* tasks. Under the hood, Rust uses asynchronous I/O via a
+per-scheduler (and hence per-thread) event loop. Synchronous I/O
+requests are implemented by descheduling the running task and
+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
+errors. It should allow errors to be handled efficiently.
+It needs to be convenient to use I/O when you don't care
+about dealing with specific errors.
+
+Rust's I/O employs a combination of techniques to reduce boilerplate
+while still providing feedback about errors. The basic strategy:
+
+* Errors are fatal by default, resulting in task failure
+* Errors raise the `io_error` conditon which provides an opportunity to inspect
+  an IoError object containing details.
+* Return values must have a sensible null or zero value which is returned
+  if a condition is handled successfully. This may be an `Option`, an empty
+  vector, or other designated error value.
+* Common traits are implemented for `Option`, e.g. `impl<R: Reader> Reader for Option<R>`,
+  so that nullable values do not have to be 'unwrapped' before use.
+
+These features combine in the API to allow for expressions like
+`File::new("diary.txt").write_line("met a girl")` without having to
+worry about whether "diary.txt" exists or whether the write
+succeeds. As written, if either `new` or `write_line` encounters
+an error the task will fail.
+
+If you wanted to handle the error though you might write
+
+    let mut error = None;
+    do io_error::cond(|e: IoError| {
+        error = Some(e);
+    }).in {
+        File::new("diary.txt").write_line("met a girl");
+    }
+
+    if error.is_some() {
+        println("failed to write my diary");
+    }
+
+XXX: Need better condition handling syntax
+
+In this case the condition handler will have the opportunity to
+inspect the IoError raised by either the call to `new` or the call to
+`write_line`, but then execution will continue.
+
+So what actually happens if `new` encounters an error? To understand
+that it's important to know that what `new` returns is not a `File`
+but an `Option<File>`.  If the file does not open, and the condition
+is handled, then `new` will simply return `None`. Because there is an
+implementation of `Writer` (the trait required ultimately required for
+types to implement `write_line`) there is no need to inspect or unwrap
+the `Option<File>` and we simply call `write_line` on it.  If `new`
+returned a `None` then the followup call to `write_line` will also
+raise an error.
+
+## Concerns about this strategy
+
+This structure will encourage a programming style that is prone
+to errors similar to null pointer dereferences.
+In particular code written to ignore errors and expect conditions to be unhandled
+will start passing around null or zero objects when wrapped in a condition handler.
+
+* XXX: How should we use condition handlers that return values?
+
+
+# Issues withi/o scheduler affinity, work stealing, task pinning
+
 # Resource management
 
 * `close` vs. RAII
 
-# Paths and URLs
+# Paths, URLs and overloaded constructors
 
-# std
+
+
+# Scope
+
+In scope for core
+
+* Url?
 
 Some I/O things don't belong in core
 
@@ -73,7 +212,12 @@ Some I/O things don't belong in core
     - http
   - flate
 
-# XXX
+Out of scope
+
+* Async I/O. We'll probably want it eventually
+
+
+# XXX Questions and issues
 
 * Should default constructors take `Path` or `&str`? `Path` makes simple cases verbose.
   Overloading would be nice.
@@ -83,6 +227,7 @@ Some I/O things don't belong in core
 * fsync
 * relationship with filesystem querying, Directory, File types etc.
 * Rename Reader/Writer to ByteReader/Writer, make Reader/Writer generic?
+* Can Port and Chan be implementations of a generic Reader<T>/Writer<T>?
 * Trait for things that are both readers and writers, Stream?
 * How to handle newline conversion
 * String conversion
@@ -92,6 +237,7 @@ Some I/O things don't belong in core
 * Do we need `close` at all? dtors might be good enough
 * How does I/O relate to the Iterator trait?
 * std::base64 filters
+* Using conditions is a big unknown since we don't have much experience with them
 
 */
 
@@ -104,25 +250,29 @@ pub use self::stdio::stderr;
 pub use self::stdio::print;
 pub use self::stdio::println;
 
-pub use self::file::open_file;
 pub use self::file::FileStream;
-pub use self::net::Listener;
 pub use self::net::ip::IpAddr;
 pub use self::net::tcp::TcpListener;
 pub use self::net::tcp::TcpStream;
 pub use self::net::udp::UdpStream;
 
 // Some extension traits that all Readers and Writers get.
-pub use self::util::ReaderUtil;
-pub use self::util::ReaderByteConversions;
-pub use self::util::WriterByteConversions;
+pub use self::extensions::ReaderUtil;
+pub use self::extensions::ReaderByteConversions;
+pub use self::extensions::WriterByteConversions;
 
 /// Synchronous, non-blocking file I/O.
 pub mod file;
 
 /// Synchronous, non-blocking network I/O.
-#[path = "net/mod.rs"]
-pub mod net;
+pub mod net {
+    pub mod tcp;
+    pub mod udp;
+    pub mod ip;
+    #[cfg(unix)]
+    pub mod unix;
+    pub mod http;
+}
 
 /// Readers and Writers for memory buffers and strings.
 #[cfg(not(stage0))] // XXX Using unsnapshotted features
@@ -131,6 +281,10 @@ pub mod mem;
 /// Non-blocking access to stdin, stdout, stderr
 pub mod stdio;
 
+/// Implementations for Option
+#[cfg(not(stage0))] // Requires condition! fixes
+mod option;
+
 /// Basic stream compression. XXX: Belongs with other flate code
 #[cfg(not(stage0))] // XXX Using unsnapshotted features
 pub mod flate;
@@ -140,10 +294,10 @@ pub mod flate;
 pub mod comm_adapters;
 
 /// Extension traits
-mod util;
+mod extensions;
 
 /// Non-I/O things needed by the I/O module
-mod misc;
+mod support;
 
 /// Thread-blocking implementations
 pub mod native {
@@ -173,12 +327,14 @@ pub struct IoError {
     detail: Option<~str>
 }
 
+#[deriving(Eq)]
 pub enum IoErrorKind {
     FileNotFound,
     FilePermission,
     ConnectionFailed,
     Closed,
-    OtherIoError
+    OtherIoError,
+    PreviousIoError
 }
 
 // XXX: Can't put doc comments on macros
@@ -211,9 +367,9 @@ pub trait Reader {
     ///         println(reader.read_line());
     ///     }
     ///
-    /// # XXX
+    /// # Failue
     ///
-    /// What does this return if the Reader is in an error state?
+    /// Returns `true` on failure.
     fn eof(&mut self) -> bool;
 }
 
@@ -253,9 +409,30 @@ pub enum SeekStyle {
 /// * Are `u64` and `i64` the right choices?
 pub trait Seek {
     fn tell(&self) -> u64;
+
+    /// Seek to an offset in a stream
+    ///
+    /// A successful seek clears the EOF indicator.
+    ///
+    /// # XXX
+    ///
+    /// * What is the behavior when seeking past the end of a stream?
     fn seek(&mut self, pos: i64, style: SeekStyle);
 }
 
+/// A listener is a value that listens for connections
+pub trait Listener<S> {
+    /// Wait for and accept an incoming connection
+    ///
+    /// Returns `None` on timeout.
+    ///
+    /// # Failure
+    ///
+    /// Raises `io_error` condition. If the condition is handled,
+    /// then `accept` returns `None`.
+    fn accept(&mut self) -> Option<S>;
+}
+
 /// Common trait for decorator types.
 ///
 /// Provides accessors to get the inner, 'decorated' values. The I/O library
@@ -281,3 +458,16 @@ pub trait Decorator<T> {
     /// Take a mutable reference to the decorated value
     fn inner_mut_ref<'a>(&'a mut self) -> &'a mut T;
 }
+
+pub fn standard_error(kind: IoErrorKind) -> IoError {
+    match kind {
+        PreviousIoError => {
+            IoError {
+                kind: PreviousIoError,
+                desc: "Failing due to a previous I/O error",
+                detail: None
+            }
+        }
+        _ => fail!()
+    }
+}

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

@@ -1,31 +0,0 @@
-// 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 prelude::*;
-
-pub mod tcp;
-pub mod udp;
-pub mod ip;
-#[cfg(unix)]
-pub mod unix;
-pub mod http;
-
-/// A listener is a value that listens for connections
-pub trait Listener<S> {
-    /// Wait for and accept an incoming connection
-    ///
-    /// Returns `None` on timeout.
-    ///
-    /// # Failure
-    ///
-    /// Raises `io_error` condition. If the condition is handled,
-    /// then `accept` returns `None`.
-    fn accept(&mut self) -> Option<S>;
-}

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

@@ -9,14 +9,13 @@
 // except according to those terms.
 
 use prelude::*;
-use super::*;
 use super::super::*;
 use super::ip::IpAddr;
 
 pub struct TcpStream;
 
 impl TcpStream {
-    pub fn connect(_addr: IpAddr) -> Result<TcpStream, IoError> {
+    pub fn connect(_addr: IpAddr) -> Option<TcpStream> {
         fail!()
     }
 }
@@ -40,7 +39,7 @@ impl Close for TcpStream {
 pub struct TcpListener;
 
 impl TcpListener {
-    pub fn new(_addr: IpAddr) -> TcpListener {
+    pub fn bind(_addr: IpAddr) -> Option<TcpListener> {
         fail!()
     }
 }
@@ -48,3 +47,28 @@ impl TcpListener {
 impl Listener<TcpStream> for TcpListener {
     fn accept(&mut self) -> Option<TcpStream> { fail!() }
 }
+
+#[cfg(test)]
+mod test {
+
+    #[test] #[ignore]
+    fn smoke_test() {
+        /*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 spawn_immediately {
+                let stream = TcpStream::connect(addr);
+                stream.write([99]);
+            }
+        }*/
+    }
+}

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

@@ -9,14 +9,13 @@
 // except according to those terms.
 
 use prelude::*;
-use super::*;
 use super::super::*;
 use super::ip::IpAddr;
 
 pub struct UdpStream;
 
 impl UdpStream {
-    pub fn connect(_addr: IpAddr) -> Result<UdpStream, IoError> {
+    pub fn connect(_addr: IpAddr) -> Option<UdpStream> {
         fail!()
     }
 }
@@ -40,7 +39,7 @@ impl Close for UdpStream {
 pub struct UdpListener;
 
 impl UdpListener {
-    pub fn new(_addr: IpAddr) -> UdpListener {
+    pub fn bind(_addr: IpAddr) -> Option<UdpListener> {
         fail!()
     }
 }

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

@@ -9,14 +9,13 @@
 // except according to those terms.
 
 use prelude::*;
-use super::*;
 use super::super::*;
-use super::super::misc::PathLike;
+use super::super::support::PathLike;
 
 pub struct UnixStream;
 
 impl UnixStream {
-    pub fn connect<P: PathLike>(_path: &P) -> Result<UnixStream, IoError> {
+    pub fn connect<P: PathLike>(_path: &P) -> Option<UnixStream> {
         fail!()
     }
 }
@@ -40,7 +39,7 @@ impl Close for UnixStream {
 pub struct UnixListener;
 
 impl UnixListener {
-    pub fn new<P: PathLike>(_path: &P) -> UnixListener {
+    pub fn bind<P: PathLike>(_path: &P) -> Option<UnixListener> {
         fail!()
     }
 }

src/libcore/rt/io/option.rs

@@ -0,0 +1,153 @@
+// 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.
+
+//! Implementations of I/O traits for the Option type
+//!
+//! I/O constructors return option types to allow errors to be handled.
+//! These implementations allow e.g. `Option<FileStream>` to be used
+//! as a `Reader` without unwrapping the option first.
+//!
+//! # XXX Seek and Close
+
+use option::*;
+use super::{Reader, Writer, Listener};
+use super::{standard_error, PreviousIoError, io_error, IoError};
+
+fn prev_io_error() -> IoError {
+    standard_error(PreviousIoError)
+}
+
+impl<W: Writer> Writer for Option<W> {
+    fn write(&mut self, buf: &[u8]) {
+        match *self {
+            Some(ref mut writer) => writer.write(buf),
+            None => io_error::cond.raise(prev_io_error())
+        }
+    }
+
+    fn flush(&mut self) {
+        match *self {
+            Some(ref mut writer) => writer.flush(),
+            None => io_error::cond.raise(prev_io_error())
+        }
+    }
+}
+
+impl<R: Reader> Reader for Option<R> {
+    fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
+        match *self {
+            Some(ref mut reader) => reader.read(buf),
+            None => {
+                io_error::cond.raise(prev_io_error());
+                None
+            }
+        }
+    }
+
+    fn eof(&mut self) -> bool {
+        match *self {
+            Some(ref mut reader) => reader.eof(),
+            None => {
+                io_error::cond.raise(prev_io_error());
+                true
+            }
+        }
+    }
+}
+
+impl<L: Listener<S>, S> Listener<S> for Option<L> {
+    fn accept(&mut self) -> Option<S> {
+        match *self {
+            Some(ref mut listener) => listener.accept(),
+            None => {
+                io_error::cond.raise(prev_io_error());
+                None
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use option::*;
+    use super::super::mem::*;
+    use rt::test::*;
+    use super::super::{PreviousIoError, io_error};
+
+    #[test]
+    fn test_option_writer() {
+        do run_in_newsched_task {
+            let mut writer: Option<MemWriter> = Some(MemWriter::new());
+            writer.write([0, 1, 2]);
+            writer.flush();
+            assert!(writer.unwrap().inner() == ~[0, 1, 2]);
+        }
+    }
+
+    #[test]
+    fn test_option_writer_error() {
+        do run_in_newsched_task {
+            let mut writer: Option<MemWriter> = None;
+
+            let mut called = false;
+            do io_error::cond.trap(|err| {
+                assert!(err.kind == PreviousIoError);
+                called = true;
+            }).in {
+                writer.write([0, 0, 0]);
+            }
+            assert!(called);
+
+            let mut called = false;
+            do io_error::cond.trap(|err| {
+                assert!(err.kind == PreviousIoError);
+                called = true;
+            }).in {
+                writer.flush();
+            }
+            assert!(called);
+        }
+    }
+
+    #[test]
+    fn test_option_reader() {
+        do run_in_newsched_task {
+            let mut reader: Option<MemReader> = Some(MemReader::new(~[0, 1, 2, 3]));
+            let mut buf = [0, 0];
+            reader.read(buf);
+            assert!(buf == [0, 1]);
+            assert!(!reader.eof());
+        }
+    }
+
+    #[test]
+    fn test_option_reader_error() {
+        let mut reader: Option<MemReader> = None;
+        let mut buf = [];
+
+        let mut called = false;
+        do io_error::cond.trap(|err| {
+            assert!(err.kind == PreviousIoError);
+            called = true;
+        }).in {
+            reader.read(buf);
+        }
+        assert!(called);
+
+        let mut called = false;
+        do io_error::cond.trap(|err| {
+            assert!(err.kind == PreviousIoError);
+            called = true;
+        }).in {
+            assert!(reader.eof());
+        }
+        assert!(called);
+    }
+}

src/libcore/rt/local_heap.rs

@@ -0,0 +1,81 @@
+// 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 local, garbage collected heap
+
+use libc::{c_void, uintptr_t, size_t};
+use ops::Drop;
+
+type MemoryRegion = c_void;
+type BoxedRegion = c_void;
+
+pub type OpaqueBox = c_void;
+pub type TypeDesc = c_void;
+
+pub struct LocalHeap {
+    memory_region: *MemoryRegion,
+    boxed_region: *BoxedRegion
+}
+
+impl LocalHeap {
+    pub fn new() -> LocalHeap {
+        unsafe {
+            // Don't need synchronization for the single-threaded local heap
+            let synchronized = false as uintptr_t;
+            // XXX: These usually come from the environment
+            let detailed_leaks = false as uintptr_t;
+            let poison_on_free = false as uintptr_t;
+            let region = rust_new_memory_region(synchronized, detailed_leaks, poison_on_free);
+            assert!(region.is_not_null());
+            let boxed = rust_new_boxed_region(region, poison_on_free);
+            assert!(boxed.is_not_null());
+            LocalHeap {
+                memory_region: region,
+                boxed_region: boxed
+            }
+        }
+    }
+
+    pub fn alloc(&mut self, td: *TypeDesc, size: uint) -> *OpaqueBox {
+        unsafe {
+            return rust_boxed_region_malloc(self.boxed_region, td, size as size_t);
+        }
+    }
+
+    pub fn free(&mut self, box: *OpaqueBox) {
+        unsafe {
+            return rust_boxed_region_free(self.boxed_region, box);
+        }
+    }
+}
+
+impl Drop for LocalHeap {
+    fn finalize(&self) {
+        unsafe {
+            rust_delete_boxed_region(self.boxed_region);
+            rust_delete_memory_region(self.memory_region);
+        }
+    }
+}
+
+extern {
+    fn rust_new_memory_region(synchronized: uintptr_t,
+                               detailed_leaks: uintptr_t,
+                               poison_on_free: uintptr_t) -> *MemoryRegion;
+    fn rust_delete_memory_region(region: *MemoryRegion);
+    fn rust_new_boxed_region(region: *MemoryRegion,
+                             poison_on_free: uintptr_t) -> *BoxedRegion;
+    fn rust_delete_boxed_region(region: *BoxedRegion);
+    fn rust_boxed_region_malloc(region: *BoxedRegion,
+                                td: *TypeDesc,
+                                size: size_t) -> *OpaqueBox;
+    fn rust_boxed_region_free(region: *BoxedRegion, box: *OpaqueBox);
+}
+

src/libcore/rt/local_services.rs

@@ -0,0 +1,223 @@
+// 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.
+
+//! Language-level runtime services that should reasonably expected
+//! to be available 'everywhere'. Local heaps, GC, unwinding,
+//! local storage, and logging. Even a 'freestanding' Rust would likely want
+//! to implement this.
+
+//! Local services may exist in at least three different contexts:
+//! when running as a task, when running in the scheduler's context,
+//! or when running outside of a scheduler but with local services
+//! (freestanding rust with local services?).
+
+use prelude::*;
+use libc::{c_void, uintptr_t};
+use cast::transmute;
+use super::sched::local_sched;
+use super::local_heap::LocalHeap;
+
+pub struct LocalServices {
+    heap: LocalHeap,
+    gc: GarbageCollector,
+    storage: LocalStorage,
+    logger: Logger,
+    unwinder: Option<Unwinder>,
+    destroyed: bool
+}
+
+pub struct GarbageCollector;
+pub struct LocalStorage(*c_void, Option<~fn(*c_void)>);
+pub struct Logger;
+
+pub struct Unwinder {
+    unwinding: bool,
+}
+
+impl LocalServices {
+    pub fn new() -> LocalServices {
+        LocalServices {
+            heap: LocalHeap::new(),
+            gc: GarbageCollector,
+            storage: LocalStorage(ptr::null(), None),
+            logger: Logger,
+            unwinder: Some(Unwinder { unwinding: false }),
+            destroyed: false
+        }
+    }
+
+    pub fn without_unwinding() -> LocalServices {
+        LocalServices {
+            heap: LocalHeap::new(),
+            gc: GarbageCollector,
+            storage: LocalStorage(ptr::null(), None),
+            logger: Logger,
+            unwinder: None,
+            destroyed: false
+        }
+    }
+
+    pub fn run(&mut self, f: &fn()) {
+        // This is just an assertion that `run` was called unsafely
+        // and this instance of LocalServices is still accessible.
+        do borrow_local_services |sched| {
+            assert!(ptr::ref_eq(sched, self));
+        }
+
+        match self.unwinder {
+            Some(ref mut unwinder) => {
+                // If there's an unwinder then set up the catch block
+                unwinder.try(f);
+            }
+            None => {
+                // Otherwise, just run the body
+                f()
+            }
+        }
+        self.destroy();
+    }
+
+    /// Must be called manually before finalization to clean up
+    /// thread-local resources. Some of the routines here expect
+    /// LocalServices to be available recursively so this must be
+    /// called unsafely, without removing LocalServices from
+    /// thread-local-storage.
+    fn destroy(&mut self) {
+        // This is just an assertion that `destroy` was called unsafely
+        // and this instance of LocalServices is still accessible.
+        do borrow_local_services |sched| {
+            assert!(ptr::ref_eq(sched, self));
+        }
+        match self.storage {
+            LocalStorage(ptr, Some(ref dtor)) => {
+                (*dtor)(ptr)
+            }
+            _ => ()
+        }
+        self.destroyed = true;
+    }
+}
+
+impl Drop for LocalServices {
+    fn finalize(&self) { assert!(self.destroyed) }
+}
+
+// Just a sanity check to make sure we are catching a Rust-thrown exception
+static UNWIND_TOKEN: uintptr_t = 839147;
+
+impl Unwinder {
+    pub fn try(&mut self, f: &fn()) {
+        use sys::Closure;
+
+        unsafe {
+            let closure: Closure = transmute(f);
+            let code = transmute(closure.code);
+            let env = transmute(closure.env);
+
+            let token = rust_try(try_fn, code, env);
+            assert!(token == 0 || token == UNWIND_TOKEN);
+        }
+
+        extern fn try_fn(code: *c_void, env: *c_void) {
+            unsafe {
+                let closure: Closure = Closure {
+                    code: transmute(code),
+                    env: transmute(env),
+                };
+                let closure: &fn() = transmute(closure);
+                closure();
+            }
+        }
+
+        extern {
+            #[rust_stack]
+            fn rust_try(f: *u8, code: *c_void, data: *c_void) -> uintptr_t;
+        }
+    }
+
+    pub fn begin_unwind(&mut self) -> ! {
+        self.unwinding = true;
+        unsafe {
+            rust_begin_unwind(UNWIND_TOKEN);
+            return transmute(());
+        }
+        extern {
+            fn rust_begin_unwind(token: uintptr_t);
+        }
+    }
+}
+
+/// Borrow a pointer to the installed local services.
+/// Fails (likely aborting the process) if local services are not available.
+pub fn borrow_local_services(f: &fn(&mut LocalServices)) {
+    do local_sched::borrow |sched| {
+        match sched.current_task {
+            Some(~ref mut task) => {
+                f(&mut task.local_services)
+            }
+            None => {
+                fail!(~"no local services for schedulers yet")
+            }
+        }
+    }
+}
+
+pub unsafe fn unsafe_borrow_local_services() -> &mut LocalServices {
+    use cast::transmute_mut_region;
+
+    match local_sched::unsafe_borrow().current_task {
+        Some(~ref mut task) => {
+            transmute_mut_region(&mut task.local_services)
+        }
+        None => {
+            fail!(~"no local services for schedulers yet")
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use rt::test::*;
+
+    #[test]
+    fn local_heap() {
+        do run_in_newsched_task() {
+            let a = @5;
+            let b = a;
+            assert!(*a == 5);
+            assert!(*b == 5);
+        }
+    }
+
+    #[test]
+    fn tls() {
+        use task::local_data::*;
+        do run_in_newsched_task() {
+            unsafe {
+                fn key(_x: @~str) { }
+                local_data_set(key, @~"data");
+                assert!(*local_data_get(key).get() == ~"data");
+                fn key2(_x: @~str) { }
+                local_data_set(key2, @~"data");
+                assert!(*local_data_get(key2).get() == ~"data");
+            }
+        }
+    }
+
+    #[test]
+    fn unwind() {
+        do run_in_newsched_task() {
+            let result = spawntask_try(||());
+            assert!(result.is_ok());
+            let result = spawntask_try(|| fail!());
+            assert!(result.is_err());
+        }
+    }
+}

src/libcore/rt/mod.rs

@@ -8,30 +8,12 @@
 // 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 */
+
 #[doc(hidden)];
 
 use libc::c_char;
 
-// Some basic logging
-macro_rules! rtdebug_ (
-    ($( $arg:expr),+) => ( {
-        dumb_println(fmt!( $($arg),+ ));
-
-        fn dumb_println(s: &str) {
-            use io::WriterUtil;
-            let dbg = ::libc::STDERR_FILENO as ::io::fd_t;
-            dbg.write_str(s);
-            dbg.write_str("\n");
-        }
-
-    } )
-)
-
-// An alternate version with no output, for turning off logging
-macro_rules! rtdebug (
-    ($( $arg:expr),+) => ( $(let _ = $arg)*; )
-)
-
 #[path = "sched/mod.rs"]
 mod sched;
 mod rtio;
@@ -48,6 +30,12 @@ mod stack;
 mod context;
 mod thread;
 pub mod env;
+pub mod local_services;
+mod local_heap;
+
+/// Tools for testing the runtime
+#[cfg(test)]
+pub mod test;
 
 #[cfg(stage0)]
 pub fn start(main: *u8, _argc: int, _argv: *c_char, _crate_map: *u8) -> int {
@@ -93,7 +81,7 @@ pub fn start(main: *u8, _argc: int, _argv: **c_char, _crate_map: *u8) -> int {
 /// Different runtime services are available depending on context.
 #[deriving(Eq)]
 pub enum RuntimeContext {
-    // Only default services, e.g. exchange heap
+    // Only the exchange heap is available
     GlobalContext,
     // The scheduler may be accessed
     SchedulerContext,
@@ -160,24 +148,3 @@ fn test_context() {
         sched.run();
     }
 }
-
-// For setting up tests of the new scheduler
-#[cfg(test)]
-pub fn run_in_newsched_task(f: ~fn()) {
-    use cell::Cell;
-    use unstable::run_in_bare_thread;
-    use self::sched::Task;
-    use self::uvio::UvEventLoop;
-
-    let f = Cell(Cell(f));
-
-    do run_in_bare_thread {
-        let mut sched = ~UvEventLoop::new_scheduler();
-        let f = f.take();
-        let task = ~do Task::new(&mut sched.stack_pool) {
-            (f.take())();
-        };
-        sched.task_queue.push_back(task);
-        sched.run();
-    }
-}

src/libcore/rt/sched/mod.rs

@@ -16,6 +16,7 @@ use super::work_queue::WorkQueue;
 use super::stack::{StackPool, StackSegment};
 use super::rtio::{EventLoop, EventLoopObject};
 use super::context::Context;
+use super::local_services::LocalServices;
 use cell::Cell;
 
 #[cfg(test)] use super::uvio::UvEventLoop;
@@ -38,7 +39,7 @@ pub struct Scheduler {
     /// Always valid when a task is executing, otherwise not
     priv saved_context: Context,
     /// The currently executing task
-    priv current_task: Option<~Task>,
+    current_task: Option<~Task>,
     /// An action performed after a context switch on behalf of the
     /// code running before the context switch
     priv cleanup_job: Option<CleanupJob>
@@ -148,7 +149,7 @@ pub impl Scheduler {
             }
         }
 
-        // Control never reaches here
+        abort!("control reached end of task");
     }
 
     fn schedule_new_task(~self, task: ~Task) {
@@ -326,10 +327,18 @@ pub struct Task {
     /// These are always valid when the task is not running, unless
     /// the task is dead
     priv saved_context: Context,
+    /// The heap, GC, unwinding, local storage, logging
+    local_services: LocalServices
 }
 
 pub impl Task {
     fn new(stack_pool: &mut StackPool, start: ~fn()) -> Task {
+        Task::with_local(stack_pool, LocalServices::new(), start)
+    }
+
+    fn with_local(stack_pool: &mut StackPool,
+                  local_services: LocalServices,
+                  start: ~fn()) -> Task {
         let start = Task::build_start_wrapper(start);
         let mut stack = stack_pool.take_segment(TASK_MIN_STACK_SIZE);
         // NB: Context holds a pointer to that ~fn
@@ -337,6 +346,7 @@ pub impl Task {
         return Task {
             current_stack_segment: stack,
             saved_context: initial_context,
+            local_services: local_services
         };
     }
 
@@ -349,9 +359,12 @@ pub impl Task {
             unsafe {
                 let sched = local_sched::unsafe_borrow();
                 sched.run_cleanup_job();
-            }
 
-            start();
+                let sched = local_sched::unsafe_borrow();
+                let task = sched.current_task.get_mut_ref();
+                // FIXME #6141: shouldn't neet to put `start()` in another closure
+                task.local_services.run(||start());
+            }
 
             let sched = local_sched::take();
             sched.terminate_current_task();

src/libcore/rt/test.rs

@@ -0,0 +1,120 @@
+// 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 cell::Cell;
+use result::{Result, Ok, Err};
+use super::io::net::ip::{IpAddr, Ipv4};
+use rt::local_services::LocalServices;
+
+/// Creates a new scheduler in a new thread and runs a task in it,
+/// then waits for the scheduler to exit. Failure of the task
+/// will abort the process.
+pub fn run_in_newsched_task(f: ~fn()) {
+    use unstable::run_in_bare_thread;
+    use super::sched::Task;
+    use super::uvio::UvEventLoop;
+
+    let f = Cell(f);
+
+    do run_in_bare_thread {
+        let mut sched = ~UvEventLoop::new_scheduler();
+        let task = ~Task::with_local(&mut sched.stack_pool,
+                                     LocalServices::without_unwinding(),
+                                     f.take());
+        sched.task_queue.push_back(task);
+        sched.run();
+    }
+}
+
+/// Test tasks will abort on failure instead of unwinding
+pub fn spawntask(f: ~fn()) {
+    use super::sched::*;
+
+    let mut sched = local_sched::take();
+    let task = ~Task::with_local(&mut sched.stack_pool,
+                                 LocalServices::without_unwinding(),
+                                 f);
+    do sched.switch_running_tasks_and_then(task) |task| {
+        let task = Cell(task);
+        let sched = local_sched::take();
+        sched.schedule_new_task(task.take());
+    }
+}
+
+/// Create a new task and run it right now. Aborts on failure
+pub fn spawntask_immediately(f: ~fn()) {
+    use super::sched::*;
+
+    let mut sched = local_sched::take();
+    let task = ~Task::with_local(&mut sched.stack_pool,
+                                 LocalServices::without_unwinding(),
+                                 f);
+    do sched.switch_running_tasks_and_then(task) |task| {
+        let task = Cell(task);
+        do local_sched::borrow |sched| {
+            sched.task_queue.push_front(task.take());
+        }
+    }
+}
+
+/// Spawn a task and wait for it to finish, returning whether it completed successfully or failed
+pub fn spawntask_try(f: ~fn()) -> Result<(), ()> {
+    use cell::Cell;
+    use super::sched::*;
+    use task;
+    use unstable::finally::Finally;
+
+    // Our status variables will be filled in from the scheduler context
+    let mut failed = false;
+    let failed_ptr: *mut bool = &mut failed;
+
+    // Switch to the scheduler
+    let f = Cell(Cell(f));
+    let mut sched = local_sched::take();
+    do sched.deschedule_running_task_and_then() |old_task| {
+        let old_task = Cell(old_task);
+        let f = f.take();
+        let mut sched = local_sched::take();
+        let new_task = ~do Task::new(&mut sched.stack_pool) {
+            do (|| {
+                (f.take())()
+            }).finally {
+                // Check for failure then resume the parent task
+                unsafe { *failed_ptr = task::failing(); }
+                let sched = local_sched::take();
+                do sched.switch_running_tasks_and_then(old_task.take()) |new_task| {
+                    let new_task = Cell(new_task);
+                    do local_sched::borrow |sched| {
+                        sched.task_queue.push_front(new_task.take());
+                    }
+                }
+            }
+        };
+
+        sched.resume_task_immediately(new_task);
+    }
+
+    if !failed { Ok(()) } else { Err(()) }
+}
+
+/// Get a port number, starting at 9600, for use in tests
+pub fn next_test_port() -> u16 {
+    unsafe {
+        return rust_dbg_next_port() as u16;
+    }
+    extern {
+        fn rust_dbg_next_port() -> ::libc::uintptr_t;
+    }
+}
+
+/// Get a unique localhost:port pair starting at 9600
+pub fn next_test_ip4() -> IpAddr {
+    Ipv4(127, 0, 0, 1, next_test_port())
+}

src/libcore/rt/uv/mod.rs

@@ -301,7 +301,8 @@ struct WatcherData {
     write_cb: Option<ConnectionCallback>,
     connect_cb: Option<ConnectionCallback>,
     close_cb: Option<NullCallback>,
-    alloc_cb: Option<AllocCallback>
+    alloc_cb: Option<AllocCallback>,
+    buf: Option<Buf>
 }
 
 pub fn install_watcher_data<H, W: Watcher + NativeHandle<*H>>(watcher: &mut W) {
@@ -311,7 +312,8 @@ pub fn install_watcher_data<H, W: Watcher + NativeHandle<*H>>(watcher: &mut W) {
             write_cb: None,
             connect_cb: None,
             close_cb: None,
-            alloc_cb: None
+            alloc_cb: None,
+            buf: None
         };
         let data = transmute::<~WatcherData, *c_void>(data);
         uvll::set_data_for_uv_handle(watcher.native_handle(), data);

src/libcore/rt/uv/net.rs

@@ -19,12 +19,10 @@ use super::{Loop, Watcher, Request, UvError, Buf, Callback, NativeHandle, NullCa
             vec_to_uv_buf, vec_from_uv_buf};
 use super::super::io::net::ip::{IpAddr, Ipv4, Ipv6};
 
-#[cfg(test)]
-use unstable::run_in_bare_thread;
-#[cfg(test)]
-use super::super::thread::Thread;
-#[cfg(test)]
-use cell::Cell;
+#[cfg(test)] use cell::Cell;
+#[cfg(test)] use unstable::run_in_bare_thread;
+#[cfg(test)] use super::super::thread::Thread;
+#[cfg(test)] use super::super::test::*;
 
 fn ip4_as_uv_ip4(addr: IpAddr, f: &fn(*sockaddr_in)) {
     match addr {
@@ -109,21 +107,25 @@ pub impl StreamWatcher {
 
         let req = WriteRequest::new();
         let buf = vec_to_uv_buf(msg);
-        // XXX: Allocation
-        let bufs = ~[buf];
+        assert!(data.buf.is_none());
+        data.buf = Some(buf);
+        let bufs = [buf];
         unsafe {
             assert!(0 == uvll::write(req.native_handle(),
                                           self.native_handle(),
-                                          &bufs, write_cb));
+                                          bufs, write_cb));
         }
-        // XXX: Freeing immediately after write. Is this ok?
-        let _v = vec_from_uv_buf(buf);
 
         extern fn write_cb(req: *uvll::uv_write_t, status: c_int) {
             let write_request: WriteRequest = NativeHandle::from_native_handle(req);
             let mut stream_watcher = write_request.stream();
             write_request.delete();
-            let cb = get_watcher_data(&mut stream_watcher).write_cb.swap_unwrap();
+            let cb = {
+                let data = get_watcher_data(&mut stream_watcher);
+                let _vec = vec_from_uv_buf(data.buf.swap_unwrap());
+                let cb = data.write_cb.swap_unwrap();
+                cb
+            };
             let status = status_to_maybe_uv_error(stream_watcher.native_handle(), status);
             cb(stream_watcher, status);
         }
@@ -361,7 +363,7 @@ fn connect_close() {
         let mut loop_ = Loop::new();
         let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
         // Connect to a port where nobody is listening
-        let addr = Ipv4(127, 0, 0, 1, 2923);
+        let addr = next_test_ip4();
         do tcp_watcher.connect(addr) |stream_watcher, status| {
             rtdebug!("tcp_watcher.connect!");
             assert!(status.is_some());
@@ -373,47 +375,13 @@ fn connect_close() {
     }
 }
 
-#[test]
-#[ignore(reason = "need a server to connect to")]
-fn connect_read() {
-    do run_in_bare_thread() {
-        let mut loop_ = Loop::new();
-        let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
-        let addr = Ipv4(127, 0, 0, 1, 2924);
-        do tcp_watcher.connect(addr) |stream_watcher, status| {
-            let mut stream_watcher = stream_watcher;
-            rtdebug!("tcp_watcher.connect!");
-            assert!(status.is_none());
-            let alloc: AllocCallback = |size| {
-                vec_to_uv_buf(vec::from_elem(size, 0))
-            };
-            do stream_watcher.read_start(alloc)
-                |stream_watcher, _nread, buf, status| {
-
-                let buf = vec_from_uv_buf(buf);
-                rtdebug!("read cb!");
-                if status.is_none() {
-                    let _bytes = buf.unwrap();
-                    rtdebug!("%s", bytes.slice(0, nread as uint).to_str());
-                } else {
-                    rtdebug!("status after read: %s", status.get().to_str());
-                    rtdebug!("closing");
-                    stream_watcher.close(||());
-                }
-            }
-        }
-        loop_.run();
-        loop_.close();
-    }
-}
-
 #[test]
 fn listen() {
     do run_in_bare_thread() {
         static MAX: int = 10;
         let mut loop_ = Loop::new();
         let mut server_tcp_watcher = { TcpWatcher::new(&mut loop_) };
-        let addr = Ipv4(127, 0, 0, 1, 2925);
+        let addr = next_test_ip4();
         server_tcp_watcher.bind(addr);
         let loop_ = loop_;
         rtdebug!("listening");

src/libcore/rt/uvio.rs

@@ -19,10 +19,9 @@ use cell::{Cell, empty_cell};
 use cast::transmute;
 use super::sched::{Scheduler, local_sched};
 
-#[cfg(test)] use super::io::net::ip::Ipv4;
-#[cfg(test)] use super::sched::Task;
-#[cfg(test)] use unstable::run_in_bare_thread;
 #[cfg(test)] use uint;
+#[cfg(test)] use unstable::run_in_bare_thread;
+#[cfg(test)] use super::test::*;
 
 pub struct UvEventLoop {
     uvio: UvIoFactory
@@ -335,38 +334,22 @@ impl Stream for UvStream {
 }
 
 #[test]
-#[ignore(reason = "ffi struct issues")]
 fn test_simple_io_no_connect() {
-    do run_in_bare_thread {
-        let mut sched = ~UvEventLoop::new_scheduler();
-        let task = ~do Task::new(&mut sched.stack_pool) {
-            let io = unsafe { local_sched::unsafe_borrow_io() };
-            let addr = Ipv4(127, 0, 0, 1, 2926);
-            let maybe_chan = io.connect(addr);
-            assert!(maybe_chan.is_none());
-        };
-        sched.task_queue.push_back(task);
-        sched.run();
+    do run_in_newsched_task {
+        let io = unsafe { local_sched::unsafe_borrow_io() };
+        let addr = next_test_ip4();
+        let maybe_chan = io.connect(addr);
+        assert!(maybe_chan.is_none());
     }
 }
 
 #[test]
-#[ignore(reason = "ffi struct issues")]
 fn test_simple_tcp_server_and_client() {
-    do run_in_bare_thread {
-        let mut sched = ~UvEventLoop::new_scheduler();
-        let addr = Ipv4(127, 0, 0, 1, 2929);
+    do run_in_newsched_task {
+        let addr = next_test_ip4();
 
-        let client_task = ~do Task::new(&mut sched.stack_pool) {
-            unsafe {
-                let io = local_sched::unsafe_borrow_io();
-                let mut stream = io.connect(addr).unwrap();
-                stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
-                stream.close();
-            }
-        };
-
-        let server_task = ~do Task::new(&mut sched.stack_pool) {
+        // Start the server first so it's listening when we connect
+        do spawntask_immediately {
             unsafe {
                 let io = local_sched::unsafe_borrow_io();
                 let mut listener = io.bind(addr).unwrap();
@@ -381,32 +364,25 @@ fn test_simple_tcp_server_and_client() {
                 stream.close();
                 listener.close();
             }
-        };
+        }
 
-        // Start the server first so it listens before the client connects
-        sched.task_queue.push_back(server_task);
-        sched.task_queue.push_back(client_task);
-        sched.run();
+        do spawntask_immediately {
+            unsafe {
+                let io = local_sched::unsafe_borrow_io();
+                let mut stream = io.connect(addr).unwrap();
+                stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
+                stream.close();
+            }
+        }
     }
 }
 
 #[test] #[ignore(reason = "busted")]
 fn test_read_and_block() {
-    do run_in_bare_thread {
-        let mut sched = ~UvEventLoop::new_scheduler();
-        let addr = Ipv4(127, 0, 0, 1, 2930);
+    do run_in_newsched_task {
+        let addr = next_test_ip4();
 
-        let client_task = ~do Task::new(&mut sched.stack_pool) {
-            let io = unsafe { local_sched::unsafe_borrow_io() };
-            let mut stream = io.connect(addr).unwrap();
-            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
-            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
-            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
-            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
-            stream.close();
-        };
-
-        let server_task = ~do Task::new(&mut sched.stack_pool) {
+        do spawntask_immediately {
             let io = unsafe { local_sched::unsafe_borrow_io() };
             let mut listener = io.bind(addr).unwrap();
             let mut stream = listener.listen().unwrap();
@@ -442,36 +418,58 @@ fn test_read_and_block() {
 
             stream.close();
             listener.close();
-        };
+        }
+
+        do spawntask_immediately {
+            let io = unsafe { local_sched::unsafe_borrow_io() };
+            let mut stream = io.connect(addr).unwrap();
+            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
+            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
+            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
+            stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
+            stream.close();
+        }
 
-        // Start the server first so it listens before the client connects
-        sched.task_queue.push_back(server_task);
-        sched.task_queue.push_back(client_task);
-        sched.run();
     }
 }
 
-#[test] #[ignore(reason = "needs server")]
+#[test]
 fn test_read_read_read() {
-    do run_in_bare_thread {
-        let mut sched = ~UvEventLoop::new_scheduler();
-        let addr = Ipv4(127, 0, 0, 1, 2931);
+    do run_in_newsched_task {
+        let addr = next_test_ip4();
+        static MAX: uint = 500000;
 
-        let client_task = ~do Task::new(&mut sched.stack_pool) {
+        do spawntask_immediately {
+            unsafe {
+                let io = local_sched::unsafe_borrow_io();
+                let mut listener = io.bind(addr).unwrap();
+                let mut stream = listener.listen().unwrap();
+                let mut buf = [1, .. 2048];
+                let mut total_bytes_written = 0;
+                while total_bytes_written < MAX {
+                    stream.write(buf);
+                    total_bytes_written += buf.len();
+                }
+                stream.close();
+                listener.close();
+            }
+        }
+
+        do spawntask_immediately {
             let io = unsafe { local_sched::unsafe_borrow_io() };
             let mut stream = io.connect(addr).unwrap();
             let mut buf = [0, .. 2048];
             let mut total_bytes_read = 0;
-            while total_bytes_read < 500000000 {
+            while total_bytes_read < MAX {
                 let nread = stream.read(buf).unwrap();
                 rtdebug!("read %u bytes", nread as uint);
                 total_bytes_read += nread;
+                for uint::range(0, nread) |i| {
+                    assert!(buf[i] == 1);
+                }
             }
-            rtdebug_!("read %u bytes total", total_bytes_read as uint);
+            rtdebug!("read %u bytes total", total_bytes_read as uint);
             stream.close();
-        };
-
-        sched.task_queue.push_back(client_task);
-        sched.run();
+        }
     }
 }

src/libcore/rt/uvll.rs

@@ -219,9 +219,9 @@ pub unsafe fn accept(server: *c_void, client: *c_void) -> c_int {
     return rust_uv_accept(server as *c_void, client as *c_void);
 }
 
-pub unsafe fn write<T>(req: *uv_write_t, stream: *T, buf_in: *~[uv_buf_t], cb: *u8) -> c_int {
-    let buf_ptr = vec::raw::to_ptr(*buf_in);
-    let buf_cnt = vec::len(*buf_in) as i32;
+pub unsafe fn write<T>(req: *uv_write_t, stream: *T, buf_in: &[uv_buf_t], cb: *u8) -> c_int {
+    let buf_ptr = vec::raw::to_ptr(buf_in);
+    let buf_cnt = vec::len(buf_in) as i32;
     return rust_uv_write(req as *c_void, stream as *c_void, buf_ptr, buf_cnt, cb);
 }
 pub unsafe fn read_start(stream: *uv_stream_t, on_alloc: *u8, on_read: *u8) -> c_int {

src/libcore/sys.rs

@@ -10,6 +10,7 @@
 
 //! Misc low level stuff
 
+use option::{Some, None};
 use cast;
 use cmp::{Eq, Ord};
 use gc;
@@ -202,6 +203,7 @@ impl FailWithCause for &'static str {
 // NOTE: remove function after snapshot
 #[cfg(stage0)]
 pub fn begin_unwind(msg: ~str, file: ~str, line: uint) -> ! {
+
     do str::as_buf(msg) |msg_buf, _msg_len| {
         do str::as_buf(file) |file_buf, _file_len| {
             unsafe {
@@ -215,10 +217,28 @@ pub fn begin_unwind(msg: ~str, file: ~str, line: uint) -> ! {
 
 // FIXME #4427: Temporary until rt::rt_fail_ goes away
 pub fn begin_unwind_(msg: *c_char, file: *c_char, line: size_t) -> ! {
-    unsafe {
-        gc::cleanup_stack_for_failure();
-        rustrt::rust_upcall_fail(msg, file, line);
-        cast::transmute(())
+    use rt::{context, OldTaskContext};
+    use rt::local_services::unsafe_borrow_local_services;
+
+    match context() {
+        OldTaskContext => {
+            unsafe {
+                gc::cleanup_stack_for_failure();
+                rustrt::rust_upcall_fail(msg, file, line);
+                cast::transmute(())
+            }
+        }
+        _ => {
+            // XXX: Need to print the failure message
+            gc::cleanup_stack_for_failure();
+            unsafe {
+                let local_services = unsafe_borrow_local_services();
+                match local_services.unwinder {
+                    Some(ref mut unwinder) => unwinder.begin_unwind(),
+                    None => abort!("failure without unwinder. aborting process")
+                }
+            }
+        }
     }
 }
 

src/libcore/task/local_data.rs

@@ -27,8 +27,7 @@ magic.
 */
 
 use prelude::*;
-use task::local_data_priv::{local_get, local_pop, local_modify, local_set};
-use task::rt;
+use task::local_data_priv::{local_get, local_pop, local_modify, local_set, Handle};
 
 /**
  * Indexes a task-local data slot. The function's code pointer is used for
@@ -53,7 +52,7 @@ pub type LocalDataKey<'self,T> = &'self fn(v: @T);
 pub unsafe fn local_data_pop<T:Durable>(
     key: LocalDataKey<T>) -> Option<@T> {
 
-    local_pop(rt::rust_get_task(), key)
+    local_pop(Handle::new(), key)
 }
 /**
  * Retrieve a task-local data value. It will also be kept alive in the
@@ -62,7 +61,7 @@ pub unsafe fn local_data_pop<T:Durable>(
 pub unsafe fn local_data_get<T:Durable>(
     key: LocalDataKey<T>) -> Option<@T> {
 
-    local_get(rt::rust_get_task(), key)
+    local_get(Handle::new(), key)
 }
 /**
  * Store a value in task-local data. If this key already has a value,
@@ -71,7 +70,7 @@ pub unsafe fn local_data_get<T:Durable>(
 pub unsafe fn local_data_set<T:Durable>(
     key: LocalDataKey<T>, data: @T) {
 
-    local_set(rt::rust_get_task(), key, data)
+    local_set(Handle::new(), key, data)
 }
 /**
  * Modify a task-local data value. If the function returns 'None', the
@@ -81,7 +80,7 @@ pub unsafe fn local_data_modify<T:Durable>(
     key: LocalDataKey<T>,
     modify_fn: &fn(Option<@T>) -> Option<@T>) {
 
-    local_modify(rt::rust_get_task(), key, modify_fn)
+    local_modify(Handle::new(), key, modify_fn)
 }
 
 #[test]

src/libcore/task/local_data_priv.rs

@@ -18,6 +18,30 @@ use task::rt;
 use task::local_data::LocalDataKey;
 
 use super::rt::rust_task;
+use rt::local_services::LocalStorage;
+
+pub enum Handle {
+    OldHandle(*rust_task),
+    NewHandle(*mut LocalStorage)
+}
+
+impl Handle {
+    pub fn new() -> Handle {
+        use rt::{context, OldTaskContext};
+        use rt::local_services::unsafe_borrow_local_services;
+        unsafe {
+            match context() {
+                OldTaskContext => {
+                    OldHandle(rt::rust_get_task())
+                }
+                _ => {
+                    let local_services = unsafe_borrow_local_services();
+                    NewHandle(&mut local_services.storage)
+                }
+            }
+        }
+    }
+}
 
 pub trait LocalData { }
 impl<T:Durable> LocalData for @T { }
@@ -25,8 +49,8 @@ impl<T:Durable> LocalData for @T { }
 impl Eq for @LocalData {
     fn eq(&self, other: &@LocalData) -> bool {
         unsafe {
-            let ptr_a: (uint, uint) = cast::transmute(*self);
-            let ptr_b: (uint, uint) = cast::transmute(*other);
+            let ptr_a: &(uint, uint) = cast::transmute(self);
+            let ptr_b: &(uint, uint) = cast::transmute(other);
             return ptr_a == ptr_b;
         }
     }
@@ -39,7 +63,7 @@ type TaskLocalElement = (*libc::c_void, *libc::c_void, @LocalData);
 // Has to be a pointer at outermost layer; the foreign call returns void *.
 type TaskLocalMap = @mut ~[Option<TaskLocalElement>];
 
-extern fn cleanup_task_local_map(map_ptr: *libc::c_void) {
+fn cleanup_task_local_map(map_ptr: *libc::c_void) {
     unsafe {
         assert!(!map_ptr.is_null());
         // Get and keep the single reference that was created at the
@@ -50,8 +74,19 @@ extern fn cleanup_task_local_map(map_ptr: *libc::c_void) {
 }
 
 // Gets the map from the runtime. Lazily initialises if not done so already.
+unsafe fn get_local_map(handle: Handle) -> TaskLocalMap {
+    match handle {
+        OldHandle(task) => get_task_local_map(task),
+        NewHandle(local_storage) => get_newsched_local_map(local_storage)
+    }
+}
+
 unsafe fn get_task_local_map(task: *rust_task) -> TaskLocalMap {
 
+    extern fn cleanup_task_local_map_extern_cb(map_ptr: *libc::c_void) {
+        cleanup_task_local_map(map_ptr);
+    }
+
     // Relies on the runtime initialising the pointer to null.
     // Note: The map's box lives in TLS invisibly referenced once. Each time
     // we retrieve it for get/set, we make another reference, which get/set
@@ -60,7 +95,7 @@ unsafe fn get_task_local_map(task: *rust_task) -> TaskLocalMap {
     if map_ptr.is_null() {
         let map: TaskLocalMap = @mut ~[];
         rt::rust_set_task_local_data(task, cast::transmute(map));
-        rt::rust_task_local_data_atexit(task, cleanup_task_local_map);
+        rt::rust_task_local_data_atexit(task, cleanup_task_local_map_extern_cb);
         // Also need to reference it an extra time to keep it for now.
         let nonmut = cast::transmute::<TaskLocalMap,
                                        @~[Option<TaskLocalElement>]>(map);
@@ -75,6 +110,27 @@ unsafe fn get_task_local_map(task: *rust_task) -> TaskLocalMap {
     }
 }
 
+unsafe fn get_newsched_local_map(local: *mut LocalStorage) -> TaskLocalMap {
+    match &mut *local {
+        &LocalStorage(map_ptr, Some(_)) => {
+            assert!(map_ptr.is_not_null());
+            let map = cast::transmute(map_ptr);
+            let nonmut = cast::transmute::<TaskLocalMap,
+            @~[Option<TaskLocalElement>]>(map);
+            cast::bump_box_refcount(nonmut);
+            return map;
+        }
+        &LocalStorage(ref mut map_ptr, ref mut at_exit) => {
+            assert!((*map_ptr).is_null());
+            let map: TaskLocalMap = @mut ~[];
+            *map_ptr = cast::transmute(map);
+            let at_exit_fn: ~fn(*libc::c_void) = |p|cleanup_task_local_map(p);
+            *at_exit = Some(at_exit_fn);
+            return map;
+        }
+    }
+}
+
 unsafe fn key_to_key_value<T:Durable>(key: LocalDataKey<T>) -> *libc::c_void {
     // Keys are closures, which are (fnptr,envptr) pairs. Use fnptr.
     // Use reintepret_cast -- transmute would leak (forget) the closure.
@@ -102,10 +158,10 @@ unsafe fn local_data_lookup<T:Durable>(
 }
 
 unsafe fn local_get_helper<T:Durable>(
-    task: *rust_task, key: LocalDataKey<T>,
+    handle: Handle, key: LocalDataKey<T>,
     do_pop: bool) -> Option<@T> {
 
-    let map = get_task_local_map(task);
+    let map = get_local_map(handle);
     // Interpreturn our findings from the map
     do local_data_lookup(map, key).map |result| {
         // A reference count magically appears on 'data' out of thin air. It
@@ -124,23 +180,23 @@ unsafe fn local_get_helper<T:Durable>(
 
 
 pub unsafe fn local_pop<T:Durable>(
-    task: *rust_task,
+    handle: Handle,
     key: LocalDataKey<T>) -> Option<@T> {
 
-    local_get_helper(task, key, true)
+    local_get_helper(handle, key, true)
 }
 
 pub unsafe fn local_get<T:Durable>(
-    task: *rust_task,
+    handle: Handle,
     key: LocalDataKey<T>) -> Option<@T> {
 
-    local_get_helper(task, key, false)
+    local_get_helper(handle, key, false)
 }
 
 pub unsafe fn local_set<T:Durable>(
-    task: *rust_task, key: LocalDataKey<T>, data: @T) {
+    handle: Handle, key: LocalDataKey<T>, data: @T) {
 
-    let map = get_task_local_map(task);
+    let map = get_local_map(handle);
     // Store key+data as *voids. Data is invisibly referenced once; key isn't.
     let keyval = key_to_key_value(key);
     // We keep the data in two forms: one as an unsafe pointer, so we can get
@@ -148,7 +204,7 @@ pub unsafe fn local_set<T:Durable>(
     // own on it can be dropped when the box is destroyed. The unsafe pointer
     // does not have a reference associated with it, so it may become invalid
     // when the box is destroyed.
-    let data_ptr = cast::transmute(data);
+    let data_ptr = *cast::transmute::<&@T, &*libc::c_void>(&data);
     let data_box = @data as @LocalData;
     // Construct new entry to store in the map.
     let new_entry = Some((keyval, data_ptr, data_box));
@@ -170,12 +226,12 @@ pub unsafe fn local_set<T:Durable>(
 }
 
 pub unsafe fn local_modify<T:Durable>(
-    task: *rust_task, key: LocalDataKey<T>,
+    handle: Handle, key: LocalDataKey<T>,
     modify_fn: &fn(Option<@T>) -> Option<@T>) {
 
     // Could be more efficient by doing the lookup work, but this is easy.
-    let newdata = modify_fn(local_pop(task, key));
+    let newdata = modify_fn(local_pop(handle, key));
     if newdata.is_some() {
-        local_set(task, key, newdata.unwrap());
+        local_set(handle, key, newdata.unwrap());
     }
 }

src/libcore/task/mod.rs

@@ -559,8 +559,31 @@ pub fn yield() {
 pub fn failing() -> bool {
     //! True if the running task has failed
 
-    unsafe {
-        rt::rust_task_is_unwinding(rt::rust_get_task())
+    use rt::{context, OldTaskContext};
+    use rt::local_services::borrow_local_services;
+
+    match context() {
+        OldTaskContext => {
+            unsafe {
+                rt::rust_task_is_unwinding(rt::rust_get_task())
+            }
+        }
+        _ => {
+            let mut unwinding = false;
+            do borrow_local_services |local| {
+                unwinding = match local.unwinder {
+                    Some(unwinder) => {
+                        unwinder.unwinding
+                    }
+                    None => {
+                        // Because there is no unwinder we can't be unwinding.
+                        // (The process will abort on failure)
+                        false
+                    }
+                }
+            }
+            return unwinding;
+        }
     }
 }
 
@@ -1194,7 +1217,7 @@ fn test_spawn_thread_on_demand() {
 
 #[test]
 fn test_simple_newsched_spawn() {
-    use rt::run_in_newsched_task;
+    use rt::test::run_in_newsched_task;
 
     do run_in_newsched_task {
         spawn(||())

src/libcore/task/spawn.rs

@@ -80,7 +80,7 @@ use prelude::*;
 use unstable;
 use ptr;
 use hashmap::HashSet;
-use task::local_data_priv::{local_get, local_set};
+use task::local_data_priv::{local_get, local_set, OldHandle};
 use task::rt::rust_task;
 use task::rt;
 use task::{Failure, ManualThreads, PlatformThread, SchedOpts, SingleThreaded};
@@ -451,7 +451,7 @@ fn gen_child_taskgroup(linked: bool, supervised: bool)
         /*##################################################################*
          * Step 1. Get spawner's taskgroup info.
          *##################################################################*/
-        let spawner_group = match local_get(spawner, taskgroup_key!()) {
+        let spawner_group = match local_get(OldHandle(spawner), taskgroup_key!()) {
             None => {
                 // Main task, doing first spawn ever. Lazily initialise here.
                 let mut members = new_taskset();
@@ -463,7 +463,7 @@ fn gen_child_taskgroup(linked: bool, supervised: bool)
                 // Main task/group has no ancestors, no notifier, etc.
                 let group =
                     @TCB(spawner, tasks, AncestorList(None), true, None);
-                local_set(spawner, taskgroup_key!(), group);
+                local_set(OldHandle(spawner), taskgroup_key!(), group);
                 group
             }
             Some(group) => group
@@ -627,7 +627,7 @@ fn spawn_raw_oldsched(opts: TaskOpts, f: ~fn()) {
                 let group = @TCB(child, child_arc, ancestors,
                                  is_main, notifier);
                 unsafe {
-                    local_set(child, taskgroup_key!(), group);
+                    local_set(OldHandle(child), taskgroup_key!(), group);
                 }
 
                 // Run the child's body.

src/libcore/unstable/lang.rs

@@ -17,6 +17,8 @@ use str;
 use sys;
 use unstable::exchange_alloc;
 use cast::transmute;
+use rt::{context, OldTaskContext};
+use rt::local_services::borrow_local_services;
 
 #[allow(non_camel_case_types)]
 pub type rust_task = c_void;
@@ -88,19 +90,58 @@ pub unsafe fn exchange_free(ptr: *c_char) {
 
 #[lang="malloc"]
 #[inline(always)]
+#[cfg(stage0)] // For some reason this isn't working on windows in stage0
 pub unsafe fn local_malloc(td: *c_char, size: uintptr_t) -> *c_char {
     return rustrt::rust_upcall_malloc_noswitch(td, size);
 }
 
+#[lang="malloc"]
+#[inline(always)]
+#[cfg(not(stage0))]
+pub unsafe fn local_malloc(td: *c_char, size: uintptr_t) -> *c_char {
+    match context() {
+        OldTaskContext => {
+            return rustrt::rust_upcall_malloc_noswitch(td, size);
+        }
+        _ => {
+            let mut alloc = ::ptr::null();
+            do borrow_local_services |srv| {
+                alloc = srv.heap.alloc(td as *c_void, size as uint) as *c_char;
+            }
+            return alloc;
+        }
+    }
+}
+
 // NB: Calls to free CANNOT be allowed to fail, as throwing an exception from
 // inside a landing pad may corrupt the state of the exception handler. If a
 // problem occurs, call exit instead.
 #[lang="free"]
 #[inline(always)]
+#[cfg(stage0)]
 pub unsafe fn local_free(ptr: *c_char) {
     rustrt::rust_upcall_free_noswitch(ptr);
 }
 
+// NB: Calls to free CANNOT be allowed to fail, as throwing an exception from
+// inside a landing pad may corrupt the state of the exception handler. If a
+// problem occurs, call exit instead.
+#[lang="free"]
+#[inline(always)]
+#[cfg(not(stage0))]
+pub unsafe fn local_free(ptr: *c_char) {
+    match context() {
+        OldTaskContext => {
+            rustrt::rust_upcall_free_noswitch(ptr);
+        }
+        _ => {
+            do borrow_local_services |srv| {
+                srv.heap.free(ptr as *c_void);
+            }
+        }
+    }
+}
+
 #[lang="borrow_as_imm"]
 #[inline(always)]
 pub unsafe fn borrow_as_imm(a: *u8) {

src/librustc/middle/trans/glue.rs

@@ -394,10 +394,15 @@ pub fn call_tydesc_glue(cx: block, v: ValueRef, t: ty::t, field: uint)
 
 pub fn make_visit_glue(bcx: block, v: ValueRef, t: ty::t) {
     let _icx = bcx.insn_ctxt("make_visit_glue");
-    let mut bcx = bcx;
-    let (visitor_trait, object_ty) = ty::visitor_object_ty(bcx.tcx());
-    let v = PointerCast(bcx, v, T_ptr(type_of::type_of(bcx.ccx(), object_ty)));
-    bcx = reflect::emit_calls_to_trait_visit_ty(bcx, t, v, visitor_trait.def_id);
+    let bcx = do with_scope(bcx, None, ~"visitor cleanup") |bcx| {
+        let mut bcx = bcx;
+        let (visitor_trait, object_ty) = ty::visitor_object_ty(bcx.tcx());
+        let v = PointerCast(bcx, v, T_ptr(type_of::type_of(bcx.ccx(), object_ty)));
+        bcx = reflect::emit_calls_to_trait_visit_ty(bcx, t, v, visitor_trait.def_id);
+        // The visitor is a boxed object and needs to be dropped
+        add_clean(bcx, v, object_ty);
+        bcx
+    };
     build_return(bcx);
 }
 

src/libsyntax/ext/expand.rs

@@ -487,7 +487,7 @@ pub fn core_macros() -> ~str {
 
         { $c:ident: $in:ty -> $out:ty; } => {
 
-            mod $c {
+            pub mod $c {
                 fn key(_x: @::core::condition::Handler<$in,$out>) { }
 
                 pub static cond :

src/rt/boxed_region.cpp

@@ -27,11 +27,11 @@ rust_opaque_box *boxed_region::malloc(type_desc *td, size_t body_size) {
     if (live_allocs) live_allocs->prev = box;
     live_allocs = box;
 
-    LOG(rust_get_current_task(), box,
+    /*LOG(rust_get_current_task(), box,
         "@malloc()=%p with td %p, size %lu==%lu+%lu, "
         "align %lu, prev %p, next %p\n",
         box, td, total_size, sizeof(rust_opaque_box), body_size,
-        td->align, box->prev, box->next);
+        td->align, box->prev, box->next);*/
 
     return box;
 }
@@ -50,9 +50,9 @@ rust_opaque_box *boxed_region::realloc(rust_opaque_box *box,
     if (new_box->next) new_box->next->prev = new_box;
     if (live_allocs == box) live_allocs = new_box;
 
-    LOG(rust_get_current_task(), box,
+    /*LOG(rust_get_current_task(), box,
         "@realloc()=%p with orig=%p, size %lu==%lu+%lu",
-        new_box, box, total_size, sizeof(rust_opaque_box), new_size);
+        new_box, box, total_size, sizeof(rust_opaque_box), new_size);*/
 
     return new_box;
 }
@@ -74,15 +74,15 @@ void boxed_region::free(rust_opaque_box *box) {
     // double frees (kind of).
     assert(box->td != NULL);
 
-    LOG(rust_get_current_task(), box,
+    /*LOG(rust_get_current_task(), box,
         "@free(%p) with td %p, prev %p, next %p\n",
-        box, box->td, box->prev, box->next);
+        box, box->td, box->prev, box->next);*/
 
     if (box->prev) box->prev->next = box->next;
     if (box->next) box->next->prev = box->prev;
     if (live_allocs == box) live_allocs = box->next;
 
-    if (env->poison_on_free) {
+    if (poison_on_free) {
         memset(box_body(box), 0xab, box->td->size);
     }
 

src/rt/boxed_region.h

@@ -24,7 +24,7 @@ struct rust_env;
  * a type descr which describes the payload (what follows the header). */
 class boxed_region {
 private:
-    rust_env *env;
+    bool poison_on_free;
     memory_region *backing_region;
     rust_opaque_box *live_allocs;
 
@@ -41,8 +41,8 @@ private:
     boxed_region& operator=(const boxed_region& rhs);
 
 public:
-    boxed_region(rust_env *e, memory_region *br)
-        : env(e)
+    boxed_region(memory_region *br, bool poison_on_free)
+        : poison_on_free(poison_on_free)
         , backing_region(br)
         , live_allocs(NULL)
     {}

src/rt/memory_region.cpp

@@ -11,7 +11,6 @@
 
 #include "sync/sync.h"
 #include "memory_region.h"
-#include "rust_env.h"
 
 #if RUSTRT_TRACK_ALLOCATIONS >= 3
 #include <execinfo.h>
@@ -35,15 +34,19 @@ void *memory_region::get_data(alloc_header *ptr) {
     return (void*)((char *)ptr + HEADER_SIZE);
 }
 
-memory_region::memory_region(rust_env *env, bool synchronized) :
-    _env(env), _parent(NULL), _live_allocations(0),
-    _detailed_leaks(env->detailed_leaks),
+memory_region::memory_region(bool synchronized,
+                             bool detailed_leaks,
+                             bool poison_on_free) :
+    _parent(NULL), _live_allocations(0),
+    _detailed_leaks(detailed_leaks),
+    _poison_on_free(poison_on_free),
     _synchronized(synchronized) {
 }
 
 memory_region::memory_region(memory_region *parent) :
-    _env(parent->_env), _parent(parent), _live_allocations(0),
+    _parent(parent), _live_allocations(0),
     _detailed_leaks(parent->_detailed_leaks),
+    _poison_on_free(parent->_poison_on_free),
     _synchronized(parent->_synchronized) {
 }
 
@@ -241,7 +244,7 @@ memory_region::claim_alloc(void *mem) {
 void
 memory_region::maybe_poison(void *mem) {
 
-    if (!_env->poison_on_free)
+    if (!_poison_on_free)
         return;
 
 #   if RUSTRT_TRACK_ALLOCATIONS >= 1

src/rt/memory_region.h

@@ -54,11 +54,11 @@ private:
     inline alloc_header *get_header(void *mem);
     inline void *get_data(alloc_header *);
 
-    rust_env *_env;
     memory_region *_parent;
     int _live_allocations;
     array_list<alloc_header *> _allocation_list;
     const bool _detailed_leaks;
+    const bool _poison_on_free;
     const bool _synchronized;
     lock_and_signal _lock;
 
@@ -75,7 +75,8 @@ private:
     memory_region& operator=(const memory_region& rhs);
 
 public:
-    memory_region(rust_env *env, bool synchronized);
+    memory_region(bool synchronized,
+                  bool detailed_leaks, bool poison_on_free);
     memory_region(memory_region *parent);
     void *malloc(size_t size, const char *tag);
     void *realloc(void *mem, size_t size);

src/rt/rust_builtin.cpp

@@ -856,6 +856,63 @@ rust_initialize_global_state() {
     }
 }
 
+extern "C" CDECL memory_region*
+rust_new_memory_region(uintptr_t synchronized,
+                       uintptr_t detailed_leaks,
+                       uintptr_t poison_on_free) {
+    return new memory_region((bool)synchronized,
+                             (bool)detailed_leaks,
+                             (bool)poison_on_free);
+}
+
+extern "C" CDECL void
+rust_delete_memory_region(memory_region *region) {
+    delete region;
+}
+
+extern "C" CDECL boxed_region*
+rust_new_boxed_region(memory_region *region,
+                      uintptr_t poison_on_free) {
+    return new boxed_region(region, poison_on_free);
+}
+
+extern "C" CDECL void
+rust_delete_boxed_region(boxed_region *region) {
+    delete region;
+}
+
+extern "C" CDECL rust_opaque_box*
+rust_boxed_region_malloc(boxed_region *region, type_desc *td, size_t size) {
+    return region->malloc(td, size);
+}
+
+extern "C" CDECL void
+rust_boxed_region_free(boxed_region *region, rust_opaque_box *box) {
+    region->free(box);
+}
+
+typedef void *(rust_try_fn)(void*, void*);
+
+extern "C" CDECL uintptr_t
+rust_try(rust_try_fn f, void *fptr, void *env) {
+    try {
+        f(fptr, env);
+    } catch (uintptr_t token) {
+        assert(token != 0);
+        return token;
+    }
+    return 0;
+}
+
+extern "C" CDECL void
+rust_begin_unwind(uintptr_t token) {
+#ifndef __WIN32__
+    throw token;
+#else
+    abort();
+#endif
+}
+
 //
 // Local Variables:
 // mode: C++

src/rt/rust_sched_loop.cpp

@@ -38,7 +38,7 @@ rust_sched_loop::rust_sched_loop(rust_scheduler *sched, int id, bool killed) :
     sched(sched),
     log_lvl(log_debug),
     min_stack_size(kernel->env->min_stack_size),
-    local_region(kernel->env, false),
+    local_region(false, kernel->env->detailed_leaks, kernel->env->poison_on_free),
     // FIXME #2891: calculate a per-scheduler name.
     name("main")
 {

src/rt/rust_task.cpp

@@ -36,7 +36,7 @@ rust_task::rust_task(rust_sched_loop *sched_loop, rust_task_state state,
     kernel(sched_loop->kernel),
     name(name),
     list_index(-1),
-    boxed(sched_loop->kernel->env, &local_region),
+    boxed(&local_region, sched_loop->kernel->env->poison_on_free),
     local_region(&sched_loop->local_region),
     unwinding(false),
     total_stack_sz(0),

src/rt/rust_test_helpers.cpp

@@ -165,3 +165,14 @@ extern "C" CDECL TwoDoubles
 rust_dbg_extern_identity_TwoDoubles(TwoDoubles u) {
     return u;
 }
+
+// Generates increasing port numbers for network testing
+extern "C" CDECL uintptr_t
+rust_dbg_next_port() {
+  static lock_and_signal dbg_port_lock;
+  static uintptr_t next_port = 9600;
+  scoped_lock with(dbg_port_lock);
+  uintptr_t this_port = next_port;
+  next_port += 1;
+  return this_port;
+}

src/rt/rust_upcall.cpp

@@ -293,7 +293,13 @@ upcall_rust_personality(int version,
     s_rust_personality_args args = {(_Unwind_Reason_Code)0,
                                     version, actions, exception_class,
                                     ue_header, context};
-    rust_task *task = rust_get_current_task();
+    rust_task *task = rust_try_get_current_task();
+
+    if (task == NULL) {
+        // Assuming we're running with the new scheduler
+        upcall_s_rust_personality(&args);
+        return args.retval;
+    }
 
     // The personality function is run on the stack of the
     // last function that threw or landed, which is going
@@ -330,8 +336,12 @@ upcall_del_stack() {
 // needs to acquire the value of the stack pointer
 extern "C" CDECL void
 upcall_reset_stack_limit() {
-    rust_task *task = rust_get_current_task();
-    task->reset_stack_limit();
+    rust_task *task = rust_try_get_current_task();
+    if (task != NULL) {
+        task->reset_stack_limit();
+    } else {
+        // We must be in a newsched task
+    }
 }
 
 //

src/rt/rustrt.def.in

@@ -224,4 +224,12 @@ rust_uv_free_ip4_addr
 rust_uv_free_ip6_addr
 rust_call_nullary_fn
 rust_initialize_global_state
-
+rust_dbg_next_port
+rust_new_memory_region
+rust_delete_memory_region
+rust_new_boxed_region
+rust_delete_boxed_region
+rust_boxed_region_malloc
+rust_boxed_region_free
+rust_try
+rust_begin_unwind