Go Ep 11: Advanced Concurrency & Worker Pools

Channel communication handles message passing, but shared memory synchronization requires standard primitives like sync.Mutex and sync.WaitGroup. Learn how to build thread-safe concurrent systems.

TL;DR (Quick Summary)
#

sync.WaitGroup: Synchronizes Goroutine completion (Add(n), Done(), Wait()).
sync.Mutex: Exclusive mutual exclusion lock preventing data races (Lock(), Unlock()).
sync.RWMutex: Multiple concurrent readers (RLock()), single exclusive writer (Lock()).
Race Detector (go run -race): Go compiler flag that dynamically detects data races at runtime.
Worker Pool: A fixed number of worker Goroutines processing a queue of jobs concurrently.

1. Thread-Safe Mutex Architecture
#


graph TD
    subgraph SharedResource ["Shared Resource"]
        Counter["Safe Counter Struct"] --> Mutex["sync.Mutex"]
        Counter --> DataValue["Value: 100"]
    end

    G1["Goroutine 1"] -->|Lock| Mutex
    G2["Goroutine 2"] -->|Blocked Waiting| Mutex
    G3["Goroutine 3"] -->|Blocked Waiting| Mutex

2. Step-by-Step Lab: Building a Concurrent Worker Pool
#

package main

import (
 "fmt"
 "sync"
 "time"
)

type Job struct {
 ID    int
 Input int
}

type Result struct {
 JobID  int
 Output int
}

// Worker Goroutine
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
 defer wg.Done()

 for job := range jobs {
  // Simulate computation time
  time.Sleep(50 * time.Millisecond)
  results <- Result{JobID: job.ID, Output: job.Input * 2}
 }
}

func main() {
 const numJobs = 10
 const numWorkers = 3

 jobs := make(chan Job, numJobs)
 results := make(chan Result, numJobs)

 var wg sync.WaitGroup

 // 1. Spawn Workers
 for w := 1; w <= numWorkers; w++ {
  wg.Add(1)
  go worker(w, jobs, results, &wg)
 }

 // 2. Enqueue Jobs
 for j := 1; j <= numJobs; j++ {
  jobs <- Job{ID: j, Input: j * 10}
 }
 close(jobs) // Signal workers no more jobs coming

 // 3. Wait for completion in separate goroutine
 go func() {
  wg.Wait()
  close(results)
 }()

 // 4. Read Results
 for res := range results {
  fmt.Printf("Job %d completed: Output = %d\n", res.JobID, res.Output)
 }
}

3. Detecting Data Races (`go run -race`)
#

A data race occurs when two Goroutines access the same memory location concurrently, and at least one access is a write.

Unsafe Data Race Example
#

// 🔴 Data Race!
func main() {
    count := 0
    for i := 0; i < 1000; i++ {
        go func() { count++ }()
    }
}

Run with the Race Detector flag:

go run -race main.go

Expected Output:

==================
WARNING: DATA RACE
Write at 0x00c000014090 by goroutine 7:
  main.main.func1()
==================

Thread-Safe Fix with Mutex
#

// 🟢 Thread-Safe!
type SafeCounter struct {
    mu    sync.Mutex
    value int
}

func (c *SafeCounter) Inc() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

4. Troubleshooting & Common Errors
#

Error 1: Copying a Mutex by Value
#

The Cause: Passing a struct containing sync.Mutex by value to a function creates a copy of the mutex lock, rendering the lock useless! The Fix: Always pass structs containing Mutexes by pointer (func (s *SafeCounter) Inc()).