Ring Buffers and Lock-Free Queues

This lesson builds the queues that low-latency Java systems use between threads: bounded, array-based, and free of locks. It contains Project 1 of the course, a single-producer single-consumer (SPSC) queue built in three stages. It uses VarHandle access modes from VarHandle, Access Modes, and Fences and CAS from Atomic Classes and CAS.

Idea	Short meaning
Ring buffer	A fixed array used as a circle: indices wrap around to 0
Sequence	An ever-increasing `long` counter; the slot is `sequence & mask`
Single-writer principle	Each variable has exactly one thread that writes it
SPSC / MPSC	Single-producer single-consumer / multi-producer single-consumer
Cached index	A thread's private copy of the other thread's counter, refreshed only when needed
False sharing	Two threads writing different variables on the same cache line slow each other down
Wait strategy	What a thread does when the queue is empty or full: spin, yield, or park

Example 1: power-of-two masking

A ring buffer of capacity 2^k maps a sequence number to a slot with sequence & (capacity - 1). The bitwise AND is one cycle, while % is an integer division that costs tens of cycles, and the AND stays correct when the sequence is a large long.

Java

public class Masking {
    static int roundUpToPowerOfTwo(int n) {
        return 1 << (32 - Integer.numberOfLeadingZeros(n - 1));
    }

    public static void main(String[] args) {
        System.out.println("capacity for 1000 requested = " + roundUpToPowerOfTwo(1000));

        int capacity = 8;
        long mask = capacity - 1;
        long[] sequences = {0, 7, 8, 9, 15, 16, 1000};
        for (long seq : sequences) {
            System.out.println("seq " + seq + " -> slot " + (seq & mask));
        }
    }
}

Output

capacity for 1000 requested = 1024
seq 0 -> slot 0
seq 7 -> slot 7
seq 8 -> slot 0
seq 9 -> slot 1
seq 15 -> slot 7
seq 16 -> slot 0
seq 1000 -> slot 0

Queues use two sequences that never wrap: tail (next slot to write) and head (next slot to read). The queue is empty when head == tail and full when tail - head == capacity. A long incremented a billion times per second takes about 292 years to overflow, so overflow is ignored.

Example 2: a single-threaded ring buffer

Before adding threads, the queue logic on its own:

Java

public class RingBuffer {
    private final Object[] buffer;
    private final int mask;
    private long head, tail;

    RingBuffer(int capacity) {
        buffer = new Object[capacity];   // capacity must be a power of two
        mask = capacity - 1;
    }

    boolean offer(Object e) {
        if (tail - head == buffer.length) return false;   // full
        buffer[(int) (tail & mask)] = e;
        tail++;
        return true;
    }

    Object poll() {
        if (head == tail) return null;                    // empty
        int i = (int) (head & mask);
        Object e = buffer[i];
        buffer[i] = null;                                 // let the GC collect it
        head++;
        return e;
    }

    public static void main(String[] args) {
        RingBuffer rb = new RingBuffer(4);
        for (int i = 1; i <= 5; i++) System.out.println("offer " + i + ": " + rb.offer(i));
        System.out.println("poll: " + rb.poll() + ", " + rb.poll());
        System.out.println("offer 6: " + rb.offer(6) + ", offer 7: " + rb.offer(7));
        Object e;
        StringBuilder rest = new StringBuilder();
        while ((e = rb.poll()) != null) rest.append(e).append(' ');
        System.out.println("drain: " + rest.toString().trim());
    }
}

Output

offer 1: true
offer 2: true
offer 3: true
offer 4: true
offer 5: false
poll: 1, 2
offer 6: true, offer 7: true
drain: 3 4 6 7

offer writes only tail and the slot, and poll writes only head and the slot it frees. That split is what makes the concurrent version possible.

The single-writer principle

A variable written by one thread only never needs CAS: there is no competing writer to lose against. The writer publishes new values with a release write, and other threads read them with an acquire read. In an SPSC queue the producer is the only writer of tail and the consumer is the only writer of head. The cost of each operation becomes one release store, which on x86 is a plain mov, compared with a locked instruction for CAS or a volatile store.

