src/test/run-pass/hashmap-memory.rs

   1 // Copyright 2012-2014 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 // ignore-emscripten No support for threads
  12
  13 #![allow(unknown_features)]
  14 #![feature(std_misc)]
  15
  16 /**
  17    A somewhat reduced test case to expose some Valgrind issues.
  18
  19    This originally came from the word-count benchmark.
  20 */
  21
  22 pub fn map(filename: String, mut emit: map_reduce::putter) {
  23     emit(filename, "1".to_string());
  24 }
  25
  26 mod map_reduce {
  27     use std::collections::HashMap;
  28     use std::sync::mpsc::{channel, Sender};
  29     use std::str;
  30     use std::thread;
  31
  32     pub type putter<'a> = Box<FnMut(String, String) + 'a>;
  33
  34     pub type mapper = extern fn(String, putter);
  35
  36     enum ctrl_proto { find_reducer(Vec<u8>, Sender<isize>), mapper_done, }
  37
  38     fn start_mappers(ctrl: Sender<ctrl_proto>, inputs: Vec<String>) {
  39         for i in &inputs {
  40             let ctrl = ctrl.clone();
  41             let i = i.clone();
  42             thread::spawn(move|| map_task(ctrl.clone(), i.clone()) );
  43         }
  44     }
  45
  46     fn map_task(ctrl: Sender<ctrl_proto>, input: String) {
  47         let mut intermediates = HashMap::new();
  48
  49         fn emit(im: &mut HashMap<String, isize>,
  50                 ctrl: Sender<ctrl_proto>, key: String,
  51                 _val: String) {
  52             if im.contains_key(&key) {
  53                 return;
  54             }
  55             let (tx, rx) = channel();
  56             println!("sending find_reducer");
  57             ctrl.send(ctrl_proto::find_reducer(key.as_bytes().to_vec(), tx)).unwrap();
  58             println!("receiving");
  59             let c = rx.recv().unwrap();
  60             println!("{}", c);
  61             im.insert(key, c);
  62         }
  63
  64         let ctrl_clone = ctrl.clone();
  65         ::map(input, Box::new(|a,b| emit(&mut intermediates, ctrl.clone(), a, b)));
  66         ctrl_clone.send(ctrl_proto::mapper_done).unwrap();
  67     }
  68
  69     pub fn map_reduce(inputs: Vec<String>) {
  70         let (tx, rx) = channel();
  71
  72         // This thread becomes the master control thread. It spawns others
  73         // to do the rest.
  74
  75         let mut reducers: HashMap<String, isize>;
  76
  77         reducers = HashMap::new();
  78
  79         start_mappers(tx, inputs.clone());
  80
  81         let mut num_mappers = inputs.len() as isize;
  82
  83         while num_mappers > 0 {
  84             match rx.recv().unwrap() {
  85               ctrl_proto::mapper_done => { num_mappers -= 1; }
  86               ctrl_proto::find_reducer(k, cc) => {
  87                 let mut c;
  88                 match reducers.get(&str::from_utf8(&k).unwrap().to_string()) {
  89                   Some(&_c) => { c = _c; }
  90                   None => { c = 0; }
  91                 }
  92                 cc.send(c).unwrap();
  93               }
  94             }
  95         }
  96     }
  97 }
  98
  99 pub fn main() {
 100     map_reduce::map_reduce(
 101         vec!["../src/test/run-pass/hashmap-memory.rs".to_string()]);
 102 }