Goroutine Coordination Patterns

Advanced ~45 min read

Coordinating multiple goroutines is essential for building scalable concurrent applications. This lesson covers production-ready patterns: worker pools, fan-out/fan-in, pipelines, and bounded concurrency. Master these patterns to write efficient, maintainable concurrent Go code.

Figure: WaitGroup Lifecycle and Coordination Patterns (Worker Pool, Fan-Out/Fan-In)

WaitGroup Coordination Pattern

The fundamental pattern for coordinating goroutines uses sync.WaitGroup to wait for a collection of goroutines to complete:

Output

Click Run to execute your code

Pattern Steps:

wg.Add(1) - Increment counter before starting goroutine
defer wg.Done() - Decrement counter when goroutine completes
wg.Wait() - Block until all goroutines finish

Worker Pool Pattern

The worker pool pattern uses a fixed number of workers to process jobs from a queue. This limits concurrency and provides better resource management:

Output

Click Run to execute your code

Worker Pool Benefits:

Limits concurrent goroutines (prevents resource exhaustion)
Reuses goroutines (reduces overhead)
Provides backpressure (buffered job channel)
Easy to scale (adjust number of workers)

Fan-Out Pattern

Fan-out distributes work across multiple goroutines to parallelize processing:

Output

Click Run to execute your code

When to Use Fan-Out: Use when you have independent tasks that can be processed in parallel, like processing multiple files, making concurrent API calls, or performing parallel computations.

Fan-In Pattern

Fan-in merges multiple input channels into a single output channel:

package main

import (
	"fmt"
	"sync"
)

func producer(id int, out chan<- int) {
	for i := 1; i <= 3; i++ {
		out <- id*10 + i
	}
	close(out)
}

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

// Start a goroutine for each input channel
	output := func(c <-chan int) {
		defer wg.Done()
		for n := range c {
			out <- n
		}
	}

wg.Add(len(inputs))
	for _, c := range inputs {
		go output(c)
	}

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

return out
}

func main() {
	// Create multiple input channels
	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: Merge all channels into one
	fmt.Println("Merging 3 channels (fan-in)...")
	merged := fanIn(ch1, ch2, ch3)

// Consume merged output
	fmt.Println("Receiving merged results:")
	for value := range merged {
		fmt.Printf("Received: %d\n", value)
	}

fmt.Println("\nAll channels merged!")
}

Output

Click Run to execute your code

Fan-In Use Cases:

Collecting results from multiple workers
Merging log streams
Aggregating data from multiple sources
Combining outputs from parallel computations

Pipeline Pattern with WaitGroup

Pipelines chain multiple stages together, with each stage processing data and passing it to the next:

Output

Click Run to execute your code

Stage	Input	Processing	Output
Generate	-	Create numbers	Channel of ints
Square	Channel of ints	Square each number	Channel of ints
Print	Channel of ints	Print results	-

Error Handling in Concurrent Code

Collecting errors from multiple goroutines requires careful coordination:

package main

import (
	"fmt"
	"sync"
)

func worker(id int, jobs <-chan int, wg *sync.WaitGroup, errors chan<- error) {
	defer wg.Done()

for job := range jobs {
		// Simulate error on job 5
		if job == 5 {
			errors <- fmt.Errorf("worker %d failed on job %d", id, job)
			continue
		}

fmt.Printf("Worker %d completed job %d\n", id, job)
	}
}

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

jobs := make(chan int, numJobs)
	errors := make(chan error, numJobs)
	var wg sync.WaitGroup

// Start workers
	for w := 1; w <= numWorkers; w++ {
		wg.Add(1)
		go worker(w, jobs, &wg, errors)
	}

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

// Wait and close errors channel
	go func() {
		wg.Wait()
		close(errors)
	}()

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

// Report results
	if len(errorList) > 0 {
		fmt.Printf("\nEncountered %d errors:\n", len(errorList))
		for _, err := range errorList {
			fmt.Printf("  - %v\n", err)
		}
	} else {
		fmt.Println("\nAll jobs completed successfully!")
	}
}

Output

Click Run to execute your code

Pro Tip: For more sophisticated error handling, consider using the golang.org/x/sync/errgroup package, which provides a WaitGroup with error collection built-in.

Bounded Concurrency

Limit the number of concurrent goroutines using a semaphore pattern:

package main

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

func worker(id int, jobs <-chan int, results chan<- int, sem chan struct{}, wg *sync.WaitGroup) {
	defer wg.Done()

for job := range jobs {
		// Acquire semaphore (limit concurrency)
		sem <- struct{}{}

fmt.Printf("Worker %d processing job %d\n", id, job)
		time.Sleep(500 * time.Millisecond)
		results <- job * 2

// Release semaphore
		<-sem
	}
}

