multithreaded_work_queue_with_priorities.rs

use std::collections::BinaryHeap;
use std::sync::{
    mpsc::{channel, Receiver, Sender},
    Arc, Mutex,
};
use std::thread;
use std::time::Duration;

#[derive(Debug, Ord, PartialOrd, PartialEq, Eq)]
enum Task {
    A,
    B,
    C,
}

type PriorityQueue = Arc<Mutex<BinaryHeap<(u32, Task)>>>;

fn worker(todo: PriorityQueue, results: Sender<String>) {
    loop {
        let task = {
            let mut queue = todo.lock().unwrap();
            if let Some((_priority, task)) = queue.pop() {
                task
            } else {
                break;
            }
        };
        let answer = match task {
            Task::A => {
                thread::sleep(Duration::from_secs(2));
                "apple".to_string()
            }
            Task::B => {
                thread::sleep(Duration::from_secs(1));
                "banana".to_string()
            }
            Task::C => "cherry".to_string(),
        };

        results.send(answer).unwrap();
    }
}

fn main() {
    let num_workers = 2;
    let (result_tx, result_rx): (Sender<String>, Receiver<String>) = channel();

    let queue = Arc::new(Mutex::new(BinaryHeap::new()));

    {
        let mut queue = queue.lock().unwrap();
        queue.push((0, Task::A));
        queue.push((0, Task::B));
        queue.push((0, Task::C));
        queue.push((1, Task::A));
    }

    let mut workers = vec![];

    for _ in 0..num_workers {
        let todo = queue.clone();
        let tx = result_tx.clone();

        let handle = thread::spawn(move || {
            worker(todo, tx);
        });

        workers.push(handle);
    }
    
    {
        let mut queue = queue.lock().unwrap();
        queue.push((0, Task::A));
        queue.push((2, Task::B));
        queue.push((2, Task::C));
        queue.push((1, Task::A));
    }


    for handle in workers {
        handle.join().unwrap();
    }

    drop(result_tx);
    
    while let Ok(answer) = result_rx.recv() {
        println!("{answer}");
    }
}