rust-lang/rust
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Make moves explicit in bench tests

Browse Source
This commit is contained in:
Tim Chevalier
2012-09-18 22:44:34 -07:00
parent e18c6bb3b6
commit f5f3a75b65
15 changed files with 124 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/test/bench/graph500-bfs.rs
View File

@@ -119,7 +119,7 @@ fn bfs(graph: graph, key: node_id) -> bfs_result {
Q.add_back(key);
marks[key] = key;
while Q.size() > 0u {
while Q.size() > 0 {
let t = Q.pop_front();
do graph[t].each() |k| {
@@ -131,7 +131,7 @@ fn bfs(graph: graph, key: node_id) -> bfs_result {
};
}
vec::from_mut(marks)
vec::from_mut(move marks)
}
/**
@@ -167,11 +167,11 @@ fn bfs2(graph: graph, key: node_id) -> bfs_result {
}
}
let mut i = 0u;
let mut i = 0;
while vec::any(colors, is_gray) {
// Do the BFS.
log(info, fmt!("PBFS iteration %?", i));
i += 1u;
i += 1;
colors = do colors.mapi() |i, c| {
let c : color = *c;
match c {
@@ -245,13 +245,13 @@ fn pbfs(&&graph: arc::ARC<graph>, key: node_id) -> bfs_result {
i += 1;
let old_len = colors.len();
let color = arc::ARC(colors);
let color = arc::ARC(move colors);
let color_vec = arc::get(&color); // FIXME #3387 requires this temp
colors = do par::mapi_factory(*color_vec) {
let colors = arc::clone(&color);
let graph = arc::clone(&graph);
fn~(+i: uint, +c: color) -> color {
fn~(move graph, move colors, +i: uint, +c: color) -> color {
let c : color = c;
let colors = arc::get(&colors);
let graph = arc::get(&graph);
@@ -388,7 +388,7 @@ fn main() {
let args = os::args();
let args = if os::getenv(~"RUST_BENCH").is_some() {
~[~"", ~"15", ~"48"]
} else if args.len() <= 1u {
} else if args.len() <= 1 {
~[~"", ~"10", ~"16"]
} else {
args
@@ -400,21 +400,21 @@ fn main() {
let do_sequential = true;
let start = time::precise_time_s();
let edges = make_edges(scale, 16u);
let edges = make_edges(scale, 16);
let stop = time::precise_time_s();
io::stdout().write_line(fmt!("Generated %? edges in %? seconds.",
vec::len(edges), stop - start));
let start = time::precise_time_s();
let graph = make_graph(1u << scale, edges);
let graph = make_graph(1 << scale, edges);
let stop = time::precise_time_s();
let mut total_edges = 0u;
let mut total_edges = 0;
vec::each(graph, |edges| { total_edges += edges.len(); true });
io::stdout().write_line(fmt!("Generated graph with %? edges in %? seconds.",
total_edges / 2u,
total_edges / 2,
stop - start));
let mut total_seq = 0.0;

14
src/test/bench/msgsend-pipes-shared.rs
View File

@@ -19,7 +19,7 @@ use io::WriterUtil;
use pipes::{Port, Chan, SharedChan};
macro_rules! move_out (
{ $x:expr } => { unsafe { let y <- *ptr::addr_of(&($x)); y } }
{ $x:expr } => { unsafe { let y <- *ptr::addr_of(&($x)); move y } }
)
enum request {
@@ -50,26 +50,26 @@ fn run(args: &[~str]) {
let (to_parent, from_child) = pipes::stream();
let (to_child, from_parent) = pipes::stream();
let to_child = SharedChan(to_child);
let to_child = SharedChan(move to_child);
let size = uint::from_str(args[1]).get();
let workers = uint::from_str(args[2]).get();
let num_bytes = 100;
let start = std::time::precise_time_s();
let mut worker_results = ~[];
for uint::range(0u, workers) |i| {
for uint::range(0, workers) |_i| {
let to_child = to_child.clone();
do task::task().future_result(|+r| {
worker_results.push(r);
}).spawn {
for uint::range(0u, size / workers) |_i| {
worker_results.push(move r);
}).spawn |move to_child| {
for uint::range(0, size / workers) |_i| {
//error!("worker %?: sending %? bytes", i, num_bytes);
to_child.send(bytes(num_bytes));
}
//error!("worker %? exiting", i);
};
}
do task::spawn {
do task::spawn |move from_parent, move to_parent| {
server(from_parent, to_parent);
}

16
src/test/bench/msgsend-pipes.rs
View File

@@ -15,7 +15,7 @@ use io::WriterUtil;
use pipes::{Port, PortSet, Chan};
macro_rules! move_out (
{ $x:expr } => { unsafe { let y <- *ptr::addr_of(&($x)); y } }
{ $x:expr } => { unsafe { let y <- *ptr::addr_of(&($x)); move y } }
)
enum request {
@@ -46,27 +46,27 @@ fn run(args: &[~str]) {
let (to_parent, from_child) = pipes::stream();
let (to_child, from_parent_) = pipes::stream();
let from_parent = PortSet();
from_parent.add(from_parent_);
from_parent.add(move from_parent_);
let size = uint::from_str(args[1]).get();
let workers = uint::from_str(args[2]).get();
let num_bytes = 100;
let start = std::time::precise_time_s();
let mut worker_results = ~[];
for uint::range(0u, workers) |i| {
for uint::range(0, workers) |_i| {
let (to_child, from_parent_) = pipes::stream();
from_parent.add(from_parent_);
from_parent.add(move from_parent_);
do task::task().future_result(|+r| {
worker_results.push(r);
}).spawn {
for uint::range(0u, size / workers) |_i| {
worker_results.push(move r);
}).spawn |move to_child| {
for uint::range(0, size / workers) |_i| {
//error!("worker %?: sending %? bytes", i, num_bytes);
to_child.send(bytes(num_bytes));
}
//error!("worker %? exiting", i);
};
}
do task::spawn {
do task::spawn |move from_parent, move to_parent| {
server(from_parent, to_parent);
}

24
src/test/bench/msgsend-ring-mutex-arcs.rs
View File

@@ -33,7 +33,7 @@ fn recv(p: &pipe) -> uint {
fn init() -> (pipe,pipe) {
let m = arc::MutexARC(~[]);
((&m).clone(), m)
((&m).clone(), move m)
}
@@ -41,18 +41,18 @@ fn thread_ring(i: uint,
count: uint,
+num_chan: pipe,
+num_port: pipe) {
let mut num_chan <- Some(num_chan);
let mut num_port <- Some(num_port);
let mut num_chan <- Some(move num_chan);
let mut num_port <- Some(move num_port);
// Send/Receive lots of messages.
for uint::range(0u, count) |j| {
//error!("task %?, iter %?", i, j);
let mut num_chan2 = option::swap_unwrap(&mut num_chan);
let mut num_port2 = option::swap_unwrap(&mut num_port);
send(&num_chan2, i * j);
num_chan = Some(num_chan2);
num_chan = Some(move num_chan2);
let _n = recv(&num_port2);
//log(error, _n);
num_port = Some(num_port2);
num_port = Some(move num_port2);
};
}
@@ -70,7 +70,7 @@ fn main() {
let msg_per_task = uint::from_str(args[2]).get();
let (num_chan, num_port) = init();
let mut num_chan = Some(num_chan);
let mut num_chan = Some(move num_chan);
let start = time::precise_time_s();
@@ -82,22 +82,22 @@ fn main() {
let (new_chan, num_port) = init();
let num_chan2 = ~mut None;
*num_chan2 <-> num_chan;
let num_port = ~mut Some(num_port);
let num_port = ~mut Some(move num_port);
let new_future = future::spawn(|move num_chan2, move num_port| {
let mut num_chan = None;
num_chan <-> *num_chan2;
let mut num_port1 = None;
num_port1 <-> *num_port;
thread_ring(i, msg_per_task,
option::unwrap(num_chan),
option::unwrap(num_port1))
option::unwrap(move num_chan),
option::unwrap(move num_port1))
});
futures.push(new_future);
num_chan = Some(new_chan);
futures.push(move new_future);
num_chan = Some(move new_chan);
};
// do our iteration
thread_ring(0u, msg_per_task, option::unwrap(num_chan), num_port);
thread_ring(0, msg_per_task, option::unwrap(move num_chan), move num_port);
// synchronize
for futures.each |f| { future::get(f) };

28
src/test/bench/msgsend-ring-pipes.rs
View File

@@ -24,7 +24,7 @@ proto! ring (
fn macros() {
#macro[
[#move_out[x],
unsafe { let y <- *ptr::addr_of(&x); y }]
unsafe { let y <- *ptr::addr_of(&x); move y }]
];
}
@@ -32,18 +32,18 @@ fn thread_ring(i: uint,
count: uint,
+num_chan: ring::client::num,
+num_port: ring::server::num) {
let mut num_chan <- Some(num_chan);
let mut num_port <- Some(num_port);
let mut num_chan <- Some(move num_chan);
let mut num_port <- Some(move num_port);
// Send/Receive lots of messages.
for uint::range(0u, count) |j| {
for uint::range(0, count) |j| {
//error!("task %?, iter %?", i, j);
let mut num_chan2 = None;
let mut num_port2 = None;
num_chan2 <-> num_chan;
num_port2 <-> num_port;
num_chan = Some(ring::client::num(option::unwrap(num_chan2), i * j));
let port = option::unwrap(num_port2);
match recv(port) {
num_chan = Some(ring::client::num(option::unwrap(move num_chan2), i * j));
let port = option::unwrap(move num_port2);
match recv(move port) {
ring::num(_n, p) => {
//log(error, _n);
num_port = Some(move_out!(p));
@@ -66,7 +66,7 @@ fn main() {
let msg_per_task = uint::from_str(args[2]).get();
let (num_chan, num_port) = ring::init();
let mut num_chan = Some(num_chan);
let mut num_chan = Some(move num_chan);
let start = time::precise_time_s();
@@ -78,7 +78,7 @@ fn main() {
let (new_chan, num_port) = ring::init();
let num_chan2 = ~mut None;
*num_chan2 <-> num_chan;
let num_port = ~mut Some(num_port);
let num_port = ~mut Some(move num_port);
let new_future = do future::spawn
|move num_chan2, move num_port| {
let mut num_chan = None;
@@ -86,15 +86,15 @@ fn main() {
let mut num_port1 = None;
num_port1 <-> *num_port;
thread_ring(i, msg_per_task,
option::unwrap(num_chan),
option::unwrap(num_port1))
option::unwrap(move num_chan),
option::unwrap(move num_port1))
};
futures.push(new_future);
num_chan = Some(new_chan);
futures.push(move new_future);
num_chan = Some(move new_chan);
};
// do our iteration
thread_ring(0u, msg_per_task, option::unwrap(num_chan), num_port);
thread_ring(0, msg_per_task, option::unwrap(move num_chan), move num_port);
// synchronize
for futures.each |f| { future::get(f) };

24
src/test/bench/msgsend-ring-rw-arcs.rs
View File

@@ -33,7 +33,7 @@ fn recv(p: &pipe) -> uint {
fn init() -> (pipe,pipe) {
let x = arc::RWARC(~[]);
((&x).clone(), x)
((&x).clone(), move x)
}
@@ -41,18 +41,18 @@ fn thread_ring(i: uint,
count: uint,
+num_chan: pipe,
+num_port: pipe) {
let mut num_chan <- Some(num_chan);
let mut num_port <- Some(num_port);
let mut num_chan <- Some(move num_chan);
let mut num_port <- Some(move num_port);
// Send/Receive lots of messages.
for uint::range(0u, count) |j| {
//error!("task %?, iter %?", i, j);
let mut num_chan2 = option::swap_unwrap(&mut num_chan);
let mut num_port2 = option::swap_unwrap(&mut num_port);
send(&num_chan2, i * j);
num_chan = Some(num_chan2);
num_chan = Some(move num_chan2);
let _n = recv(&num_port2);
//log(error, _n);
num_port = Some(num_port2);
num_port = Some(move num_port2);
};
}
@@ -70,7 +70,7 @@ fn main() {
let msg_per_task = uint::from_str(args[2]).get();
let (num_chan, num_port) = init();
let mut num_chan = Some(num_chan);
let mut num_chan = Some(move num_chan);
let start = time::precise_time_s();
@@ -82,7 +82,7 @@ fn main() {
let (new_chan, num_port) = init();
let num_chan2 = ~mut None;
*num_chan2 <-> num_chan;
let num_port = ~mut Some(num_port);
let num_port = ~mut Some(move num_port);
let new_future = do future::spawn
|move num_chan2, move num_port| {
let mut num_chan = None;
@@ -90,15 +90,15 @@ fn main() {
let mut num_port1 = None;
num_port1 <-> *num_port;
thread_ring(i, msg_per_task,
option::unwrap(num_chan),
option::unwrap(num_port1))
option::unwrap(move num_chan),
option::unwrap(move num_port1))
};
futures.push(new_future);
num_chan = Some(new_chan);
futures.push(move new_future);
num_chan = Some(move new_chan);
};
// do our iteration
thread_ring(0u, msg_per_task, option::unwrap(num_chan), num_port);
thread_ring(0, msg_per_task, option::unwrap(move num_chan), move num_port);
// synchronize
for futures.each |f| { future::get(f) };

2
src/test/bench/msgsend-ring.rs
View File

@@ -52,7 +52,7 @@ fn main() {
get_chan_chan.send(Chan(&p));
thread_ring(i, msg_per_task, num_chan, p)
};
futures.push(new_future);
futures.push(move new_future);
num_chan = get_chan.recv();
};

8
src/test/bench/msgsend.rs
View File

@@ -35,12 +35,12 @@ fn run(args: ~[~str]) {
let workers = uint::from_str(args[2]).get();
let start = std::time::precise_time_s();
let mut worker_results = ~[];
for uint::range(0u, workers) |_i| {
for uint::range(0, workers) |_i| {
do task::task().future_result(|+r| {
worker_results.push(r);
worker_results.push(move r);
}).spawn {
for uint::range(0u, size / workers) |_i| {
comm::send(to_child, bytes(100u));
for uint::range(0, size / workers) |_i| {
comm::send(to_child, bytes(100));
}
};
}

36
src/test/bench/pingpong.rs
View File

@@ -33,7 +33,7 @@ proto! pingpong_unbounded (
// This stuff should go in libcore::pipes
macro_rules! move_it (
{ $x:expr } => { let t <- *ptr::addr_of(&($x)); t }
{ $x:expr } => { let t <- *ptr::addr_of(&($x)); move t }
)
macro_rules! follow (
@@ -42,8 +42,8 @@ macro_rules! follow (
} => (
|m| match move m {
$(Some($message($($x,)* move next)) => {
let $next = next;
$e })+
let $next = move next;
move $e })+
_ => { fail }
}
);
@@ -53,8 +53,8 @@ macro_rules! follow (
} => (
|m| match move m {
$(Some($message(move next)) => {
let $next = next;
$e })+
let $next = move next;
move $e })+
_ => { fail }
}
)
@@ -62,7 +62,7 @@ macro_rules! follow (
fn switch<T: Send, Tb: Send, U>(+endp: pipes::RecvPacketBuffered<T, Tb>,
f: fn(+v: Option<T>) -> U) -> U {
f(pipes::try_recv(endp))
f(pipes::try_recv(move endp))
}
// Here's the benchmark
@@ -72,10 +72,10 @@ fn bounded(count: uint) {
let mut ch = do spawn_service(init) |ch| {
let mut count = count;
let mut ch = ch;
let mut ch = move ch;
while count > 0 {
ch = switch(ch, follow! (
ping -> next { server::pong(next) }
ch = switch(move ch, follow! (
ping -> next { server::pong(move next) }
));
count -= 1;
@@ -84,10 +84,10 @@ fn bounded(count: uint) {
let mut count = count;
while count > 0 {
let ch_ = client::ping(ch);
let ch_ = client::ping(move ch);
ch = switch(ch_, follow! (
pong -> next { next }
ch = switch(move ch_, follow! (
pong -> next { move next }
));
count -= 1;
@@ -99,10 +99,10 @@ fn unbounded(count: uint) {
let mut ch = do spawn_service(init) |ch| {
let mut count = count;
let mut ch = ch;
let mut ch = move ch;
while count > 0 {
ch = switch(ch, follow! (
ping -> next { server::pong(next) }
ch = switch(move ch, follow! (
ping -> next { server::pong(move next) }
));
count -= 1;
@@ -111,10 +111,10 @@ fn unbounded(count: uint) {
let mut count = count;
while count > 0 {
let ch_ = client::ping(ch);
let ch_ = client::ping(move ch);
ch = switch(ch_, follow! (
pong -> next { next }
ch = switch(move ch_, follow! (
pong -> next { move next }
));
count -= 1;

14
src/test/bench/shootout-k-nucleotide-pipes.rs
View File

@@ -124,7 +124,7 @@ fn make_sequence_processor(sz: uint, from_parent: pipes::Port<~[u8]>,
};
//comm::send(to_parent, fmt!("yay{%u}", sz));
to_parent.send(buffer);
to_parent.send(move buffer);
}
// given a FASTA file on stdin, process sequence THREE
@@ -143,25 +143,25 @@ fn main() {
// initialize each sequence sorter
let sizes = ~[1u,2u,3u,4u,6u,12u,18u];
let sizes = ~[1,2,3,4,6,12,18];
let streams = vec::map(sizes, |_sz| Some(stream()));
let streams = vec::to_mut(streams);
let streams = vec::to_mut(move streams);
let mut from_child = ~[];
let to_child = vec::mapi(sizes, |ii, sz| {
let sz = *sz;
let mut stream = None;
stream <-> streams[ii];
let (to_parent_, from_child_) = option::unwrap(stream);
let (to_parent_, from_child_) = option::unwrap(move stream);
from_child.push(from_child_);
from_child.push(move from_child_);
let (to_child, from_parent) = pipes::stream();
do task::spawn_with(from_parent) |from_parent| {
do task::spawn_with(move from_parent) |move to_parent_, from_parent| {
make_sequence_processor(sz, from_parent, to_parent_);
};
to_child
move to_child
});

10
src/test/bench/shootout-pfib.rs
View File

@@ -34,15 +34,15 @@ fn fib(n: int) -> int {
} else {
let p = pipes::PortSet();
let ch = p.chan();
task::spawn(|| pfib(ch, n - 1) );
task::spawn(|move ch| pfib(ch, n - 1) );
let ch = p.chan();
task::spawn(|| pfib(ch, n - 2) );
task::spawn(|move ch| pfib(ch, n - 2) );
c.send(p.recv() + p.recv());
}
}
let (ch, p) = pipes::stream();
let t = task::spawn(|| pfib(ch, n) );
let _t = task::spawn(|move ch| pfib(ch, n) );
p.recv()
}
@@ -73,7 +73,7 @@ fn stress(num_tasks: int) {
let mut results = ~[];
for range(0, num_tasks) |i| {
do task::task().future_result(|+r| {
results.push(r);
results.push(move r);
}).spawn {
stress_task(i);
}
@@ -104,7 +104,7 @@ fn main() {
let out = io::stdout();
for range(1, max + 1) |n| {
for range(0, num_trials) |i| {
for range(0, num_trials) |_i| {
let start = time::precise_time_ns();
let fibn = fib(n);
let stop = time::precise_time_ns();

5
src/test/bench/task-perf-alloc-unwind.rs
View File

@@ -78,7 +78,8 @@ fn recurse_or_fail(depth: int, st: Option<st>) {
box: @Cons((), st.box),
unique: ~Cons((), @*st.unique),
fn_box: fn@() -> @nillist { @Cons((), fn_box()) },
fn_unique: fn~() -> ~nillist { ~Cons((), @*fn_unique()) },
fn_unique: fn~(move fn_unique) -> ~nillist
{ ~Cons((), @*fn_unique()) },
tuple: (@Cons((), st.tuple.first()),
~Cons((), @*st.tuple.second())),
vec: st.vec + ~[@Cons((), st.vec.last())],
@@ -87,6 +88,6 @@ fn recurse_or_fail(depth: int, st: Option<st>) {
}
};
recurse_or_fail(depth, Some(st));
recurse_or_fail(depth, Some(move st));
}
}

10
src/test/bench/task-perf-jargon-metal-smoke.rs
View File

@@ -9,14 +9,14 @@ fn child_generation(gens_left: uint, -c: pipes::Chan<()>) {
// This used to be O(n^2) in the number of generations that ever existed.
// With this code, only as many generations are alive at a time as tasks
// alive at a time,
let c = ~mut Some(c);
do task::spawn_supervised {
let c = ~mut Some(move c);
do task::spawn_supervised |move c| {
let c = option::swap_unwrap(c);
if gens_left & 1 == 1 {
task::yield(); // shake things up a bit
}
if gens_left > 0 {
child_generation(gens_left - 1, c); // recurse
child_generation(gens_left - 1, move c); // recurse
} else {
c.send(())
}
@@ -27,14 +27,14 @@ fn main() {
let args = os::args();
let args = if os::getenv(~"RUST_BENCH").is_some() {
~[~"", ~"100000"]
} else if args.len() <= 1u {
} else if args.len() <= 1 {
~[~"", ~"100"]
} else {
copy args
};
let (c,p) = pipes::stream();
child_generation(uint::from_str(args[1]).get(), c);
child_generation(uint::from_str(args[1]).get(), move c);
if p.try_recv().is_none() {
fail ~"it happened when we slumbered";
}

4
src/test/bench/task-perf-linked-failure.rs
View File

@@ -31,9 +31,9 @@ fn grandchild_group(num_tasks: uint) {
fn spawn_supervised_blocking(myname: &str, +f: fn~()) {
let mut res = None;
task::task().future_result(|+r| res = Some(r)).supervised().spawn(f);
task::task().future_result(|+r| res = Some(move r)).supervised().spawn(move f);
error!("%s group waiting", myname);
let x = future::get(&option::unwrap(res));
let x = future::get(&option::unwrap(move res));
assert x == task::Success;
}

30
src/test/bench/task-perf-word-count-generic.rs
View File

@@ -30,7 +30,7 @@ use cmp::Eq;
use to_bytes::IterBytes;
macro_rules! move_out (
{ $x:expr } => { unsafe { let y <- *ptr::addr_of(&($x)); y } }
{ $x:expr } => { unsafe { let y <- *ptr::addr_of(&($x)); move y } }
)
trait word_reader {
@@ -90,19 +90,19 @@ impl<T> box<T> {
fn swap(f: fn(+v: T) -> T) {
let mut tmp = None;
self.contents <-> tmp;
self.contents = Some(f(option::unwrap(tmp)));
self.contents = Some(f(option::unwrap(move tmp)));
}
fn unwrap() -> T {
let mut tmp = None;
self.contents <-> tmp;
option::unwrap(tmp)
option::unwrap(move tmp)
}
}
fn box<T>(+x: T) -> box<T> {
box {
contents: Some(x)
contents: Some(move x)
}
}
@@ -151,13 +151,13 @@ mod map_reduce {
let mut tasks = ~[];
for inputs.each |i| {
let (ctrl, ctrl_server) = ctrl_proto::init();
let ctrl = box(ctrl);
let ctrl = box(move ctrl);
let i = copy *i;
let m = copy *map;
tasks.push(spawn_joinable(|move i| map_task(m, &ctrl, i)));
ctrls.push(ctrl_server);
tasks.push(spawn_joinable(|move ctrl, move i| map_task(m, &ctrl, i)));
ctrls.push(move ctrl_server);
}
return tasks;
move tasks
}
fn map_task<K1: Copy Send, K2: Hash IterBytes Eq Const Copy Send, V: Copy Send>(
@@ -177,8 +177,8 @@ mod map_reduce {
Some(_c) => { c = Some(_c); }
None => {
do ctrl.swap |ctrl| {
let ctrl = ctrl_proto::client::find_reducer(ctrl, *key);
match pipes::recv(ctrl) {
let ctrl = ctrl_proto::client::find_reducer(move ctrl, *key);
match pipes::recv(move ctrl) {
ctrl_proto::reducer(c_, ctrl) => {
c = Some(c_);
move_out!(ctrl)
@@ -250,12 +250,12 @@ mod map_reduce {
let mut num_mappers = vec::len(inputs) as int;
while num_mappers > 0 {
let (_ready, message, ctrls) = pipes::select(ctrl);
match option::unwrap(message) {
let (_ready, message, ctrls) = pipes::select(move ctrl);
match option::unwrap(move message) {
ctrl_proto::mapper_done => {
// error!("received mapper terminated.");
num_mappers -= 1;
ctrl = ctrls;
ctrl = move ctrls;
}
ctrl_proto::find_reducer(k, cc) => {
let c;
@@ -271,13 +271,13 @@ mod map_reduce {
let p = Port();
let ch = Chan(&p);
let r = reduce, kk = k;
tasks.push(spawn_joinable(|| reduce_task(~r, kk, ch) ));
tasks.push(spawn_joinable(|move r| reduce_task(~r, kk, ch) ));
c = recv(p);
reducers.insert(k, c);
}
}
ctrl = vec::append_one(
ctrls,
move ctrls,
ctrl_proto::server::reducer(move_out!(cc), c));
}
}