future.rs

/*!
 * A type representing values that may be computed concurrently and
 * operations for working with them.
 *
 * # Example
 *
 * ~~~
 * let delayed_fib = future::spawn {|| fib(5000) };
 * make_a_sandwich();
 * io::println(#fmt("fib(5000) = %?", delayed_fib.get()))
 * ~~~
 */

import either::either;

export future;
export extensions;
export from_value;
export from_port;
export from_fn;
export get;
export with;
export spawn;

// for task.rs
export future_pipe;

#[doc = "The future type"]
enum future<A> = {
    mut v: either<@A, fn@() -> A>
};

/// Methods on the `future` type
impl extensions<A:copy send> for future<A> {

    fn get() -> A {
        //! Get the value of the future

        get(self)
    }

    fn with<B>(blk: fn(A) -> B) -> B {
        //! Work with the value without copying it

        with(self, blk)
    }
}

fn from_value<A>(+val: A) -> future<A> {
    /*!
     * Create a future from a value
     *
     * The value is immediately available and calling `get` later will
     * not block.
     */

    future({
        mut v: either::left(@val)
    })
}

fn macros() {
    #macro[
        [#recv[chan],
         chan.recv()(chan)]
    ];
    #macro[
        [#move[x],
         unsafe { let y <- *ptr::addr_of(x); y }]
    ];
}

fn from_port<A:send>(-port: future_pipe::client::waiting<A>) -> future<A> {
    #[doc = "
    Create a future from a port

    The first time that the value is requested the task will block
    waiting for the result to be received on the port.
    "];
    import future_pipe::client::recv;

    let port = ~mut some(port);
    do from_fn |move port| {
        let mut port_ = none;
        port_ <-> *port;
        let port = option::unwrap(port_);
        alt (#recv(port)) {
          future_pipe::completed(data, _next) { #move(data) }
        }
    }
}

fn from_fn<A>(f: fn@() -> A) -> future<A> {
    /*!
     * Create a future from a function.
     *
     * The first time that the value is requested it will be retreived by
     * calling the function.  Note that this function is a local
     * function. It is not spawned into another task.
     */

    future({
        mut v: either::right(f)
    })
}

fn spawn<A:send>(+blk: fn~() -> A) -> future<A> {
    /*!
     * Create a future from a unique closure.
     *
     * The closure will be run in a new task and its result used as the
     * value of the future.
     */

    from_port(pipes::spawn_service_recv(future_pipe::init, |ch| {
        future_pipe::server::completed(ch, blk());
    }))
}

fn get<A:copy>(future: future<A>) -> A {
    //! Get the value of the future

    do with(future) |v| { v }
}

fn with<A,B>(future: future<A>, blk: fn(A) -> B) -> B {
    //! Work with the value without copying it

    let v = alt copy future.v {
      either::left(v) { v }
      either::right(f) {
        let v = @f();
        future.v = either::left(v);
        v
      }
    };
    blk(*v)
}

// The pipe protocol, generated by pipec
/*
proto! future_pipe {
    waiting:recv<T:send> {
        completed(T) -> terminated
    }

    terminated { }
}
*/
mod future_pipe {
    fn init<T: send>() -> (client::waiting<T>, server::waiting<T>) {
        { let (s, c) = pipes::entangle(); (c, s) }
    }
    enum waiting<T: send> { completed(T, client::terminated), }
    enum terminated { }
    mod client {
        impl recv<T: send> for waiting<T> {
            fn recv() -> extern fn(+waiting<T>) -> future_pipe::waiting<T> {
                fn recv<T: send>(+pipe: waiting<T>) ->
                   future_pipe::waiting<T> {
                    option::unwrap(pipes::recv(pipe))
                }
                recv
            }
        }
        type waiting<T: send> = pipes::recv_packet<future_pipe::waiting<T>>;
        type terminated = pipes::send_packet<future_pipe::terminated>;
    }
    mod server {
        fn completed<T: send>(+pipe: waiting<T>, +x_0: T) -> terminated {
            {
                let (s, c) = pipes::entangle();
                let message = future_pipe::completed(x_0, s);
                pipes::send(pipe, message);
                c
            }
        }
        type waiting<T: send> = pipes::send_packet<future_pipe::waiting<T>>;
        impl recv for terminated {
            fn recv() -> extern fn(+terminated) -> future_pipe::terminated {
                fn recv(+pipe: terminated) -> future_pipe::terminated {
                    option::unwrap(pipes::recv(pipe))
                }
                recv
            }
        }
        type terminated = pipes::recv_packet<future_pipe::terminated>;
    }
}

#[test]
fn test_from_value() {
    let f = from_value("snail");
    assert get(f) == "snail";
}

#[test]
fn test_from_port() {
    let (po, ch) = future_pipe::init();
    future_pipe::server::completed(ch, "whale");
    let f = from_port(po);
    assert get(f) == "whale";
}

#[test]
fn test_from_fn() {
    let f = fn@() -> str { "brail" };
    let f = from_fn(f);
    assert get(f) == "brail";
}

#[test]
fn test_iface_get() {
    let f = from_value("fail");
    assert f.get() == "fail";
}

#[test]
fn test_with() {
    let f = from_value("nail");
    assert with(f, |v| v) == "nail";
}

#[test]
fn test_iface_with() {
    let f = from_value("kale");
    assert f.with(|v| v) == "kale";
}

#[test]
fn test_spawn() {
    let f = spawn(|| "bale");
    assert get(f) == "bale";
}

#[test]
#[should_fail]
#[ignore(cfg(target_os = "win32"))]
fn test_futurefail() {
    let f = spawn(|| fail);
    let _x: str = get(f);
}