The goal of this post is to take a look at threads in Rust, really I split the topic in two posts (a full understanding requires more).
To explain the problem let’s start with a simple problem, a log message type counter in a file; nothing special but it’s enough to illustrate the concept.
Let’s start by looking at this piece of code:
fn parser_with_scope(file_name: String) -> io::Result<()> {
let file = File::open(file_name)?;
let reader = BufReader::new(file);
let mut vec: Vec<String> = Vec::new();
let mut map1: HashMap<String, i32> = HashMap::new();
let mut map2: HashMap<String, i32> = HashMap::new();
for line_result in reader.lines() {
vec.push(line_result?);
}
let half = vec.len() / 2;
let (vec1, vec2) = vec.split_at(half);
thread::scope(|s| {
s.spawn(|| {
for line in vec1 {
if let Some(level) = parse_line(line) {
*map1.entry(level.to_string()).or_insert(0) += 1;
}
}
});
s.spawn(|| {
for line in vec2 {
if let Some(level) = parse_line(line) {
*map2.entry(level.to_string()).or_insert(0) += 1;
}
}
});
}); //
for (k, v) in map2 {
*map1.entry(k).or_insert(0) += v;
}
for (level, occurrency) in map1 {
println!("{level}: {occurrency}");
}
Ok(())
}
fn parse_line(line: &str) -> Option<&'static str> {
if line.contains("ERROR") {
Some("ERROR")
} else if line.contains("WARN") {
Some("WARN")
} else if line.contains("INFO") {
Some("INFO")
} else {
None
}
}What does this function do? It reads all the lines from the file and adds them into a vector, then splits it in two.
We create two HashMaps where the key is the word found and the value the number of occurrences.
Thus, we create two threads using scope, every thread works on a different vector and map. When the threads have finished, we merge map2 into map1.
What does thread scope do?
When I first encountered thread::scope, the name didn’t help. I already knew thread::spawn, that’s the obvious entry point. But scope sounded like a scoping construct, not a concurrency primitive. It took me a while to understand that the name refers to the lifetime guarantee: the scope is what ensures all threads finish before you move on.
With the scope we have an implicit join and s.spawn creates a thread that can borrow data from the enclosing scope. Use it when threads don’t need to outlive the scope and you don’t need to clone the data.
Other solution:
fn parser_without_scope(file_name: String) -> io::Result<()> {
let map = Arc::new(Mutex::new(HashMap::<&'static str, i32>::new()));
let file = File::open(file_name)?;
let reader = BufReader::new(file);
let mut vec: Vec<String> = Vec::new();
for line_result in reader.lines() {
vec.push(line_result?);
}
let half = vec.len() / 2;
let (vec1, vec2) = vec.split_at(half);
let vec1: Vec<String> = vec1.to_vec();
let vec2: Vec<String> = vec2.to_vec();
let map_clone1 = Arc::clone(&map);
let map_clone2 = Arc::clone(&map);
let thread_join_handle_1 = thread::spawn(move || {
process_lines(&vec1, map_clone1);
});
let thread_join_handle_2 = thread::spawn(move || {
process_lines(&vec2, map_clone2);
});
thread_join_handle_1.join();
thread_join_handle_2.join();
for (level, occurrency) in map.lock().unwrap().iter() {
println!("{level}: {occurrency}");
}
Ok(())
}
fn process_lines(lines: &[String], map: Arc<Mutex<HashMap<&'static str, i32>>>) {
let mut local = HashMap::new();
for line in lines {
if let Some(level) = parse_line(line) {
*local.entry(level).or_insert(0) += 1;
}
}
let mut global = map.lock().unwrap();
for (k, v) in local {
*global.entry(k).or_insert(0) += v;
}
}Take a look at the definition of the map,
Arc::new(Mutex::new(HashMap::<&'static str,i32>::new()));
Thus, with Arc we can share the ownership between thread, mutex is self-explanatory, and the real map definition has the key defined as: &’static str, what’s mean? The lifetime is static, the string lives for all the duration of the program and it lives in the binary’s read-only data segment, not on the heap, because it has fixed values (“ERROR”, “WARN” or “INFO”). If we use String instead of the static we have to allocate on the heap every key, it works, but in this case is not the best solution, because we know the possible values at the compile time.
thread::spawn requires all captured data to be ‘static. The compiler can’t guarantee the thread won’t outlive the current scope, so it refuses to accept borrowed references.
Thus we have one map created with Arc and Mutex, we create two clones of the map, and the threads with thread::spawn, so in this case we need to call join explicitly.
Note that in this case both threads are working on the same map, we have cloned the reference to it.
Note what process_lines does, it creates a local hashmap that it’s filled with the values from the lines vector and then the values are pushed in the global map only at the end to avoid acquiring the lock for every value.
When to use this solution? When threads need to outlive the current scope, think of a long-running worker or a background task. In all other cases, prefer scope.