6.5 Patterns and Best Practices

This section covers common concurrency patterns and best practices that you'll use in real-world Go programs. These patterns solve recurring problems in concurrent programming.

Worker Pools

A worker pool limits the number of concurrent goroutines, preventing resource exhaustion.

Worker pools are needed because creating unlimited goroutines can cause problems:

• Memory exhaustion: Each goroutine consumes memory
• Resource contention: Too many goroutines compete for CPU, I/O, or network
• Diminishing returns: Beyond a certain point, more goroutines slow things down

Worker pools work as follows:

1. Fixed number of workers: Create a limited number of goroutines
2. Job queue: Send work items through a channel
3. Workers process jobs: Each worker takes jobs from the queue and processes them
4. Results collection: Workers send results through another channel

Basic worker pool example

package main

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

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d: Processing job %d\n", id, job)
        // Simulate work
        time.Sleep(100 * time.Millisecond)
        results <- job * 2
    }
}

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

    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    // Start workers
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, jobs, results)
        }(w)
    }

    // Send jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)

    // Close results channel when all workers done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for result := range results {
        fmt.Printf("Result: %d\n", result)
    }
}

This pattern is effective because it provides controlled concurrency with exactly numWorkers goroutines running concurrently. Workers are reused rather than created per job, making resource use efficient. The pattern is scalable since it's easy to adjust worker count based on system resources.

Advanced worker pool with error handling

package main

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

type Job struct {
    ID   int
    Data string
}

type Result struct {
    Job    Job
    Output string
    Error  error
}

func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        // Simulate work that might fail
        time.Sleep(50 * time.Millisecond)

        var err error
        if job.ID%3 == 0 {
            err = errors.New("simulated error")
        }

        results <- Result{
            Job:    job,
            Output: fmt.Sprintf("Processed by worker %d", id),
            Error:  err,
        }
    }
}

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

    // Start workers
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, jobs, results)
        }(w)
    }

    // Send jobs
    go func() {
        defer close(jobs)
        for i := 1; i <= 10; i++ {
            jobs <- Job{ID: i, Data: fmt.Sprintf("data-%d", i)}
        }
    }()

    // Close results when workers done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Process results
    for result := range results {
        if result.Error != nil {
            fmt.Printf("Job %d failed: %v\n", result.Job.ID, result.Error)
        } else {
            fmt.Printf("Job %d: %s\n", result.Job.ID, result.Output)
        }
    }
}

Pipeline Pattern

The pipeline pattern is a powerful way to process data through a series of stages, where each stage transforms data and passes it to the next. This pattern enables concurrent processing of different items at different stages, making it ideal for data transformation workflows.

Understanding Pipelines

A pipeline consists of multiple stages connected by channels, where each stage:

• Receives data from the previous stage via a channel
• Transforms or processes the data
• Sends the processed data to the next stage via another channel

The stages run concurrently, meaning that while one stage is processing item N, the next stage can be processing item N-1, and the previous stage can be processing item N+1. This concurrent processing maximizes throughput and resource utilization.

Pipelines provide several benefits:

• Modularity: Each stage is independent and can be tested separately
• Composability: Stages can be combined in different ways to create new pipelines
• Concurrency: Different items are processed concurrently across stages
• Clarity: Data flow is explicit and easy to understand
• Scalability: Easy to add or remove stages as needed

Basic Pipeline Structure

A simple pipeline follows this structure:

Input → Stage1 → Stage2 → Stage3 → Output

Each stage is typically a function that:

1. Takes an input channel (<-chan T)
2. Returns an output channel (<-chan U)
3. Runs a goroutine that processes values from the input and sends to the output

Basic pipeline example

package main

import (
    "fmt"
    "strings"
)

// Stage 1: Generate numbers
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// Stage 3: Convert to string
func toString(in <-chan int) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for n := range in {
            out <- fmt.Sprintf("Result: %d", n)
        }
    }()
    return out
}

func main() {
    // Build pipeline: generate → square → toString
    numbers := generate(2, 3, 4)
    squared := square(numbers)
    results := toString(squared)

    // Consume results
    for result := range results {
        fmt.Println(result)
    }
}

Each stage uses a goroutine so stages can process different items concurrently. While stage 2 processes item 1, stage 1 can already be processing item 2.

Pipeline with Error Handling

Real-world pipelines need to handle errors gracefully. Here's a pattern for error handling in pipelines:

package main

import (
    "errors"
    "fmt"
)

type Result struct {
    Value int
    Error error
}

func generate(nums ...int) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- Result{Value: n}
        }
    }()
    return out
}

func square(in <-chan Result) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for res := range in {
            if res.Error != nil {
                out <- res  // Forward errors
                continue
            }
            if res.Value < 0 {
                out <- Result{Error: errors.New("negative number")}
                continue
            }
            out <- Result{Value: res.Value * res.Value}
        }
    }()
    return out
}

