rewrite to use old C++-based mechanism
This commit is contained in:
@@ -28,8 +28,6 @@ Example:
|
|||||||
*/
|
*/
|
||||||
import cast = unsafe::reinterpret_cast;
|
import cast = unsafe::reinterpret_cast;
|
||||||
import comm;
|
import comm;
|
||||||
import option::{some, none};
|
|
||||||
import option = option::t;
|
|
||||||
import ptr;
|
import ptr;
|
||||||
import c = ctypes;
|
import c = ctypes;
|
||||||
|
|
||||||
@@ -112,10 +110,23 @@ Returns:
|
|||||||
|
|
||||||
A handle to the new task
|
A handle to the new task
|
||||||
*/
|
*/
|
||||||
fn spawn(-f: sendfn()) -> task unsafe {
|
fn spawn(-f: sendfn()) -> task {
|
||||||
|
spawn_inner(f, none)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn spawn_inner(-f: sendfn(),
|
||||||
|
notify: option<comm::chan<task_notification>>) -> task unsafe {
|
||||||
let closure: *rust_closure = unsafe::reinterpret_cast(ptr::addr_of(f));
|
let closure: *rust_closure = unsafe::reinterpret_cast(ptr::addr_of(f));
|
||||||
#debug("spawn: closure={%x,%x}", (*closure).fnptr, (*closure).envptr);
|
#debug("spawn: closure={%x,%x}", (*closure).fnptr, (*closure).envptr);
|
||||||
let id = rustrt::new_task();
|
let id = rustrt::new_task();
|
||||||
|
|
||||||
|
// set up notifications if they are enabled.
|
||||||
|
option::may(notify) {|c|
|
||||||
|
let task_ptr <- rust_task_ptr(rustrt::get_task_pointer(id));
|
||||||
|
(**task_ptr).notify_enabled = 1;
|
||||||
|
(**task_ptr).notify_chan = c;
|
||||||
|
}
|
||||||
|
|
||||||
rustrt::start_task(id, closure);
|
rustrt::start_task(id, closure);
|
||||||
unsafe::leak(f);
|
unsafe::leak(f);
|
||||||
ret id;
|
ret id;
|
||||||
@@ -129,6 +140,11 @@ A task that sends notification upon termination
|
|||||||
type joinable_task = (task, comm::port<task_notification>);
|
type joinable_task = (task, comm::port<task_notification>);
|
||||||
|
|
||||||
fn spawn_joinable(-f: sendfn()) -> joinable_task {
|
fn spawn_joinable(-f: sendfn()) -> joinable_task {
|
||||||
|
let notify_port = comm::port();
|
||||||
|
let notify_chan = comm::chan(notify_port);
|
||||||
|
let task = spawn_inner(f, some(notify_chan));
|
||||||
|
ret (task, notify_port);
|
||||||
|
/*
|
||||||
resource notify_rsrc(data: (comm::chan<task_notification>,
|
resource notify_rsrc(data: (comm::chan<task_notification>,
|
||||||
task,
|
task,
|
||||||
@mutable task_result)) {
|
@mutable task_result)) {
|
||||||
@@ -148,6 +164,7 @@ fn spawn_joinable(-f: sendfn()) -> joinable_task {
|
|||||||
};
|
};
|
||||||
let task = spawn(g);
|
let task = spawn(g);
|
||||||
ret (task, notify_port);
|
ret (task, notify_port);
|
||||||
|
*/
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ fn f(&&n: uint) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn g(&&_i: ()) { }
|
fn g() { }
|
||||||
|
|
||||||
fn main(args: [str]) {
|
fn main(args: [str]) {
|
||||||
let n =
|
let n =
|
||||||
|
|||||||
@@ -71,13 +71,12 @@ mod map_reduce {
|
|||||||
[joinable_task] {
|
[joinable_task] {
|
||||||
let tasks = [];
|
let tasks = [];
|
||||||
for i: str in inputs {
|
for i: str in inputs {
|
||||||
tasks += [task::spawn_joinable((ctrl, i), map_task)];
|
tasks += [task::spawn_joinable {|| map_task(ctrl, i)}];
|
||||||
}
|
}
|
||||||
ret tasks;
|
ret tasks;
|
||||||
}
|
}
|
||||||
|
|
||||||
fn map_task(args: (chan<ctrl_proto>, str)) {
|
fn map_task(ctrl: chan<ctrl_proto>, input: str) {
|
||||||
let (ctrl, input) = args;
|
|
||||||
// log(error, "map_task " + input);
|
// log(error, "map_task " + input);
|
||||||
let intermediates = map::new_str_hash();
|
let intermediates = map::new_str_hash();
|
||||||
|
|
||||||
@@ -106,8 +105,7 @@ mod map_reduce {
|
|||||||
send(ctrl, mapper_done);
|
send(ctrl, mapper_done);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn reduce_task(args: (str, chan<chan<reduce_proto>>)) {
|
fn reduce_task(key: str, out: chan<chan<reduce_proto>>) {
|
||||||
let (key, out) = args;
|
|
||||||
let p = port();
|
let p = port();
|
||||||
|
|
||||||
send(out, chan(p));
|
send(out, chan(p));
|
||||||
@@ -168,8 +166,9 @@ mod map_reduce {
|
|||||||
none. {
|
none. {
|
||||||
// log(error, "creating new reducer for " + k);
|
// log(error, "creating new reducer for " + k);
|
||||||
let p = port();
|
let p = port();
|
||||||
|
let ch = chan(p);
|
||||||
tasks +=
|
tasks +=
|
||||||
[task::spawn_joinable((k, chan(p)), reduce_task)];
|
[task::spawn_joinable{||reduce_task(k, ch)}];
|
||||||
c = recv(p);
|
c = recv(p);
|
||||||
reducers.insert(k, c);
|
reducers.insert(k, c);
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user