user0/rust
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

auto merge of #8387 : brson/rust/nooldrt, r=brson

Browse source
This commit is contained in:
bors 2013-08-09 18:41:13 -07:00
commit e81e81f234
62 changed files with 288 additions and 6458 deletions
Download patch file Download diff file Expand all files Collapse all files

17
src/libstd/at_vec.rs
View file

@ -275,24 +275,11 @@ pub mod raw {
}
fn local_realloc(ptr: *(), size: uint) -> *() {
use rt;
use rt::OldTaskContext;
use rt::local::Local;
use rt::task::Task;
if rt::context() == OldTaskContext {
unsafe {
return rust_local_realloc(ptr, size as libc::size_t);
}
extern {
#[fast_ffi]
fn rust_local_realloc(ptr: *(), size: libc::size_t) -> *();
}
} else {
do Local::borrow::<Task, *()> |task| {
task.heap.realloc(ptr as *libc::c_void, size) as *()
}
do Local::borrow::<Task, *()> |task| {
task.heap.realloc(ptr as *libc::c_void, size) as *()
}
}
}

19
src/libstd/cleanup.rs
View file

@ -56,13 +56,8 @@ unsafe fn each_live_alloc(read_next_before: bool,
#[cfg(unix)]
fn debug_mem() -> bool {
use rt;
use rt::OldTaskContext;
// XXX: Need to port the environment struct to newsched
match rt::context() {
OldTaskContext => ::rt::env::get().debug_mem,
_ => false
}
false
}
#[cfg(windows)]
@ -147,15 +142,3 @@ pub unsafe fn annihilate() {
dbg.write_str("\n");
}
}
/// Bindings to the runtime
pub mod rustrt {
use libc::c_void;
#[link_name = "rustrt"]
extern {
#[rust_stack]
// FIXME (#4386): Unable to make following method private.
pub fn rust_get_task() -> *c_void;
}
}

670
src/libstd/comm.rs
View file