func main() {
    numbers := generate(2, -3, 4, 5)
    squared := square(numbers)

    successCount := 0
    errorCount := 0

    for res := range squared {
        if res.Error != nil {
            errorCount++
            fmt.Printf("Error: %v\n", res.Error)
        } else {
            successCount++
            fmt.Printf("Success: %d\n", res.Value)
        }
    }

    fmt.Printf("\nSummary: %d successes, %d errors\n", successCount, errorCount)
}

Advanced Pipeline Patterns

Bounded pipeline (with buffering)

Buffering channels in pipeline stages can improve performance by allowing stages to work ahead:

func square(in <-chan int) <-chan int {
    out := make(chan int, 10)  // Buffered channel
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

Pipeline with cancellation

Using context to cancel pipeline stages:

import "context"

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                return
            case n, ok := <-in:
                if !ok {
                    return
                }
                select {
                case out <- n * n:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

Pipeline best practices

• Always close output channels when done sending
• Handle errors explicitly at each stage
• Use buffered channels when stages have different processing speeds
• Consider using context for cancellation in long-running pipelines
• Keep stages focused on a single transformation

Fan-out and Fan-in Pattern

The fan-out/fan-in pattern is a powerful concurrency pattern for distributing work across multiple workers (fan-out) and then collecting results from those workers (fan-in). This pattern is essential for parallel processing and load balancing.

Understanding Fan-out

Fan-out means distributing work from a single source to multiple workers. This pattern is useful when:

• You have a large amount of work that can be processed in parallel
• Work items are independent and can be processed concurrently
• You want to utilize multiple CPU cores or I/O resources
• You need to balance load across multiple workers

Fan-out pattern example

package main

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

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d: Processing job %d\n", id, job)
        // Simulate work
        time.Sleep(100 * time.Millisecond)
        results <- job * 2
    }
}

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

    // Fan-out: Start multiple workers
    const numWorkers = 3
    var wg sync.WaitGroup

    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, jobs, results)
        }(w)
    }

    // Send work
    go func() {
        defer close(jobs)
        for i := 1; i <= 9; i++ {
            jobs <- i
        }
    }()

    // Close results when all workers done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for result := range results {
        fmt.Printf("Result: %d\n", result)
    }
}

Understanding Fan-in

Fan-in means combining results from multiple sources into a single channel. This pattern is useful when:

• You have multiple producers generating data
• You need to aggregate results from multiple workers
• You want to merge data from different sources
• You need a single point to consume from multiple channels

Fan-in pattern example

package main

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

func producer(id int, ch chan<- int) {
    defer close(ch)
    for i := 0; i < 3; i++ {
        ch <- id*10 + i
        time.Sleep(100 * time.Millisecond)
    }
}

func fanIn(inputs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    // Forward from each input channel
    for _, input := range inputs {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for val := range ch {
                out <- val
            }
        }(input)
    }

    // Close output when all inputs done
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)
    ch3 := make(chan int)

    // Start producers
    go producer(1, ch1)
    go producer(2, ch2)
    go producer(3, ch3)

    // Fan-in: Combine all channels
    merged := fanIn(ch1, ch2, ch3)

    // Receive from merged channel
    for val := range merged {
        fmt.Printf("Received: %d\n", val)
    }
}

Fan-in uses goroutines because each input channel needs its own goroutine to forward values. Without separate goroutines, reading from one channel would block reading from others.

Combining Fan-out and Fan-in

The most powerful pattern combines both fan-out and fan-in:

package main

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

// Fan-out: Distribute work to workers
func fanOut(jobs <-chan int, numWorkers int) []<-chan int {
    outputs := make([]<-chan int, numWorkers)

    for i := 0; i < numWorkers; i++ {
        out := make(chan int)
        outputs[i] = out

        go func(workerID int) {
            defer close(out)
            for job := range jobs {
                // Simulate work
                time.Sleep(100 * time.Millisecond)
                result := job * 2
                fmt.Printf("Worker %d processed job %d -> %d\n", workerID, job, result)
                out <- result
            }
        }(i)
    }

    return outputs
}

// Fan-in: Combine results from multiple channels
func fanIn(inputs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, input := range inputs {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for val := range ch {
                out <- val
            }
        }(input)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    jobs := make(chan int, 10)

    // Send jobs
    go func() {
        defer close(jobs)
        for i := 1; i <= 9; i++ {
            jobs <- i
        }
    }()

    // Fan-out: Distribute to 3 workers
    workerOutputs := fanOut(jobs, 3)

    // Fan-in: Combine results
    results := fanIn(workerOutputs...)

    // Collect results
    for result := range results {
        fmt.Printf("Final result: %d\n", result)
    }
}

Best Practices for Fan-out/Fan-in

• Control worker count: Limit the number of workers to avoid resource exhaustion
• Use buffered channels: Buffered channels can improve performance by reducing blocking
• Handle errors: Consider using error channels or Result types that include errors
• Close channels properly: Always close channels when done sending to signal completion
• Monitor worker health: Consider adding timeouts or health checks for long-running workers
• Balance work distribution: Ensure work is distributed evenly across workers

Context Package

The context package provides a standard way to handle cancellation, timeouts, and request-scoped values in concurrent programs.