Example 3: Project 1, stage 1, SPSC with AtomicLong

The first correct concurrent version keeps head and tail in AtomicLongs. The producer offers the numbers 1 to 1,000,000 and the consumer sums them.

Java

import java.util.concurrent.atomic.AtomicLong;

public class SpscStage1 {
    static final class SpscQueue<E> {
        private final Object[] buffer;
        private final int mask;
        private final AtomicLong head = new AtomicLong();
        private final AtomicLong tail = new AtomicLong();

        SpscQueue(int capacity) {
            buffer = new Object[capacity];
            mask = capacity - 1;
        }

        boolean offer(E e) {
            long t = tail.get();
            if (t - head.get() == buffer.length) return false;
            buffer[(int) (t & mask)] = e;
            tail.set(t + 1);                // volatile write publishes the slot
            return true;
        }

        @SuppressWarnings("unchecked")
        E poll() {
            long h = head.get();
            if (h == tail.get()) return null;   // volatile read
            int i = (int) (h & mask);
            E e = (E) buffer[i];
            buffer[i] = null;
            head.set(h + 1);                // frees the slot for the producer
            return e;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final int n = 1_000_000;
        SpscQueue<Integer> q = new SpscQueue<>(1024);
        long[] result = new long[2];

        Thread consumer = new Thread(() -> {
            long count = 0, sum = 0;
            while (count < n) {
                Integer v = q.poll();
                if (v == null) { Thread.onSpinWait(); continue; }
                count++;
                sum += v;
            }
            result[0] = count;
            result[1] = sum;
        });
        consumer.start();
        for (int i = 1; i <= n; i++) {
            while (!q.offer(i)) Thread.onSpinWait();
        }
        consumer.join();
        System.out.println("received " + result[0] + ", sum " + result[1]);
    }
}

Output

received 1000000, sum 500000500000

The slot write happens before tail.set in program order, and tail.set happens-before the consumer's tail.get that sees it, so the consumer sees the element. The same reasoning in reverse protects the slot the consumer frees. The program is correct, but every set is a full volatile store with a StoreLoad barrier, which is more than the algorithm needs.

Example 4: Project 1, stage 2, release and acquire

Each counter has one writer, so the owner can read its own counter with a plain read and publish with setRelease. The other side reads with getAcquire. Only the queue class changes; main stays as in stage 1.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;

public class SpscStage2 {
    static final class SpscQueue<E> {
        private final Object[] buffer;
        private final int mask;
        private long head;   // written only by the consumer
        private long tail;   // written only by the producer

        private static final VarHandle HEAD, TAIL;
        static {
            try {
                var lookup = MethodHandles.lookup();
                HEAD = lookup.findVarHandle(SpscQueue.class, "head", long.class);
                TAIL = lookup.findVarHandle(SpscQueue.class, "tail", long.class);
            } catch (ReflectiveOperationException e) {
                throw new ExceptionInInitializerError(e);
            }
        }

        SpscQueue(int capacity) {
            buffer = new Object[capacity];
            mask = capacity - 1;
        }

        boolean offer(E e) {
            long t = tail;                                   // own counter: plain
            if (t - (long) HEAD.getAcquire(this) == buffer.length) return false;
            buffer[(int) (t & mask)] = e;
            TAIL.setRelease(this, t + 1);
            return true;
        }

