Implementing Lock Free Queues

Welcome back to our journey through lock-free programming in Go. In the last lesson, you learned about implementing a lock-free stack. Now, we're progressing to another essential data structure: the queue. This lesson will focus on implementing a thread-safe lock-free queue, building upon the foundation of atomic operations explored in the previous lessons. By the end of this unit, you'll be equipped with the skills to implement a robust, efficient queue that operates without traditional locking mechanisms.

In this lesson, you'll learn how to build a lock-free queue using Go's sync/atomic package to handle concurrency efficiently across multiple goroutines.

The following code example is a sneak peek into what you'll be creating:

package main

import (
    "fmt"
    "sync/atomic"
    "time"
    "unsafe"
)

type LockFreeQueue struct {
    head unsafe.Pointer // *node
    tail unsafe.Pointer // *node
}

type node struct {
    data int
    next unsafe.Pointer // *node
}

func newNode(value int) *node {
    return &node{
        data: value,
        next: nil,
    }
}

func NewLockFreeQueue() *LockFreeQueue {
    dummy := newNode(0) // Dummy node to simplify push/pop logic
    return &LockFreeQueue{
        head: unsafe.Pointer(dummy),
        tail: unsafe.Pointer(dummy),
    }
}

func (q *LockFreeQueue) Push(value int) {
    newNode := newNode(value)

    for {
        tail := (*node)(atomic.LoadPointer(&q.tail))
        next := (*node)(atomic.LoadPointer(&tail.next))

        // Check if tail hasn't changed
        if tail == (*node)(atomic.LoadPointer(&q.tail)) {
            if next == nil { // Tail is truly the last node
                // Try to link new node at the end of the list
                if atomic.CompareAndSwapPointer(&tail.next, nil, unsafe.Pointer(newNode)) {
                    // Push was successful, try to swing tail to the new node
                    atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(newNode))
                    fmt.Printf("Pushed: %d\n", value)
                    return
                }
            } else {
                // Tail is lagging; try to advance it
                atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(next))
            }
        }
    }
}

func (q *LockFreeQueue) Pop() (int, bool) {
    for {
        head := (*node)(atomic.LoadPointer(&q.head))
        tail := (*node)(atomic.LoadPointer(&q.tail))
        next := (*node)(atomic.LoadPointer(&head.next))

        // Consistency check
        if head == (*node)(atomic.LoadPointer(&q.head)) {
            if head == tail { // Queue might be empty
                if next == nil { // Queue is empty
                    fmt.Println("Pop failed - queue is empty.")
                    return 0, false
                }
                // Tail is lagging; try to advance it
                atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(next))
            } else {
                // Queue is not empty, read value before CAS
                result := next.data
                // Try to swing head to the next node
                if atomic.CompareAndSwapPointer(&q.head, unsafe.Pointer(head), unsafe.Pointer(next)) {
                    fmt.Printf("Popped: %d\n", result)
                    return result, true
                }
            }
        }
    }
}

Let's break down the implementation step by step:

1. Node structure: The queue is implemented using a linked list of nodes, where each node contains the data and an unsafe.Pointer to the next node. The newNode function creates a new node with the provided value and a nil pointer for the next node.
2. Head and tail pointers: The queue maintains two unsafe.Pointer fields, head and tail, representing the front and back of the queue, respectively. Both pointers are initially set to a dummy node to simplify the logic. We use unsafe.Pointer instead of regular pointers because Go's sync/atomic package requires this type for atomic pointer operations.
3. Push operation: The Push method adds a new node to the queue. It creates a new node with the provided value and then attempts to add it to the queue. The method uses a loop to handle concurrent updates to the tail pointer and ensures that the new node is correctly linked to the queue using atomic.CompareAndSwapPointer.
4. Pop operation: The Pop method removes and returns the front node from the queue. It also uses a loop to handle concurrent updates to the head and tail pointers. The method checks for empty and non-empty queue conditions, updating the pointers accordingly. It returns both the value and a bool indicating success.
5. Memory management: Unlike manual memory management, Go's garbage collector automatically reclaims memory for nodes that are no longer referenced, so we do not need explicit cleanup code.

Let's now see how to use this lock-free queue in a concurrent environment with goroutines:

func pushItems(queue *LockFreeQueue, start, end int) {
    for i := start; i < end; i++ {
        queue.Push(i)
        time.Sleep(50 * time.Millisecond) // Delay for interleaving
    }
}

func popItems(queue *LockFreeQueue, count int) {
    for i := 0; i < count; i++ {
        queue.Pop()
        time.Sleep(30 * time.Millisecond) // Delay for interleaving
    }
}

func main() {
    queue := NewLockFreeQueue()

    // Create goroutines to push and pop concurrently
    go pushItems(queue, 1, 6)
    go popItems(queue, 3)
    go pushItems(queue, 6, 11)
    go popItems(queue, 5)

    // Wait for goroutines to complete
    time.Sleep(2 * time.Second)
}

In this example, we create a lock-free queue of integers and spawn multiple goroutines to push and pop items concurrently. The pushItems and popItems functions simulate the enqueue and dequeue operations, respectively, with a delay to introduce interleaving. The main function creates goroutines to perform these operations and waits for them to complete before exiting. Note that the output may vary due to the interleaving of operations, and the printed messages may be mixed up because we do not use any synchronization for the output.

Lock-free queues are vital for high-performance systems that require concurrent processing without delays caused by mutual exclusion locks. Unlike sync.Mutex-based queues, lock-free queues are more scalable, allowing multiple goroutines to perform enqueue and dequeue operations simultaneously without blocking.

In Go's runtime, goroutines are multiplexed onto a smaller number of OS threads. When a goroutine blocks on a mutex, the Go scheduler must manage this contention, potentially limiting parallelism. Lock-free data structures, on the other hand, allow goroutines to make progress without blocking, working harmoniously with Go's scheduler to maximize concurrency.

By mastering the design of these data structures, you'll enhance your ability to write concurrent Go applications that fully utilize the Go runtime's capabilities for better responsiveness and throughput. This is especially important for building high-performance services, real-time systems, and applications that need to handle many concurrent operations efficiently.

Exciting, isn't it? Let's move on to the practice section to solidify your understanding through hands-on implementation!