How to loop over thread handles and join if finished, within another loop?

Question

I have a program that creates threads in a loop, and also checks if they have finished and cleans them up if they have. See below for a minimal example:

use std::thread;

fn main() {    

    let mut v = Vec::<std::thread::JoinHandle<()>>::new();
    for _ in 0..10 {
        let jh = thread::spawn(|| {
            thread::sleep(std::time::Duration::from_secs(1));
        });
        v.push(jh);
        for jh in v.iter_mut() {
            if jh.is_finished() {
                jh.join().unwrap();
            }
        } 
    }
}

This gives the error:

error[E0507]: cannot move out of `*jh` which is behind a mutable reference
    --> src\main.rs:13:17
     |
13   |                 jh.join().unwrap();
     |                 ^^^------
     |                 |  |
     |                 |  `*jh` moved due to this method call
     |                 move occurs because `*jh` has type `JoinHandle<()>`, which does not implement the `Copy` trait
     |
note: this function takes ownership of the receiver `self`, which moves `*jh`
    --> D:\rust\.rustup\toolchains\stable-x86_64-pc-windows-msvc\lib/rustlib/src/rust\library\std\src\thread\mod.rs:1461:17
     |
1461 |     pub fn join(self) -> Result<T> {

How can I get the borrow checker to allow this?

Answer 1

JoinHandle::join actually consumes the JoinHandle. iter_mut(), however, only borrows the elements of the vector and keeps the vector alive. Therefore your JoinHandles are only borrowed, and you cannot call consuming methods on borrowed objects.

What you need to do is to take the ownership of the elements while iterating over the vector, so they can be then consumed by join(). This is achieved by using into_iter() instead of iter_mut().

The second mistake is that you (probably accidentally) wrote the two for loops inside of each other, while they should be independent loops.

The third problem is a little more complex. You cannot check if a thread has finished and then join it the way you did. Therefore I removed the is_finished() check for now and will talk about this further down again.

Here is your fixed code:

use std::thread;

fn main() {
    let mut v = Vec::<std::thread::JoinHandle<()>>::new();
    for _ in 0..10 {
        let jh = thread::spawn(|| {
            thread::sleep(std::time::Duration::from_secs(1));
        });
        v.push(jh);
    }

    for jh in v.into_iter() {
        jh.join().unwrap();
    }
}

---

Reacting to finished threads

This one is harder. If you just want to wait until all of them are finished, the code above is the way to go.

However, if you have to react to finished threads right away, you basically have to set up some kind of event propagation. You don't want to loop over all threads over and over again until they are all finished, because that is something called idle-waiting and consumes a lot of computational power.

So if you want to achieve that there are two problems that have to be dealt with:

• join() consumes the JoinHandle(), which would leave behind an incomplete Vec of JoinHandles. This isn't possible, so we need to wrap JoinHandle in a type that can actually be ripped out of the vector partially, like Option.
• we need a way to signal to the main thread that a new child thread is finished, so that the main thread doesn't have to continuously iterate over the threads.

All in all this is very complex and tricky to implement.

Here is my attempt:

use std::{
    thread::{self, JoinHandle},
    time::Duration,
};

fn main() {
    let mut v: Vec<Option<JoinHandle<()>>> = Vec::new();
    let (send_finished_thread, receive_finished_thread) = std::sync::mpsc::channel();

    for i in 0..10 {
        let send_finished_thread = send_finished_thread.clone();

        let join_handle = thread::spawn(move || {
            println!("Thread {} started.", i);

            thread::sleep(Duration::from_millis(2000 - i as u64 * 100));

            println!("Thread {} finished.", i);

            // Signal that we are finished.
            // This will wake up the main thread.
            send_finished_thread.send(i).unwrap();
        });
        v.push(Some(join_handle));
    }

    loop {
        // Check if all threads are finished
        let num_left = v.iter().filter(|th| th.is_some()).count();
        if num_left == 0 {
            break;
        }

        // Wait until a thread is finished, then join it
        let i = receive_finished_thread.recv().unwrap();
        let join_handle = std::mem::take(&mut v[i]).unwrap();
        println!("Joining {} ...", i);
        join_handle.join().unwrap();
        println!("{} joined.", i);
    }

    println!("All joined.");
}

Important

This code is just a demonstration. It will deadlock if one of the threads panic. But this shows how complicated that problem is.

It could be solved by utilizing a drop guard, but I think this answer is convoluted enough ;)