diff --git a/src/libcore/iter-trait/dlist.rs b/src/libcore/iter-trait/dlist.rs
index 1cf7fb980cd2..a94ed6a54dc1 100644
--- a/src/libcore/iter-trait/dlist.rs
+++ b/src/libcore/iter-trait/dlist.rs
@@ -7,7 +7,7 @@ type IMPL_T<A> = dlist::dlist<A>;
  * e.g. breadth-first search with in-place enqueues), but removing the current
  * node is forbidden.
  */
-fn EACH<A>(self: IMPL_T<A>, f: fn(A) -> bool) {
+pure fn EACH<A>(self: IMPL_T<A>, f: fn(A) -> bool) {
     import dlist::extensions;
 
     let mut link = self.peek_n();
diff --git a/src/libcore/iter-trait/option.rs b/src/libcore/iter-trait/option.rs
index 530de27c614c..87763b273da6 100644
--- a/src/libcore/iter-trait/option.rs
+++ b/src/libcore/iter-trait/option.rs
@@ -1,6 +1,6 @@
 type IMPL_T<A> = option<A>;
 
-fn EACH<A>(self: IMPL_T<A>, f: fn(A) -> bool) {
+pure fn EACH<A>(self: IMPL_T<A>, f: fn(A) -> bool) {
     alt self {
       none { }
       some(a) { f(a); }
diff --git a/src/libcore/pipes.rs b/src/libcore/pipes.rs
index 44e54fb0fd4b..2e463df9e900 100644
--- a/src/libcore/pipes.rs
+++ b/src/libcore/pipes.rs
@@ -5,17 +5,22 @@ import either::{either, left, right};
 import option::unwrap;
 import arc::methods;
 
-/* Use this after the snapshot
+// Things used by code generated by the pipe compiler.
+export entangle;
+
+// User-level things
+export send_packet, recv_packet, send, recv, try_recv, peek;
+export select, select2, selecti, select2i, selectable;
+export spawn_service, spawn_service_recv;
+export stream, port, chan, shared_chan, port_set, channel;
+
 macro_rules! move {
     { $x:expr } => { unsafe { let y <- *ptr::addr_of($x); y } }
 }
-*/
 
-fn macros() {
-    #macro[
-        [#move(x), { unsafe { let y <- *ptr::addr_of(x); y } }]
-    ];
-}
+// 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<T>(-x: T) -> T { x }
 
 enum state {
     empty,
@@ -465,6 +470,19 @@ proto! streamp {
     }
 }
 
+// It'd be nice to call this send, but it'd conflict with the built in
+// send kind.
+trait channel<T: send> {
+    fn send(+x: T);
+}
+
+trait recv<T: send> {
+    fn recv() -> T;
+    fn try_recv() -> option<T>;
+    // This should perhaps be a new trait
+    pure fn peek() -> bool;
+}
+
 type chan_<T:send> = { mut endp: option<streamp::client::open<T>> };
 
 enum chan<T:send> {
@@ -483,7 +501,7 @@ fn stream<T:send>() -> (chan<T>, port<T>) {
     (chan_({ mut endp: some(c) }), port_({ mut endp: some(s) }))
 }
 
-impl chan<T: send> for chan<T> {
+impl chan<T: send> of channel<T> for chan<T> {
     fn send(+x: T) {
         let mut endp = none;
         endp <-> self.endp;
@@ -492,7 +510,7 @@ impl chan<T: send> for chan<T> {
     }
 }
 
-impl port<T: send> for port<T> {
+impl port<T: send> of recv<T> for port<T> {
     fn recv() -> T {
         let mut endp = none;
         endp <-> self.endp;
@@ -504,10 +522,10 @@ impl port<T: send> for port<T> {
     fn try_recv() -> option<T> {
         let mut endp = none;
         endp <-> self.endp;
-        alt pipes::try_recv(unwrap(endp)) {
+        alt move(pipes::try_recv(unwrap(endp))) {
           some(streamp::data(x, endp)) {
-            self.endp = some(#move(endp));
-            some(#move(x))
+            self.endp = some(move!{endp});
+            some(move!{x})
           }
           none { none }
         }
@@ -528,7 +546,7 @@ impl port<T: send> for port<T> {
 }
 
 // Treat a whole bunch of ports as one.
-class port_set<T: send> {
+class port_set<T: send> : recv<T> {
     let mut ports: ~[pipes::port<T>];
 
     new() { self.ports = ~[]; }
@@ -540,12 +558,12 @@ class port_set<T: send> {
     fn try_recv() -> option<T> {
         let mut result = none;
         while result == none && self.ports.len() > 0 {
-            let i = pipes::wait_many(self.ports.map(|p| p.header()));
+            let i = wait_many(self.ports.map(|p| p.header()));
             // dereferencing an unsafe pointer nonsense to appease the
             // borrowchecker.
-            alt unsafe {(*ptr::addr_of(self.ports[i])).try_recv()} {
+            alt move(unsafe {(*ptr::addr_of(self.ports[i])).try_recv()}) {
               some(m) {
-                result = some(#move(m));
+                  result = some(move!{m});
               }
               none {
                 // Remove this port.
@@ -562,10 +580,19 @@ class port_set<T: send> {
     fn recv() -> T {
         option::unwrap(self.try_recv())
     }
+
+    pure fn peek() -> bool {
+        // It'd be nice to use self.port.each, but that version isn't
+        // pure.
+        for vec::each(self.ports) |p| {
+            if p.peek() { ret true }
+        }
+        false
+    }
 }
 
-impl<T: send> of selectable for pipes::port<T> {
-    pure fn header() -> *pipes::packet_header unchecked {
+impl<T: send> of selectable for port<T> {
+    pure fn header() -> *packet_header unchecked {
         alt self.endp {
           some(endp) {
             endp.header()
@@ -576,13 +603,9 @@ impl<T: send> of selectable for pipes::port<T> {
 }
 
 
-type shared_chan<T: send> = arc::exclusive<pipes::chan<T>>;
+type shared_chan<T: send> = arc::exclusive<chan<T>>;
 
-trait send_on_shared_chan<T> {
-    fn send(+x: T);
-}
-
-impl chan<T: send> of send_on_shared_chan<T> for shared_chan<T> {
+impl chan<T: send> of channel<T> for shared_chan<T> {
     fn send(+x: T) {
         let mut xx = some(x);
         do self.with |_c, chan| {
@@ -593,6 +616,6 @@ impl chan<T: send> of send_on_shared_chan<T> for shared_chan<T> {
     }
 }
 
-fn shared_chan<T:send>(+c: pipes::chan<T>) -> shared_chan<T> {
+fn shared_chan<T:send>(+c: chan<T>) -> shared_chan<T> {
     arc::exclusive(c)
 }