func main() {
	const maxConcurrent = 3 // Limit to 3 concurrent workers
	const numWorkers = 10
	const numJobs = 10

jobs := make(chan int, numJobs)
	results := make(chan int, numJobs)
	sem := make(chan struct{}, maxConcurrent) // Semaphore
	var wg sync.WaitGroup

// Start workers (more than max concurrent)
	fmt.Printf("Starting %d workers (max %d concurrent)...\n", numWorkers, maxConcurrent)
	for w := 1; w <= numWorkers; w++ {
		wg.Add(1)
		go worker(w, jobs, results, sem, &wg)
	}

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

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

// Collect results
	fmt.Println("\nResults:")
	for result := range results {
		fmt.Println(result)
	}

fmt.Println("\nAll jobs completed with bounded concurrency!")
}

Output

Click Run to execute your code

Semaphore Pattern: A buffered channel acts as a semaphore. Sending to the channel acquires a slot, receiving releases it. The buffer size limits concurrency.

Common Mistakes

1. Adding to WaitGroup inside goroutine

// ❌ Wrong - race condition
for i := 0; i < 10; i++ {
    go func() {
        wg.Add(1)  // Too late! Main might call Wait() first
        defer wg.Done()
        // work
    }()
}

// ✅ Correct - Add before starting goroutine
for i := 0; i < 10; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        // work
    }()
}

2. Closing channel too early

// ❌ Wrong - channel closed before workers finish
close(jobs)
wg.Wait()  // Workers might still be reading!

// ✅ Correct - close after sending all jobs
for j := 1; j <= numJobs; j++ {
    jobs <- j
}
close(jobs)  // Now safe to close
wg.Wait()

3. Not closing results channel

// ❌ Wrong - range will block forever
for result := range results {
    fmt.Println(result)
}

// ✅ Correct - close results when done
go func() {
    wg.Wait()
    close(results)  // Signal no more results
}()

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

Exercise: Concurrent File Processor

Task: Build a concurrent file processor using the worker pool pattern.

Requirements:

Process 20 "files" (simulated with numbers)
Use 4 workers in a worker pool
Each file takes 200-500ms to process
Collect and display all results
Handle errors gracefully

Show Solution

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type FileResult struct {
    FileID int
    Size   int
    Err    error
}

func processFile(fileID int) (int, error) {
    // Simulate processing time
    time.Sleep(time.Duration(200+rand.Intn(300)) * time.Millisecond)
    
    // Simulate occasional errors
    if fileID%7 == 0 {
        return 0, fmt.Errorf("failed to process file %d", fileID)
    }
    
    return rand.Intn(1000), nil
}

func worker(id int, files <-chan int, results chan<- FileResult, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for fileID := range files {
        fmt.Printf("Worker %d processing file %d\n", id, fileID)
        size, err := processFile(fileID)
        results <- FileResult{FileID: fileID, Size: size, Err: err}
    }
}

func main() {
    const numWorkers = 4
    const numFiles = 20
    
    files := make(chan int, numFiles)
    results := make(chan FileResult, numFiles)
    var wg sync.WaitGroup
    
    // Start worker pool
    fmt.Printf("Starting %d workers...\n", numWorkers)
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, files, results, &wg)
    }
    
    // Send files to process
    for f := 1; f <= numFiles; f++ {
        files <- f
    }
    close(files)
    
    // Close results when all workers done
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // Collect results
    var totalSize int
    var errors []error
    
    for result := range results {
        if result.Err != nil {
            errors = append(errors, result.Err)
        } else {
            totalSize += result.Size
        }
    }
    
    // Display summary
    fmt.Printf("\n=== Processing Complete ===\n")
    fmt.Printf("Files processed: %d\n", numFiles-len(errors))
    fmt.Printf("Total size: %d bytes\n", totalSize)
    if len(errors) > 0 {
        fmt.Printf("Errors: %d\n", len(errors))
        for _, err := range errors {
            fmt.Printf("  - %v\n", err)
        }
    }
}

Summary

WaitGroup coordinates multiple goroutines (Add, Done, Wait)
Worker Pool limits concurrency with fixed workers
Fan-Out distributes work across multiple goroutines
Fan-In merges multiple channels into one
Pipeline chains processing stages together
Error Collection requires dedicated error channel
Bounded Concurrency uses semaphore pattern
Always Add() before starting goroutine
Always defer Done() to ensure cleanup
Close channels when no more data will be sent

What's Next?

You've mastered goroutine coordination patterns! These patterns form the foundation of production Go concurrency. Continue exploring advanced topics like generics and reflection to complete your Go expertise.

Previous Sync & Context

Next Packages & Modules

WaitGroup Coordination Pattern

Worker Pool Pattern

Fan-Out Pattern

Fan-In Pattern

Pipeline Pattern with WaitGroup

Error Handling in Concurrent Code

Bounded Concurrency

Common Mistakes

1. Adding to WaitGroup inside goroutine

2. Closing channel too early

3. Not closing results channel

Exercise: Concurrent File Processor

Summary

What's Next?

Enjoying these tutorials?