src/test/ui/hashmap-memory.rs

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