Multithreaded list iteration while using Mutex to prevent dealing with the same type at the same time

user_665794 0
Viewed 57

I am writing an application that needs to run on many threads at the same time. It will process a long list of items where one property of each item is a user_id. I am trying to make sure that items belonging to the same user_id are never processed at the same time. This means that the closure running the sub threads needs to wait until no other thread is processing data for the same user.

I do not understand how to solve this. My simplified, current example, looks like this:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use threadpool::ThreadPool;

fn main() {
    let pool = ThreadPool::new(num_cpus::get());

    let mut locks: HashMap<String, Mutex<bool>> = HashMap::new();
    let queue = Arc::new(vec![
        "1".to_string(),
        "1".to_string(),
        "2".to_string(),
        "1".to_string(),
        "3".to_string(),
    ]);
    let count = queue.len();

    for i in 0..count {
        let user_id = queue[i].clone();
        // Problem: cannot borrow `locks` as mutable more than once at a time
        // mutable borrow starts here in previous iteration of loop
        let lock = locks.entry(user_id).or_insert(Mutex::new(true));

        pool.execute(move || {
            // Wait until the user_id becomes free.
            lock.lock().unwrap();

            // Do stuff with user_id, but never process
            // the same user_id more than once at the same time.
            println!("{:?}", user_id);
        });
    }

    pool.join();
}

I am trying to keep a list of Mutex which I then use to wait for the user_id to become free, but the borrow checker does not allow this. The queue items and the item process code is much more complex in the actual application I am working on.

I am not allowed to change the order of the items in the queue (but some variations will be allowed because of waiting for the lock).

How to solve this scenario?

borrow-checker rust multithreading
1 Answers

First of all, HashMap::entry() consumes the key, so since you want to use it in the closure as well, you'll need to clone it, i.e. .entry(user_id.clone()).

Since you need to share the Mutex<bool> between the main thread and worker threads, then you need to likewise wrap that in an Arc. You can also use Entry::or_insert_with(), so you avoid needlessly creating a new Mutex unless needed.

let mut locks: HashMap<String, Arc<Mutex<bool>>> = HashMap::new();
// ...
let lock = locks
    .entry(user_id.clone())
    .or_insert_with(|| Arc::new(Mutex::new(true)))
    .clone();

Lastly, you must store the guard returned by lock(), otherwise it is immediately released.

let _guard = lock.lock().unwrap();

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use threadpool::ThreadPool;

fn main() {
    let pool = ThreadPool::new(num_cpus::get());

    let mut locks: HashMap<String, Arc<Mutex<bool>>> = HashMap::new();
    let queue = Arc::new(vec![
        "1".to_string(),
        "1".to_string(),
        "2".to_string(),
        "1".to_string(),
        "3".to_string(),
    ]);
    let count = queue.len();

    for i in 0..count {
        let user_id = queue[i].clone();

        let lock = locks
            .entry(user_id.clone())
            .or_insert_with(|| Arc::new(Mutex::new(true)))
            .clone();

        pool.execute(move || {
            // Wait until the user_id becomes free.
            let _guard = lock.lock().unwrap();

            // Do stuff with user_id, but never process
            // the same user_id more than once at the same time.
            println!("{:?}", user_id);
        });
    }

    pool.join();
}
user_24708180
Related