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

   1 #![allow(non_camel_case_types)]
   2 #![allow(dead_code)]
   3 #![allow(unused_mut)]
   4 // ignore-emscripten No support for threads
   5
   6 /**
   7    A somewhat reduced test case to expose some Valgrind issues.
   8
   9    This originally came from the word-count benchmark.
  10 */
  11
  12 pub fn map(filename: String, mut emit: map_reduce::putter) {
  13     emit(filename, "1".to_string());
  14 }
  15
  16 mod map_reduce {
  17     use std::collections::HashMap;
  18     use std::sync::mpsc::{channel, Sender};
  19     use std::str;
  20     use std::thread;
  21
  22     pub type putter<'a> = Box<dyn FnMut(String, String) + 'a>;
  23
  24     pub type mapper = extern fn(String, putter);
  25
  26     enum ctrl_proto { find_reducer(Vec<u8>, Sender<isize>), mapper_done, }
  27
  28     fn start_mappers(ctrl: Sender<ctrl_proto>, inputs: Vec<String>) {
  29         for i in &inputs {
  30             let ctrl = ctrl.clone();
  31             let i = i.clone();
  32             thread::spawn(move|| map_task(ctrl.clone(), i.clone()) );
  33         }
  34     }
  35
  36     fn map_task(ctrl: Sender<ctrl_proto>, input: String) {
  37         let mut intermediates = HashMap::new();
  38
  39         fn emit(im: &mut HashMap<String, isize>,
  40                 ctrl: Sender<ctrl_proto>, key: String,
  41                 _val: String) {
  42             if im.contains_key(&key) {
  43                 return;
  44             }
  45             let (tx, rx) = channel();
  46             println!("sending find_reducer");
  47             ctrl.send(ctrl_proto::find_reducer(key.as_bytes().to_vec(), tx)).unwrap();
  48             println!("receiving");
  49             let c = rx.recv().unwrap();
  50             println!("{}", c);
  51             im.insert(key, c);
  52         }
  53
  54         let ctrl_clone = ctrl.clone();
  55         ::map(input, Box::new(|a,b| emit(&mut intermediates, ctrl.clone(), a, b)));
  56         ctrl_clone.send(ctrl_proto::mapper_done).unwrap();
  57     }
  58
  59     pub fn map_reduce(inputs: Vec<String>) {
  60         let (tx, rx) = channel();
  61
  62         // This thread becomes the master control thread. It spawns others
  63         // to do the rest.
  64
  65         let mut reducers: HashMap<String, isize>;
  66
  67         reducers = HashMap::new();
  68
  69         start_mappers(tx, inputs.clone());
  70
  71         let mut num_mappers = inputs.len() as isize;
  72
  73         while num_mappers > 0 {
  74             match rx.recv().unwrap() {
  75               ctrl_proto::mapper_done => { num_mappers -= 1; }
  76               ctrl_proto::find_reducer(k, cc) => {
  77                 let mut c;
  78                 match reducers.get(&str::from_utf8(&k).unwrap().to_string()) {
  79                   Some(&_c) => { c = _c; }
  80                   None => { c = 0; }
  81                 }
  82                 cc.send(c).unwrap();
  83               }
  84             }
  85         }
  86     }
  87 }
  88
  89 pub fn main() {
  90     map_reduce::map_reduce(
  91         vec!["../src/test/run-pass/hashmap-memory.rs".to_string()]);
  92 }