---

yaml
---
r: 36091
b: refs/heads/try2
c: 61bb357
h: refs/heads/master
i:
  36089: 8f3a73f
  36087: bb3f606
v: v3

Author: pcwalton

[refs]

@@ -5,7 +5,7 @@ refs/heads/snap-stage3: eb8fd119c65c67f3b1b8268cc7341c22d39b7b61
 refs/heads/try: d324a424d8f84b1eb049b12cf34182bda91b0024
 refs/tags/release-0.1: 1f5c5126e96c79d22cb7862f75304136e204f105
 refs/heads/ndm: f3868061cd7988080c30d6d5bf352a5a5fe2460b
-refs/heads/try2: d29962f0eb2772c208a299efae86ad16bc25f7de
+refs/heads/try2: 61bb3571a59f4659a0a46565c71fa7ecfa352811
 refs/heads/incoming: d9317a174e434d4c99fc1a37fd7dc0d2f5328d37
 refs/heads/dist-snap: 22efa39382d41b084fde1719df7ae8ce5697d8c9
 refs/tags/release-0.2: c870d2dffb391e14efb05aa27898f1f6333a9596

branches/try2/src/libcore/core.rc

@@ -181,6 +181,7 @@ pub mod task {
     pub mod spawn;
     pub mod rt;
 }
+pub mod future;
 pub mod pipes;
 
 // Runtime and language-primitive support

branches/try2/src/libstd/future.rs renamed to branches/try2/src/libcore/future.rs

@@ -17,7 +17,7 @@
  */
 
 use either::Either;
-use pipes::{recv, oneshot, ChanOne, PortOne, send_one, recv_one};
+use pipes::recv;
 use cast::copy_lifetime;
 
 #[doc = "The future type"]
@@ -67,7 +67,7 @@ pub fn from_value<A>(val: A) -> Future<A> {
     Future {state: Forced(~(move val))}
 }
 
