How to use Condvar to limit multithreading?

Question

I am trying to use Condvar to limit the number of threads that are active at any given time, and have difficulty finding good examples of how to use Condvar. So far I:

use std::sync::{Arc, Mutex, Condvar};
use std::thread;

fn main() {
    let thread_count_arc = Arc::new((Mutex::new(0), Condvar::new()));
    let mut i = 0;
    while i < 100
    {
        let thread_count = thread_count_arc.clone();
        thread::spawn(move || {
            let &(ref num, ref cvar) = &*thread_count;
            {
                let mut start = num.lock().unwrap();
                if *start >= 20
                {
                    cvar.wait(start);
                }
                *start += 1;
            }
            println!("hello");
            cvar.notify_one();
        });
        i += 1;
    }
}

Indicated compiler error:

<anon>:18:18: 18:23 error: use of moved value: `start`
<anon>:18                 *start += 1;
                           ^~~~~
<anon>:16:31: 16:36 note: `start` moved here because it has type `std::sync::mutex::MutexGuard<'_, i32>`, which is non-copyable
<anon>:16                     cvar.wait(start);
                                        ^~~~~

I am completely unsure if I am using it correctly Condvar. I tried to stay as close as possible to the example in the Rust API. Does anyone have any suggestions on the right way to implement this?

Answer 1

Here is the version that compiles:

use std::sync::{Arc, Mutex, Condvar};
use std::thread;

fn main() {
    let thread_count_arc = Arc::new((Mutex::new(0u8), Condvar::new()));
    let mut i = 0;
    while i < 100 {
        let thread_count = thread_count_arc.clone();
        thread::spawn(move || {
            let &(ref num, ref cvar) = &*thread_count;
            {
                let mut start = num.lock().unwrap();
                while *start >= 20 {
                    start = cvar.wait(start).unwrap()
                }
                *start += 1;
            }
            println!("hello");
            cvar.notify_one();
        });
        i += 1;
    }
}

The important part can be seen from the waitAPI :

fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> LockResult<MutexGuard<'a, T>>

, wait guard, , - start, !

, condvars - ( start), , .

, , wait. if while, dbaupp.

Answer 2

- Semaphore:

let sema = Arc::new(Semaphore::new(20)); 

for i in 0..100 {
    let sema = sema.clone();
    thread::spawn(move || {
        let _guard = sema.acquire();
        println!("{}", i);
    })
}

(, : , .)

, Semaphore #[unstable], , -.

Answer 3

This is just editing Shepmaster code. I realized that the code I provided did not do exactly what I wanted, so I put it here for future reference.

use std::sync::{Arc, Mutex, Condvar};
use std::thread;

fn main() {
    let thread_count_arc = Arc::new((Mutex::new(0u8), Condvar::new()));
    let mut i = 0;
    while i < 150 {
        let thread_count = thread_count_arc.clone();
        thread::spawn(move || {
            let x;
            let &(ref num, ref cvar) = &*thread_count;
            {
                let start = num.lock().unwrap();
                let mut start =
                    if *start >= 20 {
                        cvar.wait(start).unwrap()
                    } else {
                        start
                    };
                *start += 1;
                x = *start;
            }
            println!("{}", x);
            {
                let mut counter = num.lock().unwrap();
                *counter -= 1;
            }
            cvar.notify_one();
        });
        i += 1;
    }
    println!("done");
}

Again, not my code. Doing this in the playpen should show more or less expected behavior.