lf-queue

A lock-free multi-producer multi-consumer unbounded queue.

Examples

[dependencies]
lf-queue = "0.1"

Single Producer - Single Consumer:

use lf_queue::Queue;

fn main() {
    const COUNT: usize = 1_000;
    let queue: Queue<usize> = Queue::new();

    for i in 0..COUNT {
        queue.push(i);
    }

    for i in 0..COUNT {
        assert_eq!(i, queue.pop().unwrap());
    }

    assert!(queue.pop().is_none());
}

Multi Producer - Single Consumer:

use lf_queue::Queue;
use std::thread;

fn main() {
    const COUNT: usize = 1_000;
    const CONCURRENCY: usize = 4;

    let queue: Queue<usize> = Queue::new();

    let ths: Vec<_> = (0..CONCURRENCY)
        .map(|_| {
            let q = queue.clone();
            thread::spawn(move || {
                for i in 0..COUNT {
                    q.push(i);
                }
            })
        })
        .collect();

    for th in ths {
        th.join().unwrap();
    }

    for _ in 0..COUNT * CONCURRENCY {
        assert!(queue.pop().is_some());
    }

    assert!(queue.pop().is_none());
}

Single Producer - Multi Consumer:

use lf_queue::Queue;
use std::thread;

fn main() {
    const COUNT: usize = 1_000;
    const CONCURRENCY: usize = 4;

    let queue: Queue<usize> = Queue::new();

    for i in 0..COUNT * CONCURRENCY {
        queue.push(i);
    }

    let ths: Vec<_> = (0..CONCURRENCY)
        .map(|_| {
            let q = queue.clone();
            thread::spawn(move || {
                for _ in 0..COUNT {
                    loop {
                        if q.pop().is_some() {
                            break;
                        }
                    }
                }
            })
        })
        .collect();

    for th in ths {
        th.join().unwrap();
    }

    assert!(queue.pop().is_none());
}

Multi Producer - Multi Consumer:

use lf_queue::Queue;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;

fn main() {
    const COUNT: usize = 1_000;
    const CONCURRENCY: usize = 4;

    let queue: Queue<usize> = Queue::new();
    let items = Arc::new((0..COUNT).map(|_| AtomicUsize::new(0)).collect::<Vec<_>>());

    let ths: Vec<_> = (0..CONCURRENCY)
        .map(|_| {
            let q = queue.clone();
            let its = items.clone();
            thread::spawn(move || {
                for _ in 0..COUNT {
                    let n = loop {
                        if let Some(x) = q.pop() {
                            break x;
                        } else {
                            thread::yield_now();
                        }
                    };
                    its[n].fetch_add(1, Ordering::SeqCst);
                }
            })
        })
        .map(|_| {
            let q = queue.clone();
            thread::spawn(move || {
                for i in 0..COUNT {
                    q.push(i);
                }
            })
        })
        .collect();

    for th in ths {
        th.join().unwrap();
    }

    thread::sleep(std::time::Duration::from_millis(10));

    for c in &*items {
        assert_eq!(c.load(Ordering::SeqCst), CONCURRENCY);
    }

    assert!(queue.pop().is_none());
}

Acknowledgement

This implementation of a lock-free queue in Rust took inspiration from the concurrent-queue crate and aims to be used for educational purposes. The code documentation help you to discover the algorithm used to implement a concurrent lock-free queue in Rust, but might not yet be beginner-friendly. More details and learning materials will be added over time.

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, shall be licensed as above, without any additional terms or conditions.

Note that, as of now, my focus is on improving the documentation of this crate, not adding any additional feature. Please open an issue and start a discussion before working on any significant PR.

Contributions are welcome.