        @SuppressWarnings("unchecked")
        E poll() {
            long h = head;                                   // own counter: plain
            if (h == (long) TAIL.getAcquire(this)) return null;
            int i = (int) (h & mask);
            E e = (E) buffer[i];
            buffer[i] = null;
            HEAD.setRelease(this, h + 1);
            return e;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final int n = 1_000_000;
        SpscQueue<Integer> q = new SpscQueue<>(1024);
        long[] result = new long[2];
        Thread consumer = new Thread(() -> {
            long count = 0, sum = 0;
            while (count < n) {
                Integer v = q.poll();
                if (v == null) { Thread.onSpinWait(); continue; }
                count++;
                sum += v;
            }
            result[0] = count;
            result[1] = sum;
        });
        consumer.start();
        for (int i = 1; i <= n; i++) {
            while (!q.offer(i)) Thread.onSpinWait();
        }
        consumer.join();
        System.out.println("received " + result[0] + ", sum " + result[1]);
    }
}

Output

received 1000000, sum 500000500000

No StoreLoad order is needed: neither side writes one counter and then must see the other side's newest value before continuing. Seeing a slightly old value is safe, because an old head only makes the queue look fuller and an old tail only makes it look emptier. The thread then retries. Both counters are still read by the other thread on every call, which moves their cache lines between cores. Stage 3 removes most of those reads.

Example 5: Project 1, stage 3, cached indices and padding

Two changes finish the queue:

Cached indices. The producer keeps cachedHead, its last view of head, and re-reads head only when the queue looks full against that cached value. The consumer does the same with cachedTail. With a 1024-slot queue the producer reads head about once per 1024 offers instead of on every offer.
Padding. If tail and head share a 64-byte cache line, every producer write invalidates the consumer's copy of the line, which is false sharing (Cache Coherence and False Sharing). Seven or more long fields on each side keep the hot fields on separate lines. HotSpot lays out superclass fields before subclass fields, so the padding uses a class hierarchy that the JVM cannot reorder.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;

public class SpscQueueProject {
    abstract static class Pad0 { long p00, p01, p02, p03, p04, p05, p06, p07; }
    abstract static class ProducerFields extends Pad0 { long tail; long cachedHead; }
    abstract static class Pad1 extends ProducerFields { long p10, p11, p12, p13, p14, p15, p16, p17; }
    abstract static class ConsumerFields extends Pad1 { long head; long cachedTail; }
    abstract static class Pad2 extends ConsumerFields { long p20, p21, p22, p23, p24, p25, p26, p27; }

    static final class SpscQueue<E> extends Pad2 {
        private final Object[] buffer;
        private final int mask;
        private final int capacity;

        private static final VarHandle HEAD, TAIL;
        static {
            try {
                var lookup = MethodHandles.lookup();
                HEAD = lookup.findVarHandle(ConsumerFields.class, "head", long.class);
                TAIL = lookup.findVarHandle(ProducerFields.class, "tail", long.class);
            } catch (ReflectiveOperationException e) {
                throw new ExceptionInInitializerError(e);
            }
        }

        SpscQueue(int requested) {
            capacity = 1 << (32 - Integer.numberOfLeadingZeros(requested - 1));
            buffer = new Object[capacity];
            mask = capacity - 1;
        }

        /** Called only by the producer thread. */
        boolean offer(E e) {
            long t = tail;
            if (t - cachedHead >= capacity) {
                cachedHead = (long) HEAD.getAcquire(this);   // refresh only when needed
                if (t - cachedHead >= capacity) return false;
            }
            buffer[(int) (t & mask)] = e;
            TAIL.setRelease(this, t + 1);
            return true;
        }

