Concurrent Work Queue

A bounded concurrent priority work queue coordinates multiple producer and consumer threads using a single Mutex<QueueState<T>> paired with two condition variables (not_empty and not_full), enforcing backpressure when the queue reaches capacity and ordering queued items by priority before FIFO sequence number.

Producer-Consumer Synchronization State Machine

Producers: push(item, priority)           Consumers: pop() -> Option<T>
       |                                         |
       v                                         v
  Lock Mutex<State>                         Lock Mutex<State>
  While heap.len() == cap && !closed:       While heap.is_empty() && !closed:
    not_full.wait(guard)                      not_empty.wait(guard)
  Push into BinaryHeap                      Pop highest-priority item
  not_empty.notify_one()                    not_full.notify_one()

Complete Implementation

use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;

struct PrioritizedItem<T> {
    priority: u8,
    seq: u64,
    item: T,
}

impl<T> PartialEq for PrioritizedItem<T> {
    fn eq(&self, other: &Self) -> bool {
        self.priority == other.priority && self.seq == other.seq
    }
}

impl<T> Eq for PrioritizedItem<T> {}

impl<T> PartialOrd for PrioritizedItem<T> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl<T> Ord for PrioritizedItem<T> {
    fn cmp(&self, other: &Self) -> Ordering {
        // Higher `priority` first; ties broken by smaller `seq` (FIFO order)
        self.priority
            .cmp(&other.priority)
            .then_with(|| other.seq.cmp(&self.seq))
    }
}

struct QueueState<T> {
    heap: BinaryHeap<PrioritizedItem<T>>,
    next_seq: u64,
    closed: bool,
}

pub struct BoundedWorkQueue<T> {
    capacity: usize,
    state: Mutex<QueueState<T>>,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedWorkQueue<T> {
    pub fn with_capacity(capacity: usize) -> Self {
        assert!(capacity > 0, "capacity must be non-zero");
        Self {
            capacity,
            state: Mutex::new(QueueState {
                heap: BinaryHeap::with_capacity(capacity),
                next_seq: 0,
                closed: false,
            }),
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    pub fn push(&self, item: T, priority: u8) -> Result<(), T> {
        let mut guard = self.state.lock().unwrap();
        while guard.heap.len() == self.capacity && !guard.closed {
            guard = self.not_full.wait(guard).unwrap();
        }
        if guard.closed {
            return Err(item);
        }
        let seq = guard.next_seq;
        guard.next_seq += 1;
        guard.heap.push(PrioritizedItem {
            priority,
            seq,
            item,
        });
        self.not_empty.notify_one();
        Ok(())
    }

    pub fn pop(&self) -> Option<(T, u8)> {
        let mut guard = self.state.lock().unwrap();
        while guard.heap.is_empty() && !guard.closed {
            guard = self.not_empty.wait(guard).unwrap();
        }
        let popped = guard.heap.pop().map(|entry| (entry.item, entry.priority));
        if popped.is_some() {
            self.not_full.notify_one();
        }
        popped
    }

    pub fn close(&self) {
        let mut guard = self.state.lock().unwrap();
        guard.closed = true;
        self.not_empty.notify_all();
        self.not_full.notify_all();
    }
}

fn main() {
    let queue = Arc::new(BoundedWorkQueue::<&'static str>::with_capacity(8));

    // Enqueue mixed-priority jobs before starting consumers to demonstrate deterministic ordering
    let jobs = [
        ("batch-backup", 10),
        ("payment-webhook", 90),
        ("audit-log", 10),
        ("fraud-alert", 99),
        ("invoice-pdf", 50),
        ("auth-token-refresh", 90),
    ];

    for (name, prio) in jobs {
        queue.push(name, prio).unwrap();
    }
    queue.close();

    let results = Arc::new(Mutex::new(Vec::new()));
    thread::scope(|s| {
        for worker_id in 0..2 {
            let q = Arc::clone(&queue);
            let out = Arc::clone(&results);
            s.spawn(move || {
                while let Some((job, prio)) = q.pop() {
                    out.lock().unwrap().push((worker_id, prio, job));
                }
            });
        }
    });

    let mut processed = results.lock().unwrap().clone();
    processed.sort_by(|a, b| b.1.cmp(&a.1));
    println!("Processed jobs in priority order:");
    for (worker_id, prio, job) in processed {
        println!("  [prio={:>2}] {:<20} (handled by worker {})", prio, job, worker_id);
    }
}

Processed jobs in priority order:
  [prio=99] fraud-alert          (handled by worker 0)
  [prio=90] payment-webhook      (handled by worker 0)
  [prio=90] auth-token-refresh   (handled by worker 1)
  [prio=50] invoice-pdf          (handled by worker 0)
  [prio=10] batch-backup         (handled by worker 1)
  [prio=10] audit-log            (handled by worker 0)