Context is important because it enables cancellation propagation to cancel operations across goroutines, timeouts to set deadlines for operations, and request scoping to pass request-specific data through call chains.

Context works by being passed through function calls and goroutines. When a context is cancelled, all operations using that context should stop.

Basic context usage

package main

import (
    "context"
    "fmt"
    "time"
)

func longRunningTask(ctx context.Context) {
    for i := 0; i < 10; i++ {
        select {
        case <-ctx.Done():
            fmt.Println("Task cancelled")
            return
        default:
            fmt.Printf("Working... %d\n", i)
            time.Sleep(200 * time.Millisecond)
        }
    }
    fmt.Println("Task completed")
}

func main() {
    // Context with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()

    longRunningTask(ctx)
}

Context with timeout

package main

import (
    "context"
    "fmt"
    "time"
)

func fetchData(ctx context.Context) (string, error) {
    // Simulate slow operation
    select {
    case <-time.After(2 * time.Second):
        return "Data", nil
    case <-ctx.Done():
        return "", ctx.Err()
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()

    data, err := fetchData(ctx)
    if err != nil {
        fmt.Printf("Error: %v\n", err)  // context deadline exceeded
    } else {
        fmt.Printf("Data: %s\n", data)
    }
}

Context with cancellation

package main

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

func worker(ctx context.Context, id int, wg *sync.WaitGroup) {
    defer wg.Done()

    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d: Cancelled\n", id)
            return
        default:
            fmt.Printf("Worker %d: Working...\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go worker(ctx, i, &wg)
    }

    // Cancel after 2 seconds
    time.Sleep(2 * time.Second)
    fmt.Println("Cancelling all workers...")
    cancel()

    wg.Wait()
    fmt.Println("All workers stopped")
}

Context is powerful because it provides hierarchical cancellation where cancelling a parent context cancels all children. It offers a standard interface since all Go libraries use context for cancellation, and it's composable since contexts can be combined (timeout + cancellation + values).

Context with values

package main

import (
    "context"
    "fmt"
)

func processRequest(ctx context.Context) {
    // Retrieve value from context
    userID := ctx.Value("userID")
    if userID != nil {
        fmt.Printf("Processing request for user: %v\n", userID)
    }
}

func main() {
    ctx := context.WithValue(context.Background(), "userID", 12345)
    processRequest(ctx)
}

Best practices for context

1. Pass context as first parameter: Standard Go convention
2. Don't store context in structs: Pass it explicitly
3. Use context for cancellation, not for passing optional parameters: Use function parameters for optional data
4. Always check ctx.Done(): Respect cancellation signals

Error Handling in Concurrent Code

Handling errors in concurrent code requires special care because errors can occur in multiple goroutines simultaneously.

Error handling in concurrent code is challenging because errors can come from multiple goroutines, errors might arrive at different times creating timing issues, and you need to collect and handle multiple errors through error aggregation.

Pattern 1: Error channel

package main

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

func worker(id int, errChan chan<- error) {
    if id%2 == 0 {
        errChan <- fmt.Errorf("worker %d failed", id)
        return
    }
    fmt.Printf("Worker %d: Success\n", id)
}

func main() {
    errChan := make(chan error, 10)
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, errChan)
        }(i)
    }

    // Close error channel when done
    go func() {
        wg.Wait()
        close(errChan)
    }()

    // Collect errors
    var errors []error
    for err := range errChan {
        errors = append(errors, err)
    }

    if len(errors) > 0 {
        fmt.Printf("Encountered %d errors:\n", len(errors))
        for _, err := range errors {
            fmt.Printf("  - %v\n", err)
        }
    }
}

Pattern 2: Result with error

package main

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

type Result struct {
    Value int
    Error error
}

func compute(id int) Result {
    if id%3 == 0 {
        return Result{Error: errors.New("computation failed")}
    }
    return Result{Value: id * 2}
}

func main() {
    results := make(chan Result, 10)
    var wg sync.WaitGroup

    for i := 1; i <= 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            results <- compute(id)
        }(i)
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    successCount := 0
    errorCount := 0

    for res := range results {
        if res.Error != nil {
            errorCount++
            fmt.Printf("Error: %v\n", res.Error)
        } else {
            successCount++
            fmt.Printf("Success: %d\n", res.Value)
        }
    }

    fmt.Printf("\nSummary: %d successes, %d errors\n", successCount, errorCount)
}

Pattern 3: Error group (using errgroup package)

The golang.org/x/sync/errgroup package provides a better way to handle errors from multiple goroutines. It combines the functionality of sync.WaitGroup with error handling and automatic cancellation through context integration.

Error groups are useful because they provide automatic cancellation where when one task fails, others are cancelled. They return the first error encountered and offer a clean API that's simpler than managing channels and WaitGroups manually.

For a comprehensive discussion of ErrGroup, including detailed explanations, context integration, and best practices, see section 6.7.

These patterns and practices form the foundation of writing robust, efficient concurrent Go programs. Understanding when and how to use each pattern is key to mastering Go concurrency.

Previous: 6.4 Select Statement Next: 6.6 Sync Package