        /** Called only by the consumer thread. */
        @SuppressWarnings("unchecked")
        E poll() {
            long h = head;
            if (h >= cachedTail) {
                cachedTail = (long) TAIL.getAcquire(this);
                if (h >= cachedTail) return null;
            }
            int i = (int) (h & mask);
            E e = (E) buffer[i];
            buffer[i] = null;
            HEAD.setRelease(this, h + 1);
            return e;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final int n = 1_000_000;
        SpscQueue<Integer> q = new SpscQueue<>(1000);    // rounded up to 1024
        long[] result = new long[2];
        Thread consumer = new Thread(() -> {
            long count = 0, sum = 0;
            while (count < n) {
                Integer v = q.poll();
                if (v == null) { Thread.onSpinWait(); continue; }
                count++;
                sum += v;
            }
            result[0] = count;
            result[1] = sum;
        }, "consumer");
        consumer.start();
        for (int i = 1; i <= n; i++) {
            while (!q.offer(i)) Thread.onSpinWait();
        }
        consumer.join();
        System.out.println("capacity " + q.capacity);
        System.out.println("received " + result[0] + ", sum " + result[1]);
    }
}

Output

capacity 1024
received 1000000, sum 500000500000

Part	Purpose
`tail`, `head` with `setRelease`/`getAcquire`	Publish slots and freed slots across threads
Plain read of own counter	The owner is the only writer, so it always sees its latest value
`cachedHead`, `cachedTail`	Avoid reading the other thread's cache line on most calls
`Pad0`, `Pad1`, `Pad2`	Keep producer fields, consumer fields, and neighbouring objects on separate cache lines
`buffer[i] = null`	Release the element for garbage collection
Power-of-two capacity	Slot index by mask

The buffer array itself can still false-share at its ends with other objects, and adjacent slots share lines between producer and consumer when the queue is nearly empty. JCTools pads the array as well. @jdk.internal.vm.annotation.Contended, covered in Cache Coherence and False Sharing, is the alternative: application code needs --add-exports java.base/jdk.internal.vm.annotation=ALL-UNNAMED to compile and -XX:-RestrictContended for the JVM to apply it.

Extension tasks:

Add drain(Consumer<E> handler, int limit) that reads tail once and processes a batch, publishing head once at the end.
Store primitive long values in a long[] instead of Integer objects so the hot path allocates nothing.
Add a size() method and state in its Javadoc why the result is only an estimate.

Example 6: MPSC queue with CAS on the tail

With several producers, tail has several writers, so the single-writer principle no longer holds. Producers claim a slot with a CAS on tail, then publish the element into the slot with a release write. The consumer treats a null slot as "not yet published".

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;

public class MpscDemo {
    static final class MpscQueue<E> {
        private final Object[] buffer;
        private final int mask;
        private volatile long tail;   // CAS by producers
        private volatile long head;   // written by the single consumer

        private static final VarHandle TAIL, HEAD;
        private static final VarHandle SLOT = MethodHandles.arrayElementVarHandle(Object[].class);
        static {
            try {
                var lookup = MethodHandles.lookup();
                TAIL = lookup.findVarHandle(MpscQueue.class, "tail", long.class);
                HEAD = lookup.findVarHandle(MpscQueue.class, "head", long.class);
            } catch (ReflectiveOperationException e) {
                throw new ExceptionInInitializerError(e);
            }
        }

        MpscQueue(int capacity) {
            buffer = new Object[capacity];
            mask = capacity - 1;
        }

        boolean offer(E e) {
            while (true) {
                long t = tail;
                long h = (long) HEAD.getAcquire(this);
                if (t - h >= buffer.length) return false;            // full
                if (TAIL.compareAndSet(this, t, t + 1)) {            // claim slot t
                    SLOT.setRelease(buffer, (int) (t & mask), (Object) e);
                    return true;
                }
            }
        }

