core: Remove oldcomm
This commit is contained in:
Brian Anderson
parent
5633783f47
commit
542bf20414
|
|
@ -144,7 +144,6 @@ pub mod hashmap;
|
|||
|
||||
/* Tasks and communication */
|
||||
|
||||
pub mod oldcomm;
|
||||
#[path = "task/mod.rs"]
|
||||
pub mod task;
|
||||
pub mod pipes;
|
||||
|
|
|
|||
|
|
@ -1,547 +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.
|
||||
|
||||
/*!
|
||||
|
||||
Deprecated communication between tasks
|
||||
|
||||
Communication between tasks is facilitated by ports (in the receiving
|
||||
task), and channels (in the sending task). Any number of channels may
|
||||
feed into a single port. Ports and channels may only transmit values
|
||||
of unique types; that is, values that are statically guaranteed to be
|
||||
accessed by a single 'owner' at a time. Unique types include scalars,
|
||||
vectors, strings, and records, tags, tuples and unique boxes (`~T`)
|
||||
thereof. Most notably, shared boxes (`@T`) may not be transmitted
|
||||
across channels.
|
||||
|
||||
# Example
|
||||
|
||||
~~~
|
||||
let po = oldcomm::Port();
|
||||
let ch = oldcomm::Chan(&po);
|
||||
|
||||
do task::spawn {
|
||||
oldcomm::send(ch, ~"Hello, World");
|
||||
}
|
||||
|
||||
io::println(oldcomm::recv(po));
|
||||
~~~
|
||||
|
||||
# Note
|
||||
|
||||
Use of this module is deprecated in favor of `core::pipes`. In the
|
||||
`core::comm` will likely be rewritten with pipes, at which point it
|
||||
will once again be the preferred module for intertask communication.
|
||||
|
||||
*/
|
||||
|
||||
// NB: transitionary, de-mode-ing.
|
||||
#[forbid(deprecated_mode)];
|
||||
#[forbid(deprecated_pattern)];
|
||||
|
||||
use cast;
|
||||
use either;
|
||||
use either::Either;
|
||||
use iter;
|
||||
use libc;
|
||||
use libc::size_t;
|
||||
use prelude::*;
|
||||
use ptr;
|
||||
use result;
|
||||
use sys;
|
||||
use task;
|
||||
use vec;
|
||||
|
||||
// After snapshot, change p2::addr_of => addr_of
|
||||
|
||||
/**
|
||||
* A communication endpoint that can receive messages
|
||||
*
|
||||
* Each port has a unique per-task identity and may not be replicated or
|
||||
* transmitted. If a port value is copied, both copies refer to the same
|
||||
* port. Ports may be associated with multiple `chan`s.
|
||||
*/
|
||||
pub enum Port<T> {
|
||||
Port_(@PortPtr<T>)
|
||||
}
|
||||
|
||||
// It's critical that this only have one variant, so it has a record
|
||||
// layout, and will work in the rust_task structure in task.rs.
|
||||
/**
|
||||
* A communication endpoint that can send messages
|
||||
*
|
||||
* Each channel is bound to a port when the channel is constructed, so
|
||||
* the destination port for a channel must exist before the channel
|
||||
* itself. Channels are weak: a channel does not keep the port it is
|
||||
* bound to alive. If a channel attempts to send data to a dead port that
|
||||
* data will be silently dropped. Channels may be duplicated and
|
||||
* themselves transmitted over other channels.
|
||||
*/
|
||||
pub enum Chan<T> {
|
||||
Chan_(port_id)
|
||||
}
|
||||
|
||||
/// Constructs a port
|
||||
pub fn Port<T: Owned>() -> Port<T> {
|
||||
unsafe {
|
||||
Port_(@PortPtr(rustrt::new_port(sys::nonzero_size_of::<T>()
|
||||
as size_t)))
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Owned> Port<T> {
|
||||
|
||||
fn chan() -> Chan<T> { Chan(&self) }
|
||||
fn send(v: T) { self.chan().send(move v) }
|
||||
fn recv() -> T { recv(self) }
|
||||
fn peek() -> bool { peek(self) }
|
||||
|
||||
}
|
||||
|
||||
impl<T: Owned> Chan<T> {
|
||||
|
||||
fn chan() -> Chan<T> { self }
|
||||
fn send(v: T) { send(self, move v) }
|
||||
fn recv() -> T { recv_chan(self) }
|
||||
fn peek() -> bool { peek_chan(self) }
|
||||
|
||||
}
|
||||
|
||||
/// Open a new receiving channel for the duration of a function
|
||||
pub fn listen<T: Owned, U>(f: fn(Chan<T>) -> U) -> U {
|
||||
let po = Port();
|
||||
f(po.chan())
|
||||
}
|
||||
|
||||
struct PortPtr<T> {
|
||||
po: *rust_port,
|
||||
drop {
|
||||
unsafe {
|
||||
do task::unkillable {
|
||||
// Once the port is detached it's guaranteed not to receive further
|
||||
// messages
|
||||
let yield = 0;
|
||||
let yieldp = ptr::addr_of(&yield);
|
||||
rustrt::rust_port_begin_detach(self.po, yieldp);
|
||||
if yield != 0 {
|
||||
// Need to wait for the port to be detached
|
||||
task::yield();
|
||||
}
|
||||
rustrt::rust_port_end_detach(self.po);
|
||||
|
||||
// Drain the port so that all the still-enqueued items get dropped
|
||||
while rustrt::rust_port_size(self.po) > 0 as size_t {
|
||||
recv_::<T>(self.po);
|
||||
}
|
||||
rustrt::del_port(self.po);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn PortPtr<T: Owned>(po: *rust_port) -> PortPtr<T> {
|
||||
PortPtr {
|
||||
po: po
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal function for converting from a channel to a port
|
||||
*
|
||||
* # Failure
|
||||
*
|
||||
* Fails if the port is detached or dead. Fails if the port
|
||||
* is owned by a different task.
|
||||
*/
|
||||
fn as_raw_port<T: Owned, U>(ch: Chan<T>, f: fn(*rust_port) -> U) -> U {
|
||||
|
||||
struct PortRef {
|
||||
p: *rust_port,
|
||||
drop {
|
||||
unsafe {
|
||||
if !ptr::is_null(self.p) {
|
||||
rustrt::rust_port_drop(self.p);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn PortRef(p: *rust_port) -> PortRef {
|
||||
PortRef {
|
||||
p: p
|
||||
}
|
||||
}
|
||||
|
||||
unsafe {
|
||||
let p = PortRef(rustrt::rust_port_take(*ch));
|
||||
|
||||
if ptr::is_null(p.p) {
|
||||
die!(~"unable to locate port for channel")
|
||||
} else if rustrt::get_task_id() != rustrt::rust_port_task(p.p) {
|
||||
die!(~"unable to access unowned port")
|
||||
}
|
||||
|
||||
f(p.p)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructs a channel. The channel is bound to the port used to
|
||||
* construct it.
|
||||
*/
|
||||
pub fn Chan<T: Owned>(p: &Port<T>) -> Chan<T> {
|
||||
unsafe {
|
||||
Chan_(rustrt::get_port_id((**p).po))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends data over a channel. The sent data is moved into the channel,
|
||||
* whereupon the caller loses access to it.
|
||||
*/
|
||||
pub fn send<T: Owned>(ch: Chan<T>, data: T) {
|
||||
unsafe {
|
||||
let Chan_(p) = ch;
|
||||
let data_ptr = ptr::addr_of(&data) as *();
|
||||
let res = rustrt::rust_port_id_send(p, data_ptr);
|
||||
if res != 0 {
|
||||
// Data sent successfully
|
||||
cast::forget(move data);
|
||||
}
|
||||
task::yield();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Receive from a port. If no data is available on the port then the
|
||||
* task will block until data becomes available.
|
||||
*/
|
||||
pub fn recv<T: Owned>(p: Port<T>) -> T { recv_((**p).po) }
|
||||
|
||||
/// Returns true if there are messages available
|
||||
pub fn peek<T: Owned>(p: Port<T>) -> bool { peek_((**p).po) }
|
||||
|
||||
#[doc(hidden)]
|
||||
pub fn recv_chan<T: Owned>(ch: Chan<T>) -> T {
|
||||
as_raw_port(ch, |x|recv_(x))
|
||||
}
|
||||
|
||||
fn peek_chan<T: Owned>(ch: Chan<T>) -> bool {
|
||||
as_raw_port(ch, |x|peek_(x))
|
||||
}
|
||||
|
||||
/// Receive on a raw port pointer
|
||||
fn recv_<T>(p: *rust_port) -> T {
|
||||
unsafe {
|
||||
let yield = 0;
|
||||
let yieldp = ptr::addr_of(&yield);
|
||||
let mut res;
|
||||
res = rusti::init::<T>();
|
||||
rustrt::port_recv(ptr::addr_of(&res) as *uint, p, yieldp);
|
||||
|
||||
if yield != 0 {
|
||||
// Data isn't available yet, so res has not been initialized.
|
||||
task::yield();
|
||||
} else {
|
||||
// In the absence of compiler-generated preemption points
|
||||
// this is a good place to yield
|
||||
task::yield();
|
||||
}
|
||||
move res
|
||||
}
|
||||
}
|
||||
|
||||
fn peek_(p: *rust_port) -> bool {
|
||||
unsafe {
|
||||
// Yield here before we check to see if someone sent us a message
|
||||
// FIXME #524, if the compiler generates yields, we don't need this
|
||||
task::yield();
|
||||
rustrt::rust_port_size(p) != 0 as libc::size_t
|
||||
}
|
||||
}
|
||||
|
||||
/// Receive on one of two ports
|
||||
pub fn select2<A: Owned, B: Owned>(p_a: Port<A>, p_b: Port<B>)
|
||||
-> Either<A, B> {
|
||||
unsafe {
|
||||
let ports = ~[(**p_a).po, (**p_b).po];
|
||||
let yield = 0, yieldp = ptr::addr_of(&yield);
|
||||
|
||||
let mut resport: *rust_port;
|
||||
resport = rusti::init::<*rust_port>();
|
||||
do vec::as_imm_buf(ports) |ports, n_ports| {
|
||||
rustrt::rust_port_select(ptr::addr_of(&resport), ports,
|
||||
n_ports as size_t, yieldp);
|
||||
}
|
||||
|
||||
if yield != 0 {
|
||||
// Wait for data
|
||||
task::yield();
|
||||
} else {
|
||||
// As in recv, this is a good place to yield anyway until
|
||||
// the compiler generates yield calls
|
||||
task::yield();
|
||||
}
|
||||
|
||||
// Now we know the port we're supposed to receive from
|
||||
assert resport != ptr::null();
|
||||
|
||||
if resport == (**p_a).po {
|
||||
either::Left(recv(p_a))
|
||||
} else if resport == (**p_b).po {
|
||||
either::Right(recv(p_b))
|
||||
} else {
|
||||
die!(~"unexpected result from rust_port_select");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/* Implementation details */
|
||||
|
||||
#[allow(non_camel_case_types)] // runtime type
|
||||
enum rust_port {}
|
||||
|
||||
#[allow(non_camel_case_types)] // runtime type
|
||||
type port_id = int;
|
||||
|
||||
#[abi = "cdecl"]
|
||||
extern mod rustrt {
|
||||
unsafe fn rust_port_id_send(target_port: port_id, data: *())
|
||||
-> libc::uintptr_t;
|
||||
|
||||
unsafe fn new_port(unit_sz: libc::size_t) -> *rust_port;
|
||||
unsafe fn del_port(po: *rust_port);
|
||||
unsafe fn rust_port_begin_detach(po: *rust_port,
|
||||
yield: *libc::uintptr_t);
|
||||
unsafe fn rust_port_end_detach(po: *rust_port);
|
||||
unsafe fn get_port_id(po: *rust_port) -> port_id;
|
||||
unsafe fn rust_port_size(po: *rust_port) -> libc::size_t;
|
||||
unsafe fn port_recv(dptr: *uint, po: *rust_port,
|
||||
yield: *libc::uintptr_t);
|
||||
unsafe fn rust_port_select(dptr: **rust_port, ports: **rust_port,
|
||||
n_ports: libc::size_t,
|
||||
yield: *libc::uintptr_t);
|
||||
unsafe fn rust_port_take(port_id: port_id) -> *rust_port;
|
||||
unsafe fn rust_port_drop(p: *rust_port);
|
||||
unsafe fn rust_port_task(p: *rust_port) -> libc::uintptr_t;
|
||||
unsafe fn get_task_id() -> libc::uintptr_t;
|
||||
}
|
||||
|
||||
#[abi = "rust-intrinsic"]
|
||||
extern mod rusti {
|
||||
fn init<T>() -> T;
|
||||
}
|
||||
|
||||
|
||||
/* Tests */
|
||||
|
||||
|
||||
#[test]
|
||||
fn create_port_and_chan() { let p = Port::<int>(); Chan(&p); }
|
||||
|
||||
#[test]
|
||||
fn send_int() {
|
||||
let p = Port::<int>();
|
||||
let c = Chan(&p);
|
||||
send(c, 22);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_recv_fn() {
|
||||
let p = Port::<int>();
|
||||
let c = Chan::<int>(&p);
|
||||
send(c, 42);
|
||||
assert (recv(p) == 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_recv_fn_infer() {
|
||||
let p = Port();
|
||||
let c = Chan(&p);
|
||||
send(c, 42);
|
||||
assert (recv(p) == 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chan_chan_infer() {
|
||||
let p = Port(), p2 = Port::<int>();
|
||||
let c = Chan(&p);
|
||||
send(c, Chan(&p2));
|
||||
recv(p);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chan_chan() {
|
||||
let p = Port::<Chan<int>>(), p2 = Port::<int>();
|
||||
let c = Chan(&p);
|
||||
send(c, Chan(&p2));
|
||||
recv(p);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_peek() {
|
||||
let po = Port();
|
||||
let ch = Chan(&po);
|
||||
assert !peek(po);
|
||||
send(ch, ());
|
||||
assert peek(po);
|
||||
recv(po);
|
||||
assert !peek(po);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_select2_available() {
|
||||
let po_a = Port();
|
||||
let po_b = Port();
|
||||
let ch_a = Chan(&po_a);
|
||||
let ch_b = Chan(&po_b);
|
||||
|
||||
send(ch_a, ~"a");
|
||||
|
||||
assert select2(po_a, po_b) == either::Left(~"a");
|
||||
|
||||
send(ch_b, ~"b");
|
||||
|
||||
assert select2(po_a, po_b) == either::Right(~"b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_select2_rendezvous() {
|
||||
let po_a = Port();
|
||||
let po_b = Port();
|
||||
let ch_a = Chan(&po_a);
|
||||
let ch_b = Chan(&po_b);
|
||||
|
||||
for iter::repeat(10) {
|
||||
do task::spawn {
|
||||
for iter::repeat(10) { task::yield() }
|
||||
send(ch_a, ~"a");
|
||||
};
|
||||
|
||||
assert select2(po_a, po_b) == either::Left(~"a");
|
||||
|
||||
do task::spawn {
|
||||
for iter::repeat(10) { task::yield() }
|
||||
send(ch_b, ~"b");
|
||||
};
|
||||
|
||||
assert select2(po_a, po_b) == either::Right(~"b");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_select2_stress() {
|
||||
let po_a = Port();
|
||||
let po_b = Port();
|
||||
let ch_a = Chan(&po_a);
|
||||
let ch_b = Chan(&po_b);
|
||||
|
||||
let msgs = 100;
|
||||
let times = 4u;
|
||||
|
||||
for iter::repeat(times) {
|
||||
do task::spawn {
|
||||
for iter::repeat(msgs) {
|
||||
send(ch_a, ~"a")
|
||||
}
|
||||
};
|
||||
do task::spawn {
|
||||
for iter::repeat(msgs) {
|
||||
send(ch_b, ~"b")
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
let mut as_ = 0;
|
||||
let mut bs = 0;
|
||||
for iter::repeat(msgs * times * 2u) {
|
||||
match select2(po_a, po_b) {
|
||||
either::Left(~"a") => as_ += 1,
|
||||
either::Right(~"b") => bs += 1,
|
||||
_ => die!(~"test_select_2_stress failed")
|
||||
}
|
||||
}
|
||||
|
||||
assert as_ == 400;
|
||||
assert bs == 400;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_recv_chan() {
|
||||
let po = Port();
|
||||
let ch = Chan(&po);
|
||||
send(ch, ~"flower");
|
||||
assert recv_chan(ch) == ~"flower";
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_fail]
|
||||
#[ignore(cfg(windows))]
|
||||
fn test_recv_chan_dead() {
|
||||
let ch = Chan(&Port());
|
||||
send(ch, ~"flower");
|
||||
recv_chan(ch);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore(cfg(windows))]
|
||||
fn test_recv_chan_wrong_task() {
|
||||
let po = Port();
|
||||
let ch = Chan(&po);
|
||||
send(ch, ~"flower");
|
||||
assert result::is_err(&task::try(||
|
||||
recv_chan(ch)
|
||||
))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_port_send() {
|
||||
let po = Port();
|
||||
po.send(());
|
||||
po.recv();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_chan_peek() {
|
||||
let po = Port();
|
||||
let ch = po.chan();
|
||||
ch.send(());
|
||||
assert ch.peek();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_listen() {
|
||||
do listen |parent| {
|
||||
do task::spawn {
|
||||
parent.send(~"oatmeal-salad");
|
||||
}
|
||||
assert parent.recv() == ~"oatmeal-salad";
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore(cfg(windows))]
|
||||
fn test_port_detach_fail() {
|
||||
for iter::repeat(100) {
|
||||
do task::spawn_unlinked {
|
||||
let po = Port();
|
||||
let ch = po.chan();
|
||||
|
||||
do task::spawn {
|
||||
die!();
|
||||
}
|
||||
|
||||
do task::spawn {
|
||||
ch.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -69,7 +69,6 @@ pub use io;
|
|||
pub use iter;
|
||||
pub use libc;
|
||||
pub use num;
|
||||
pub use oldcomm;
|
||||
pub use ops;
|
||||
pub use option;
|
||||
pub use os;
|
||||
|
|
|
|||
|
|
@ -75,7 +75,6 @@
|
|||
|
||||
use cast;
|
||||
use container::Map;
|
||||
use oldcomm;
|
||||
use option;
|
||||
use pipes::{Chan, GenericChan, GenericPort, Port, stream};
|
||||
use pipes;
|
||||
|
|
|
|||
Loading…
Reference in New Issue