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