        @SuppressWarnings("unchecked")
        E poll() {
            long h = head;
            int i = (int) (h & mask);
            Object o = SLOT.getAcquire(buffer, i);
            if (o == null) return null;          // empty, or claimed but not yet written
            buffer[i] = null;
            HEAD.setRelease(this, h + 1);
            return (E) o;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final int producers = 4, perProducer = 250_000;
        MpscQueue<Integer> q = new MpscQueue<>(1024);
        Thread[] threads = new Thread[producers];
        for (int p = 0; p < producers; p++) {
            threads[p] = new Thread(() -> {
                for (int i = 1; i <= perProducer; i++) {
                    while (!q.offer(i)) Thread.onSpinWait();
                }
            });
            threads[p].start();
        }
        long count = 0, sum = 0;
        while (count < (long) producers * perProducer) {
            Integer v = q.poll();
            if (v == null) { Thread.onSpinWait(); continue; }
            count++;
            sum += v;
        }
        for (Thread t : threads) t.join();
        System.out.println("received " + count + ", sum " + sum);
    }
}

Output

received 1000000, sum 125000500000

The sum is fixed (4 times 1 + 2 + ... + 250,000), but the order in which elements from different producers arrive is not, so the program does not print it. A producer descheduled between its CAS and its setRelease leaves a claimed empty slot; the consumer cannot pass it, even if later slots are full. This queue is therefore not strictly lock-free on the consumer side, which is a known property of this design (JCTools MpscArrayQueue has the same gap). The CAS on tail is also a contention point: under many producers it costs retries and cache-line transfers, which is why MPSC queues are slower than SPSC.

Example 7: wait strategies

When poll returns null the consumer must wait. The choice trades latency against CPU use.

Strategy	Wake-up latency	CPU while idle	Fits
Busy-spin (`Thread.onSpinWait()`)	Lowest, tens of nanoseconds	One full core	Pinned threads on dedicated cores
`Thread.yield()`	Low, depends on the scheduler	High	Fewer spare cores than spinning threads
`LockSupport.parkNanos(n)`	Microseconds, timer slack applies	Near zero	Background or bursty work
Blocking (`take` on a lock and condition)	Microseconds plus a wake-up by the producer	Zero	General-purpose code

A backoff strategy combines them: spin first, then yield, then park, and reset after work arrives. This is the shape of Agrona's BackoffIdleStrategy.

Java

import java.util.concurrent.locks.LockSupport;

public class BackoffIdle {
    private final int maxSpins, maxYields;
    private int idleCount;

    BackoffIdle(int maxSpins, int maxYields) {
        this.maxSpins = maxSpins;
        this.maxYields = maxYields;
    }

    /** workCount is the number of items processed by the last poll loop. */
    String idle(int workCount) {
        if (workCount > 0) { idleCount = 0; return "reset"; }
        idleCount++;
        if (idleCount <= maxSpins) { Thread.onSpinWait(); return "spin"; }
        if (idleCount <= maxSpins + maxYields) { Thread.yield(); return "yield"; }
        LockSupport.parkNanos(1_000);
        return "park";
    }

    public static void main(String[] args) {
        BackoffIdle idle = new BackoffIdle(3, 2);
        int[] work = {0, 0, 0, 0, 0, 0, 0, 5, 0};
        StringBuilder trace = new StringBuilder();
        for (int w : work) trace.append(idle.idle(w)).append(' ');
        System.out.println(trace.toString().trim());
    }
}

Output

spin spin spin yield yield park park reset spin

The trace shows the escalation over seven empty polls and the reset when five items arrive. A real consumer loop calls idle(drained) after each attempt to drain the queue. Parking needs care: if a blocking producer has to unpark the consumer, the queue needs extra state to know the consumer is parked, which adds cost to every offer; a timed parkNanos avoids that at the price of latency.

Interview points

Power-of-two capacity turns seq % capacity into seq & mask; head and tail are unbounded long sequences, and tail - head is the size.
The single-writer principle removes CAS: one owner per counter, published with setRelease, read with getAcquire.
SPSC needs no StoreLoad: stale counters only make the queue look fuller or emptier, and the caller retries.
Cached indices cut cross-core reads to about one per buffer length; padding keeps head and tail off the same cache line.
MPSC claims slots by CAS on the tail and publishes by a release write to the slot; a stalled producer blocks the consumer at its slot.
Busy-spin gives the lowest latency at the cost of a core; backoff strategies spin, then yield, then park.

Summary

A ring buffer is full when tail - head == capacity and empty when head == tail.
Producer and consumer each null or fill slots and publish with one release write.
False sharing comes from independent variables on one cache line; class-hierarchy padding prevents it in user code.
JCTools (SpscArrayQueue, MpscArrayQueue) and Agrona provide production versions of these queues.
The next lesson, the LMAX Disruptor pattern, extends these ideas with preallocated events and batching.