@ -14,13 +14,11 @@ Message passing
#[allow(missing_doc)];
use either::{Either, Left, Right};
use clone::Clone;
use kinds::Send;
use option::{Option, Some};
use unstable::sync::Exclusive;
use option::Option;
pub use rt::comm::SendDeferred;
use rtcomm = rt::comm;
use rt;
/// A trait for things that can send multiple messages.
pub trait GenericChan<T> {
@ -52,614 +50,146 @@ pub trait Peekable<T> {
fn peek(&self) -> bool;
}
/// An endpoint that can send many messages.
pub struct Chan<T> {
inner: Either<pipesy::Chan<T>, rtcomm::Chan<T>>
pub struct PortOne<T> { x: rtcomm::PortOne<T> }
pub struct ChanOne<T> { x: rtcomm::ChanOne<T> }
pub fn oneshot<T: Send>() -> (PortOne<T>, ChanOne<T>) {
let (p, c) = rtcomm::oneshot();
(PortOne { x: p }, ChanOne { x: c })
}
/// An endpoint that can receive many messages.
pub struct Port<T> {
inner: Either<pipesy::Port<T>, rtcomm::Port<T>>
pub struct Port<T> { x: rtcomm::Port<T> }
pub struct Chan<T> { x: rtcomm::Chan<T> }
pub fn stream<T: Send>() -> (Port<T>, Chan<T>) {
let (p, c) = rtcomm::stream();
(Port { x: p }, Chan { x: c })
}
/** Creates a `(Port, Chan)` pair.
pub struct SharedChan<T> { x: rtcomm::SharedChan<T> }
These allow sending or receiving an unlimited number of messages.
impl<T: Send> SharedChan<T> {
pub fn new(c: Chan<T>) -> SharedChan<T> {
let Chan { x: c } = c;
SharedChan { x: rtcomm::SharedChan::new(c) }
}
}
*/
pub fn stream<T:Send>() -> (Port<T>, Chan<T>) {
let (port, chan) = match rt::context() {
rt::OldTaskContext => match pipesy::stream() {
(p, c) => (Left(p), Left(c))
},
_ => match rtcomm::stream() {
(p, c) => (Right(p), Right(c))
}
};
let port = Port { inner: port };
let chan = Chan { inner: chan };
return (port, chan);
impl<T: Send> ChanOne<T> {
pub fn send(self, val: T) {
let ChanOne { x: c } = self;
c.send(val)
}
pub fn try_send(self, val: T) -> bool {
let ChanOne { x: c } = self;
c.try_send(val)
}
pub fn send_deferred(self, val: T) {
let ChanOne { x: c } = self;
c.send_deferred(val)
}
pub fn try_send_deferred(self, val: T) -> bool {
let ChanOne{ x: c } = self;
c.try_send_deferred(val)
}
}
impl<T: Send> PortOne<T> {
pub fn recv(self) -> T {
let PortOne { x: p } = self;
p.recv()
}
pub fn try_recv(self) -> Option<T> {
let PortOne { x: p } = self;
p.try_recv()
}
}
impl<T: Send> Peekable<T> for PortOne<T> {
fn peek(&self) -> bool {
let &PortOne { x: ref p } = self;
p.peek()
}
}
impl<T: Send> GenericChan<T> for Chan<T> {
fn send(&self, x: T) {
match self.inner {
Left(ref chan) => chan.send(x),
Right(ref chan) => chan.send(x)
}
fn send(&self, val: T) {
let &Chan { x: ref c } = self;
c.send(val)
}
}
impl<T: Send> GenericSmartChan<T> for Chan<T> {
fn try_send(&self, x: T) -> bool {
match self.inner {
Left(ref chan) => chan.try_send(x),
Right(ref chan) => chan.try_send(x)
}
fn try_send(&self, val: T) -> bool {
let &Chan { x: ref c } = self;
c.try_send(val)
}
}
impl<T: Send> SendDeferred<T> for Chan<T> {
fn send_deferred(&self, x: T) {
match self.inner {
Left(ref chan) => chan.send(x),
Right(ref chan) => chan.send_deferred(x)
}
fn send_deferred(&self, val: T) {
let &Chan { x: ref c } = self;
c.send_deferred(val)
}
fn try_send_deferred(&self, x: T) -> bool {
match self.inner {
Left(ref chan) => chan.try_send(x),
Right(ref chan) => chan.try_send_deferred(x)
}
fn try_send_deferred(&self, val: T) -> bool {
let &Chan { x: ref c } = self;
c.try_send_deferred(val)
}
}
impl<T: Send> GenericPort<T> for Port<T> {
fn recv(&self) -> T {
match self.inner {
Left(ref port) => port.recv(),
Right(ref port) => port.recv()
}
let &Port { x: ref p } = self;
p.recv()
}
fn try_recv(&self) -> Option<T> {
match self.inner {
Left(ref port) => port.try_recv(),
Right(ref port) => port.try_recv()
}
let &Port { x: ref p } = self;
p.try_recv()
}
}
impl<T: Send> Peekable<T> for Port<T> {
fn peek(&self) -> bool {
match self.inner {
Left(ref port) => port.peek(),
Right(ref port) => port.peek()
}
}
}
/// A channel that can be shared between many senders.
pub struct SharedChan<T> {
inner: Either<Exclusive<pipesy::Chan<T>>, rtcomm::SharedChan<T>>
}
impl<T: Send> SharedChan<T> {
/// Converts a `chan` into a `shared_chan`.
pub fn new(c: Chan<T>) -> SharedChan<T> {
let Chan { inner } = c;
let c = match inner {
Left(c) => Left(Exclusive::new(c)),
Right(c) => Right(rtcomm::SharedChan::new(c))
};
SharedChan { inner: c }
let &Port { x: ref p } = self;
p.peek()
}
}
impl<T: Send> GenericChan<T> for SharedChan<T> {
fn send(&self, x: T) {
match self.inner {
Left(ref chan) => {
unsafe {
let mut xx = Some(x);
do chan.with_imm |chan| {
chan.send(xx.take_unwrap())
}
}
}
Right(ref chan) => chan.send(x)
}
fn send(&self, val: T) {
let &SharedChan { x: ref c } = self;
c.send(val)
}
}
impl<T: Send> GenericSmartChan<T> for SharedChan<T> {
fn try_send(&self, x: T) -> bool {
match self.inner {
Left(ref chan) => {
unsafe {
let mut xx = Some(x);
do chan.with_imm |chan| {
chan.try_send(xx.take_unwrap())
}
}
}
Right(ref chan) => chan.try_send(x)
}
fn try_send(&self, val: T) -> bool {
let &SharedChan { x: ref c } = self;
c.try_send(val)
}
}
impl<T: Send> ::clone::Clone for SharedChan<T> {
impl<T: Send> SendDeferred<T> for SharedChan<T> {
fn send_deferred(&self, val: T) {
let &SharedChan { x: ref c } = self;
c.send_deferred(val)
}
fn try_send_deferred(&self, val: T) -> bool {
let &SharedChan { x: ref c } = self;
c.try_send_deferred(val)
}
}
impl<T> Clone for SharedChan<T> {
fn clone(&self) -> SharedChan<T> {
SharedChan { inner: self.inner.clone() }
}
}
pub struct PortOne<T> {
inner: Either<pipesy::PortOne<T>, rtcomm::PortOne<T>>
}
pub struct ChanOne<T> {
inner: Either<pipesy::ChanOne<T>, rtcomm::ChanOne<T>>
}
pub fn oneshot<T: Send>() -> (PortOne<T>, ChanOne<T>) {
let (port, chan) = match rt::context() {
rt::OldTaskContext => match pipesy::oneshot() {
(p, c) => (Left(p), Left(c)),
},
_ => match rtcomm::oneshot() {
(p, c) => (Right(p), Right(c))
}
};
let port = PortOne { inner: port };
let chan = ChanOne { inner: chan };
return (port, chan);
}
impl<T: Send> PortOne<T> {
pub fn recv(self) -> T {
let PortOne { inner } = self;
match inner {
Left(p) => p.recv(),
Right(p) => p.recv()
}
}
pub fn try_recv(self) -> Option<T> {
let PortOne { inner } = self;
match inner {
Left(p) => p.try_recv(),
Right(p) => p.try_recv()
}
}
}
impl<T: Send> ChanOne<T> {
pub fn send(self, data: T) {
let ChanOne { inner } = self;
match inner {
Left(p) => p.send(data),
Right(p) => p.send(data)
}
}
pub fn try_send(self, data: T) -> bool {
let ChanOne { inner } = self;
match inner {
Left(p) => p.try_send(data),
Right(p) => p.try_send(data)
}
}
pub fn send_deferred(self, data: T) {
let ChanOne { inner } = self;
match inner {
Left(p) => p.send(data),
Right(p) => p.send_deferred(data)
}
}
pub fn try_send_deferred(self, data: T) -> bool {
let ChanOne { inner } = self;
match inner {
Left(p) => p.try_send(data),
Right(p) => p.try_send_deferred(data)
}
}
}
pub fn recv_one<T: Send>(port: PortOne<T>) -> T {
let PortOne { inner } = port;
match inner {
Left(p) => pipesy::recv_one(p),
Right(p) => p.recv()
}
}
pub fn try_recv_one<T: Send>(port: PortOne<T>) -> Option<T> {
let PortOne { inner } = port;
match inner {
Left(p) => pipesy::try_recv_one(p),
Right(p) => p.try_recv()
}
}
pub fn send_one<T: Send>(chan: ChanOne<T>, data: T) {
let ChanOne { inner } = chan;
match inner {
Left(c) => pipesy::send_one(c, data),
Right(c) => c.send(data)
}
}
pub fn try_send_one<T: Send>(chan: ChanOne<T>, data: T) -> bool {
let ChanOne { inner } = chan;
match inner {
Left(c) => pipesy::try_send_one(c, data),
Right(c) => c.try_send(data)
}
}
mod pipesy {
use kinds::Send;
use option::{Option, Some, None};
use pipes::{recv, try_recv, peek};
use super::{GenericChan, GenericSmartChan, GenericPort, Peekable};
use cast::transmute_mut;
/*proto! oneshot (
Oneshot:send<T:Send> {
send(T) -> !
}
)*/
#[allow(non_camel_case_types)]
pub mod oneshot {
use std::kinds::Send;
use ptr::to_mut_unsafe_ptr;
pub fn init<T: Send>() -> (server::Oneshot<T>, client::Oneshot<T>) {
pub use std::pipes::HasBuffer;
let buffer = ~::std::pipes::Buffer {
header: ::std::pipes::BufferHeader(),
data: __Buffer {
Oneshot: ::std::pipes::mk_packet::<Oneshot<T>>()
},
};
do ::std::pipes::entangle_buffer(buffer) |buffer, data| {
data.Oneshot.set_buffer(buffer);
to_mut_unsafe_ptr(&mut data.Oneshot)
}
}
#[allow(non_camel_case_types)]
pub enum Oneshot<T> { pub send(T), }
#[allow(non_camel_case_types)]
pub struct __Buffer<T> {
Oneshot: ::std::pipes::Packet<Oneshot<T>>,
}
#[allow(non_camel_case_types)]
pub mod client {
use std::kinds::Send;
#[allow(non_camel_case_types)]
pub fn try_send<T: Send>(pipe: Oneshot<T>, x_0: T) ->
::std::option::Option<()> {
{
use super::send;
let message = send(x_0);
if ::std::pipes::send(pipe, message) {
::std::pipes::rt::make_some(())
} else { ::std::pipes::rt::make_none() }
}
}
#[allow(non_camel_case_types)]
pub fn send<T: Send>(pipe: Oneshot<T>, x_0: T) {
{
use super::send;
let message = send(x_0);
::std::pipes::send(pipe, message);
}
}
#[allow(non_camel_case_types)]
pub type Oneshot<T> =
::std::pipes::SendPacketBuffered<super::Oneshot<T>,
super::__Buffer<T>>;
}
#[allow(non_camel_case_types)]
pub mod server {
#[allow(non_camel_case_types)]
pub type Oneshot<T> =
::std::pipes::RecvPacketBuffered<super::Oneshot<T>,
super::__Buffer<T>>;
}
}
/// The send end of a oneshot pipe.
pub struct ChanOne<T> {
contents: oneshot::client::Oneshot<T>
}
impl<T> ChanOne<T> {
pub fn new(contents: oneshot::client::Oneshot<T>) -> ChanOne<T> {
ChanOne {
contents: contents
}
}
}
/// The receive end of a oneshot pipe.
pub struct PortOne<T> {
contents: oneshot::server::Oneshot<T>
}
impl<T> PortOne<T> {
pub fn new(contents: oneshot::server::Oneshot<T>) -> PortOne<T> {
PortOne {
contents: contents
}
}
}
/// Initialiase a (send-endpoint, recv-endpoint) oneshot pipe pair.
pub fn oneshot<T: Send>() -> (PortOne<T>, ChanOne<T>) {
let (port, chan) = oneshot::init();
(PortOne::new(port), ChanOne::new(chan))
}
impl<T: Send> PortOne<T> {
pub fn recv(self) -> T { recv_one(self) }
pub fn try_recv(self) -> Option<T> { try_recv_one(self) }
pub fn unwrap(self) -> oneshot::server::Oneshot<T> {
match self {
PortOne { contents: s } => s
}
}
}
impl<T: Send> ChanOne<T> {
pub fn send(self, data: T) { send_one(self, data) }
pub fn try_send(self, data: T) -> bool { try_send_one(self, data) }
pub fn unwrap(self) -> oneshot::client::Oneshot<T> {
match self {
ChanOne { contents: s } => s
}
}
}
/**
* Receive a message from a oneshot pipe, failing if the connection was
* closed.
*/
pub fn recv_one<T: Send>(port: PortOne<T>) -> T {
match port {
PortOne { contents: port } => {
let oneshot::send(message) = recv(port);
message
}
}
}
/// Receive a message from a oneshot pipe unless the connection was closed.
pub fn try_recv_one<T: Send> (port: PortOne<T>) -> Option<T> {
match port {
PortOne { contents: port } => {
let message = try_recv(port);
if message.is_none() {
None
} else {
let oneshot::send(message) = message.unwrap();
Some(message)
}
}
}
}
/// Send a message on a oneshot pipe, failing if the connection was closed.
pub fn send_one<T: Send>(chan: ChanOne<T>, data: T) {
match chan {
ChanOne { contents: chan } => oneshot::client::send(chan, data),
}
}
/**
* Send a message on a oneshot pipe, or return false if the connection was
* closed.
*/
pub fn try_send_one<T: Send>(chan: ChanOne<T>, data: T) -> bool {
match chan {
ChanOne { contents: chan } => {
oneshot::client::try_send(chan, data).is_some()
}
}
}
// Streams - Make pipes a little easier in general.
/*proto! streamp (
Open:send<T: Send> {
data(T) -> Open<T>
}
)*/
#[allow(non_camel_case_types)]
pub mod streamp {
use std::kinds::Send;
pub fn init<T: Send>() -> (server::Open<T>, client::Open<T>) {
pub use std::pipes::HasBuffer;
::std::pipes::entangle()
}
#[allow(non_camel_case_types)]
pub enum Open<T> { pub data(T, server::Open<T>), }
#[allow(non_camel_case_types)]
pub mod client {
use std::kinds::Send;
#[allow(non_camel_case_types)]
pub fn try_data<T: Send>(pipe: Open<T>, x_0: T) ->
::std::option::Option<Open<T>> {
{
use super::data;
let (s, c) = ::std::pipes::entangle();
let message = data(x_0, s);
if ::std::pipes::send(pipe, message) {
::std::pipes::rt::make_some(c)
} else { ::std::pipes::rt::make_none() }
}
}
#[allow(non_camel_case_types)]
pub fn data<T: Send>(pipe: Open<T>, x_0: T) -> Open<T> {
{
use super::data;
let (s, c) = ::std::pipes::entangle();
let message = data(x_0, s);
::std::pipes::send(pipe, message);
c
}
}
#[allow(non_camel_case_types)]
pub type Open<T> = ::std::pipes::SendPacket<super::Open<T>>;
}
#[allow(non_camel_case_types)]
pub mod server {
#[allow(non_camel_case_types)]
pub type Open<T> = ::std::pipes::RecvPacket<super::Open<T>>;
}
}
/// An endpoint that can send many messages.
#[unsafe_mut_field(endp)]
pub struct Chan<T> {
endp: Option<streamp::client::Open<T>>
}
/// An endpoint that can receive many messages.
#[unsafe_mut_field(endp)]
pub struct Port<T> {
endp: Option<streamp::server::Open<T>>,
}
/** Creates a `(Port, Chan)` pair.
These allow sending or receiving an unlimited number of messages.
*/
pub fn stream<T:Send>() -> (Port<T>, Chan<T>) {
let (s, c) = streamp::init();
(Port {
endp: Some(s)
}, Chan {
endp: Some(c)
})
}
impl<T: Send> GenericChan<T> for Chan<T> {
#[inline]
fn send(&self, x: T) {
unsafe {
let self_endp = transmute_mut(&self.endp);
*self_endp = Some(streamp::client::data(self_endp.take_unwrap(), x))
}
}
}
impl<T: Send> GenericSmartChan<T> for Chan<T> {
#[inline]
fn try_send(&self, x: T) -> bool {
unsafe {
let self_endp = transmute_mut(&self.endp);
match streamp::client::try_data(self_endp.take_unwrap(), x) {
Some(next) => {
*self_endp = Some(next);
true
}
None => false
}
}
}
}
impl<T: Send> GenericPort<T> for Port<T> {
#[inline]
fn recv(&self) -> T {
unsafe {
let self_endp = transmute_mut(&self.endp);
let endp = self_endp.take();
let streamp::data(x, endp) = recv(endp.unwrap());
*self_endp = Some(endp);
x
}
}
#[inline]
fn try_recv(&self) -> Option<T> {
unsafe {
let self_endp = transmute_mut(&self.endp);
let endp = self_endp.take();
match try_recv(endp.unwrap()) {
Some(streamp::data(x, endp)) => {
*self_endp = Some(endp);
Some(x)
}
None => None
}
}
}
}
impl<T: Send> Peekable<T> for Port<T> {
#[inline]
fn peek(&self) -> bool {
unsafe {
let self_endp = transmute_mut(&self.endp);
let mut endp = self_endp.take();
let peek = match endp {
Some(ref mut endp) => peek(endp),
None => fail!("peeking empty stream")
};
*self_endp = endp;
peek
}
}
}
}
#[cfg(test)]
mod test {
use either::Right;
use super::{Chan, Port, oneshot, stream};
#[test]
fn test_oneshot() {
let (p, c) = oneshot();
c.send(());
p.recv()
}
#[test]
fn test_peek_terminated() {
let (port, chan): (Port<int>, Chan<int>) = stream();
{
// Destroy the channel
let _chan = chan;
}
assert!(!port.peek());
let &SharedChan { x: ref c } = self;
SharedChan { x: c.clone() }
}
}

52
src/libstd/logging.rs
View file

@ -14,18 +14,11 @@ use option::*;
use os;
use either::*;
use rt;
use rt::OldTaskContext;
use rt::logging::{Logger, StdErrLogger};
/// Turns on logging to stdout globally
pub fn console_on() {
if rt::context() == OldTaskContext {
unsafe {
rustrt::rust_log_console_on();
}
} else {
rt::logging::console_on();
}
rt::logging::console_on();
}
/**
@ -41,45 +34,24 @@ pub fn console_off() {
return;
}
if rt::context() == OldTaskContext {
unsafe {
rustrt::rust_log_console_off();
}
} else {
rt::logging::console_off();
}
rt::logging::console_off();
}
#[cfg(not(test))]
#[lang="log_type"]
#[allow(missing_doc)]
pub fn log_type<T>(level: u32, object: &T) {
use cast;
use container::Container;
pub fn log_type<T>(_level: u32, object: &T) {
use io;
use libc;
use repr;
use rt;
use str;
use vec;
let bytes = do io::with_bytes_writer |writer| {
repr::write_repr(writer, object);
};
match rt::context() {
rt::OldTaskContext => {
unsafe {
let len = bytes.len() as libc::size_t;
rustrt::rust_log_str(level, cast::transmute(vec::raw::to_ptr(bytes)), len);
}
}
_ => {
// XXX: Bad allocation
let msg = str::from_bytes(bytes);
newsched_log_str(msg);
}
}
// XXX: Bad allocation
let msg = str::from_bytes(bytes);
newsched_log_str(msg);
}
fn newsched_log_str(msg: ~str) {
@ -100,15 +72,3 @@ fn newsched_log_str(msg: ~str) {
}
}
}
pub mod rustrt {
use libc;
extern {
pub fn rust_log_console_on();
pub fn rust_log_console_off();
pub fn rust_log_str(level: u32,
string: *libc::c_char,
size: libc::size_t);
}
}

28
src/libstd/os.rs
View file

@ -61,11 +61,8 @@ pub mod rustrt {
use libc;
extern {
pub fn rust_get_argc() -> c_int;
pub fn rust_get_argv() -> **c_char;
pub fn rust_path_is_dir(path: *libc::c_char) -> c_int;
pub fn rust_path_exists(path: *libc::c_char) -> c_int;
pub fn rust_set_exit_status(code: libc::intptr_t);
}
}
@ -1104,15 +1101,7 @@ pub fn last_os_error() -> ~str {
*/
pub fn set_exit_status(code: int) {
use rt;
use rt::OldTaskContext;
if rt::context() == OldTaskContext {
unsafe {
rustrt::rust_set_exit_status(code as libc::intptr_t);
}
} else {
rt::util::set_exit_status(code);
}
rt::util::set_exit_status(code);
}
unsafe fn load_argc_and_argv(argc: c_int, argv: **c_char) -> ~[~str] {
@ -1142,19 +1131,10 @@ pub fn real_args() -> ~[~str] {
#[cfg(target_os = "freebsd")]
pub fn real_args() -> ~[~str] {
use rt;
use rt::NewRtContext;
if rt::context() == NewRtContext {
match rt::args::clone() {
Some(args) => args,
None => fail!("process arguments not initialized")
}
} else {
unsafe {
let argc = rustrt::rust_get_argc();
let argv = rustrt::rust_get_argv();
load_argc_and_argv(argc, argv)
}
match rt::args::clone() {
Some(args) => args,
None => fail!("process arguments not initialized")
}
}

870
src/libstd/pipes.rs
View file

@ -1,870 +0,0 @@
// 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.
/*! Runtime support for message passing with protocol enforcement.
Pipes consist of two endpoints. One endpoint can send messages and
the other can receive messages. The set of legal messages and which
directions they can flow at any given point are determined by a
protocol. Below is an example protocol.
~~~ {.rust}
proto! pingpong (
ping: send {
ping -> pong
}
pong: recv {
pong -> ping
}
)
~~~
The `proto!` syntax extension will convert this into a module called
`pingpong`, which includes a set of types and functions that can be
used to write programs that follow the pingpong protocol.
*/
/* IMPLEMENTATION NOTES
The initial design for this feature is available at:
https://github.com/eholk/rust/wiki/Proposal-for-channel-contracts
Much of the design in that document is still accurate. There are
several components for the pipe implementation. First of all is the
syntax extension. To see how that works, it is best see comments in
libsyntax/ext/pipes.rs.
This module includes two related pieces of the runtime
implementation: support for unbounded and bounded
protocols. The main difference between the two is the type of the
buffer that is carried along in the endpoint data structures.
The heart of the implementation is the packet type. It contains a
header and a payload field. Much of the code in this module deals with
the header field. This is where the synchronization information is
stored. In the case of a bounded protocol, the header also includes a
pointer to the buffer the packet is contained in.
Packets represent a single message in a protocol. The payload field
gets instatiated at the type of the message, which is usually an enum
generated by the pipe compiler. Packets are conceptually single use,
although in bounded protocols they are reused each time around the
loop.
Packets are usually handled through a send_packet_buffered or
recv_packet_buffered object. Each packet is referenced by one
send_packet and one recv_packet, and these wrappers enforce that only
one end can send and only one end can receive. The structs also
include a destructor that marks packets are terminated if the sender
or receiver destroys the object before sending or receiving a value.
The *_packet_buffered structs take two type parameters. The first is
the message type for the current packet (or state). The second
represents the type of the whole buffer. For bounded protocols, the
protocol compiler generates a struct with a field for each protocol
state. This generated struct is used as the buffer type parameter. For
unbounded protocols, the buffer is simply one packet, so there is a
shorthand struct called send_packet and recv_packet, where the buffer
type is just `packet<T>`. Using the same underlying structure for both
bounded and unbounded protocols allows for less code duplication.
*/
#[allow(missing_doc)];
use container::Container;
use cast::{forget, transmute, transmute_copy, transmute_mut};
use either::{Either, Left, Right};
use iterator::{Iterator, IteratorUtil};
use kinds::Send;
use libc;
use ops::Drop;
use option::{None, Option, Some};
use unstable::finally::Finally;
use unstable::intrinsics;
use ptr;
use ptr::RawPtr;
use task;
use vec::{OwnedVector, MutableVector};
use util::replace;
static SPIN_COUNT: uint = 0;
#[deriving(Eq)]
enum State {
Empty,
Full,
Blocked,
Terminated
}
pub struct BufferHeader {
// Tracks whether this buffer needs to be freed. We can probably
// get away with restricting it to 0 or 1, if we're careful.
ref_count: int,
// We may want a drop, and to be careful about stringing this
// thing along.
}
pub fn BufferHeader() -> BufferHeader {
BufferHeader {
ref_count: 0
}
}
// This is for protocols to associate extra data to thread around.
pub struct Buffer<T> {
header: BufferHeader,
data: T,
}
pub struct PacketHeader {
state: State,
blocked_task: *rust_task,
// This is a transmute_copy of a ~buffer, that can also be cast
// to a buffer_header if need be.
buffer: *libc::c_void,
}
pub fn PacketHeader() -> PacketHeader {
PacketHeader {
state: Empty,
blocked_task: ptr::null(),
buffer: ptr::null()
}
}
impl PacketHeader {
// Returns the old state.
pub unsafe fn mark_blocked(&mut self, this: *rust_task) -> State {
rustrt::rust_task_ref(this);
let old_task = swap_task(&mut self.blocked_task, this);
assert!(old_task.is_null());
swap_state_acq(&mut self.state, Blocked)
}
pub unsafe fn unblock(&mut self) {
let old_task = swap_task(&mut self.blocked_task, ptr::null());
if !old_task.is_null() {
rustrt::rust_task_deref(old_task)
}
match swap_state_acq(&mut self.state, Empty) {
Empty | Blocked => (),
Terminated => self.state = Terminated,
Full => self.state = Full
}
}
// unsafe because this can do weird things to the space/time
// continuum. It ends making multiple unique pointers to the same
// thing. You'll probably want to forget them when you're done.
pub unsafe fn buf_header(&mut self) -> ~BufferHeader {
assert!(self.buffer.is_not_null());
transmute_copy(&self.buffer)
}
pub fn set_buffer<T:Send>(&mut self, b: ~Buffer<T>) {
unsafe {
self.buffer = transmute_copy(&b);
}
}
}
pub struct Packet<T> {
header: PacketHeader,
payload: Option<T>,
}
pub trait HasBuffer {
fn set_buffer(&mut self, b: *libc::c_void);
}
impl<T:Send> HasBuffer for Packet<T> {
fn set_buffer(&mut self, b: *libc::c_void) {
self.header.buffer = b;
}
}
pub fn mk_packet<T:Send>() -> Packet<T> {
Packet {
header: PacketHeader(),
payload: None,
}
}
fn unibuffer<T>() -> ~Buffer<Packet<T>> {
let mut b = ~Buffer {
header: BufferHeader(),
data: Packet {
header: PacketHeader(),
payload: None,
}
};
unsafe {
b.data.header.buffer = transmute_copy(&b);
}
b
}
pub fn packet<T>() -> *mut Packet<T> {
let mut b = unibuffer();
let p = ptr::to_mut_unsafe_ptr(&mut b.data);
// We'll take over memory management from here.
unsafe {
forget(b);
}
p
}
pub fn entangle_buffer<T:Send,Tstart:Send>(
mut buffer: ~Buffer<T>,
init: &fn(*libc::c_void, x: &mut T) -> *mut Packet<Tstart>)
-> (RecvPacketBuffered<Tstart, T>, SendPacketBuffered<Tstart, T>) {
unsafe {
let p = init(transmute_copy(&buffer), &mut buffer.data);
forget(buffer);
(RecvPacketBuffered(p), SendPacketBuffered(p))
}
}
pub fn swap_task(dst: &mut *rust_task, src: *rust_task) -> *rust_task {
// It might be worth making both acquire and release versions of
// this.
unsafe {
transmute(intrinsics::atomic_xchg(transmute(dst), src as int))
}
}
#[allow(non_camel_case_types)]
pub type rust_task = libc::c_void;
pub mod rustrt {
use libc;
use super::rust_task;
extern {
#[rust_stack]
pub fn rust_get_task() -> *rust_task;
#[rust_stack]
pub fn rust_task_ref(task: *rust_task);
pub fn rust_task_deref(task: *rust_task);
#[rust_stack]
pub fn task_clear_event_reject(task: *rust_task);
pub fn task_wait_event(this: *rust_task, killed: &mut *libc::c_void)
-> bool;
pub fn task_signal_event(target: *rust_task, event: *libc::c_void);
}
}
fn wait_event(this: *rust_task) -> *libc::c_void {
unsafe {
let mut event = ptr::null();
let killed = rustrt::task_wait_event(this, &mut event);
if killed && !task::failing() {
fail!("killed")
}
event
}
}
fn swap_state_acq(dst: &mut State, src: State) -> State {
unsafe {
transmute(intrinsics::atomic_xchg_acq(transmute(dst), src as int))
}
}
fn swap_state_rel(dst: &mut State, src: State) -> State {
unsafe {
transmute(intrinsics::atomic_xchg_rel(transmute(dst), src as int))
}
}
pub unsafe fn get_buffer<T>(p: *mut PacketHeader) -> ~Buffer<T> {
transmute((*p).buf_header())
}
// This could probably be done with SharedMutableState to avoid move_it!().
struct BufferResource<T> {
buffer: ~Buffer<T>,
}
#[unsafe_destructor]
impl<T> Drop for BufferResource<T> {
fn drop(&self) {
unsafe {
// FIXME(#4330) Need self by value to get mutability.
let this: &mut BufferResource<T> = transmute_mut(self);
let null_buffer: ~Buffer<T> = transmute(ptr::null::<Buffer<T>>());
let mut b = replace(&mut this.buffer, null_buffer);
//let p = ptr::to_unsafe_ptr(*b);
//error!("drop %?", p);
let old_count = intrinsics::atomic_xsub_rel(
&mut b.header.ref_count,
1);
//let old_count = atomic_xchng_rel(b.header.ref_count, 0);
if old_count == 1 {
// The new count is 0.
// go go gadget drop glue
}
else {
forget(b)
}
}
}
}
fn BufferResource<T>(mut b: ~Buffer<T>) -> BufferResource<T> {
//let p = ptr::to_unsafe_ptr(*b);
//error!("take %?", p);
unsafe {
intrinsics::atomic_xadd_acq(&mut b.header.ref_count, 1);
}
BufferResource {
// tjc: ????
buffer: b
}
}
pub fn send<T,Tbuffer>(mut p: SendPacketBuffered<T,Tbuffer>,
payload: T)
-> bool {
let header = p.header();
let p_ = p.unwrap();
let p = unsafe { &mut *p_ };
assert_eq!(ptr::to_unsafe_ptr(&(p.header)), header);
assert!(p.payload.is_none());
p.payload = Some(payload);
let old_state = swap_state_rel(&mut p.header.state, Full);
match old_state {
Empty => {
// Yay, fastpath.
// The receiver will eventually clean this up.
//unsafe { forget(p); }
return true;
}
Full => fail!("duplicate send"),
Blocked => {
debug!("waking up task for %?", p_);
let old_task = swap_task(&mut p.header.blocked_task, ptr::null());
if !old_task.is_null() {
unsafe {
rustrt::task_signal_event(
old_task,
ptr::to_unsafe_ptr(&(p.header)) as *libc::c_void);
rustrt::rust_task_deref(old_task);
}
}
// The receiver will eventually clean this up.
//unsafe { forget(p); }
return true;
}
Terminated => {
// The receiver will never receive this. Rely on drop_glue
// to clean everything up.
return false;
}
}
}
/** Receives a message from a pipe.
Fails if the sender closes the connection.
*/
pub fn recv<T:Send,Tbuffer:Send>(
p: RecvPacketBuffered<T, Tbuffer>) -> T {
try_recv(p).expect("connection closed")
}
/** Attempts to receive a message from a pipe.
Returns `None` if the sender has closed the connection without sending
a message, or `Some(T)` if a message was received.
*/
pub fn try_recv<T:Send,Tbuffer:Send>(mut p: RecvPacketBuffered<T, Tbuffer>)
-> Option<T> {
let p_ = p.unwrap();
let p = unsafe { &mut *p_ };
do (|| {
try_recv_(p)
}).finally {
unsafe {
if task::failing() {
p.header.state = Terminated;
let old_task = swap_task(&mut p.header.blocked_task, ptr::null());
if !old_task.is_null() {
rustrt::rust_task_deref(old_task);
}
}
}
}
}
fn try_recv_<T:Send>(p: &mut Packet<T>) -> Option<T> {
// optimistic path
match p.header.state {
Full => {
let payload = p.payload.take();
p.header.state = Empty;
return Some(payload.unwrap())
},
Terminated => return None,
_ => {}
}
// regular path
let this = unsafe { rustrt::rust_get_task() };
unsafe {
rustrt::task_clear_event_reject(this);
rustrt::rust_task_ref(this);
};
debug!("blocked = %x this = %x", p.header.blocked_task as uint,
this as uint);
let old_task = swap_task(&mut p.header.blocked_task, this);
debug!("blocked = %x this = %x old_task = %x",
p.header.blocked_task as uint,
this as uint, old_task as uint);
assert!(old_task.is_null());
let mut first = true;
let mut count = SPIN_COUNT;
loop {
unsafe {
rustrt::task_clear_event_reject(this);
}
let old_state = swap_state_acq(&mut p.header.state,
Blocked);
match old_state {
Empty => {
debug!("no data available on %?, going to sleep.", p);
if count == 0 {
wait_event(this);
}
else {
count -= 1;
// FIXME (#524): Putting the yield here destroys a lot
// of the benefit of spinning, since we still go into
// the scheduler at every iteration. However, without
// this everything spins too much because we end up
// sometimes blocking the thing we are waiting on.
task::yield();
}
debug!("woke up, p.state = %?", p.header.state);
}
Blocked => if first {
fail!("blocking on already blocked packet")
},
Full => {
let payload = p.payload.take();
let old_task = swap_task(&mut p.header.blocked_task, ptr::null());
if !old_task.is_null() {
unsafe {
rustrt::rust_task_deref(old_task);
}
}
p.header.state = Empty;
return Some(payload.unwrap())
}
Terminated => {
// This assert detects when we've accidentally unsafely
// casted too big of a number to a state.
assert_eq!(old_state, Terminated);
let old_task = swap_task(&mut p.header.blocked_task, ptr::null());
if !old_task.is_null() {
unsafe {
rustrt::rust_task_deref(old_task);
}
}
return None;
}
}
first = false;
}
}
/// Returns true if messages are available.
pub fn peek<T:Send,Tb:Send>(p: &mut RecvPacketBuffered<T, Tb>) -> bool {
unsafe {
match (*p.header()).state {
Empty | Terminated => false,
Blocked => fail!("peeking on blocked packet"),
Full => true
}
}
}
fn sender_terminate<T:Send>(p: *mut Packet<T>) {
let p = unsafe {
&mut *p
};
match swap_state_rel(&mut p.header.state, Terminated) {
Empty => {
// The receiver will eventually clean up.
}
Blocked => {
// wake up the target
let old_task = swap_task(&mut p.header.blocked_task, ptr::null());
if !old_task.is_null() {
unsafe {
rustrt::task_signal_event(
old_task,
ptr::to_unsafe_ptr(&(p.header)) as *libc::c_void);
rustrt::rust_task_deref(old_task);
}
}
// The receiver will eventually clean up.
}
Full => {
// This is impossible
fail!("you dun goofed")
}
Terminated => {
assert!(p.header.blocked_task.is_null());
// I have to clean up, use drop_glue
}
}
}
fn receiver_terminate<T:Send>(p: *mut Packet<T>) {
let p = unsafe {
&mut *p
};
match swap_state_rel(&mut p.header.state, Terminated) {
Empty => {
assert!(p.header.blocked_task.is_null());
// the sender will clean up
}
Blocked => {
let old_task = swap_task(&mut p.header.blocked_task, ptr::null());
if !old_task.is_null() {
unsafe {
rustrt::rust_task_deref(old_task);
assert_eq!(old_task, rustrt::rust_get_task());
}
}
}
Terminated | Full => {
assert!(p.header.blocked_task.is_null());
// I have to clean up, use drop_glue
}
}
}
/** Returns when one of the packet headers reports data is available.
This function is primarily intended for building higher level waiting
functions, such as `select`, `select2`, etc.
It takes a vector slice of packet_headers and returns an index into
that vector. The index points to an endpoint that has either been
closed by the sender or has a message waiting to be received.
*/
pub fn wait_many<T: Selectable>(pkts: &mut [T]) -> uint {
let this = unsafe {
rustrt::rust_get_task()
};
unsafe {
rustrt::task_clear_event_reject(this);
}
let mut data_avail = false;
let mut ready_packet = pkts.len();
for (i, p) in pkts.mut_iter().enumerate() {
unsafe {
let p = &mut *p.header();
let old = p.mark_blocked(this);
match old {
Full | Terminated => {
data_avail = true;
ready_packet = i;
(*p).state = old;
break;
}
Blocked => fail!("blocking on blocked packet"),
Empty => ()
}
}
}
while !data_avail {
debug!("sleeping on %? packets", pkts.len());
let event = wait_event(this) as *PacketHeader;
let mut pos = None;
for (i, p) in pkts.mut_iter().enumerate() {
if p.header() == event {
pos = Some(i);
break;
}
};
match pos {
Some(i) => {
ready_packet = i;
data_avail = true;
}
None => debug!("ignoring spurious event, %?", event)
}
}
debug!("%?", &mut pkts[ready_packet]);
for p in pkts.mut_iter() {
unsafe {
(*p.header()).unblock()
}
}
debug!("%?, %?", ready_packet, &mut pkts[ready_packet]);
unsafe {
assert!((*pkts[ready_packet].header()).state == Full
|| (*pkts[ready_packet].header()).state == Terminated);
}
ready_packet
}
/** The sending end of a pipe. It can be used to send exactly one
message.
*/
pub type SendPacket<T> = SendPacketBuffered<T, Packet<T>>;
pub fn SendPacket<T>(p: *mut Packet<T>) -> SendPacket<T> {
SendPacketBuffered(p)
}
pub struct SendPacketBuffered<T, Tbuffer> {
p: Option<*mut Packet<T>>,
buffer: Option<BufferResource<Tbuffer>>,
}
#[unsafe_destructor]
impl<T:Send,Tbuffer:Send> Drop for SendPacketBuffered<T,Tbuffer> {
fn drop(&self) {
unsafe {
let this: &mut SendPacketBuffered<T,Tbuffer> = transmute(self);
if this.p != None {
sender_terminate(this.p.take_unwrap());
}
}
}
}
pub fn SendPacketBuffered<T,Tbuffer>(p: *mut Packet<T>)
-> SendPacketBuffered<T,Tbuffer> {
SendPacketBuffered {
p: Some(p),
buffer: unsafe {
Some(BufferResource(get_buffer(&mut (*p).header)))
}
}
}
impl<T,Tbuffer> SendPacketBuffered<T,Tbuffer> {
pub fn unwrap(&mut self) -> *mut Packet<T> {
self.p.take_unwrap()
}
pub fn header(&mut self) -> *mut PacketHeader {
match self.p {
Some(packet) => unsafe {
let packet = &mut *packet;
let header = ptr::to_mut_unsafe_ptr(&mut packet.header);
header
},
None => fail!("packet already consumed")
}
}
pub fn reuse_buffer(&mut self) -> BufferResource<Tbuffer> {
//error!("send reuse_buffer");
self.buffer.take_unwrap()
}
}
/// Represents the receive end of a pipe. It can receive exactly one
/// message.
pub type RecvPacket<T> = RecvPacketBuffered<T, Packet<T>>;
pub fn RecvPacket<T>(p: *mut Packet<T>) -> RecvPacket<T> {
RecvPacketBuffered(p)
}
pub struct RecvPacketBuffered<T, Tbuffer> {
p: Option<*mut Packet<T>>,
buffer: Option<BufferResource<Tbuffer>>,
}
#[unsafe_destructor]
impl<T:Send,Tbuffer:Send> Drop for RecvPacketBuffered<T,Tbuffer> {
fn drop(&self) {
unsafe {
let this: &mut RecvPacketBuffered<T,Tbuffer> = transmute(self);
if this.p != None {
receiver_terminate(this.p.take_unwrap())
}
}
}
}
impl<T:Send,Tbuffer:Send> RecvPacketBuffered<T, Tbuffer> {
pub fn unwrap(&mut self) -> *mut Packet<T> {
self.p.take_unwrap()
}
pub fn reuse_buffer(&mut self) -> BufferResource<Tbuffer> {
self.buffer.take_unwrap()
}
}
impl<T:Send,Tbuffer:Send> Selectable for RecvPacketBuffered<T, Tbuffer> {
fn header(&mut self) -> *mut PacketHeader {
match self.p {
Some(packet) => unsafe {
let packet = &mut *packet;
let header = ptr::to_mut_unsafe_ptr(&mut packet.header);
header
},
None => fail!("packet already consumed")
}
}
}
pub fn RecvPacketBuffered<T,Tbuffer>(p: *mut Packet<T>)
-> RecvPacketBuffered<T,Tbuffer> {
RecvPacketBuffered {
p: Some(p),
buffer: unsafe {
Some(BufferResource(get_buffer(&mut (*p).header)))
}
}
}
pub fn entangle<T>() -> (RecvPacket<T>, SendPacket<T>) {
let p = packet();
(RecvPacket(p), SendPacket(p))
}
/** Receives a message from one of two endpoints.
The return value is `left` if the first endpoint received something,
or `right` if the second endpoint receives something. In each case,
the result includes the other endpoint as well so it can be used
again. Below is an example of using `select2`.
~~~ {.rust}
match select2(a, b) {
left((none, b)) {
// endpoint a was closed.
}
right((a, none)) {
// endpoint b was closed.
}
left((Some(_), b)) {
// endpoint a received a message
}
right(a, Some(_)) {
// endpoint b received a message.
}
}
~~~
Sometimes messages will be available on both endpoints at once. In
this case, `select2` may return either `left` or `right`.
*/
pub fn select2<A:Send,Ab:Send,B:Send,Bb:Send>(
mut a: RecvPacketBuffered<A, Ab>,
mut b: RecvPacketBuffered<B, Bb>)
-> Either<(Option<A>, RecvPacketBuffered<B, Bb>),
(RecvPacketBuffered<A, Ab>, Option<B>)> {
let mut endpoints = [ a.header(), b.header() ];
let i = wait_many(endpoints);
match i {
0 => Left((try_recv(a), b)),
1 => Right((a, try_recv(b))),
_ => fail!("select2 return an invalid packet")
}
}
pub trait Selectable {
fn header(&mut self) -> *mut PacketHeader;
}
impl Selectable for *mut PacketHeader {
fn header(&mut self) -> *mut PacketHeader { *self }
}
/// Returns the index of an endpoint that is ready to receive.
pub fn selecti<T:Selectable>(endpoints: &mut [T]) -> uint {
wait_many(endpoints)
}
/// Returns 0 or 1 depending on which endpoint is ready to receive
pub fn select2i<A:Selectable,B:Selectable>(a: &mut A, b: &mut B)
-> Either<(), ()> {
let mut endpoints = [ a.header(), b.header() ];
match wait_many(endpoints) {
0 => Left(()),
1 => Right(()),
_ => fail!("wait returned unexpected index")
}
}
/// Waits on a set of endpoints. Returns a message, its index, and a
/// list of the remaining endpoints.
pub fn select<T:Send,Tb:Send>(mut endpoints: ~[RecvPacketBuffered<T, Tb>])
-> (uint,
Option<T>,
~[RecvPacketBuffered<T, Tb>]) {
let mut endpoint_headers = ~[];
for endpoint in endpoints.mut_iter() {
endpoint_headers.push(endpoint.header());
}
let ready = wait_many(endpoint_headers);
let mut remaining = endpoints;
let port = remaining.swap_remove(ready);
let result = try_recv(port);
(ready, result, remaining)
}
pub mod rt {
use option::{None, Option, Some};
// These are used to hide the option constructors from the
// compiler because their names are changing
pub fn make_some<T>(val: T) -> Option<T> { Some(val) }
pub fn make_none<T>() -> Option<T> { None }
}

50
src/libstd/rt/borrowck.rs
View file

@ -9,7 +9,7 @@
// except according to those terms.
use cast::transmute;
use libc::{c_char, c_void, size_t, STDERR_FILENO};
use libc::{c_char, size_t, STDERR_FILENO};
use io;
use io::{Writer, WriterUtil};
use option::{Option, None, Some};
@ -20,9 +20,6 @@ use sys;
use unstable::raw;
use vec::ImmutableVector;
#[allow(non_camel_case_types)]
type rust_task = c_void;
pub static FROZEN_BIT: uint = 1 << (uint::bits - 1);
pub static MUT_BIT: uint = 1 << (uint::bits - 2);
static ALL_BITS: uint = FROZEN_BIT | MUT_BIT;
@ -35,34 +32,12 @@ struct BorrowRecord {
}
fn try_take_task_borrow_list() -> Option<~[BorrowRecord]> {
unsafe {
let cur_task: *rust_task = rust_try_get_task();
if cur_task.is_not_null() {
let ptr = rust_take_task_borrow_list(cur_task);
if ptr.is_null() {
None
} else {
let v: ~[BorrowRecord] = transmute(ptr);
Some(v)
}
} else {
None
}
}
// XXX
None
}
fn swap_task_borrow_list(f: &fn(~[BorrowRecord]) -> ~[BorrowRecord]) {
unsafe {
let cur_task: *rust_task = rust_try_get_task();
if cur_task.is_not_null() {
let mut borrow_list: ~[BorrowRecord] = {
let ptr = rust_take_task_borrow_list(cur_task);
if ptr.is_null() { ~[] } else { transmute(ptr) }
};
borrow_list = f(borrow_list);
rust_set_task_borrow_list(cur_task, transmute(borrow_list));
}
}
fn swap_task_borrow_list(_f: &fn(~[BorrowRecord]) -> ~[BorrowRecord]) {
// XXX
}
pub unsafe fn clear_task_borrow_list() {
@ -113,7 +88,8 @@ unsafe fn debug_borrow<T>(tag: &'static str,
//! A useful debugging function that prints a pointer + tag + newline
//! without allocating memory.
if ENABLE_DEBUG && ::rt::env::get().debug_borrow {
// XXX
if false {
debug_borrow_slow(tag, p, old_bits, new_bits, filename, line);
}
@ -269,15 +245,3 @@ pub unsafe fn check_not_borrowed(a: *u8,
fail_borrowed(a, file, line);
}
}
extern {
#[rust_stack]
pub fn rust_take_task_borrow_list(task: *rust_task) -> *c_void;
#[rust_stack]
pub fn rust_set_task_borrow_list(task: *rust_task, map: *c_void);
#[rust_stack]
pub fn rust_try_get_task() -> *rust_task;
}

41
src/libstd/rt/env.rs
View file

@ -11,50 +11,9 @@
//! Runtime environment settings
use from_str::FromStr;
use libc::{size_t, c_char, c_int};
use option::{Some, None};
use os;
// OLD RT stuff
pub struct Environment {
/// The number of threads to use by default
num_sched_threads: size_t,
/// The minimum size of a stack segment
min_stack_size: size_t,
/// The maximum amount of total stack per task before aborting
max_stack_size: size_t,
/// The default logging configuration
logspec: *c_char,
/// Record and report detailed information about memory leaks
detailed_leaks: bool,
/// Seed the random number generator
rust_seed: *c_char,
/// Poison allocations on free
poison_on_free: bool,
/// The argc value passed to main
argc: c_int,
/// The argv value passed to main
argv: **c_char,
/// Print GC debugging info (true if env var RUST_DEBUG_MEM is set)
debug_mem: bool,
/// Print GC debugging info (true if env var RUST_DEBUG_BORROW is set)
debug_borrow: bool,
}
/// Get the global environment settings
/// # Safety Note
/// This will abort the process if run outside of task context
pub fn get() -> &Environment {
unsafe { rust_get_rt_env() }
}
extern {
fn rust_get_rt_env() -> &Environment;
}
// NEW RT stuff
// Note that these are all accessed without any synchronization.
// They are expected to be initialized once then left alone.

3
src/libstd/rt/io/comm_adapters.rs
View file

@ -8,7 +8,8 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use prelude::*;
use option::Option;
use comm::{GenericPort, GenericChan};
use super::{Reader, Writer};
struct PortReader<P>;

31
src/libstd/rt/local_heap.rs
View file

@ -13,9 +13,6 @@
use libc;
use libc::{c_void, uintptr_t, size_t};
use ops::Drop;
use option::{Some, None};
use rt;
use rt::OldTaskContext;
use rt::local::Local;
use rt::task::Task;
use unstable::raw;
@ -87,32 +84,14 @@ impl Drop for LocalHeap {
// A little compatibility function
pub unsafe fn local_free(ptr: *libc::c_char) {
// XXX: Unsafe borrow for speed. Lame.
match Local::try_unsafe_borrow::<Task>() {
Some(task) => {
(*task).heap.free(ptr as *libc::c_void);
}
None => {
rust_upcall_free_noswitch(ptr);
extern {
#[fast_ffi]
fn rust_upcall_free_noswitch(ptr: *libc::c_char);
}
}
do Local::borrow::<Task,()> |task| {
task.heap.free(ptr as *libc::c_void);
}
}
pub fn live_allocs() -> *raw::Box<()> {
let region = match rt::context() {
OldTaskContext => {
unsafe { rust_current_boxed_region() }
}
_ => {
do Local::borrow::<Task, *BoxedRegion> |task| {
task.heap.boxed_region
}
}
let region = do Local::borrow::<Task, *BoxedRegion> |task| {
task.heap.boxed_region
};
return unsafe { (*region).live_allocs };
@ -140,8 +119,6 @@ extern {
size: size_t) -> *OpaqueBox;
#[fast_ffi]
fn rust_boxed_region_free(region: *BoxedRegion, box: *OpaqueBox);
#[fast_ffi]
fn rust_current_boxed_region() -> *BoxedRegion;
}
#[cfg(test)]

33
src/libstd/rt/mod.rs
View file

@ -120,7 +120,7 @@ mod context;
/// Bindings to system threading libraries.
mod thread;
/// The runtime configuration, read from environment variables
/// The runtime configuration, read from environment variables.
pub mod env;
/// The local, managed heap
@ -401,35 +401,6 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int {
}
}
/// Possible contexts in which Rust code may be executing.
/// Different runtime services are available depending on context.
/// Mostly used for determining if we're using the new scheduler
/// or the old scheduler.
#[deriving(Eq)]
pub enum RuntimeContext {
// Running in an old-style task
OldTaskContext,
// Not old task context
NewRtContext
}
/// Determine the current RuntimeContext
pub fn context() -> RuntimeContext {
use task::rt::rust_task;
if unsafe { rust_try_get_task().is_not_null() } {
return OldTaskContext;
} else {
return NewRtContext;
}
extern {
#[rust_stack]
pub fn rust_try_get_task() -> *rust_task;
}
}
pub fn in_sched_context() -> bool {
unsafe {
match Local::try_unsafe_borrow::<Task>() {
@ -456,4 +427,4 @@ pub fn in_green_task_context() -> bool {
None => false
}
}
}
}

4
src/libstd/rt/task.rs
View file

@ -515,8 +515,8 @@ mod test {
do run_in_newsched_task {
let (port, chan) = oneshot();
send_one(chan, 10);
assert!(recv_one(port) == 10);
chan.send(10);
assert!(port.recv() == 10);
}
}

9
src/libstd/rt/uv/uvll.rs
View file

@ -448,13 +448,6 @@ pub unsafe fn get_base_from_buf(buf: uv_buf_t) -> *u8 {
pub unsafe fn get_len_from_buf(buf: uv_buf_t) -> size_t {
return rust_uv_get_len_from_buf(buf);
}
pub unsafe fn malloc_buf_base_of(suggested_size: size_t) -> *u8 {
return rust_uv_malloc_buf_base_of(suggested_size);
}
pub unsafe fn free_base_of_buf(buf: uv_buf_t) {
rust_uv_free_base_of_buf(buf);
}
pub unsafe fn get_last_err_info(uv_loop: *c_void) -> ~str {
let err = last_error(uv_loop);
let err_ptr = ptr::to_unsafe_ptr(&err);
@ -558,8 +551,6 @@ extern {
repeat: libc::uint64_t) -> c_int;
fn rust_uv_timer_stop(handle: *uv_timer_t) -> c_int;
fn rust_uv_malloc_buf_base_of(sug_size: size_t) -> *u8;
fn rust_uv_free_base_of_buf(buf: uv_buf_t);
fn rust_uv_get_stream_handle_from_connect_req(connect_req: *uv_connect_t) -> *uv_stream_t;
fn rust_uv_get_stream_handle_from_write_req(write_req: *uv_write_t) -> *uv_stream_t;
fn rust_uv_get_loop_for_uv_handle(handle: *c_void) -> *c_void;

2
src/libstd/std.rs
View file

@ -164,7 +164,6 @@ pub mod trie;
pub mod task;
pub mod comm;
pub mod pipes;
pub mod local_data;
@ -213,7 +212,6 @@ mod std {
pub use kinds;
pub use local_data;
pub use sys;
pub use pipes;
pub use unstable;
pub use str;
pub use os;

78
src/libstd/sys.rs
View file

@ -21,15 +21,6 @@ use str::StrSlice;
use str;
use unstable::intrinsics;
pub mod rustrt {
use libc::{c_char, size_t};
extern {
#[rust_stack]
pub fn rust_upcall_fail(expr: *c_char, file: *c_char, line: size_t);
}
}
/// Returns the size of a type
#[inline]
pub fn size_of<T>() -> uint {
@ -136,55 +127,44 @@ impl FailWithCause for &'static str {
pub fn begin_unwind_(msg: *c_char, file: *c_char, line: size_t) -> ! {
use either::Left;
use option::{Some, None};
use rt::{context, OldTaskContext, in_green_task_context};
use rt::in_green_task_context;
use rt::task::Task;
use rt::local::Local;
use rt::logging::Logger;
use str::Str;
let context = context();
match context {
OldTaskContext => {
unsafe {
rustrt::rust_upcall_fail(msg, file, line);
cast::transmute(())
unsafe {
// XXX: Bad re-allocations. fail! needs some refactoring
let msg = str::raw::from_c_str(msg);
let file = str::raw::from_c_str(file);
// XXX: Logging doesn't work correctly in non-task context because it
// invokes the local heap
if in_green_task_context() {
// XXX: Logging doesn't work here - the check to call the log
// function never passes - so calling the log function directly.
do Local::borrow::<Task, ()> |task| {
let msg = match task.name {
Some(ref name) =>
fmt!("task '%s' failed at '%s', %s:%i",
name.as_slice(), msg, file, line as int),
None =>
fmt!("task <unnamed> failed at '%s', %s:%i",
msg, file, line as int)
};
task.logger.log(Left(msg));
}
} else {
rterrln!("failed in non-task context at '%s', %s:%i",
msg, file, line as int);
}
_ => {
unsafe {
// XXX: Bad re-allocations. fail! needs some refactoring
let msg = str::raw::from_c_str(msg);
let file = str::raw::from_c_str(file);
// XXX: Logging doesn't work correctly in non-task context because it
// invokes the local heap
if in_green_task_context() {
// XXX: Logging doesn't work here - the check to call the log
// function never passes - so calling the log function directly.
do Local::borrow::<Task, ()> |task| {
let msg = match task.name {
Some(ref name) =>
fmt!("task '%s' failed at '%s', %s:%i",
name.as_slice(), msg, file, line as int),
None =>
fmt!("task <unnamed> failed at '%s', %s:%i",
msg, file, line as int)
};
task.logger.log(Left(msg));
}
} else {
rterrln!("failed in non-task context at '%s', %s:%i",
msg, file, line as int);
}
let task = Local::unsafe_borrow::<Task>();
if (*task).unwinder.unwinding {
rtabort!("unwinding again");
}
(*task).unwinder.begin_unwind();
}
let task = Local::unsafe_borrow::<Task>();
if (*task).unwinder.unwinding {
rtabort!("unwinding again");
}
(*task).unwinder.begin_unwind();
}
}

36
src/libstd/task/local_data_priv.rs
View file

@ -15,32 +15,21 @@ use libc;
use local_data;
use prelude::*;
use ptr;
use task::rt;
use unstable::raw;
use util;
use super::rt::rust_task;
use rt::task::{Task, LocalStorage};
pub enum Handle {
OldHandle(*rust_task),
NewHandle(*mut LocalStorage)
}
impl Handle {
pub fn new() -> Handle {
use rt::{context, OldTaskContext};
use rt::local::Local;
unsafe {
match context() {
OldTaskContext => {
OldHandle(rt::rust_get_task())
}
_ => {
let task = Local::unsafe_borrow::<Task>();
NewHandle(&mut (*task).storage)
}
}
let task = Local::unsafe_borrow::<Task>();
NewHandle(&mut (*task).storage)
}
}
}
@ -109,26 +98,6 @@ 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) -> &mut TaskLocalMap {
unsafe fn oldsched_map(task: *rust_task) -> &mut TaskLocalMap {
extern fn cleanup_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 is an owned pointer and is "owned" by TLS. It is moved
// into the tls slot for this task, and then mutable loans are taken
// from this slot to modify the map.
let map_ptr = rt::rust_get_task_local_data(task);
if (*map_ptr).is_null() {
// First time TLS is used, create a new map and set up the necessary
// TLS information for its safe destruction
let map: TaskLocalMap = ~[];
*map_ptr = cast::transmute(map);
rt::rust_task_local_data_atexit(task, cleanup_extern_cb);
}
return cast::transmute(map_ptr);
}
unsafe fn newsched_map(local: *mut LocalStorage) -> &mut TaskLocalMap {
// This is based on the same idea as the oldsched code above.
match &mut *local {
@ -152,7 +121,6 @@ unsafe fn get_local_map(handle: Handle) -> &mut TaskLocalMap {
}
match handle {
OldHandle(task) => oldsched_map(task),
NewHandle(local_storage) => newsched_map(local_storage)
}
}

110
src/libstd/task/mod.rs
View file

@ -42,7 +42,7 @@ use cmp::Eq;
use comm::{stream, Chan, GenericChan, GenericPort, Port};
use result::Result;
use result;
use rt::{context, OldTaskContext, in_green_task_context};
use rt::in_green_task_context;
use rt::local::Local;
use unstable::finally::Finally;
use util;
@ -54,7 +54,6 @@ use util;
#[cfg(test)] use task;
mod local_data_priv;
pub mod rt;
pub mod spawn;
/**
@ -535,35 +534,21 @@ pub fn with_task_name<U>(blk: &fn(Option<&str>) -> U) -> U {
}
}
} else {
fail!("no task name exists in %?", context())
fail!("no task name exists in non-green task context")
}
}
pub fn yield() {
//! Yield control to the task scheduler
use rt::{context, OldTaskContext};
use rt::local::Local;
use rt::sched::Scheduler;
unsafe {
match context() {
OldTaskContext => {
let task_ = rt::rust_get_task();
let killed = rt::rust_task_yield(task_);
if killed && !failing() {
fail!("killed");
}
}
_ => {
// XXX: What does yield really mean in newsched?
// FIXME(#7544): Optimize this, since we know we won't block.
let sched = Local::take::<Scheduler>();
do sched.deschedule_running_task_and_then |sched, task| {
sched.enqueue_blocked_task(task);
}
}
}
// XXX: What does yield really mean in newsched?
// FIXME(#7544): Optimize this, since we know we won't block.
let sched = Local::take::<Scheduler>();
do sched.deschedule_running_task_and_then |sched, task| {
sched.enqueue_blocked_task(task);
}
}
@ -572,17 +557,8 @@ pub fn failing() -> bool {
use rt::task::Task;
match context() {
OldTaskContext => {
unsafe {
rt::rust_task_is_unwinding(rt::rust_get_task())
}
}
_ => {
do Local::borrow::<Task, bool> |local| {
local.unwinder.unwinding
}
}
do Local::borrow::<Task, bool> |local| {
local.unwinder.unwinding
}
}
@ -605,29 +581,19 @@ pub fn unkillable<U>(f: &fn() -> U) -> U {
use rt::task::Task;
unsafe {
match context() {
OldTaskContext => {
let t = rt::rust_get_task();
do (|| {
rt::rust_task_inhibit_kill(t);
f()
}).finally {
rt::rust_task_allow_kill(t);
}
}
_ if in_green_task_context() => {
// The inhibits/allows might fail and need to borrow the task.
let t = Local::unsafe_borrow::<Task>();
do (|| {
(*t).death.inhibit_kill((*t).unwinder.unwinding);
f()
}).finally {
(*t).death.allow_kill((*t).unwinder.unwinding);
}
if in_green_task_context() {
// The inhibits/allows might fail and need to borrow the task.
let t = Local::unsafe_borrow::<Task>();
do (|| {
(*t).death.inhibit_kill((*t).unwinder.unwinding);
f()
}).finally {
(*t).death.allow_kill((*t).unwinder.unwinding);
}
} else {
// FIXME(#3095): This should be an rtabort as soon as the scheduler
// no longer uses a workqueue implemented with an Exclusive.
_ => f()
f()
}
}
}
@ -636,27 +602,17 @@ pub fn unkillable<U>(f: &fn() -> U) -> U {
pub unsafe fn rekillable<U>(f: &fn() -> U) -> U {
use rt::task::Task;
match context() {
OldTaskContext => {
let t = rt::rust_get_task();
do (|| {
rt::rust_task_allow_kill(t);
f()
}).finally {
rt::rust_task_inhibit_kill(t);
}
}
_ if in_green_task_context() => {
let t = Local::unsafe_borrow::<Task>();
do (|| {
(*t).death.allow_kill((*t).unwinder.unwinding);
f()
}).finally {
(*t).death.inhibit_kill((*t).unwinder.unwinding);
}
if in_green_task_context() {
let t = Local::unsafe_borrow::<Task>();
do (|| {
(*t).death.allow_kill((*t).unwinder.unwinding);
f()
}).finally {
(*t).death.inhibit_kill((*t).unwinder.unwinding);
}
} else {
// FIXME(#3095): As in unkillable().
_ => f()
f()
}
}
@ -1034,14 +990,8 @@ fn test_try_fail() {
#[cfg(test)]
fn get_sched_id() -> int {
if context() == OldTaskContext {
unsafe {
rt::rust_get_sched_id() as int
}
} else {
do Local::borrow::<::rt::sched::Scheduler, int> |sched| {
sched.sched_id() as int
}
do Local::borrow::<::rt::sched::Scheduler, int> |sched| {
sched.sched_id() as int
}
}

66
src/libstd/task/rt.rs
View file

@ -1,66 +0,0 @@
// 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.
/*!
The task interface to the runtime
*/
#[doc(hidden)];
use libc;
#[allow(non_camel_case_types)] // runtime type
pub type sched_id = int;
#[allow(non_camel_case_types)] // runtime type
pub type task_id = int;
// These are both opaque runtime/compiler types that we don't know the
// structure of and should only deal with via unsafe pointer
#[allow(non_camel_case_types)] // runtime type
pub type rust_task = libc::c_void;
#[allow(non_camel_case_types)] // runtime type
pub type rust_closure = libc::c_void;
extern {
#[rust_stack]
pub fn rust_task_yield(task: *rust_task) -> bool;
pub fn rust_get_sched_id() -> sched_id;
pub fn rust_new_sched(num_threads: libc::uintptr_t) -> sched_id;
pub fn get_task_id() -> task_id;
#[rust_stack]
pub fn rust_get_task() -> *rust_task;
pub fn new_task() -> *rust_task;
pub fn rust_new_task_in_sched(id: sched_id) -> *rust_task;
pub fn start_task(task: *rust_task, closure: *rust_closure);
pub fn rust_task_is_unwinding(task: *rust_task) -> bool;
pub fn rust_osmain_sched_id() -> sched_id;
#[rust_stack]
pub fn rust_task_inhibit_kill(t: *rust_task);
#[rust_stack]
pub fn rust_task_allow_kill(t: *rust_task);
#[rust_stack]
pub fn rust_task_inhibit_yield(t: *rust_task);
#[rust_stack]
pub fn rust_task_allow_yield(t: *rust_task);
pub fn rust_task_kill_other(task: *rust_task);
pub fn rust_task_kill_all(task: *rust_task);
#[rust_stack]
pub fn rust_get_task_local_data(task: *rust_task) -> *mut *libc::c_void;
#[rust_stack]
pub fn rust_task_local_data_atexit(task: *rust_task, cleanup_fn: *u8);
}

143
src/libstd/task/spawn.rs
View file

@ -81,9 +81,6 @@ use container::MutableMap;
use comm::{Chan, GenericChan, oneshot};
use hashmap::{HashSet, HashSetConsumeIterator};
use local_data;
use task::local_data_priv::{local_get, local_set, OldHandle};
use task::rt::rust_task;
use task::rt;
use task::{Failure, SingleThreaded};
use task::{Success, TaskOpts, TaskResult};
use task::unkillable;
@ -91,7 +88,7 @@ use to_bytes::IterBytes;
use uint;
use util;
use unstable::sync::Exclusive;
use rt::{OldTaskContext, NewRtContext, context, in_green_task_context};
use rt::in_green_task_context;
use rt::local::Local;
use rt::task::{Task, Sched};
use rt::kill::KillHandle;
@ -107,14 +104,12 @@ use rt::work_queue::WorkQueue;
// Transitionary.
#[deriving(Eq)]
enum TaskHandle {
OldTask(*rust_task),
NewTask(KillHandle),
}
impl Clone for TaskHandle {
fn clone(&self) -> TaskHandle {
match *self {
OldTask(x) => OldTask(x),
NewTask(ref x) => NewTask(x.clone()),
}
}
@ -123,7 +118,6 @@ impl Clone for TaskHandle {
impl IterBytes for TaskHandle {
fn iter_bytes(&self, lsb0: bool, f: &fn(buf: &[u8]) -> bool) -> bool {
match *self {
OldTask(ref x) => x.iter_bytes(lsb0, f),
NewTask(ref x) => x.iter_bytes(lsb0, f),
}
}
@ -498,7 +492,6 @@ struct RuntimeGlue;
impl RuntimeGlue {
unsafe fn kill_task(task: TaskHandle) {
match task {
OldTask(ptr) => rt::rust_task_kill_other(ptr),
NewTask(handle) => {
let mut handle = handle;
do handle.kill().map_move |killed_task| {
@ -513,7 +506,6 @@ impl RuntimeGlue {
unsafe fn kill_all_tasks(task: &TaskHandle) {
match *task {
OldTask(ptr) => rt::rust_task_kill_all(ptr),
// FIXME(#7544): Remove the kill_all feature entirely once the
// oldsched goes away.
NewTask(ref _handle) => rtabort!("can't kill_all in newsched"),
@ -521,12 +513,8 @@ impl RuntimeGlue {
}
fn with_task_handle_and_failing(blk: &fn(TaskHandle, bool)) {
match context() {
OldTaskContext => unsafe {
let me = rt::rust_get_task();
blk(OldTask(me), rt::rust_task_is_unwinding(me))
},
NewRtContext if in_green_task_context() => unsafe {
if in_green_task_context() {
unsafe {
// Can't use safe borrow, because the taskgroup destructor needs to
// access the scheduler again to send kill signals to other tasks.
let me = Local::unsafe_borrow::<Task>();
@ -534,36 +522,15 @@ impl RuntimeGlue {
// Will probably have to wait until the old rt is gone.
blk(NewTask((*me).death.kill_handle.get_ref().clone()),
(*me).unwinder.unwinding)
},
NewRtContext => rtabort!("task dying in bad context"),
}
} else {
rtabort!("task dying in bad context")
}
}
fn with_my_taskgroup<U>(blk: &fn(&Taskgroup) -> U) -> U {
match context() {
OldTaskContext => unsafe {
let me = rt::rust_get_task();
do local_get(OldHandle(me), taskgroup_key()) |g| {
match g {
None => {
// Main task, doing first spawn ever. Lazily initialise here.
let mut members = TaskSet::new();
members.insert(OldTask(me));
let tasks = Exclusive::new(Some(TaskGroupData {
members: members,
descendants: TaskSet::new(),
}));
// Main task/group has no ancestors, no notifier, etc.
let group = @@mut Taskgroup(tasks, AncestorList(None),
true, None);
local_set(OldHandle(me), taskgroup_key(), group);
blk(&**group)
}
Some(&group) => blk(&**group)
}
}
},
NewRtContext if in_green_task_context() => unsafe {
if in_green_task_context() {
unsafe {
// Can't use safe borrow, because creating new hashmaps for the
// tasksets requires an rng, which needs to borrow the sched.
let me = Local::unsafe_borrow::<Task>();
@ -587,8 +554,9 @@ impl RuntimeGlue {
}
Some(ref group) => group,
})
},
NewRtContext => rtabort!("spawning in bad context"),
}
} else {
rtabort!("spawning in bad context")
}
}
}
@ -598,7 +566,7 @@ fn gen_child_taskgroup(linked: bool, supervised: bool)
-> Option<(TaskGroupArc, AncestorList, bool)> {
// FIXME(#7544): Not safe to lazily initialize in the old runtime. Remove
// this context check once 'spawn_raw_oldsched' is gone.
if context() == OldTaskContext || linked || supervised {
if linked || supervised {
// with_my_taskgroup will lazily initialize the parent's taskgroup if
// it doesn't yet exist. We don't want to call it in the unlinked case.
do RuntimeGlue::with_my_taskgroup |spawner_group| {
@ -665,10 +633,10 @@ fn enlist_many(child: TaskHandle, child_arc: &TaskGroupArc,
}
pub fn spawn_raw(opts: TaskOpts, f: ~fn()) {
match context() {
OldTaskContext => spawn_raw_oldsched(opts, f),
_ if in_green_task_context() => spawn_raw_newsched(opts, f),
_ => fail!("can't spawn from this context")
if in_green_task_context() {
spawn_raw_newsched(opts, f)
} else {
fail!("can't spawn from this context")
}
}
@ -810,85 +778,6 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) {
}
fn spawn_raw_oldsched(mut opts: TaskOpts, f: ~fn()) {
let (child_tg, ancestors, is_main) =
gen_child_taskgroup(opts.linked, opts.supervised).expect("old runtime needs TG");
unsafe {
let child_data = Cell::new((child_tg, ancestors, f));
// Being killed with the unsafe task/closure pointers would leak them.
do unkillable {
let (child_tg, ancestors, f) = child_data.take(); // :(
// Create child task.
let new_task = match opts.sched.mode {
DefaultScheduler => rt::new_task(),
_ => new_task_in_sched()
};
assert!(!new_task.is_null());
// Getting killed after here would leak the task.
let child_wrapper = make_child_wrapper(new_task, child_tg,
ancestors, is_main, opts.notify_chan.take(), f);
let closure = cast::transmute(&child_wrapper);
// Getting killed between these two calls would free the child's
// closure. (Reordering them wouldn't help - then getting killed
// between them would leak.)
rt::start_task(new_task, closure);
cast::forget(child_wrapper);
}
}
// This function returns a closure-wrapper that we pass to the child task.
// (1) It sets up the notification channel.
// (2) It attempts to enlist in the child's group and all ancestor groups.
// (3a) If any of those fails, it leaves all groups, and does nothing.
// (3b) Otherwise it builds a task control structure and puts it in TLS,
// (4) ...and runs the provided body function.
fn make_child_wrapper(child: *rust_task, child_arc: TaskGroupArc,
ancestors: AncestorList, is_main: bool,
notify_chan: Option<Chan<TaskResult>>,
f: ~fn())
-> ~fn() {
let child_data = Cell::new((notify_chan, child_arc, ancestors));
let result: ~fn() = || {
let (notify_chan, child_arc, ancestors) = child_data.take(); // :(
let mut ancestors = ancestors;
// Child task runs this code.
// Even if the below code fails to kick the child off, we must
// send Something on the notify channel.
let notifier = notify_chan.map_move(|c| AutoNotify(c));
if enlist_many(OldTask(child), &child_arc, &mut ancestors) {
let group = @@mut Taskgroup(child_arc, ancestors, is_main, notifier);
unsafe {
local_set(OldHandle(child), taskgroup_key(), group);
}
// Run the child's body.
f();
// TLS cleanup code will exit the taskgroup.
}
// Run the box annihilator.
// FIXME #4428: Crashy.
// unsafe { cleanup::annihilate(); }
};
return result;
}
fn new_task_in_sched() -> *rust_task {
unsafe {
let sched_id = rt::rust_new_sched(1);
rt::rust_new_task_in_sched(sched_id)
}
}
}
#[test]
fn test_spawn_raw_simple() {
let (po, ch) = stream();

58
src/libstd/unstable/lang.rs
View file

@ -11,34 +11,13 @@
//! Runtime calls emitted by the compiler.
use cast::transmute;
use libc::{c_char, c_uchar, c_void, size_t, uintptr_t, c_int};
use option::{Some, None};
use libc::{c_char, c_uchar, c_void, size_t, uintptr_t};
use str;
use sys;
use rt::task::Task;
use rt::local::Local;
use rt::borrowck;
#[allow(non_camel_case_types)]
pub type rust_task = c_void;
pub mod rustrt {
use unstable::lang::rust_task;
use libc::{c_char, uintptr_t};
extern {
#[rust_stack]
pub fn rust_upcall_malloc(td: *c_char, size: uintptr_t) -> *c_char;
#[rust_stack]
pub fn rust_upcall_free(ptr: *c_char);
#[fast_ffi]
pub fn rust_upcall_malloc_noswitch(td: *c_char, size: uintptr_t)
-> *c_char;
#[rust_stack]
pub fn rust_try_get_task() -> *rust_task;
}
}
#[lang="fail_"]
pub fn fail_(expr: *c_char, file: *c_char, line: size_t) -> ! {
sys::begin_unwind_(expr, file, line);
@ -56,15 +35,14 @@ pub fn fail_bounds_check(file: *c_char, line: size_t,
#[lang="malloc"]
pub unsafe fn local_malloc(td: *c_char, size: uintptr_t) -> *c_char {
// XXX: Unsafe borrow for speed. Lame.
match Local::try_unsafe_borrow::<Task>() {
Some(task) => {
(*task).heap.alloc(td as *c_void, size as uint) as *c_char
}
None => {
rustrt::rust_upcall_malloc_noswitch(td, size)
}
let mut alloc = ::ptr::null();
do Local::borrow::<Task,()> |task| {
rtdebug!("task pointer: %x, heap pointer: %x",
::borrow::to_uint(task),
::borrow::to_uint(&task.heap));
alloc = task.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
@ -129,23 +107,11 @@ pub unsafe fn annihilate() {
pub fn start(main: *u8, argc: int, argv: **c_char,
crate_map: *u8) -> int {
use rt;
use os;
unsafe {
let use_old_rt = os::getenv("RUST_OLDRT").is_some();
if use_old_rt {
return rust_start(main as *c_void, argc as c_int, argv,
crate_map as *c_void) as int;
} else {
return do rt::start(argc, argv as **u8, crate_map) {
let main: extern "Rust" fn() = transmute(main);
main();
};
}
}
extern {
fn rust_start(main: *c_void, argc: c_int, argv: **c_char,
crate_map: *c_void) -> c_int;
return do rt::start(argc, argv as **u8, crate_map) {
let main: extern "Rust" fn() = transmute(main);
main();
};
}
}

40
src/libstd/unstable/sync.rs
View file

@ -280,39 +280,19 @@ impl<T> Drop for UnsafeAtomicRcBox<T>{
// FIXME(#8140) should not be pub
pub unsafe fn atomically<U>(f: &fn() -> U) -> U {
use rt::task::Task;
use task::rt;
use rt::local::Local;
use rt::{context, OldTaskContext};
use rt::in_green_task_context;
match context() {
OldTaskContext => {
let t = rt::rust_get_task();
do (|| {
rt::rust_task_inhibit_kill(t);
rt::rust_task_inhibit_yield(t);
f()
}).finally {
rt::rust_task_allow_yield(t);
rt::rust_task_allow_kill(t);
}
}
_ => {
let t = Local::try_unsafe_borrow::<Task>();
match t {
Some(t) => {
do (|| {
(*t).death.inhibit_yield();
f()
}).finally {
(*t).death.allow_yield();
}
}
None => {
// FIXME(#3095): As in unkillable().
f()
}
}
if in_green_task_context() {
let t = Local::unsafe_borrow::<Task>();
do (|| {
(*t).death.inhibit_yield();
f()
}).finally {
(*t).death.allow_yield();
}
} else {
f()
}
}