workspace/src/part12.rs

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