---

yaml
---
r: 20790
b: refs/heads/snap-stage3
c: 7192761
h: refs/heads/master
v: v3

Author: erickt

[refs]

@@ -1,7 +1,7 @@
 ---
 refs/heads/master: e430a699f2c60890d9b86069fd0c68a70ece7120
 refs/heads/snap-stage1: e33de59e47c5076a89eadeb38f4934f58a3618a6
-refs/heads/snap-stage3: e7a69fbe4e33602dd6f78d140aafa5125b8eec71
+refs/heads/snap-stage3: 71927615e0a2cd2d9f96038ba0f403a4d833a2f5
 refs/heads/try: ffbe0e0e00374358b789b0037bcb3a577cd218be
 refs/tags/release-0.1: 1f5c5126e96c79d22cb7862f75304136e204f105
 refs/heads/ndm: f3868061cd7988080c30d6d5bf352a5a5fe2460b

branches/snap-stage3/src/libcore/pipes.rs

@@ -1,47 +1,4 @@
-/*! 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.
-
-~~~
-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. There is 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.
-
-FIXME (#3072) - This is still incomplete
-
-*/
+// Runtime support for pipes.
 
 import unsafe::{forget, reinterpret_cast, transmute};
 import either::{either, left, right};
@@ -54,7 +11,7 @@ export send_packet_buffered, recv_packet_buffered;
 export packet, mk_packet, entangle_buffer, has_buffer, buffer_header;
 
 // export these so we can find them in the buffer_resource
-// destructor. This is probably a symptom of #3005.
+// destructor. This is probably another metadata bug.
 export atomic_add_acq, atomic_sub_rel;
 
 // User-level things
@@ -63,22 +20,24 @@ export select, select2, selecti, select2i, selectable;
 export spawn_service, spawn_service_recv;
 export stream, port, chan, shared_chan, port_set, channel;
 
-#[doc(hidden)]
 const SPIN_COUNT: uint = 0;
 
 macro_rules! move_it {
     { $x:expr } => { unsafe { let y <- *ptr::addr_of($x); y } }
 }
 
-#[doc(hidden)]
+// This is to help make sure we only move out of enums in safe
+// places. Once there is unary move, it can be removed.
+fn move_it<T>(-x: T) -> T { x }
+
 enum state {
     empty,
     full,
     blocked,
     terminated
 }
 
