Concurrency Patterns and Pitfalls: Worker Pools, errgroup and Race Detection

Learning Objectives

Core Concepts

2.1 Worker Pool and Pipeline

What: a Worker Pool is a concurrency pattern that limits concurrency by pre-creating a fixed number of goroutines (workers) that consume a task queue. A Pipeline chains worker pools together, feeding one stage's output into the next stage's input.

Why: in Go, creating a goroutine is cheap, but creating them without bound eventually exhausts resources (memory, file descriptors). A Worker Pool limits the degree of concurrency and protects backend resources such as database connections.

How: 1. Define a task struct. 2. Create a task channel jobs and a result channel results. 3. Start N workers that read tasks from jobs, process them, and write to results. 4. The main goroutine closes the channels and collects results.

2.2 errgroup and Concurrent Error Collection

What: errgroup is a utility in the golang.org/x/sync package that coordinates a group of goroutines. It provides a Go method to submit tasks and a Wait method to wait for all of them.

Why: the standard sync.WaitGroup only counts; it does not handle errors. In concurrent scenarios, when one goroutine fails you usually want to cancel the rest. errgroup combines with context.Context to implement "first error triggers cancellation".

How: 1. Call g, ctx := errgroup.WithContext(ctx). 2. Submit tasks with g.Go(func() error { ... }). 3. Call g.Wait() to block until all tasks finish; if any task returns a non-nil error, that error is returned.

2.3 Race Detection and Leak Diagnosis

What: - Race detector: a runtime checker built into the Go toolchain that finds unsynchronized concurrent reads and writes of the same memory address. - Leak: a goroutine that stays resident in memory for too long because it never exits properly. - Deadlock: two or more goroutines wait on each other to release resources, so none of them can make progress.

Why: data races are among the most subtle bugs in Go and can corrupt data. Leaks cause memory to grow until OOM. Deadlocks hang the program forever.

How: - Race detection: use go run -race main.go or go test -race. Note: -race has existed since Go 1.1 (no special version needed); it relies on CGO, so a working C compiler must be available on the machine. - Leak diagnosis: use pprof's goroutine profile. -race only detects data races, not goroutine leaks; leaks are diagnosed with pprof's goroutine/heap profiles or runtime.NumGoroutine(). The two tools are complementary, not interchangeable. - Deadlock diagnosis: use go run -race or pprof's goroutine profile to inspect stacks, or analyze them with go tool pprof.

Example Code

Example 1: The Worker Pool Pattern

This example processes tasks with a fixed number of workers and collects the results.

package main

import (
    "fmt"
    "sync"
)

func main() {
    const numWorkers = 3
    jobs := make(chan int, 10)
    results := make(chan int, 10)

    // start the worker pool
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for j := range jobs {
                // simulate processing
                results <- j * 2
            }
        }(w)
    }

    // send tasks
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)

    // wait for all workers, then close results
    go func() {
        wg.Wait()
        close(results)
    }()

    // collect results
    var resultSlice []int
    for r := range results {
        resultSlice = append(resultSlice, r)
    }

    // sort for stable output so the result is easy to verify
    for i := 0; i < len(resultSlice); i++ {
        for j := i + 1; j < len(resultSlice); j++ {
            if resultSlice[i] > resultSlice[j] {
                resultSlice[i], resultSlice[j] = resultSlice[j], resultSlice[i]
            }
        }
    }
    fmt.Println("Results:", resultSlice)
}

Expected output:

Results: [2 4 6 8 10]

Example 2: errgroup Concurrent Error Handling

This example runs tasks concurrently with errgroup and cancels the rest when one fails.

package main

import (
    "context"
    "errors"
    "fmt"
    "sync"

    "golang.org/x/sync/errgroup"
)

func main() {
    // use errgroup.WithContext directly; no outer WithCancel needed
    g, ctx := errgroup.WithContext(context.Background())

    var mu sync.Mutex
    results := make([]string, 0, 3)

    // Task 1: succeeds
    g.Go(func() error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            mu.Lock()
            results = append(results, "task1-ok")
            mu.Unlock()
            return nil
        }
    })

    // Task 2: fails
    g.Go(func() error {
        return errors.New("task2-failed")
    })

    // Task 3: checks the context
    g.Go(func() error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            mu.Lock()
            results = append(results, "task3-ok")
            mu.Unlock()
            return nil
        }
    })

    if err := g.Wait(); err != nil {
        fmt.Println("Error:", err)
    } else {
        fmt.Println("All tasks succeeded")
    }
}

Expected output:

Error: task2-failed

Common Mistakes & Pitfalls

Best Practices

Exercises

1. Concept question: briefly describe the difference between sync.WaitGroup and errgroup.Group when handling concurrent errors.
2. Coding exercise: write a function FetchData(urls []string) ([]string, error) that uses errgroup to request several URLs concurrently (simulate the requests by returning the URL's content); if any request fails, return an error and cancel the others.
3. Advanced/debugging exercise: the following code has a data race. Find the problem and fix it. ```go package main

import ( "fmt" "sync" )

func main() { var counter int var wg sync.WaitGroup

   for i := 0; i < 10; i++ {
       wg.Add(1)
       go func() {
           defer wg.Done()
           counter++
       }()
   }
   wg.Wait()
   fmt.Println("Counter:", counter)

} ```

Reference Answers

package main

import (
    "context"
    "fmt"
    "sync"

    "golang.org/x/sync/errgroup"
)

// FetchData simulates concurrent requests
func FetchData(urls []string) ([]string, error) {
    g, ctx := errgroup.WithContext(context.Background())
    var mu sync.Mutex
    results := make([]string, 0, len(urls))

    for _, url := range urls {
        u := url // capture the loop variable
        g.Go(func() error {
            select {
            case <-ctx.Done():
                return ctx.Err()
            default:
                // simulate the request
                result := fmt.Sprintf("data-from-%s", u)
                mu.Lock()
                results = append(results, result)
                mu.Unlock()
                return nil
            }
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

func main() {
    urls := []string{"url1", "url2", "url3"}
    results, err := FetchData(urls)
    if err != nil {
        fmt.Println("Error:", err)
        return
    }
    fmt.Println("Results:", results)
}

package main

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

func main() {
    var counter int64
    var wg sync.WaitGroup

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            atomic.AddInt64(&counter, 1)
        }()
    }
    wg.Wait()
    fmt.Println("Counter:", counter)
}

Lesson Summary