I am writing a small app for rooting out duplicate files (and for practicing my Rust, which I am relatively new to!).
As part of the learning (and because I have a background in media management, involving very large files) I am determined to make it perform efficiently across the cores of my machine. To this end, I am using the Tokio runtime crate and deadqueue as my queue implementation.
So, my worker thread code looks like this (worker_count is currently initialised to the core count as given by num_cores, that will be customisable from commandline options):
println!("Starting up {} worker threads...", worker_count);
for n in 0..worker_count {
let rxq = queue.clone();
tokio::spawn(async move {
eprintln!("INFO Thread {}", n);
let _ = io::stderr().flush();
loop {
let ff = rxq.pop().await;
eprintln!("Popped {}", ff.fullpath().display());
process_item(ff).await;
}
});
}
and the process_item function is defined thusly:
async fn process_item(mut item:FoundFile) -> () {
match item.calculate_sha().await {
Ok(_)=>{}
Err(e)=>{
println!("ERROR Could not checksum: {}", e)
}
}
()
}
(there is a seperate thread created with spawn_blocking which performs the async scan and pushes FoundFile objects onto the queue)
calculate_sha() is defined in the impl FoundFile block:
pub async fn calculate_sha(&mut self) -> Result<String, Box<dyn Error> > {
let mut file = tokio::fs::File::open(self.fullpath()).await?;
let mut hasher = Box::new(Sha256::new());
//let mut buffer: [u8; BUFFER_SIZE] = [0; BUFFER_SIZE];
let mut buffer = vec![0_u8; BUFFER_SIZE];
eprintln!("SHA calculation on {}", self.fullpath().display());
while file.read(&mut buffer).await? > 0 {
hasher.update(&buffer);
}
let final_result = encode(hasher.finalize());
self.sha = Some(final_result.clone());
Ok( final_result )
}
which is basically the same as I would do it in C, C++, Scala etc. - open the file, create a hasher, set up a local buffer for 1 chunk, then repeatedly fill the buffer from the file and push into the hasher (updating the hasher state) until we run out of data in the file. That way my memory requirement should be bounded to 1 chunk and the overhead of the hasher state. RAII should ensure that everything gets cleaned up. (The encode function is hex::encode btw - not that that's relevant here!)
The compiler accepts all this fine, but when I actually try to run it I get the rather unhelpful:
Starting up 8 worker threads...
thread 'tokio-runtime-worker' has overflowed its stack
fatal runtime error: stack overflow
Commenting out the file.read() / hasher.update() block means it runs fine (though without actually doing anything, of course!).
I am at a loss to see what could be causing the overflow, though. I am not recursing anything, just repeatedly calling hasher.update with a local buffer. As you can see from the above code, I've tried ensuring that both hasher and buffer are on the heap by using Box and Vec respectively - and ensured that buffer is initialised - but nothing other than commenting out the loop seems to make any difference.
The oddest thing is that when I trace execution using eprintln! statements (making the assumption that STDERR output is not buffered, as with other langs I've used) then I don't even see the INFO Thread {n} line.
There is obviously something more subtle going on under the hood but I am at a bit of a loss as to what it is - can someone who knows Tokio better than me please shed some light on it?
Thanks a lot!
There isn't some unbounded stack usage in your code, it just needs a bigger stack.
There isn't a portable way to increase the main thread's stack size (though it is possible with linker arguments), but for spawned threads, Rust provides std::thread::Builder::stack_size().
Your threads, however, are tokio workers, so you need another way. Luckily, tokio also provides a way to configure the stack size of workers, by using tokio::runtime::Builder::thread_stack_size() when you build the runtime.
Also note that optimized builds usually use less stack storage, and your code may work on them unmodified.