-struct buffer_header {
+class buffer_header {
     // Tracks whether this buffer needs to be freed. We can probably
     // get away with restricting it to 0 or 1, if we're careful.
     let mut ref_count: int;
@@ -90,13 +49,12 @@ struct buffer_header {
 }
 
 // This is for protocols to associate extra data to thread around.
-#[doc(hidden)]
 type buffer<T: send> = {
     header: buffer_header,
     data: T,
 };
 
-struct packet_header {
+class packet_header {
     let mut state: state;
     let mut blocked_task: option<*rust_task>;
 
@@ -137,7 +95,6 @@ struct packet_header {
     }
 }
 
-#[doc(hidden)]
 type packet<T: send> = {
     header: packet_header,
     mut payload: option<T>,
@@ -176,7 +133,6 @@ fn unibuffer<T: send>() -> ~buffer<packet<T>> {
     b
 }
 
-#[doc(hidden)]
 fn packet<T: send>() -> *packet<T> {
     let b = unibuffer();
     let p = ptr::addr_of(b.data);
@@ -185,7 +141,6 @@ fn packet<T: send>() -> *packet<T> {
     p
 }
 
-#[doc(hidden)]
 fn entangle_buffer<T: send, Tstart: send>(
     -buffer: ~buffer<T>,
     init: fn(*libc::c_void, x: &T) -> *packet<Tstart>)
@@ -208,22 +163,18 @@ extern mod rusti {
 
 // If I call the rusti versions directly from a polymorphic function,
 // I get link errors. This is a bug that needs investigated more.
-#[doc(hidden)]
 fn atomic_xchng_rel(&dst: int, src: int) -> int {
     rusti::atomic_xchng_rel(dst, src)
 }
 
-#[doc(hidden)]
 fn atomic_add_acq(&dst: int, src: int) -> int {
     rusti::atomic_add_acq(dst, src)
 }
 
-#[doc(hidden)]
 fn atomic_sub_rel(&dst: int, src: int) -> int {
     rusti::atomic_sub_rel(dst, src)
 }
 
-#[doc(hidden)]
 type rust_task = libc::c_void;
 
 extern mod rustrt {
@@ -237,7 +188,6 @@ extern mod rustrt {
     pure fn task_signal_event(target: *rust_task, event: *libc::c_void);
 }
 
-#[doc(hidden)]
 fn wait_event(this: *rust_task) -> *libc::c_void {
     let mut event = ptr::null();
 
@@ -248,7 +198,6 @@ fn wait_event(this: *rust_task) -> *libc::c_void {
     event
 }
 
-#[doc(hidden)]
 fn swap_state_acq(&dst: state, src: state) -> state {
     unsafe {
         reinterpret_cast(rusti::atomic_xchng_acq(
@@ -257,7 +206,6 @@ fn swap_state_acq(&dst: state, src: state) -> state {
     }
 }
 
-#[doc(hidden)]
 fn swap_state_rel(&dst: state, src: state) -> state {
     unsafe {
         reinterpret_cast(rusti::atomic_xchng_rel(
@@ -266,12 +214,11 @@ fn swap_state_rel(&dst: state, src: state) -> state {
     }
 }
 
-#[doc(hidden)]
 unsafe fn get_buffer<T: send>(p: *packet_header) -> ~buffer<T> {
     transmute((*p).buf_header())
 }
 
-struct buffer_resource<T: send> {
+class buffer_resource<T: send> {
     let buffer: ~buffer<T>;
     new(+b: ~buffer<T>) {
         //let p = ptr::addr_of(*b);
@@ -297,7 +244,6 @@ struct buffer_resource<T: send> {
     }
 }
 
-#[doc(hidden)]
 fn send<T: send, Tbuffer: send>(-p: send_packet_buffered<T, Tbuffer>,
                                 -payload: T) {
     let header = p.header();
@@ -335,21 +281,10 @@ fn send<T: send, Tbuffer: send>(-p: send_packet_buffered<T, Tbuffer>,
     }
 }
 
-/** Receives a message from a pipe.
-
-Fails if the sender closes the connection.
-
-*/
 fn recv<T: send, Tbuffer: send>(-p: recv_packet_buffered<T, Tbuffer>) -> T {
     option::unwrap(try_recv(p))
 }
 
-/** 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.
-
-*/
 fn try_recv<T: send, Tbuffer: send>(-p: recv_packet_buffered<T, Tbuffer>)
     -> option<T>
 {
@@ -416,7 +351,6 @@ impl peek<T: send, Tb: send> for recv_packet_buffered<T, Tb> {
     }
 }
 
-#[doc(hidden)]
 fn sender_terminate<T: send>(p: *packet<T>) {
     let p = unsafe { &*p };
     alt swap_state_rel(p.header.state, terminated) {
@@ -443,7 +377,6 @@ fn sender_terminate<T: send>(p: *packet<T>) {
     }
 }
 
-#[doc(hidden)]
 fn receiver_terminate<T: send>(p: *packet<T>) {
     let p = unsafe { &*p };
     alt swap_state_rel(p.header.state, terminated) {
@@ -461,16 +394,8 @@ fn receiver_terminate<T: send>(p: *packet<T>) {
     }
 }
 
-/** 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.
-
-*/
+#[doc = "Returns when one of the packet headers reports data is
+available."]
 fn wait_many(pkts: &[*packet_header]) -> uint {
     let this = rustrt::rust_get_task();
 
@@ -522,34 +447,6 @@ fn wait_many(pkts: &[*packet_header]) -> uint {
     ready_packet
 }
 
-/** 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`.
-
-~~~
-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`.
-
-*/
 fn select2<A: send, Ab: send, B: send, Bb: send>(
     +a: recv_packet_buffered<A, Ab>,
     +b: recv_packet_buffered<B, Bb>)
@@ -603,16 +500,13 @@ fn select<T: send, Tb: send>(+endpoints: ~[recv_packet_buffered<T, Tb>])
     (ready, result, remaining)
 }
 
-/// The sending end of a pipe. It can be used to send exactly one
-/// message.
 type send_packet<T: send> = send_packet_buffered<T, packet<T>>;
 
-#[doc(hidden)]
 fn send_packet<T: send>(p: *packet<T>) -> send_packet<T> {
     send_packet_buffered(p)
 }
 
-struct send_packet_buffered<T: send, Tbuffer: send> {
+class send_packet_buffered<T: send, Tbuffer: send> {
     let mut p: option<*packet<T>>;
     let mut buffer: option<buffer_resource<Tbuffer>>;
     new(p: *packet<T>) {
@@ -666,16 +560,13 @@ struct send_packet_buffered<T: send, Tbuffer: send> {
     }
 }
 
-/// Represents the receive end of a pipe. It can receive exactly one
-/// message.
 type recv_packet<T: send> = recv_packet_buffered<T, packet<T>>;
 
-#[doc(hidden)]
 fn recv_packet<T: send>(p: *packet<T>) -> recv_packet<T> {
     recv_packet_buffered(p)
 }
 
-struct recv_packet_buffered<T: send, Tbuffer: send> : selectable {
+class recv_packet_buffered<T: send, Tbuffer: send> : selectable {
     let mut p: option<*packet<T>>;
     let mut buffer: option<buffer_resource<Tbuffer>>;
     new(p: *packet<T>) {
@@ -729,7 +620,6 @@ struct recv_packet_buffered<T: send, Tbuffer: send> : selectable {
     }
 }
 
-#[doc(hidden)]
 fn entangle<T: send>() -> (send_packet<T>, recv_packet<T>) {
     let p = packet();
     (send_packet(p), recv_packet(p))
@@ -796,14 +686,12 @@ trait recv<T: send> {
     pure fn peek() -> bool;
 }
 
-#[doc(hidden)]
 type chan_<T:send> = { mut endp: option<streamp::client::open<T>> };
 
 enum chan<T:send> {
     chan_(chan_<T>)
 }
 
-#[doc(hidden)]
 type port_<T:send> = { mut endp: option<streamp::server::open<T>> };
 
 enum port<T:send> {
@@ -837,7 +725,7 @@ impl port<T: send> of recv<T> for port<T> {
     fn try_recv() -> option<T> {
         let mut endp = none;
         endp <-> self.endp;
-        alt move pipes::try_recv(unwrap(endp)) {
+        alt move_it(pipes::try_recv(unwrap(endp))) {
           some(streamp::data(x, endp)) {
             self.endp = some(move_it!{endp});
             some(move_it!{x})
@@ -861,7 +749,7 @@ impl port<T: send> of recv<T> for port<T> {
 }
 
 // Treat a whole bunch of ports as one.
-struct port_set<T: send> : recv<T> {
+class port_set<T: send> : recv<T> {
     let mut ports: ~[pipes::port<T>];
 
     new() { self.ports = ~[]; }
@@ -882,7 +770,7 @@ struct port_set<T: send> : recv<T> {
             let i = wait_many(self.ports.map(|p| p.header()));
             // dereferencing an unsafe pointer nonsense to appease the
             // borrowchecker.
-            alt move unsafe {(*ptr::addr_of(self.ports[i])).try_recv()} {
+            alt move_it(unsafe {(*ptr::addr_of(self.ports[i])).try_recv()}) {
               some(m) {
                   result = some(move_it!{m});
               }