workspace/src/part12.rs

   1 // Rust-101, Part 12: Concurrency (WIP)
   2 // =================
   3
   4 use std::io::prelude::*;
   5 use std::{io, fs, thread};
   6 use std::sync::mpsc::{sync_channel, SyncSender, Receiver};
   7 use std::sync::Arc;
   8
   9
  10 // Before we come to the actual code, we define a data-structure `Options` to store all the information we need
  11 // to complete the job: Which files to work on, which pattern to look for, and how to output. <br/>
  12 // Besides just printing all the matching lines, we will also offer to count them, or alternatively to sort them.
  13 #[derive(Clone,Copy)]
  14 enum OutputMode {
  15     Print,
  16     SortAndPrint,
  17     Count,
  18 }
  19 use self::OutputMode::*;
  20
  21 struct Options {
  22     files: Vec<String>,
  23     pattern: String,
  24     output_mode: OutputMode,
  25 }
  26
  27
  28 // The first functions reads the files, and sends every line over the `out_channel`.
  29 fn read_files(options: Arc<Options>, out_channel: SyncSender<String>) {
  30     for file in options.files.iter() {
  31         // First, we open the file, ignoring any errors.
  32         let file = fs::File::open(file).unwrap();
  33         // Then we obtain a `BufReader` for it, which provides the `lines` function.
  34         let file = io::BufReader::new(file);
  35         for line in file.lines() {
  36             let line = line.unwrap();
  37             // Now we send the line over the channel, ignoring the possibility of `send` failing.
  38             out_channel.send(line).unwrap();
  39         }
  40     }
  41     // When we drop the `out_channel`, it will be closed, which the other end can notice.
  42 }
  43
  44 // The second function filters the lines it receives through `in_channel` with the pattern, and sends
  45 // matches via `out_channel`.
  46 fn filter_lines(options: Arc<Options>, in_channel: Receiver<String>, out_channel: SyncSender<String>) {
  47     // We can simply iterate over the channel, which will stop when the channel is closed.
  48     for line in in_channel.iter() {
  49         // `contains` works on lots of types of patterns, but in particular, we can use it to test whether
  50         // one string is contained in another.
  51         if line.contains(&options.pattern) {
  52             unimplemented!()
  53         }
  54     }
  55 }
  56
  57 // The third function performs the output operations, receiving the relevant lines on its `in_channel`.
  58 fn output_lines(options: Arc<Options>, in_channel: Receiver<String>) {
  59     match options.output_mode {
  60         Print => {
  61             // Here, we just print every line we see.
  62             for line in in_channel.iter() {
  63                 unimplemented!()
  64             }
  65         },
  66         Count => {
  67             // We are supposed to count the number of matching lines. There's a convenient iterator adapter that
  68             // we can use for this job.
  69             unimplemented!()
  70         },
  71         SortAndPrint => {
  72             // We are asked to sort the matching lines before printing. So let's collect them all in a local vector...
  73             let data: Vec<String> = in_channel.iter().collect();
  74             // ...and implement the actual sorting later.
  75             unimplemented!()
  76         }
  77     }
  78 }
  79
  80 // With the operations of the three threads defined, we can now implement a function that performs grepping according
  81 // to some given options.
  82 fn run(options: Options) {
  83     // We move the `options` into an `Arc`, as that's what the thread workers expect.
  84     let options = Arc::new(options);
  85
  86     // Set up the channels. Use `sync_channel` with buffer-size of 16 to avoid needlessly filling RAM.
  87     let (line_sender, line_receiver) = sync_channel(16);
  88     let (filtered_sender, filtered_receiver) = sync_channel(16);
  89
  90     // Spawn the read thread: `thread::spawn` takes a closure that is run in a new thread.
  91     let options1 = options.clone();
  92     let handle1 = thread::spawn(move || read_files(options1, line_sender));
  93
  94     // Same with the filter thread.
  95     let options2 = options.clone();
  96     let handle2 = thread::spawn(move || filter_lines(options2, line_receiver, filtered_sender));
  97
  98     // And the output thread.
  99     let options3 = options.clone();
 100     let handle3 = thread::spawn(move || output_lines(options3, filtered_receiver));
 101
 102     // Finally, wait until all three threads did their job.
 103     handle1.join().unwrap();
 104     handle2.join().unwrap();
 105     handle3.join().unwrap();
 106 }
 107
 108 // Now we have all the pieces together for testing our `rgrep` with some hard-coded options.
 109 pub fn main() {
 110     let options = Options {
 111         files: vec!["src/part10.rs".to_string(), "src/part11.rs".to_string(), "src/part12.rs".to_string()],
 112         pattern: "let".to_string(),
 113         output_mode: Print
 114     };
 115     run(options);
 116 }
 117
 118 // **Exercise 12.1**: Change `rgrep` such that it prints now only the matching lines, but also the name of the file
 119 // and the number of the line in the file. You will have to change the type of the channels from `String` to something
 120 // that records this extra information.
 121
 122
 123