lockfree-mpmc-bounded-queue

Answer

The implementation is a lock-free bounded MPMC queue based on Dmitry Vyukov's ring buffer algorithm, adapted for Go using sync/atomic. The core idea: each ring buffer slot carries a monotonically increasing sequence counter that encodes ownership state. Enqueuers and dequeuers atomically claim slots via CAS on shared head/tail cursors, then verify slot readiness by comparing slot sequence against the claimed cursor value.

How the sequence counter prevents ABA

Each slot i starts with sequence = i. The counter flows through a deterministic lifecycle:

Because the expected sequence for any position changes every cycle, a stale pointer (ABA) cannot trick a goroutine into thinking a slot is free when it's still occupied.

Code: queue.go

package queue

import (
    "errors"
    "sync/atomic"
)

var (
    ErrFull  = errors.New("queue is full")
    ErrEmpty = errors.New("queue is empty")
)

type slot[T any] struct {
    sequence uint64
    value    T
}

// Queue is a lock-free bounded MPMC queue.
type Queue[T any] struct {
    capacity uint64
    buffer   []slot[T]
    head     uint64   // next enqueue index
    tail     uint64   // next dequeue index
    _        [56]byte // padding to prevent false sharing
}

// New creates a Queue with the given capacity. Panics if capacity is 0.
func New[T any](capacity uint64) *Queue[T] {
    if capacity == 0 {
        panic("queue: capacity must be positive")
    }
    q := &Queue[T]{
        capacity: capacity,
        buffer:   make([]slot[T], capacity),
    }
    for i := uint64(0); i < capacity; i++ {
        q.buffer[i].sequence = i
    }
    return q
}

// Enqueue adds a value. Returns ErrFull if at capacity.
func (q *Queue[T]) Enqueue(value T) error {
    for {
        head := atomic.LoadUint64(&q.head)
        pos := head % q.capacity
        seq := atomic.LoadUint64(&q.buffer[pos].sequence)

        if seq != head {
            return ErrFull // slot still occupied from prior cycle
        }

        if atomic.CompareAndSwapUint64(&q.head, head, head+1) {
            q.buffer[pos].value = value
            atomic.StoreUint64(&q.buffer[pos].sequence, head+1)
            return nil
        }
        // CAS failed → retry with fresh head
    }
}

// Dequeue removes and returns a value. Returns ErrEmpty if empty.
func (q *Queue[T]) Dequeue() (T, error) {
    for {
        tail := atomic.LoadUint64(&q.tail)
        pos := tail % q.capacity
        seq := atomic.LoadUint64(&q.buffer[pos].sequence)

        if seq != tail+1 {
            var zero T
            return zero, ErrEmpty // no data written yet
        }

        if atomic.CompareAndSwapUint64(&q.tail, tail, tail+1) {
            val := q.buffer[pos].value
            var zero T
            q.buffer[pos].value = zero // avoid memory leak
            atomic.StoreUint64(&q.buffer[pos].sequence, tail+q.capacity)
            return val, nil
        }
        // CAS failed → retry with fresh tail
    }
}

func (q *Queue[T]) Capacity() uint64 { return q.capacity }

func (q *Queue[T]) Len() uint64 {
    return atomic.LoadUint64(&q.head) - atomic.LoadUint64(&q.tail)
}

Key design decisions

1. Check-then-claim — Enqueue/Dequeue first check the slot's sequence against the expected value, and only claim (CAS) if the slot is ready. If the sequence doesn't match, return error immediately (non-blocking). If CAS fails, retry.

2. Release via sequence store — After writing data, atomic.StoreUint64(..., head+1) acts as a release barrier, ensuring the data write is visible to the dequeuer that reads the sequence. Go's sequentially-consistent atomics provide a total happens-before order: the store to sequence happens-before any load that reads the updated value.

3. False sharing mitigation — A 56-byte padding field separates head and tail onto different cache lines, preventing the performance collapse that occurs when producer and consumer goroutines contend on the same cache line.

4. Generics — Type-safe at zero runtime cost; no interface{} boxing.