Rust Task Queue

Question

The purpose of this program is to have one thread (main thread) working on I/O reading lines from a file and feeding it to a pool of worker threads whose job is to perform some processing on each line provided. In this case, the processing is running the String.contains() method.

Please note that as test input, I used a giant .txt file of English dictionary words found here:

Also note that you need to specify the file name of the file to be scanned as the sole command line argument to the program. I am mostly concerned with the threading model and multithreading implementation but welcome other feedback as well; new Rust programmer.

spmc crate ver is "0.2.2" for your cargo.toml

extern crate spmc;
use spmc::channel;
use std::thread;
use std::io::BufReader;
use std::fs::File;
use std::io::BufRead;
use std::u32::MAX;
use std::env;
use std::sync::Arc;

fn main() -> Result<(), std::io::Error> 
{
    let args: Vec<String> = env::args().collect();
    match args.len()
    {
        2 => {},
        _ => return Err(std::io::Error::new(std::io::ErrorKind::Other, "Incorrect number of args"))
    }
    let filename = &args[1] as &str;
    let f1 = File::open(filename)?;

    let mut br = BufReader::new(f1);

    let mut vecData: Vec<String> = Vec::new();     

    let (tx, rx) = spmc::channel();

    let mut handles = Vec::new();
    for n in 0..5 {
        let rx = rx.clone();
        handles.push(thread::spawn(move || {
            loop
            {
                let mut line_to_check: Arc<String> = rx.recv().unwrap();
                if line_to_check.contains("test")
                {
                    println!("HIT: {}", line_to_check);
                }
            }

        }));
    }
    let mut input_str = String::new();
    let mut bytes_read: usize = 1;
    while bytes_read != 0 {
        let mut is_copy = input_str.clone();
        bytes_read = 
        match br.read_line(&mut is_copy)
        {
            Ok(num) => num,
            Err(err) => return Err(std::io::Error::new(std::io::ErrorKind::Other, "read_line failed...\n"))
        };

        let str_arc : Arc<String> = Arc::new(is_copy);
        tx.send(str_arc);
    }


    Ok(())
}

Answer 1

Your code has a number of warnings when compiled under rustc. Those are free code review hints, listen to them!

You use std::io::Error for all your errors. This is dubious since many of your errors are not I/O related. Personally, I like to use Error from the failure crate. All other errors can be automatically converted to failure::Error by the ? operator.

A block of code towards the end has a number of issues:

let mut input_str = String::new();
let mut bytes_read: usize = 1;
while bytes_read != 0 {
    let mut is_copy = input_str.clone();

The way your code works, input_str is never modified. As such you are always cloning an empty string for every iteration of the loop. You should just create an empty string here, and get rid of input_str altogether.

    bytes_read = 
    match br.read_line(&mut is_copy)
    {
        Ok(num) => num,
        Err(err) => return Err(std::io::Error::new(std::io::ErrorKind::Other, "read_line failed...\n"))
    };

The whole point of the read_line function is that you can reuse the memory allocated in String between iterations of your loop. Since you don't do this, this is the wrong interface. Instead, use for line in br.lines() to get an iterator over the lines in the file:

    let str_arc : Arc<String> = Arc::new(is_copy);
    tx.send(str_arc);

There is no reason to use an Arc here. You only need an Arc if both the sending and receiving threads are going to keep a reference to the String. However, that's not the case here. Here you can just send the String from one thread to another, no Arc required.

On a general note, this is a bad use of threading. Your program is going to be limited by how quickly it can read the data in from the disk. As such, you cannot get a speed boost by multi-threading it like this.