diff options
Diffstat (limited to 'src/libcore/rt')
| -rw-r--r-- | src/libcore/rt/io/extensions.rs (renamed from src/libcore/rt/io/util.rs) | 0 | ||||
| -rw-r--r-- | src/libcore/rt/io/file.rs | 10 | ||||
| -rw-r--r-- | src/libcore/rt/io/mem.rs | 69 | ||||
| -rw-r--r-- | src/libcore/rt/io/mod.rs | 246 | ||||
| -rw-r--r-- | src/libcore/rt/io/net/mod.rs | 31 | ||||
| -rw-r--r-- | src/libcore/rt/io/net/tcp.rs | 30 | ||||
| -rw-r--r-- | src/libcore/rt/io/net/udp.rs | 5 | ||||
| -rw-r--r-- | src/libcore/rt/io/net/unix.rs | 7 | ||||
| -rw-r--r-- | src/libcore/rt/io/option.rs | 153 | ||||
| -rw-r--r-- | src/libcore/rt/io/support.rs (renamed from src/libcore/rt/io/misc.rs) | 0 | ||||
| -rw-r--r-- | src/libcore/rt/local_heap.rs | 81 | ||||
| -rw-r--r-- | src/libcore/rt/local_services.rs | 223 | ||||
| -rw-r--r-- | src/libcore/rt/mod.rs | 51 | ||||
| -rw-r--r-- | src/libcore/rt/sched/mod.rs | 21 | ||||
| -rw-r--r-- | src/libcore/rt/test.rs | 120 | ||||
| -rw-r--r-- | src/libcore/rt/uv/mod.rs | 6 | ||||
| -rw-r--r-- | src/libcore/rt/uv/net.rs | 64 | ||||
| -rw-r--r-- | src/libcore/rt/uvio.rs | 124 | ||||
| -rw-r--r-- | src/libcore/rt/uvll.rs | 6 |
19 files changed, 1002 insertions, 245 deletions
diff --git a/src/libcore/rt/io/util.rs b/src/libcore/rt/io/extensions.rs index cff224a80be..cff224a80be 100644 --- a/src/libcore/rt/io/util.rs +++ b/src/libcore/rt/io/extensions.rs diff --git a/src/libcore/rt/io/file.rs b/src/libcore/rt/io/file.rs index e041183b584..85dc180452f 100644 --- a/src/libcore/rt/io/file.rs +++ b/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!() } } diff --git a/src/libcore/rt/io/mem.rs b/src/libcore/rt/io/mem.rs index 600968a3c71..06e1466831d 100644 --- a/src/libcore/rt/io/mem.rs +++ b/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; } } + + 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()); - fn eof(&mut self) -> bool { fail!() } + 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!() } -} \ No newline at end of file +} + +#[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()); + } +} diff --git a/src/libcore/rt/io/mod.rs b/src/libcore/rt/io/mod.rs index b035532144c..ced4ba0ee23 100644 --- a/src/libcore/rt/io/mod.rs +++ b/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 - -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'? +* 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 + +# 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 + + + +# Scope -# std +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!() + } +} diff --git a/src/libcore/rt/io/net/mod.rs b/src/libcore/rt/io/net/mod.rs deleted file mode 100644 index 130ff6b38fa..00000000000 --- a/src/libcore/rt/io/net/mod.rs +++ /dev/null @@ -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>; -} diff --git a/src/libcore/rt/io/net/tcp.rs b/src/libcore/rt/io/net/tcp.rs index e3f71dca8c8..c95b4344fe7 100644 --- a/src/libcore/rt/io/net/tcp.rs +++ b/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]); + } + }*/ + } +} diff --git a/src/libcore/rt/io/net/udp.rs b/src/libcore/rt/io/net/udp.rs index f76bb58a45e..0cb2978fb1a 100644 --- a/src/libcore/rt/io/net/udp.rs +++ b/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!() } } diff --git a/src/libcore/rt/io/net/unix.rs b/src/libcore/rt/io/net/unix.rs index 35eabe21b2a..262816beecc 100644 --- a/src/libcore/rt/io/net/unix.rs +++ b/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!() } } diff --git a/src/libcore/rt/io/option.rs b/src/libcore/rt/io/option.rs new file mode 100644 index 00000000000..95f8711cb5b --- /dev/null +++ b/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); + } +} diff --git a/src/libcore/rt/io/misc.rs b/src/libcore/rt/io/support.rs index 7bace5d6df2..7bace5d6df2 100644 --- a/src/libcore/rt/io/misc.rs +++ b/src/libcore/rt/io/support.rs diff --git a/src/libcore/rt/local_heap.rs b/src/libcore/rt/local_heap.rs new file mode 100644 index 00000000000..fbd4a77d79b --- /dev/null +++ b/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); +} + diff --git a/src/libcore/rt/local_services.rs b/src/libcore/rt/local_services.rs new file mode 100644 index 00000000000..a03bc6c409f --- /dev/null +++ b/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()); + } + } +} \ No newline at end of file diff --git a/src/libcore/rt/mod.rs b/src/libcore/rt/mod.rs index e93e0c6fc6c..56ed7dc95b6 100644 --- a/src/libcore/rt/mod.rs +++ b/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(); - } -} diff --git a/src/libcore/rt/sched/mod.rs b/src/libcore/rt/sched/mod.rs index 28946281628..d7543ae138c 100644 --- a/src/libcore/rt/sched/mod.rs +++ b/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(); diff --git a/src/libcore/rt/test.rs b/src/libcore/rt/test.rs new file mode 100644 index 00000000000..63db7054088 --- /dev/null +++ b/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()) +} diff --git a/src/libcore/rt/uv/mod.rs b/src/libcore/rt/uv/mod.rs index cb7925abdcd..013a28abf28 100644 --- a/src/libcore/rt/uv/mod.rs +++ b/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); diff --git a/src/libcore/rt/uv/net.rs b/src/libcore/rt/uv/net.rs index bcfe8b2cfdf..04b9008b067 100644 --- a/src/libcore/rt/uv/net.rs +++ b/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()); @@ -374,46 +376,12 @@ 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"); diff --git a/src/libcore/rt/uvio.rs b/src/libcore/rt/uvio.rs index abdd8d6619a..94f8c0bf707 100644 --- a/src/libcore/rt/uvio.rs +++ b/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); - - 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(); - } - }; + do run_in_newsched_task { + let addr = next_test_ip4(); - 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(); + } } } diff --git a/src/libcore/rt/uvll.rs b/src/libcore/rt/uvll.rs index c9a696fcd15..3eb7f8006b9 100644 --- a/src/libcore/rt/uvll.rs +++ b/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 { |