-pub fn from_port<A:Send>(port: PortOne<A>) ->
+pub fn from_port<A:Send>(port: future_pipe::client::waiting<A>) ->
         Future<A> {
     /*!
      * Create a future from a port
@@ -82,7 +82,7 @@ pub fn from_port<A:Send>(port: PortOne<A>) ->
         port_ <-> *port;
         let port = option::unwrap(move port_);
         match recv(move port) {
-            oneshot::send(move data) => move data
+            future_pipe::completed(move data) => move data
         }
     }
 }
@@ -107,15 +107,9 @@ pub fn spawn<A:Send>(blk: fn~() -> A) -> Future<A> {
      * value of the future.
      */
 
-    let (chan, port) = oneshot::init();
-
-    let chan = ~mut Some(move chan);
-    do task::spawn |move blk, move chan| {
-        let chan = option::swap_unwrap(&mut *chan);
-        send_one(move chan, blk());
-    }
-
-    return from_port(move port);
+    from_port(pipes::spawn_service_recv(future_pipe::init, |move blk, ch| {
+        future_pipe::server::completed(move ch, blk());
+    }))
 }
 
 pub fn get_ref<A>(future: &r/Future<A>) -> &r/A {
@@ -168,6 +162,12 @@ pub fn with<A,B>(future: &Future<A>, blk: fn((&A)) -> B) -> B {
     blk(get_ref(future))
 }
 
+proto! future_pipe (
+    waiting:recv<T:Send> {
+        completed(T) -> !
+    }
+)
+
 #[allow(non_implicitly_copyable_typarams)]
 pub mod test {
     #[test]
@@ -178,8 +178,8 @@ pub mod test {
 
     #[test]
     pub fn test_from_port() {
-        let (ch, po) = oneshot::init();
-        send_one(move ch, ~"whale");
+        let (po, ch) = future_pipe::init();
+        future_pipe::server::completed(move ch, ~"whale");
         let f = from_port(move po);
         assert get(&f) == ~"whale";
     }

branches/try2/src/libcore/private.rs

@@ -581,20 +581,16 @@ pub mod tests {
 
         for uint::range(0, num_tasks) |_i| {
             let total = total.clone();
-            let (chan, port) = pipes::stream();
-            futures.push(move port);
-
-            do task::spawn |move total, move chan| {
+            futures.push(future::spawn(|move total| {
                 for uint::range(0, count) |_i| {
                     do total.with |count| {
                         **count += 1;
                     }
                 }
-                chan.send(());
-            }
+            }));
         };
 
-        for futures.each |f| { f.recv() }
+        for futures.each |f| { f.get() }
 
         do total.with |total| {
             assert **total == num_tasks * count
@@ -646,7 +642,7 @@ pub mod tests {
         // Have to get rid of our reference before blocking.
         { let _x = move x; } // FIXME(#3161) util::ignore doesn't work here
         let res = option::swap_unwrap(&mut res);
-        res.recv();
+        future::get(&res);
     }
 
     #[test] #[should_fail] #[ignore(cfg(windows))]
@@ -661,7 +657,7 @@ pub mod tests {
         }
         assert unwrap_exclusive(move x) == ~~"hello";
         let res = option::swap_unwrap(&mut res);
-        res.recv();
+        future::get(&res);
     }
 
     #[test] #[ignore(cfg(windows))]

branches/try2/src/libcore/task.rs

@@ -32,7 +32,6 @@ use cmp::Eq;
 use result::Result;
 use pipes::{stream, Chan, Port};
 use local_data_priv::{local_get, local_set};
-use util::replace;
 
 use rt::task_id;
 use rt::rust_task;
@@ -73,6 +72,25 @@ impl TaskResult : Eq {
     pure fn ne(other: &TaskResult) -> bool { !self.eq(other) }
 }
 
+/// A message type for notifying of task lifecycle events
+pub enum Notification {
+    /// Sent when a task exits with the task handle and result
+    Exit(Task, TaskResult)
+}
+
+impl Notification : cmp::Eq {
+    pure fn eq(other: &Notification) -> bool {
+        match self {
+            Exit(e0a, e1a) => {
+                match (*other) {
+                    Exit(e0b, e1b) => e0a == e0b && e1a == e1b
+                }
+            }
+        }
+    }
+    pure fn ne(other: &Notification) -> bool { !self.eq(other) }
+}
+
 /// Scheduler modes
 pub enum SchedMode {
     /// All tasks run in the same OS thread
@@ -182,7 +200,7 @@ pub type SchedOpts = {
 pub type TaskOpts = {
     linked: bool,
     supervised: bool,
-    mut notify_chan: Option<Chan<TaskResult>>,
+    mut notify_chan: Option<Chan<Notification>>,
     sched: Option<SchedOpts>,
 };
 
@@ -228,7 +246,11 @@ priv impl TaskBuilder {
             fail ~"Cannot copy a task_builder"; // Fake move mode on self
         }
         self.consumed = true;
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            Some(option::swap_unwrap(&mut self.opts.notify_chan))
+        };
         TaskBuilder({
             opts: {
                 linked: self.opts.linked,
@@ -249,7 +271,11 @@ impl TaskBuilder {
      * the other will not be killed.
      */
     fn unlinked() -> TaskBuilder {
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            Some(option::swap_unwrap(&mut self.opts.notify_chan))
+        };
         TaskBuilder({
             opts: {
                 linked: false,
@@ -267,7 +293,11 @@ impl TaskBuilder {
      * the child.
      */
     fn supervised() -> TaskBuilder {
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            Some(option::swap_unwrap(&mut self.opts.notify_chan))
+        };
         TaskBuilder({
             opts: {
                 linked: false,
@@ -284,7 +314,11 @@ impl TaskBuilder {
      * other will be killed.
      */
     fn linked() -> TaskBuilder {
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            Some(option::swap_unwrap(&mut self.opts.notify_chan))
+        };
         TaskBuilder({
             opts: {
                 linked: true,
@@ -314,7 +348,7 @@ impl TaskBuilder {
      * # Failure
      * Fails if a future_result was already set for this task.
      */
-    fn future_result(blk: fn(v: Port<TaskResult>)) -> TaskBuilder {
+    fn future_result(blk: fn(v: future::Future<TaskResult>)) -> TaskBuilder {
         // FIXME (#3725): Once linked failure and notification are
         // handled in the library, I can imagine implementing this by just
         // registering an arbitrary number of task::on_exit handlers and
@@ -325,9 +359,13 @@ impl TaskBuilder {
         }
 
         // Construct the future and give it to the caller.
-        let (notify_pipe_ch, notify_pipe_po) = stream::<TaskResult>();
+        let (notify_pipe_ch, notify_pipe_po) = stream::<Notification>();
 
-        blk(move notify_pipe_po);
+        blk(do future::from_fn |move notify_pipe_po| {
+            match notify_pipe_po.recv() {
+              Exit(_, result) => result
+            }
+        });
 
         // Reconfigure self to use a notify channel.
         TaskBuilder({
@@ -343,7 +381,11 @@ impl TaskBuilder {
     }
     /// Configure a custom scheduler mode for the task.
     fn sched_mode(mode: SchedMode) -> TaskBuilder {
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            Some(option::swap_unwrap(&mut self.opts.notify_chan))
+        };
         TaskBuilder({
             opts: {
                 linked: self.opts.linked,
@@ -370,7 +412,11 @@ impl TaskBuilder {
      */
     fn add_wrapper(wrapper: fn@(v: fn~()) -> fn~()) -> TaskBuilder {
         let prev_gen_body = self.gen_body;
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            Some(option::swap_unwrap(&mut self.opts.notify_chan))
+        };
         TaskBuilder({
             opts: {
                 linked: self.opts.linked,
@@ -401,7 +447,13 @@ impl TaskBuilder {
      * must be greater than zero.
      */
     fn spawn(f: fn~()) {
-        let notify_chan = replace(&mut self.opts.notify_chan, None);
+        let notify_chan = if self.opts.notify_chan.is_none() {
+            None
+        } else {
+            let swapped_notify_chan =
+                option::swap_unwrap(&mut self.opts.notify_chan);
+            Some(move swapped_notify_chan)
+        };
         let x = self.consume();
         let opts = {
             linked: x.opts.linked,
@@ -480,7 +532,7 @@ impl TaskBuilder {
         do fr_task_builder.spawn |move f| {
             comm::send(ch, f());
         }
-        match option::unwrap(move result).recv() {
+        match future::get(&option::unwrap(move result)) {
             Success => result::Ok(comm::recv(po)),
             Failure => result::Err(())
         }
@@ -897,14 +949,14 @@ fn test_add_wrapper() {
 fn test_future_result() {
     let mut result = None;
     do task().future_result(|+r| { result = Some(move r); }).spawn { }
-    assert option::unwrap(move result).recv() == Success;
+    assert future::get(&option::unwrap(move result)) == Success;
 
     result = None;
     do task().future_result(|+r|
         { result = Some(move r); }).unlinked().spawn {
         fail;
     }
-    assert option::unwrap(move result).recv() == Failure;
+    assert future::get(&option::unwrap(move result)) == Failure;
 }
 
 #[test] #[should_fail] #[ignore(cfg(windows))]

branches/try2/src/libcore/task/spawn.rs

@@ -320,15 +320,15 @@ fn TCB(me: *rust_task, tasks: TaskGroupArc, ancestors: AncestorList,
 }
 
 struct AutoNotify {
-    notify_chan: Chan<TaskResult>,
+    notify_chan: Chan<Notification>,
     mut failed:  bool,
     drop {
         let result = if self.failed { Failure } else { Success };
-        self.notify_chan.send(result);
+        self.notify_chan.send(Exit(get_task(), result));
     }
 }
 
-fn AutoNotify(chan: Chan<TaskResult>) -> AutoNotify {
+fn AutoNotify(chan: Chan<Notification>) -> AutoNotify {
     AutoNotify {
         notify_chan: move chan,
         failed: true // Un-set above when taskgroup successfully made.
@@ -532,7 +532,7 @@ pub fn spawn_raw(opts: TaskOpts, f: fn~()) {
     // (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>>,
+                          notify_chan: Option<Chan<Notification>>,
                           f: fn~()) -> fn~() {
         let child_data = ~mut Some((move child_arc, move ancestors));
         return fn~(move notify_chan, move child_data, move f) {
@@ -660,30 +660,36 @@ fn test_spawn_raw_unsupervise() {
 #[test]
 #[ignore(cfg(windows))]
 fn test_spawn_raw_notify_success() {
+    let (task_ch, task_po) = pipes::stream();
     let (notify_ch, notify_po) = pipes::stream();
 
     let opts = {
         notify_chan: Some(move notify_ch),
         .. default_task_opts()
     };
-    do spawn_raw(move opts) {
+    do spawn_raw(move opts) |move task_ch| {
+        task_ch.send(get_task());
     }
-    assert notify_po.recv() == Success;
+    let task_ = task_po.recv();
+    assert notify_po.recv() == Exit(task_, Success);
 }
 
 #[test]
 #[ignore(cfg(windows))]
 fn test_spawn_raw_notify_failure() {
     // New bindings for these
+    let (task_ch, task_po) = pipes::stream();
     let (notify_ch, notify_po) = pipes::stream();
 
     let opts = {
         linked: false,
         notify_chan: Some(move notify_ch),
         .. default_task_opts()
     };
-    do spawn_raw(move opts) {
+    do spawn_raw(move opts) |move task_ch| {
+        task_ch.send(get_task());
         fail;
     }
-    assert notify_po.recv() == Failure;
+    let task_ = task_po.recv();
+    assert notify_po.recv() == Exit(task_, Failure);
 }

branches/try2/src/libstd/arc.rs

@@ -651,7 +651,7 @@ mod tests {
         }
 
         // Wait for children to pass their asserts
-        for vec::each(children) |r| { r.recv(); }
+        for vec::each(children) |r| { future::get(r); }
 
         // Wait for writer to finish
         p.recv();