Designing and Implementing a Concurrent Thread-Safe Queue from Scratch

Queues are fundamental building blocks in modern software architecture. They sit at the heart of message brokers, event-driven architectures, worker pools, and task scheduling systems. However, in concurrent environments where multiple threads or coroutines read and write simultaneously, a naive queue implementation breaks down rapidly.

Without explicit synchronization, concurrent access causes race conditions, corrupted internal state, dropped data, and runtime exceptions. In this guide, we break down why standard queues fail under concurrency, examine the non-atomic nature of basic slice/array operations, and implement an idiomatic, thread-safe queue from scratch using pessimistic locking.

---

1. The Anatomy of a Naive Queue

A standard first-in, first-out (FIFO) queue typically exposes two primary operations:

1. Enqueue (Push): Appends an element to the tail of the queue.
2. Dequeue (Pop): Extracts and returns the element at the head of the queue.

In an array-backed queue, enqueueing often looks like this conceptually:

queue[tail] = item
tail++

And dequeuing behaves similarly:

item = queue[head]
head++

flowchart LR
    subgraph NonAtomicEnqueue [Non-Atomic Enqueue]
        T1[Thread 1 Reads tail = 4] --> W1[Thread 1 Writes queue at 4]
        T2[Thread 2 Reads tail = 4] --> W2[Thread 2 Overwrites queue at 4!]
        W1 --> I1[tail Incremented to 5]
        W2 --> I2[tail Incremented to 5 or 6]
    end

Why Naive Operations Are Not Thread-Safe

Neither tail++, head++, nor Go’s built-in append() operation is atomic. At the CPU level, an increment operation consists of three distinct machine instructions:

1. Load the current value from memory into a CPU register.
2. Increment the register value.
3. Store the updated register value back to memory.

If two threads execute this sequence concurrently without coordination, both may read the identical tail index simultaneously. Thread A writes its value, and Thread B immediately overwrites it at the exact same memory offset. Both threads subsequently increment the counter, resulting in lost writes and internal memory corruption.

---

2. Demonstrating the Race Condition in Go

To observe this failure mode directly, consider a naive queue implementation using Go slices.

package main

import (
	"errors"
	"fmt"
	"sync"
)

type NaiveQueue struct {
	items []int32
}

func (q *NaiveQueue) Enqueue(item int32) {
	q.items = append(q.items, item)
}

func (q *NaiveQueue) Dequeue() (int32, error) {
	if len(q.items) == 0 {
		return 0, errors.New("cannot dequeue from an empty queue")
	}
	item := q.items[0]
	q.items = q.items[1:]
	return item, nil
}

func (q *NaiveQueue) Size() int {
	return len(q.items)
}

Simulating High Concurrency

We spin up $1{,}000{,}000$ concurrent goroutines, each attempting to enqueue a single integer into the shared queue:

func main() {
	q := &NaiveQueue{}
	var wg sync.WaitGroup
	numOperations := 1_000_000

	for i := 0; i < numOperations; i++ {
		wg.Add(1)
		go func(val int32) {
			defer wg.Done()
			q.Enqueue(val)
		}(int32(i))
	}

	wg.Wait()
	fmt.Printf("Expected size: %d, Actual size: %d\n", numOperations, q.Size())
}

The Result: Silent Data Loss

Expected size: 1000000, Actual size: 958214

Rather than $1{,}000{,}000$ elements, the resulting slice holds roughly $958{,}000$ items. More than $40{,}000$ enqueue operations were silently dropped. In many environments, this race condition also triggers panic crashes such as runtime memory violations during slice reallocation.

---

3. Engineering a Thread-Safe Concurrent Queue

To ensure consistency, we must serialize concurrent access across the critical section—the block of code mutating the underlying slice.

Mutual Exclusion (sync.Mutex)

A mutual exclusion lock (mutex) guarantees that only one thread can enter the critical section at any given time. Any other thread attempting to access the queue will block until the active thread releases the lock.

sequenceDiagram
    autonumber
    participant T1 as Thread 1
    participant Lock as sync.Mutex
    participant Queue as Queue Slice
    participant T2 as Thread 2

    T1->>Lock: Lock()
    Note over Lock: Lock Acquired by T1
    T2->>Lock: Lock()
    Note over T2: Blocked / Waiting
    T1->>Queue: append(items, item)
    T1->>Lock: Unlock()
    Note over Lock: Lock Released
    Note over Lock: Lock Granted to T2
    T2->>Queue: append(items, item)
    T2->>Lock: Unlock()

Complete Implementation

package main

import (
	"errors"
	"fmt"
	"sync"
)

// ConcurrentQueue represents a thread-safe FIFO queue.
type ConcurrentQueue struct {
	mu    sync.Mutex
	items []int32
}

// NewConcurrentQueue initializes an empty queue.
func NewConcurrentQueue() *ConcurrentQueue {
	return &ConcurrentQueue{
		items: make([]int32, 0),
	}
}

// Enqueue appends an item to the end of the queue safely.
func (q *ConcurrentQueue) Enqueue(item int32) {
	q.mu.Lock()
	defer q.mu.Unlock()

	q.items = append(q.items, item)
}

