0a6b9219d1
This, the Nth rewrite of channels, is not a rewrite of the core logic behind channels, but rather their API usage. In the past, we had the distinction between oneshot, stream, and shared channels, but the most recent rewrite dropped oneshots in favor of streams and shared channels. This distinction of stream vs shared has shown that it's not quite what we'd like either, and this moves the `std::comm` module in the direction of "one channel to rule them all". There now remains only one Chan and one Port. This new channel is actually a hybrid oneshot/stream/shared channel under the hood in order to optimize for the use cases in question. Additionally, this also reduces the cognitive burden of having to choose between a Chan or a SharedChan in an API. My simple benchmarks show no reduction in efficiency over the existing channels today, and a 3x improvement in the oneshot case. I sadly don't have a pre-last-rewrite compiler to test out the old old oneshots, but I would imagine that the performance is comparable, but slightly slower (due to atomic reference counting). This commit also brings the bonus bugfix to channels that the pending queue of messages are all dropped when a Port disappears rather then when both the Port and the Chan disappear.
97 lines
2.8 KiB
Rust
97 lines
2.8 KiB
Rust
// ignore-fast
|
|
|
|
// Copyright 2012-2014 The Rust Project Developers. See the COPYRIGHT
|
|
// file at the top-level directory of this distribution and at
|
|
// http://rust-lang.org/COPYRIGHT.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
|
|
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
|
|
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
|
|
// option. This file may not be copied, modified, or distributed
|
|
// except according to those terms.
|
|
|
|
#[feature(managed_boxes)];
|
|
|
|
/**
|
|
A somewhat reduced test case to expose some Valgrind issues.
|
|
|
|
This originally came from the word-count benchmark.
|
|
*/
|
|
|
|
pub fn map(filename: ~str, emit: map_reduce::putter) { emit(filename, ~"1"); }
|
|
|
|
mod map_reduce {
|
|
use std::hashmap::HashMap;
|
|
use std::str;
|
|
use std::task;
|
|
|
|
pub type putter<'a> = 'a |~str, ~str|;
|
|
|
|
pub type mapper = extern fn(~str, putter);
|
|
|
|
enum ctrl_proto { find_reducer(~[u8], Chan<int>), mapper_done, }
|
|
|
|
fn start_mappers(ctrl: Chan<ctrl_proto>, inputs: ~[~str]) {
|
|
for i in inputs.iter() {
|
|
let ctrl = ctrl.clone();
|
|
let i = i.clone();
|
|
task::spawn(proc() map_task(ctrl.clone(), i.clone()) );
|
|
}
|
|
}
|
|
|
|
fn map_task(ctrl: Chan<ctrl_proto>, input: ~str) {
|
|
let mut intermediates = HashMap::new();
|
|
|
|
fn emit(im: &mut HashMap<~str, int>,
|
|
ctrl: Chan<ctrl_proto>, key: ~str,
|
|
_val: ~str) {
|
|
if im.contains_key(&key) {
|
|
return;
|
|
}
|
|
let (pp, cc) = Chan::new();
|
|
error!("sending find_reducer");
|
|
ctrl.send(find_reducer(key.as_bytes().to_owned(), cc));
|
|
error!("receiving");
|
|
let c = pp.recv();
|
|
error!("{:?}", c);
|
|
im.insert(key, c);
|
|
}
|
|
|
|
let ctrl_clone = ctrl.clone();
|
|
::map(input, |a,b| emit(&mut intermediates, ctrl.clone(), a, b) );
|
|
ctrl_clone.send(mapper_done);
|
|
}
|
|
|
|
pub fn map_reduce(inputs: ~[~str]) {
|
|
let (ctrl_port, ctrl_chan) = Chan::new();
|
|
|
|
// This task becomes the master control task. It spawns others
|
|
// to do the rest.
|
|
|
|
let mut reducers: HashMap<~str, int>;
|
|
|
|
reducers = HashMap::new();
|
|
|
|
start_mappers(ctrl_chan, inputs.clone());
|
|
|
|
let mut num_mappers = inputs.len() as int;
|
|
|
|
while num_mappers > 0 {
|
|
match ctrl_port.recv() {
|
|
mapper_done => { num_mappers -= 1; }
|
|
find_reducer(k, cc) => {
|
|
let mut c;
|
|
match reducers.find(&str::from_utf8(k).unwrap().to_owned()) {
|
|
Some(&_c) => { c = _c; }
|
|
None => { c = 0; }
|
|
}
|
|
cc.send(c);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn main() {
|
|
map_reduce::map_reduce(~[~"../src/test/run-pass/hashmap-memory.rs"]);
|
|
}
|