Thread Safe Go Queues

Welcome back to another exciting lesson on concurrent data structures in Go! In our previous unit, we explored how to implement a thread-safe stack using sync.Mutex. As we progress, we're going to elevate our synchronization skills by exploring condition variables (sync.Cond) alongside mutexes.

This lesson will help you build a thread-safe queue that enables effective communication between goroutines when specific conditions are met. It's like giving goroutines the ability to "wait" for the right moment to act. Let's explore this thrilling aspect of concurrent programming in Go.

In this lesson, you'll grasp the concepts of implementing a thread-safe queue using sync.Mutex and sync.Cond. While Go also provides channels as a higher-level synchronization mechanism, understanding condition variables is valuable for scenarios that require fine-grained control over thread synchronization.

Here's a glimpse into what our thread-safe queue using sync.Mutex and sync.Cond looks like:

type ThreadsafeQueue struct {
    mu    sync.Mutex
    cond  *sync.Cond
    queue []int
}

func NewThreadsafeQueue() *ThreadsafeQueue {
    tsq := &ThreadsafeQueue{
        queue: make([]int, 0),
    }
    tsq.cond = sync.NewCond(&tsq.mu)
    return tsq
}

func (tsq *ThreadsafeQueue) Push(value int) {
    tsq.mu.Lock()
    defer tsq.mu.Unlock()
    
    tsq.queue = append(tsq.queue, value)
    tsq.cond.Signal() // Notify one waiting goroutine
}

func (tsq *ThreadsafeQueue) WaitAndPop() int {
    tsq.mu.Lock()
    defer tsq.mu.Unlock()
    
    // Wait until the queue is not empty
    for len(tsq.queue) == 0 {
        tsq.cond.Wait()
    }
    
    value := tsq.queue[0]
    tsq.queue = tsq.queue[1:]
    return value
}

func (tsq *ThreadsafeQueue) TryPop() (int, bool) {
    tsq.mu.Lock()
    defer tsq.mu.Unlock()
    
    if len(tsq.queue) == 0 {
        return 0, false
    }
    
    value := tsq.queue[0]
    tsq.queue = tsq.queue[1:]
    return value, true
}

func (tsq *ThreadsafeQueue) Empty() bool {
    tsq.mu.Lock()
    defer tsq.mu.Unlock()
    
    return len(tsq.queue) == 0
}

Notice how sync.Cond is used alongside sync.Mutex to allow goroutines to wait for a queue item to become available before proceeding. This unlocks more sophisticated synchronization techniques compared to simply using locks.

Let's examine the methods in our ThreadsafeQueue struct:

• Push(value int): Adds a new item to the queue and signals a waiting goroutine if one exists. We use defer to ensure the mutex is unlocked even if a panic occurs. The condition variable cond is used to signal waiting goroutines that new data is available.
• WaitAndPop() int: Waits for the queue to become non-empty and then removes and returns the front element. The cond.Wait() method blocks the goroutine until another goroutine calls Signal() or Broadcast(). We use a for loop to check the condition because Wait() can wake up spuriously.
• TryPop() (int, bool): Attempts to pop an item from the queue if it is not empty. This method does not block the calling goroutine. It returns the value and a boolean indicating success, following Go's error handling pattern.
• Empty() bool: Checks if the queue is empty in a thread-safe manner.

Let's see how these methods work together to create a thread-safe queue that can be safely accessed by multiple goroutines pushing and popping items concurrently:

func main() {
    tsq := NewThreadsafeQueue()
    
    var wg sync.WaitGroup
    wg.Add(2)
    
    // Producer goroutine
    go func() {
        defer wg.Done()
        for i := 0; i < 10; i++ {
            tsq.Push(i)
            time.Sleep(100 * time.Millisecond)
        }
    }()
    
    // Consumer goroutine
    go func() {
        defer wg.Done()
        for i := 0; i < 10; i++ {
            value := tsq.WaitAndPop()
            fmt.Printf("Consumer got: %d\n", value)
        }
    }()
    
    wg.Wait()
}

In this example, we have a producer goroutine that pushes values to the queue and a consumer goroutine that waits for the queue to become non-empty before popping the front element. The producer goroutine adds values to the queue every 100 milliseconds, and the consumer goroutine retrieves them as they become available. This demonstrates how condition variables can be used to synchronize goroutines effectively.

Mastering thread-safe queues with condition variables is crucial for scenarios where goroutines must collaborate closely to complete tasks. For instance, in a producer-consumer setup, producers add items to a queue while consumers retrieve them. Utilizing condition variables ensures that consumers do not waste resources by actively polling the queue, but instead wait patiently for new data. This efficiency is essential for robust, high-performance applications.

It's worth noting that Go also provides channels as a higher-level, idiomatic abstraction for communication between goroutines. Channels are often the preferred choice for simple producer-consumer patterns because they encapsulate synchronization. However, sync.Cond gives you more fine-grained control and is valuable when working with custom data structures, complex waiting conditions, or existing codebases that use this pattern.

By learning these techniques, you're equipping yourself with the tools to build reliable software systems that can perform seamlessly under concurrent workloads. This knowledge is a stepping stone to understanding complex multi-threaded systems and tackling real-world programming challenges with confidence. Ready to see this in action? Let's move to the practice section and put your new skills to the test!