// Dequeue removes and returns the head element of the queue.
func (q *ConcurrentQueue) Dequeue() (int32, error) {
	q.mu.Lock()
	defer q.mu.Unlock()

	if len(q.items) == 0 {
		return 0, errors.New("cannot dequeue from an empty queue")
	}

	item := q.items[0]
	q.items = q.items[1:]
	return item, nil
}

// Size returns the instantaneous number of elements in the queue.
func (q *ConcurrentQueue) Size() int {
	q.mu.Lock()
	defer q.mu.Unlock()

	return len(q.items)
}

Key Implementation Details

1. Locking the Reader (Size): Even read-only methods like Size() must acquire the lock. Reading a slice’s length while another goroutine reallocates its backing array during an append() produces undefined behavior and data races.
2. defer q.mu.Unlock(): Using Go’s defer statement ensures that whether the method returns cleanly or panics, the lock is guaranteed to be released, preventing deadlocks.

---

4. Verification Under Load

Testing the synchronized implementation with symmetric concurrent enqueues and dequeues demonstrates absolute correctness:

func main() {
	q := NewConcurrentQueue()
	numOps := 1_000_000

	var wgEnqueue sync.WaitGroup
	for i := 0; i < numOps; i++ {
		wgEnqueue.Add(1)
		go func(val int32) {
			defer wgEnqueue.Done()
			q.Enqueue(val)
		}(int32(i))
	}
	wgEnqueue.Wait()
	fmt.Printf("Post-Enqueue Size: %d (Expected: %d)\n", q.Size(), numOps)

	var wgDequeue sync.WaitGroup
	for i := 0; i < numOps; i++ {
		wgDequeue.Add(1)
		go func() {
			defer wgDequeue.Done()
			_, err := q.Dequeue()
			if err != nil {
				panic(err)
			}
		}()
	}
	wgDequeue.Wait()
	fmt.Printf("Post-Dequeue Size: %d (Expected: 0)\n", q.Size())
}

Post-Enqueue Size: 1000000 (Expected: 1000000)
Post-Dequeue Size: 0 (Expected: 0)

---

5. Architectural Trade-offs: Correctness vs. Contention

While wrapping operations in a sync.Mutex ensures linearizability and correctness, it introduces specific performance characteristics:

Attribute	Naive Queue	Mutex-Protected Queue
Thread Safety	❌ No (data races, lost updates)	✅ Yes (fully linearizable)
Throughput	High (unsynchronized CPU registers)	Bound by lock contention
Latency Profile	Low (no waiting)	High tail latency under heavy lock contention
Complexity	Minimal	Low-to-moderate

The Serialization Bottleneck

Pessimistic locking forces parallel goroutines into a serialized bottleneck. When $N$ goroutines compete for the same lock, $N-1$ threads are suspended or spinning, introducing context-switch overhead and cache invalidation penalties across CPU cores.

In systems requiring extreme throughput, alternative concurrency designs are employed:

• Fine-Grained Locking: Using separate locks for the head and the tail (e.g., the Michael-Scott non-blocking queue algorithm).
• Lock-Free Queues: Leveraging atomic compare-and-swap (CAS) primitives from sync/atomic.
• Channels (in Go): Go’s native buffered channels provide built-in, highly optimized thread-safe communication using internal ring buffers and runtime scheduler integration.

Despite the synchronization overhead, correctness supersedes performance. A high-throughput system that silently corrupts or discards 5% of its records is functionally broken.

---

6. Real-World Applications

Concurrent queues are ubiquitous across real-world distributed architectures and system internals:

1. Thread and Worker Pools

Thread pools maintain an internal blocking concurrent queue. Producer threads drop incoming tasks (e.g., HTTP requests) into the queue, while a fixed pool of consumer worker threads continuously dequeues and executes jobs without creating thread-per-request overhead.

2. High-Throughput Batch Processing & Ingestion Pipelines

Consider a distributed web scraper or metrics collector:

• Thousands of scraper goroutines pull data concurrently from external services.
• All workers deposit parsed payloads into a shared concurrent queue.
• A dedicated background worker periodically pulls batches of $B$ records from the queue and writes them via bulk INSERT statements to a database.

flowchart LR
    S1[Scraper Worker 1] -->|Enqueue| CQ[(Shared Concurrent Queue)]
    S2[Scraper Worker 2] -->|Enqueue| CQ
    S3[Scraper Worker N] -->|Enqueue| CQ
    CQ -->|Batch Dequeue| DBW[Batch Database Writer]
    DBW -->|Bulk INSERT| DB[(Storage Engine)]

If the intermediate queue is not thread-safe, scraped data is silently discarded, resulting in incomplete datasets and corrupted pipeline state.

---

Summary

• Multi-threaded code executing simple operations (like slice updates or increments) causes race conditions because these actions are decomposed into multiple non-atomic CPU instructions.
• In Go, uncoordinated writes to a slice cause overlapping array assignments and lost updates.
• Wrapping read and write methods with a mutual exclusion lock (sync.Mutex) guarantees serial execution of critical sections, ensuring end-to-end data integrity.
• Writing correct concurrent code requires recognizing shared mutable state and prioritizing correctness before optimizing throughput.