Lock-Free Data Structure Fundamentals

A lock-free data structure coordinates concurrent state transitions through hardware atomic instructions (compare_exchange_weak, fetch_add, Acquire/Release memory orderings) rather than mutual exclusion locks, guaranteeing system-wide progress: even if an arbitrary thread is suspended mid-operation by the OS scheduler, other threads continue completing operations.

Treiber Stack and SPSC Ring Buffer Architecture

1. Lock-Free Treiber Push (AtomicPtr<Node<T>>):
   loop {
       old_head = head.load(Relaxed);
       new_node.next = old_head;
       if head.compare_exchange_weak(old_head, new_node, Release, Relaxed).is_ok() {
           break;
       }
   }

2. Lock-Free Single-Producer Single-Consumer (SPSC) Ring Buffer:
   Producer owns `tail` (writes slot, then stores `tail + 1` with Release)
   Consumer owns `head` (loads `tail` with Acquire, reads slot, stores `head + 1` with Release)

Complete Lock-Free SPSC Ring Buffer and Treiber Stack Implementation

use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::ptr;
use std::sync::Arc;
use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
use std::thread;

pub struct SpscRingBuffer<T, const N: usize> {
    buffer: [UnsafeCell<MaybeUninit<T>>; N],
    head: AtomicUsize,
    tail: AtomicUsize,
}

unsafe impl<T: Send, const N: usize> Sync for SpscRingBuffer<T, N> {}
unsafe impl<T: Send, const N: usize> Send for SpscRingBuffer<T, N> {}

impl<T, const N: usize> SpscRingBuffer<T, N> {
    pub const fn new() -> Self {
        assert!(N >= 2 && N.is_power_of_two(), "N must be a power of two >= 2");
        Self {
            // SAFETY: An array of `UnsafeCell<MaybeUninit<T>>` needs no initialization.
            buffer: unsafe { MaybeUninit::uninit().assume_init() },
            head: AtomicUsize::new(0),
            tail: AtomicUsize::new(0),
        }
    }

    pub fn try_push(&self, value: T) -> Result<(), T> {
        let tail = self.tail.load(Ordering::Relaxed);
        let head = self.head.load(Ordering::Acquire);
        if tail.wrapping_sub(head) == N {
            return Err(value);
        }
        let slot = tail & (N - 1);
        // SAFETY: Producer uniquely owns `slot` while `tail - head < N`.
        unsafe {
            (*self.buffer[slot].get()).write(value);
        }
        self.tail.store(tail.wrapping_add(1), Ordering::Release);
        Ok(())
    }

    pub fn try_pop(&self) -> Option<T> {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Acquire);
        if head == tail {
            return None;
        }
        let slot = head & (N - 1);
        // SAFETY: Consumer uniquely owns `slot` after `Acquire` load observed `head != tail`.
        let value = unsafe { (*self.buffer[slot].get()).assume_init_read() };
        self.head.store(head.wrapping_add(1), Ordering::Release);
        Some(value)
    }
}

impl<T, const N: usize> Drop for SpscRingBuffer<T, N> {
    fn drop(&mut self) {
        while self.try_pop().is_some() {}
    }
}

struct Node<T> {
    value: T,
    next: *mut Node<T>,
}

pub struct AppendOnlyTreiberBag<T> {
    head: AtomicPtr<Node<T>>,
}

impl<T> AppendOnlyTreiberBag<T> {
    pub const fn new() -> Self {
        Self {
            head: AtomicPtr::new(ptr::null_mut()),
        }
    }

    pub fn push(&self, value: T) {
        let new_node = Box::into_raw(Box::new(Node {
            value,
            next: ptr::null_mut(),
        }));
        let mut current = self.head.load(Ordering::Relaxed);
        loop {
            // SAFETY: `new_node` is uniquely owned by this thread until CAS succeeds.
            unsafe {
                (*new_node).next = current;
            }
            match self.head.compare_exchange_weak(
                current,
                new_node,
                Ordering::Release,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(actual) => current = actual,
            }
        }
    }

    pub fn into_vec(mut self) -> Vec<T> {
        // Exclusive `&mut self` via ownership transfer eliminates ABA reclamation hazards on drain
        let mut curr = *self.head.get_mut();
        *self.head.get_mut() = ptr::null_mut();
        let mut out = Vec::new();
        while !curr.is_null() {
            // SAFETY: `curr` was allocated via `Box::into_raw` in `push`.
            let boxed = unsafe { Box::from_raw(curr) };
            out.push(boxed.value);
            curr = boxed.next;
        }
        out
    }
}

impl<T> Drop for AppendOnlyTreiberBag<T> {
    fn drop(&mut self) {
        let mut curr = *self.head.get_mut();
        while !curr.is_null() {
            unsafe {
                let boxed = Box::from_raw(curr);
                curr = boxed.next;
            }
        }
    }
}

fn main() {
    // 1. Lock-free SPSC ring buffer transferring 1,000 items between producer and consumer
    let ring = Arc::new(SpscRingBuffer::<u64, 64>::new());
    let ring_prod = Arc::clone(&ring);
    let ring_cons = Arc::clone(&ring);

    let producer = thread::spawn(move || {
        for i in 1..=1000 {
            let mut val = i;
            while let Err(returned) = ring_prod.try_push(val) {
                val = returned;
                thread::yield_now();
            }
        }
    });

    let consumer = thread::spawn(move || {
        let mut sum = 0u64;
        let mut received = 0usize;
        while received < 1000 {
            if let Some(v) = ring_cons.try_pop() {
                sum += v;
                received += 1;
            } else {
                thread::yield_now();
            }
        }
        sum
    });

    producer.join().unwrap();
    let spsc_sum = consumer.join().unwrap();
    println!("SPSC ring buffer sum (1..=1000) = {}", spsc_sum);

    // 2. Lock-free multi-producer Treiber bag
    let bag = Arc::new(AppendOnlyTreiberBag::new());
    thread::scope(|s| {
        for t in 0..4 {
            let b = Arc::clone(&bag);
            s.spawn(move || {
                for i in 0..25 {
                    b.push(t * 100 + i);
                }
            });
        }
    });

    let items = Arc::try_unwrap(bag).ok().unwrap().into_vec();
    println!(
        "Treiber bag collected {} items across 4 lock-free producer threads",
        items.len()
    );
}

SPSC ring buffer sum (1..=1000) = 500500
Treiber bag collected 100 items across 4 lock-free producer threads