Go Concurrency Patterns(Timeouts and Cancellation Pattern)

Overview

The Timeouts and Cancellation pattern uses context.Context or channels to control the lifecycle of concurrent operations. This pattern is essential for preventing goroutines from running indefinitely, graceful shutdown of operations, resource cleanup and management, and building responsive applications.

NOTE: For other posts on concurrency patterns, check out the index post to this series of concurrency patterns.

Implementation Details

Structure

The timeouts and cancellation implementation in examples/timeout_cancellation.go demonstrates three main techniques:

  1. Context-based Timeout – Using context.WithTimeout for deadline-based cancellation
  2. Channel-based Timeout – Using time.After for simple timeout scenarios
  3. Context Cancellation – Using context.WithCancel for manual cancellation

Code Analysis

Let’s break down the main function and understand how each component works:

func RunTimeoutCancellation() {
    // Example 1: Context-based timeout
    fmt.Println("\n1. Context-based timeout example:")
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    result := make(chan string, 1)
    go longRunningTask(ctx, result)

    select {
    case res := <-result:
        fmt.Printf("Task completed: %s\n", res)
    case <-ctx.Done():
        fmt.Printf("Task timed out: %v\n", ctx.Err())
    }

    // Example 2: Channel-based timeout
    fmt.Println("\n2. Channel-based timeout example:")
    ch := make(chan string, 1)
    go func() {
        time.Sleep(3 * time.Second)
        ch <- "Channel task completed"
    }()

    select {
    case res := <-ch:
        fmt.Printf("Channel task: %s\n", res)
    case <-time.After(1 * time.Second):
        fmt.Println("Channel task timed out")
    }

    // Example 3: Context cancellation
    fmt.Println("\n3. Context cancellation example:")
    ctx2, cancel2 := context.WithCancel(context.Background())
    defer cancel2()

    go func() {
        time.Sleep(500 * time.Millisecond)
        fmt.Println("Cancelling context...")
        cancel2()
    }()

    select {
    case <-time.After(2 * time.Second):
        fmt.Println("Context cancellation example completed")
    case <-ctx2.Done():
        fmt.Printf("Context cancelled: %v\n", ctx2.Err())
    }
}

Step-by-step breakdown:

  1. Example 1: Context-based Timeout Setup:
    • ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) creates a context with a 2-second deadline
    • context.Background() provides the root context
    • defer cancel() ensures the context is cancelled when the function exits (cleanup)
    • This demonstrates the standard Go pattern for timeout management
  2. Context-based Task Execution:
    • result := make(chan string, 1) creates a buffered channel for the task result
    • Buffering prevents goroutine leaks if the task completes after timeout
    • go longRunningTask(ctx, result) launches the task with the timeout context
    • The task receives the context and can check for cancellation
  3. Context-based Result Handling:
    • Uses select statement to wait for either task completion or timeout
    • case res := <-result: handles successful task completion
    • case <-ctx.Done(): handles timeout or cancellation
    • ctx.Err() provides detailed error information about why the context was cancelled
  4. Example 2: Channel-based Timeout Setup:
    • ch := make(chan string, 1) creates a buffered channel for the channel-based example
    • This demonstrates an alternative approach without using context
  5. Channel-based Task Execution:
    • Launches an anonymous goroutine that sleeps for 3 seconds
    • time.Sleep(3 * time.Second) simulates a long-running task
    • ch <- "Channel task completed" sends the result when complete
    • This shows a simpler approach for basic timeout scenarios
  6. Channel-based Result Handling:
    • Uses select to race between task completion and timeout
    • case res := <-ch: handles successful completion
    • case <-time.After(1 * time.Second): handles timeout after 1 second
    • Note: task will timeout (1s) before completion (3s), demonstrating timeout behavior
  7. Example 3: Context Cancellation Setup:
    • ctx2, cancel2 := context.WithCancel(context.Background()) creates a cancellable context
    • defer cancel2() ensures cleanup
    • This demonstrates manual cancellation rather than timeout-based cancellation
  8. Context Cancellation Trigger:
    • Launches a goroutine that waits 500ms then calls cancel2()
    • time.Sleep(500 * time.Millisecond) simulates some condition that triggers cancellation
    • cancel2() manually cancels the context
    • This shows how external events can trigger cancellation
  9. Context Cancellation Detection:
    • Uses select to wait for either timeout or cancellation
    • case <-time.After(2 * time.Second): provides a fallback timeout
    • case <-ctx2.Done(): detects manual cancellation
    • ctx2.Err() provides the cancellation reason

Long-running Task Implementation

func longRunningTask(ctx context.Context, result chan<- string) {
    // Simulate work with random duration
    workTime := time.Duration(rand.Intn(3000)+1000) * time.Millisecond
    fmt.Printf("Starting long task (will take %v)...\n", workTime)

    select {
    case <-time.After(workTime):
        // Simulate random failure
        if rand.Float32() < 0.6 {
            fmt.Println("Long task cancelled!")
            return
        }
        result <- "Long task completed successfully"
    case <-ctx.Done():
        fmt.Printf("Long task cancelled: %v\n", ctx.Err())
        return
    }
}

Long-running task breakdown:

  1. Function Signature:
    • ctx context.Context: Receives the timeout context from the caller
    • result chan<- string: Write-only channel for sending the result
    • Directional channel prevents the task from accidentally reading from the result channel
  2. Work Simulation Setup:
    • workTime := time.Duration(rand.Intn(3000)+1000) * time.Millisecond creates variable work duration
    • Random duration between 1000-4000ms simulates real-world processing variability
    • This makes timeout behavior visible and demonstrates different scenarios
  3. Work Execution with Cancellation Check:
    • Uses select statement to handle both work completion and cancellation
    • case <-time.After(workTime): simulates the actual work taking time
    • case <-ctx.Done(): checks for context cancellation or timeout
    • This ensures the task can be cancelled at any time
  4. Random Failure Simulation:
    • if rand.Float32() < 0.6 creates a 60% probability of simulated failure
    • High failure rate demonstrates cancellation behavior effectively
    • return exits without sending a result when failure occurs
    • This simulates internal task failures
  5. Successful Completion Path:
    • If no failure occurs, result <- "Long task completed successfully" sends the result
    • Only sends result if the task actually completes successfully
    • Buffered channel prevents blocking if the main function has already timed out
  6. Cancellation Handling:
    • case <-ctx.Done(): detects when the context is cancelled or times out
    • ctx.Err() provides detailed error information (timeout, cancellation, etc.)
    • return exits immediately when cancellation is detected
    • This ensures the task stops work immediately when cancelled

Key Design Patterns:

  1. Context Propagation: The context is passed down to the task, enabling cancellation from the caller.
  2. Select Statement: Both the main function and task use select to handle multiple events non-blockingly.
  3. Buffered Result Channels: Buffering prevents goroutine leaks when timeouts occur before task completion.
  4. Defer Cleanup: defer cancel() ensures context cleanup even if the function panics or returns early.
  5. Error Information: ctx.Err() provides rich error information about why cancellation occurred.
  6. Immediate Cancellation: Tasks check ctx.Done() and exit immediately when cancellation is detected.
  7. Race Condition Handling: The select statement naturally handles the race between completion and timeout/cancellation.

How It Works

Context-based Timeout

  1. Context Creation: context.WithTimeout creates a context with a deadline
  2. Task Execution: The long-running task receives the context and monitors it
  3. Timeout Detection: The task checks ctx.Done() to detect timeout
  4. Graceful Cancellation: When timeout occurs, the task stops work and returns

Channel-based Timeout

  1. Task Launch: A goroutine starts the potentially long-running task
  2. Timeout Setup: time.After creates a channel that will receive after the timeout
  3. Race Condition: select waits for either task completion or timeout
  4. Timeout Handling: If timeout occurs first, the task is abandoned

Context Cancellation

  1. Context Creation: context.WithCancel creates a cancellable context
  2. Cancellation Trigger: Another goroutine calls cancel() after a delay
  3. Cancellation Detection: The main goroutine detects cancellation via ctx.Done()
  4. Immediate Response: Cancellation is detected immediately when triggered

Why This Implementation?

Context-based Approach

  • Standard Pattern: Uses Go’s standard context package
  • Propagation: Context can be passed down through call chains
  • Rich Information: Context provides error information and cancellation reasons
  • Integration: Works well with HTTP servers, gRPC, and other Go libraries

Channel-based Approach

  • Simplicity: Simple and straightforward for basic timeout scenarios
  • Performance: Minimal overhead for simple cases
  • Flexibility: Can be combined with other channel operations
  • Learning: Good for understanding timeout concepts

Select Statement

  • Non-blocking: Allows waiting for multiple events without blocking
  • Race Handling: Naturally handles the race between completion and timeout
  • Clean Code: Provides clear, readable timeout logic
  • Efficiency: Efficient event-driven waiting

Buffered Result Channels

  • Non-blocking Sends: Tasks can send results without blocking
  • Cleanup Prevention: Prevents goroutine leaks when timeout occurs
  • Flexibility: Allows for both synchronous and asynchronous result handling

Key Design Decisions

  1. Multiple Examples: Demonstrates different timeout and cancellation techniques
  2. Simulated Work: Random work duration makes timeout behavior visible
  3. Error Information: Context provides detailed error information
  4. Graceful Handling: Tasks check for cancellation and stop work appropriately
  5. Resource Cleanup: Proper defer statements ensure cleanup

Performance Characteristics

Context Overhead

  • Minimal Impact: Context operations have minimal performance overhead
  • Memory Usage: Context objects are lightweight
  • Propagation Cost: Context propagation through call chains is efficient

Channel Overhead

  • Low Latency: Channel operations are fast
  • Memory Usage: Channels use minimal memory
  • Scalability: Scales well with multiple concurrent operations

Timeout Accuracy

  • System Dependent: Accuracy depends on system timer resolution
  • Go Runtime: Go’s runtime provides good timer accuracy
  • Practical Limits: Sub-millisecond timeouts may not be accurate

Common Use Cases

HTTP Request Handling

  • Request Timeouts: Cancel requests that take too long
  • Client Timeouts: Set timeouts for outgoing HTTP requests
  • Server Timeouts: Cancel long-running server operations

Database Operations

  • Query Timeouts: Cancel slow database queries
  • Connection Timeouts: Timeout database connection attempts
  • Transaction Timeouts: Cancel long-running transactions

External API Calls

  • API Timeouts: Cancel slow external API calls
  • Rate Limiting: Cancel requests when rate limits are exceeded
  • Circuit Breaker: Cancel requests when external services are down

File Operations

  • Read Timeouts: Cancel slow file read operations
  • Write Timeouts: Cancel slow file write operations
  • Network File Systems: Handle slow network file system operations

Background Processing

  • Job Timeouts: Cancel long-running background jobs
  • Batch Processing: Timeout batch processing operations
  • Data Processing: Cancel slow data processing tasks

User Interface

  • User Input: Cancel operations when user cancels
  • UI Responsiveness: Ensure UI remains responsive during long operations
  • Progress Indicators: Cancel operations based on user feedback

System Operations

  • Backup Operations: Cancel slow backup operations
  • System Maintenance: Timeout maintenance operations
  • Resource Cleanup: Cancel resource cleanup operations

Microservices

  • Service Communication: Timeout inter-service communication
  • Health Checks: Cancel slow health check operations
  • Configuration Updates: Timeout configuration update operations

Best Practices

Context Usage

  • Always Pass Context: Pass context to all functions that can be cancelled
  • Check Cancellation: Regularly check ctx.Done() in long-running operations
  • Propagate Context: Pass context down through call chains
  • Use Timeouts: Set appropriate timeouts for all operations

Error Handling

  • Check Context Errors: Always check ctx.Err() for cancellation reasons
  • Log Cancellations: Log when operations are cancelled for debugging
  • Cleanup Resources: Ensure resources are cleaned up on cancellation
  • Return Errors: Return appropriate errors when operations are cancelled

Timeout Values

  • Appropriate Timeouts: Set timeouts based on expected operation duration
  • User Experience: Consider user experience when setting timeouts
  • System Resources: Consider system resource constraints
  • External Dependencies: Account for external service response times

Resource Management

  • Goroutine Cleanup: Ensure goroutines exit when cancelled
  • Memory Cleanup: Clean up memory allocations on cancellation
  • Connection Cleanup: Close connections when operations are cancelled
  • File Handle Cleanup: Close file handles on cancellation

The timeouts and cancellation pattern is particularly effective when you have:

  • Long-running Operations: Operations that may take longer than expected
  • External Dependencies: Operations that depend on external services
  • Resource Constraints: Need to limit resource usage
  • User Experience: Need to maintain responsive user interfaces
  • System Stability: Need to prevent resource exhaustion

This pattern provides essential tools for building robust, responsive applications that can handle failures gracefully and maintain system stability under various conditions.

The final example implementation looks like this:

package examples

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

// RunTimeoutCancellation demonstrates timeouts and cancellation patterns.
func RunTimeoutCancellation() {
	fmt.Println("=== Timeouts and Cancellation Pattern Example ===")

	// Example 1: Context-based timeout
	fmt.Println("\n1. Context-based timeout example:")
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	result := make(chan string, 1)
	go longRunningTask(ctx, result)

	select {
	case res := <-result:
		fmt.Printf("Task completed: %s\n", res)
	case <-ctx.Done():
		fmt.Printf("Task timed out: %v\n", ctx.Err())
	}

	// Example 2: Channel-based timeout
	fmt.Println("\n2. Channel-based timeout example:")
	ch := make(chan string, 1)
	go func() {
		time.Sleep(3 * time.Second)
		ch <- "Channel task completed"
	}()

	select {
	case res := <-ch:
		fmt.Printf("Channel task: %s\n", res)
	case <-time.After(1 * time.Second):
		fmt.Println("Channel task timed out")
	}

	// Example 3: Cancellation with context
	fmt.Println("\n3. Context cancellation example:")
	ctx2, cancel2 := context.WithCancel(context.Background())
	defer cancel2()

	go func() {
		time.Sleep(500 * time.Millisecond)
		fmt.Println("Cancelling context...")
		cancel2()
	}()

	select {
	case <-time.After(2 * time.Second):
		fmt.Println("Context cancellation example completed")
	case <-ctx2.Done():
		fmt.Printf("Context cancelled: %v\n", ctx2.Err())
	}

	fmt.Println("\nTimeouts and Cancellation example completed!")
}

// longRunningTask simulates a long-running task that respects context cancellation
func longRunningTask(ctx context.Context, result chan<- string) {
	// Simulate work with random duration
	workTime := time.Duration(rand.Intn(3000)+1000) * time.Millisecond
	fmt.Printf("Starting long task (will take %v)...\n", workTime)

	select {
	case <-time.After(workTime):
		result <- "Long task completed successfully"
	case <-ctx.Done():
		fmt.Printf("Long task cancelled: %v\n", ctx.Err())
		return
	}
}

To run this example, and build the code yourself, check out this and other examples in the go-fluency-concurrency-model-patterns repo. That’s it for this topic, tomorrow I’ll post on the rate limiting pattern.

Go Concurrency Patterns(Publish-Subscribe (Pub/Sub) Pattern)

Overview

The Publish-Subscribe (Pub/Sub) pattern decouples message producers (publishers) from consumers (subscribers). Publishers send messages to a topic or channel, and all subscribers to that topic receive the messages. This pattern is essential for event-driven architectures, broadcasting messages to multiple consumers, and decoupling senders and receivers.

NOTE: For other posts on concurrency patterns, check out the index post to this series of concurrency patterns.

Implementation Details

Structure

The publish-subscribe implementation in examples/pubsub.go consists of three main components:

  1. Broadcaster – Manages subscriptions and message distribution
  2. Publishers – Send messages to the broadcaster
  3. Subscribers – Receive messages from the broadcaster

Code Analysis

Let’s break down the main function and understand how each component works:

func RunPubSub() {
    // Create a broadcaster
    b := newBroadcaster()

    numSubscribers := 3
    var wg sync.WaitGroup

    // Start subscribers
    for i := 1; i <= numSubscribers; i++ {
        ch := b.subscribe()
        wg.Add(1)
        go func(id int, ch <-chan string) {
            defer wg.Done()
            for msg := range ch {
                fmt.Printf("Subscriber %d received: %s\n", id, msg)
            }
            fmt.Printf("Subscriber %d done.\n", id)
        }(i, ch)
    }

    // Start publisher
    go func() {
        for i := 1; i <= 5; i++ {
            msg := fmt.Sprintf("Message %d", i)
            fmt.Printf("Publisher sending: %s\n", msg)
            b.publish(msg)
            time.Sleep(400 * time.Millisecond)
        }
        b.close()
    }()

    wg.Wait()
}

Step-by-step breakdown:

  1. Broadcaster Initialization:
    • b := newBroadcaster() creates a new broadcaster instance
    • The broadcaster manages all subscriptions and message distribution
    • This is the central hub that decouples publishers from subscribers
  2. Subscriber Configuration:
    • numSubscribers := 3 defines how many subscribers to create
    • var wg sync.WaitGroup tracks when all subscribers finish processing
    • This ensures the main function waits for all subscribers to complete
  3. Subscriber Launch Loop:
    • Launches numSubscribers goroutines (3 in this case)
    • ch := b.subscribe() creates a new subscription channel for each subscriber
    • Each subscriber gets a unique ID (1, 2, 3) for tracking and debugging
    • Uses closure func(id int, ch <-chan string) { ... }(i, ch) to capture the subscriber ID and channel
  4. Subscriber Goroutine Implementation:
    • defer wg.Done() ensures the subscriber signals completion when it exits
    • for msg := range ch continuously reads messages until the channel closes
    • Each subscriber processes messages independently at its own pace
    • Prints completion message when the channel closes
  5. Publisher Goroutine Launch:
    • Runs in background to send messages asynchronously
    • Publishes 5 messages with 400ms delays between them
    • This simulates real-world message publishing with timing
  6. Message Publishing Loop:
    • for i := 1; i <= 5; i++ sends exactly 5 messages
    • msg := fmt.Sprintf("Message %d", i) creates numbered messages
    • b.publish(msg) broadcasts the message to all subscribers
    • time.Sleep(400 * time.Millisecond) simulates message generation time
  7. Graceful Shutdown:
    • b.close() closes the broadcaster and all subscriber channels
    • This signals all subscribers to stop processing and exit
    • wg.Wait() waits for all subscribers to finish before the main function exits

Broadcaster Implementation

type broadcaster struct {
    subscribers []chan string
    closed     bool
    mu         sync.Mutex
}

func newBroadcaster() *broadcaster {
    return &broadcaster{
        subscribers: make([]chan string, 0),
    }
}

func (b *broadcaster) subscribe() <-chan string {
    b.mu.Lock()
    defer b.mu.Unlock()
    ch := make(chan string, 2)
    b.subscribers = append(b.subscribers, ch)
    return ch
}

func (b *broadcaster) publish(msg string) {
    b.mu.Lock()
    defer b.mu.Unlock()
    if b.closed {
        return
    }
    for _, ch := range b.subscribers {
        ch <- msg
    }
}

func (b *broadcaster) close() {
    b.mu.Lock()
    defer b.mu.Unlock()
    if b.closed {
        return
    }
    for _, ch := range b.subscribers {
        close(ch)
    }
    b.closed = true
}

Broadcaster function breakdown:

  1. Data Structure Design:
    • subscribers []chan string: Slice of channels, one per subscriber
    • closed bool: Flag to prevent operations after shutdown
    • mu sync.Mutex: Protects concurrent access to subscriber list
    • Simple slice-based design is efficient for typical use cases
  2. Constructor Function:
    • newBroadcaster() creates a new broadcaster instance
    • Initializes empty subscriber slice with make([]chan string, 0)
    • Returns pointer to broadcaster for method calls
  3. Subscription Management:
    • subscribe() creates a new subscription channel for each subscriber
    • b.mu.Lock() and defer b.mu.Unlock() ensure thread-safe subscriber list access
    • ch := make(chan string, 2) creates a buffered channel with capacity 2
    • Buffering prevents blocking if subscriber is temporarily slow
    • b.subscribers = append(b.subscribers, ch) adds the new channel to the list
    • Returns <-chan string (read-only) for encapsulation
  4. Message Publishing Logic:
    • publish(msg string) broadcasts a message to all subscribers
    • Thread-safe access with mutex protection
    • if b.closed { return } prevents publishing after shutdown
    • for _, ch := range b.subscribers iterates through all subscriber channels
    • ch <- msg sends the message to each subscriber
    • Non-blocking due to buffered channels
  5. Graceful Shutdown Process:
    • close() coordinates shutdown of all subscribers
    • Thread-safe access with mutex protection
    • if b.closed { return } prevents double-closing
    • for _, ch := range b.subscribers { close(ch) } closes all subscriber channels
    • b.closed = true marks the broadcaster as closed
    • Closing channels signals subscribers to stop processing

Key Design Patterns:

  1. Slice-based Subscriber Management: Simple and efficient for typical pub/sub use cases, easy to add/remove subscribers.
  2. Mutex Protection: Ensures thread-safe access to the subscriber list during concurrent subscribe/publish operations.
  3. Buffered Channels: Each subscriber gets a buffered channel (capacity 2) to prevent blocking during message distribution.
  4. Read-only Channel Returns: subscribe() returns <-chan string to prevent subscribers from accidentally closing their channels.
  5. Immediate Message Distribution: Messages are sent to all subscribers immediately upon publishing, providing real-time delivery.
  6. Graceful Shutdown: Proper channel closing ensures all subscribers exit cleanly without goroutine leaks.
  7. Closure Pattern: func(id int, ch <-chan string) { ... }(i, ch) captures loop variables in each subscriber goroutine.

How It Works

  1. Broadcaster Creation: A broadcaster is created to manage subscriptions and message distribution
  2. Subscriber Registration: Subscribers call subscribe() to get their own message channel
  3. Message Publishing: Publishers call publish() to send messages to all subscribers
  4. Message Distribution: The broadcaster sends each message to all subscriber channels
  5. Graceful Shutdown: When the broadcaster is closed, all subscriber channels are closed

The pattern enables loose coupling between publishers and subscribers, allowing for flexible message distribution.

Why This Implementation?

Individual Subscriber Channels

  • Isolation: Each subscriber has its own channel, preventing blocking
  • Independent Processing: Subscribers can process messages at their own pace
  • Scalability: Easy to add or remove subscribers without affecting others

Mutex-based Synchronization

  • Thread Safety: Protects subscriber list during concurrent operations
  • Simple Implementation: Straightforward synchronization for the subscriber list
  • Performance: Minimal overhead for the typical pub/sub use case

Buffered Subscriber Channels

  • Non-blocking Publishing: Publishers don’t block if subscribers are slow
  • Message Buffering: Can handle temporary subscriber slowdowns
  • Flow Control: Natural backpressure through channel buffering

Centralized Message Distribution

  • Broadcast Semantics: All subscribers receive all messages
  • Simple Coordination: Single point of message distribution
  • Consistent Delivery: All subscribers receive messages in the same order

Graceful Shutdown

  • Clean Termination: Properly closes all subscriber channels
  • Resource Cleanup: Prevents goroutine leaks
  • Coordinated Exit: All subscribers exit when broadcaster closes

Key Design Decisions

  1. Slice-based Subscriber Management: Simple and efficient for typical use cases
  2. Read-only Channel Returns: Subscribers receive read-only channels for safety
  3. Buffered Channels: Prevents blocking during message distribution
  4. Mutex Protection: Ensures thread-safe subscriber management
  5. Immediate Message Distribution: Messages are sent to all subscribers immediately

Performance Characteristics

Throughput

  • Linear Scaling: Throughput scales linearly with the number of subscribers
  • Channel Overhead: Each subscriber adds minimal overhead
  • Publishing Performance: Publishing is O(n) where n is the number of subscribers

Latency

  • Immediate Distribution: Messages are distributed immediately upon publishing
  • Subscriber Processing: Latency depends on individual subscriber processing speed
  • Channel Buffering: Buffered channels can absorb temporary processing delays

Memory Usage

  • Per-Subscriber Overhead: Each subscriber requires a channel and goroutine
  • Message Buffering: Buffered channels use additional memory
  • Scalability: Memory usage grows linearly with subscriber count

Common Use Cases

Event-Driven Systems

  • User Interface Events: Mouse clicks, keyboard input, window events
  • System Events: File changes, network events, timer events
  • Application Events: User actions, state changes, error conditions

Message Broadcasting

  • Chat Applications: Broadcast messages to all connected users
  • Notification Systems: Send notifications to multiple recipients
  • Status Updates: Broadcast system status to all monitoring clients

Data Distribution

  • Real-time Data: Distribute sensor data, market data, or metrics
  • Configuration Changes: Broadcast configuration updates to all services
  • Cache Invalidation: Notify all cache instances of data changes

Microservices Communication

  • Service Discovery: Notify services of new service instances
  • Health Updates: Broadcast health status changes
  • Load Balancing: Distribute load information to all instances

Monitoring and Logging

  • Metric Collection: Distribute metrics to multiple monitoring systems
  • Log Aggregation: Send logs to multiple destinations
  • Alert Distribution: Broadcast alerts to multiple notification channels

Gaming and Real-time Applications

  • Game State Updates: Broadcast game state changes to all players
  • Multiplayer Events: Distribute player actions to all participants
  • Real-time Collaboration: Share changes across multiple users

IoT and Sensor Networks

  • Sensor Data: Distribute sensor readings to multiple processors
  • Device Status: Broadcast device status changes
  • Control Commands: Distribute control commands to multiple devices

Financial Systems

  • Market Data: Distribute real-time market data to multiple traders
  • Trade Updates: Broadcast trade execution to all relevant systems
  • Risk Alerts: Distribute risk alerts to multiple risk managers

Advanced Patterns

Topic-based Pub/Sub

  • Multiple Topics: Support for different message categories
  • Selective Subscription: Subscribers can choose which topics to receive
  • Hierarchical Topics: Support for topic hierarchies (e.g., “sports.football.nfl”)

Filtered Pub/Sub

  • Message Filtering: Subscribers can filter messages based on content
  • Conditional Delivery: Only deliver messages that match subscriber criteria
  • Performance Optimization: Reduce unnecessary message processing

Persistent Pub/Sub

  • Message Persistence: Store messages for offline subscribers
  • Reliable Delivery: Ensure messages are delivered even if subscribers are temporarily unavailable
  • Message Ordering: Maintain message order across subscriber reconnections

Distributed Pub/Sub

  • Multi-node Distribution: Distribute messages across multiple nodes
  • Load Balancing: Balance message distribution across multiple brokers
  • Fault Tolerance: Continue operation even if some nodes fail

The publish-subscribe pattern is particularly effective when you have:

  • Loose Coupling: Components that should be independent of each other
  • Multiple Consumers: Multiple components that need the same information
  • Event-driven Architecture: Systems that respond to events rather than direct calls
  • Scalable Communication: Need to add/remove consumers without affecting publishers
  • Broadcast Requirements: Need to send the same message to multiple recipients

This pattern provides a flexible foundation for building decoupled, event-driven systems that can scale to handle multiple publishers and subscribers efficiently.

The final example implementation looks like this:

package examples

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

// RunPubSub demonstrates the publish-subscribe (pub/sub) pattern.
func RunPubSub() {
	fmt.Println("=== Publish-Subscribe (Pub/Sub) Pattern Example ===")

	// Create a broadcaster
	b := newBroadcaster()

	numSubscribers := 3
	var wg sync.WaitGroup

	// Start subscribers
	for i := 1; i <= numSubscribers; i++ {
		ch := b.subscribe()
		wg.Add(1)
		go func(id int, ch <-chan string) {
			defer wg.Done()
			for msg := range ch {
				fmt.Printf("Subscriber %d received: %s\n", id, msg)
			}
			fmt.Printf("Subscriber %d done.\n", id)
		}(i, ch)
	}

	// Start publisher
	go func() {
		for i := 1; i <= 5; i++ {
			msg := fmt.Sprintf("Message %d", i)
			fmt.Printf("Publisher sending: %s\n", msg)
			b.publish(msg)
			time.Sleep(400 * time.Millisecond)
		}
		b.close()
	}()

	wg.Wait()
	fmt.Println("Pub/Sub example completed!")
}

// broadcaster manages subscriptions and publishing
// Not thread-safe for subscribe after close

type broadcaster struct {
	subscribers []chan string
	closed      bool
	mu          sync.Mutex
}

func newBroadcaster() *broadcaster {
	return &broadcaster{
		subscribers: make([]chan string, 0),
	}
}

func (b *broadcaster) subscribe() <-chan string {
	b.mu.Lock()
	defer b.mu.Unlock()
	ch := make(chan string, 2)
	b.subscribers = append(b.subscribers, ch)
	return ch
}

func (b *broadcaster) publish(msg string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		return
	}
	for _, ch := range b.subscribers {
		ch <- msg
	}
}

func (b *broadcaster) close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		return
	}
	for _, ch := range b.subscribers {
		close(ch)
	}
	b.closed = true
}

To run this example, and build the code yourself, check out this and other examples in the go-fluency-concurrency-model-patterns repo. That’s it for this topic, tomorrow I’ll post on the timeouts and cancellations pattern.

Go Concurrency Patterns(Supervisor/Restart Pattern)

Overview

The Supervisor/Restart pattern involves a supervisor goroutine that monitors one or more worker goroutines. If a worker fails (panics or exits unexpectedly), the supervisor restarts it. This pattern is essential for building resilient and fault-tolerant systems, automatically recovering from transient errors, and ensuring critical tasks are always running.

NOTE: For other posts on concurrency patterns, check out the index post to this series of concurrency patterns.

Implementation Details

Structure

The supervisor/restart implementation in examples/supervisor.go consists of three main components:

  1. Supervisor – A goroutine that monitors and manages workers
  2. Worker – A goroutine that performs work and may fail
  3. Coordination – Channels and synchronization for monitoring and restart

Code Analysis

Let’s break down the main function and understand how each component works:

func RunSupervisor() {
    var restarts int32
    stop := make(chan struct{})
    done := make(chan struct{})

    // Supervisor goroutine
    go func() {
        for {
            workerDone := make(chan struct{})
            go workerWithFailure(workerDone, stop)
            select {
            case <-workerDone:
                atomic.AddInt32(&restarts, 1)
                fmt.Println("Supervisor: Worker failed, restarting...")
                // Restart after a short delay
                time.Sleep(500 * time.Millisecond)
            case <-stop:
                fmt.Println("Supervisor: Stopping worker supervision.")
                close(done)
                return
            }
        }
    }()

    // Let the supervisor run for a while
    time.Sleep(4 * time.Second)
    close(stop)
    <-done

    fmt.Printf("Supervisor example completed! Worker was restarted %d times.\n", restarts-1)
}

Step-by-step breakdown:

  1. State and Channel Setup:
    • var restarts int32 creates an atomic counter to track restart attempts
    • stop := make(chan struct{}) creates a signal channel for graceful shutdown
    • done := make(chan struct{}) creates a completion channel for supervisor coordination
    • These channels use struct{} as the type for efficiency (zero memory allocation)
  2. Supervisor Goroutine Launch:
    • The supervisor runs in its own goroutine to monitor workers continuously
    • Uses an infinite for loop to continuously restart workers as needed
    • This ensures the supervisor is always ready to handle worker failures
  3. Worker Lifecycle Management:
    • workerDone := make(chan struct{}) creates a fresh channel for each worker
    • Each worker gets its own done channel to signal completion or failure
    • go workerWithFailure(workerDone, stop) launches a new worker goroutine
    • The worker receives both the done channel (to signal back) and stop channel (for shutdown)
  4. Supervisor Monitoring Loop:
    • Uses select statement to wait for either worker completion or stop signal
    • case <-workerDone: handles worker failure or completion
    • case <-stop: handles graceful shutdown request
    • This provides non-blocking monitoring of multiple events
  5. Failure Handling and Restart Logic:
    • atomic.AddInt32(&restarts, 1) safely increments the restart counter
    • Atomic operation ensures thread safety without mutex overhead
    • time.Sleep(500 * time.Millisecond) provides a restart delay
    • Delay prevents “thundering herd” – rapid restart loops that could overwhelm the system
  6. Graceful Shutdown Process:
    • close(done) signals that the supervisor has finished its work
    • return exits the supervisor goroutine
    • This ensures clean termination without goroutine leaks
  7. Main Function Coordination:
    • time.Sleep(4 * time.Second) lets the supervisor run for a demonstration period
    • close(stop) signals the supervisor to stop monitoring
    • <-done waits for the supervisor to finish its shutdown process
    • restarts-1 accounts for the initial worker launch (not a restart)

Worker Implementation

func workerWithFailure(done chan<- struct{}, stop <-chan struct{}) {
    fmt.Println("Worker: Started")
    workTime := time.Duration(rand.Intn(1200)+400) * time.Millisecond
    select {
    case <-time.After(workTime):
        // Simulate random failure
        if rand.Float32() < 0.6 {
            fmt.Println("Worker: Simulated failure!")
            done <- struct{}{}
            return
        }
        fmt.Println("Worker: Completed work successfully.")
    case <-stop:
        fmt.Println("Worker: Received stop signal.")
    }
    // Signal normal exit
    done <- struct{}{}
}

Worker function breakdown:

  1. Function Signature:
    • done chan<- struct{}: Write-only channel for signaling completion/failure to supervisor
    • stop <-chan struct{}: Read-only channel for receiving shutdown signals
    • Directional channels provide encapsulation and prevent misuse
  2. Worker Initialization:
    • Prints startup message for debugging and monitoring
    • workTime := time.Duration(rand.Intn(1200)+400) * time.Millisecond creates variable work duration
    • Random duration between 400-1600ms simulates real-world processing variability
  3. Work Execution with Timeout:
    • Uses select statement to handle both work completion and stop signals
    • case <-time.After(workTime): simulates work that takes a variable amount of time
    • case <-stop: handles graceful shutdown requests
  4. Failure Simulation Logic:
    • if rand.Float32() < 0.6 creates a 60% probability of failure
    • High failure rate (60%) demonstrates the restart mechanism effectively
    • done <- struct{}{} immediately signals failure to supervisor
    • return exits the worker without sending a second done signal
  5. Successful Completion Path:
    • If no failure occurs, worker prints success message
    • Continues to the final done <- struct{}{} to signal normal completion
    • This distinguishes between failure and successful completion
  6. Stop Signal Handling:
    • If stop signal is received, worker prints acknowledgment
    • Continues to final done <- struct{}{} to signal graceful shutdown
    • This ensures supervisor knows the worker has stopped
  7. Final Signal:
    • done <- struct{}{} signals completion for both success and graceful shutdown cases
    • Ensures supervisor always receives a signal when worker exits
    • Prevents supervisor from waiting indefinitely

Key Design Patterns:

  1. Channel Direction Safety: Using directional channels (chan<- and <-chan) prevents accidental misuse and provides clear API contracts.
  2. Atomic Counter: Using atomic.AddInt32 for restart counting provides thread safety without mutex overhead.
  3. Select Statement: The select statement in both supervisor and worker provides non-blocking event handling for multiple channels.
  4. Fresh Channel per Worker: Creating a new workerDone channel for each worker ensures clean communication and prevents signal confusion.
  5. Graceful Shutdown: Both supervisor and worker respond to stop signals, ensuring clean resource cleanup.
  6. Failure Simulation: High failure rate (60%) with variable work time demonstrates the restart mechanism effectively in a short demonstration period.

How It Works

  1. Supervisor Initialization: The supervisor starts and launches the first worker
  2. Worker Execution: The worker performs its task and may fail randomly
  3. Failure Detection: When a worker fails, it signals the supervisor via a channel
  4. Restart Logic: The supervisor waits briefly, then launches a new worker
  5. Continuous Monitoring: This cycle continues until the supervisor receives a stop signal
  6. Graceful Shutdown: The supervisor stops launching new workers and waits for current workers to finish

The pattern ensures that critical work continues even when individual workers fail.

Why This Implementation?

Channel-based Communication

  • Non-blocking Monitoring: Supervisor can monitor multiple workers without blocking
  • Immediate Failure Detection: Workers can signal failure immediately
  • Clean Coordination: Channels provide natural synchronization

Atomic Counter for Restarts

  • Thread Safety: Atomic operations ensure accurate restart counting
  • Performance: Atomic operations are more efficient than mutex-based counting
  • Simplicity: No need for complex synchronization around the counter

Separate Stop Channel

  • Graceful Shutdown: Supervisor can stop monitoring without waiting for failures
  • Clean Termination: Workers receive stop signals and can clean up properly
  • Resource Management: Prevents infinite restart loops

Worker Done Channel

  • Failure Signaling: Workers signal completion or failure to supervisor
  • Immediate Response: Supervisor can react immediately to worker state changes
  • Flexible Communication: Can distinguish between normal completion and failure

Restart Delay

  • Prevent Thundering Herd: Brief delay prevents rapid restart loops
  • Resource Recovery: Gives system time to recover from transient issues
  • Stability: Prevents overwhelming the system with restart attempts

Key Design Decisions

  1. Simulated Failures: Random failures (60% probability) demonstrate the restart mechanism
  2. Variable Work Time: Random work duration simulates real-world variability
  3. Immediate Restart: Supervisor restarts workers immediately after failure detection
  4. Stop Signal Handling: Both supervisor and workers respond to stop signals
  5. Restart Counting: Tracks the number of restarts for monitoring purposes

Failure Handling Strategies

Immediate Restart

  • Pros: Minimal downtime, immediate recovery
  • Cons: May restart into the same failure condition
  • Use Case: Transient failures, network issues

Exponential Backoff

  • Pros: Prevents overwhelming the system, allows for recovery
  • Cons: Longer downtime for legitimate failures
  • Use Case: Persistent failures, resource exhaustion

Circuit Breaker

  • Pros: Prevents cascading failures, allows system recovery
  • Cons: More complex implementation
  • Use Case: External service failures, resource constraints

Health Checks

  • Pros: Proactive failure detection, better monitoring
  • Cons: Additional complexity, false positives
  • Use Case: Long-running processes, critical systems

Common Use Cases

System Services

  • Background Workers: Ensure critical background tasks continue running
  • Data Processing: Restart failed data processing jobs
  • Scheduled Tasks: Ensure scheduled operations complete successfully

Microservices

  • Service Health: Monitor and restart unhealthy service instances
  • API Endpoints: Ensure critical API endpoints remain available
  • Message Processors: Restart failed message processing workers

Monitoring Systems

  • Health Monitors: Ensure monitoring agents continue running
  • Alert Processors: Restart failed alert processing workers
  • Metric Collectors: Ensure metric collection continues

Database Operations

  • Connection Managers: Restart failed database connection pools
  • Migration Workers: Ensure database migrations complete
  • Backup Jobs: Restart failed backup operations

File Processing

  • Upload Handlers: Restart failed file upload processors
  • Download Workers: Ensure file downloads complete
  • Processing Jobs: Restart failed file processing tasks

Network Services

  • Connection Handlers: Restart failed network connection handlers
  • Protocol Processors: Ensure protocol processing continues
  • Load Balancers: Restart failed load balancing workers

IoT Applications

  • Device Managers: Restart failed device communication workers
  • Sensor Processors: Ensure sensor data processing continues
  • Control Systems: Restart failed control system workers

Financial Systems

  • Trading Engines: Restart failed trading algorithm workers
  • Risk Calculators: Ensure risk calculations continue
  • Settlement Processors: Restart failed settlement workers

Best Practices

Failure Detection

  • Immediate Signaling: Workers should signal failure immediately
  • Graceful Degradation: Workers should handle partial failures gracefully
  • Health Reporting: Workers should report their health status

Restart Strategy

  • Appropriate Delays: Use delays to prevent restart storms
  • Backoff Policies: Implement exponential backoff for persistent failures
  • Maximum Restarts: Limit the number of restarts to prevent infinite loops

Resource Management

  • Cleanup: Ensure workers clean up resources before exiting
  • Memory Management: Monitor memory usage to prevent leaks
  • Connection Management: Properly close connections and file handles

Monitoring and Logging

  • Restart Tracking: Log restart events for monitoring
  • Performance Metrics: Track worker performance and restart frequency
  • Alerting: Alert on excessive restart rates

The supervisor/restart pattern is particularly effective when you have:

  • Critical Processes: Tasks that must continue running
  • Unreliable Workers: Workers that may fail due to external factors
  • Long-running Systems: Systems that need to operate continuously
  • Fault Tolerance: Requirements for high availability and reliability
  • Automatic Recovery: Need for self-healing systems

This pattern provides a robust foundation for building resilient systems that can automatically recover from failures and maintain continuous operation.

The final example implementation looks like this:

package examples

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

// RunSupervisor demonstrates the supervisor/restart pattern.
func RunSupervisor() {
	fmt.Println("=== Supervisor/Restart Pattern Example ===")

	var restarts int32
	stop := make(chan struct{})
	done := make(chan struct{})

	// Supervisor goroutine
	go func() {
		for {
			workerDone := make(chan struct{})
			go workerWithFailure(workerDone, stop)
			select {
			case <-workerDone:
				atomic.AddInt32(&restarts, 1)
				fmt.Println("Supervisor: Worker failed, restarting...")
				// Restart after a short delay
				time.Sleep(500 * time.Millisecond)
			case <-stop:
				fmt.Println("Supervisor: Stopping worker supervision.")
				close(done)
				return
			}
		}
	}()

	// Let the supervisor run for a while
	time.Sleep(4 * time.Second)
	close(stop)
	<-done

	fmt.Printf("Supervisor example completed! Worker was restarted %d times.\n", restarts-1)
}

// workerWithFailure simulates a worker that randomly fails
func workerWithFailure(done chan<- struct{}, stop <-chan struct{}) {
	fmt.Println("Worker: Started")
	workTime := time.Duration(rand.Intn(1200)+400) * time.Millisecond
	select {
	case <-time.After(workTime):
		// Simulate random failure
		if rand.Float32() < 0.6 {
			fmt.Println("Worker: Simulated failure!")
			done <- struct{}{}
			return
		}
		fmt.Println("Worker: Completed work successfully.")
	case <-stop:
		fmt.Println("Worker: Received stop signal.")
	}
	// Signal normal exit
	done <- struct{}{}
}

To run this example, and build the code yourself, check out this and other examples in the go-fluency-concurrency-model-patterns repo. That’s it for this topic, tomorrow I’ll post on the pub/sub pattern.

Go Concurrency Patterns(Producer-Consumer Pattern)

Overview

The Producer-Consumer pattern decouples the production of data from its consumption using a shared buffer (channel). Multiple producers can generate data and send it to the buffer, while multiple consumers read from the buffer and process the data. This pattern is essential for buffering bursts of work, decoupling work rates between producers and consumers, and load leveling.

NOTE: For other posts on concurrency patterns, check out the index post to this series of concurrency patterns.

Implementation Details

Structure

The producer-consumer implementation in examples/producer_consumer.go consists of three main components:

  1. Producers – Multiple goroutines that generate data
  2. Buffer – A shared channel that holds the data
  3. Consumers – Multiple goroutines that process the data

Code Analysis

Let’s break down the main function and understand how each component works:

func RunProducerConsumer() {
    bufferSize := 5
    numProducers := 2
    numConsumers := 3
    numItems := 10

    buffer := make(chan int, bufferSize)
    var wg sync.WaitGroup

    // Start producers
    for p := 1; p <= numProducers; p++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for i := 0; i < numItems; i++ {
                item := rand.Intn(100)
                buffer <- item
                fmt.Printf("Producer %d produced: %d\n", id, item)
                time.Sleep(time.Duration(rand.Intn(200)+100) * time.Millisecond)
            }
        }(p)
    }

    // Start consumers
    var consumerWg sync.WaitGroup
    for c := 1; c <= numConsumers; c++ {
        consumerWg.Add(1)
        go func(id int) {
            defer consumerWg.Done()
            for item := range buffer {
                fmt.Printf("Consumer %d consumed: %d\n", id, item)
                time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond)
            }
        }(c)
    }

    // Wait for all producers to finish, then close the buffer
    wg.Wait()
    close(buffer)

    // Wait for all consumers to finish
    consumerWg.Wait()
}

Step-by-step breakdown:

  1. Configuration Setup:
    • bufferSize := 5 defines the buffer capacity – can hold up to 5 items
    • numProducers := 2 launches 2 producer goroutines
    • numConsumers := 3 launches 3 consumer goroutines
    • numItems := 10 specifies how many items each producer will create
    • These values create a scenario where producers generate more data than consumers can immediately process
  2. Buffer and Synchronization Setup:
    • buffer := make(chan int, bufferSize) creates a buffered channel with capacity 5
    • Buffering allows producers to continue working even when consumers are slow
    • var wg sync.WaitGroup tracks when all producers complete
    • var consumerWg sync.WaitGroup tracks when all consumers complete
  3. Producer Launch Loop:
    • Launches numProducers goroutines (2 in this case)
    • Each producer gets a unique ID (1, 2) for tracking
    • Uses closure func(id int) { ... }(p) to capture the producer ID
    • wg.Add(1) increments the producer wait group before each goroutine
  4. Producer Goroutine Implementation:
    • defer wg.Done() ensures the producer signals completion when it exits
    • for i := 0; i < numItems; i++ loop creates exactly 10 items per producer
    • item := rand.Intn(100) generates random data (0-99) to simulate real data
    • buffer <- item sends the item to the shared buffer (may block if buffer is full)
    • time.Sleep(time.Duration(rand.Intn(200)+100) * time.Millisecond) simulates variable production time (100-300ms)
  5. Consumer Launch Loop:
    • Launches numConsumers goroutines (3 in this case)
    • Each consumer gets a unique ID (1, 2, 3) for tracking
    • Uses closure to capture the consumer ID
    • consumerWg.Add(1) increments the consumer wait group before each goroutine
  6. Consumer Goroutine Implementation:
    • defer consumerWg.Done() ensures the consumer signals completion when it exits
    • for item := range buffer continuously reads from the buffer until it closes
    • time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond) simulates variable processing time (100-400ms)
    • Consumers compete for items from the shared buffer (automatic load balancing)
  7. Graceful Shutdown Process:
    • wg.Wait() waits for all producers to finish generating their items
    • close(buffer) signals to consumers that no more data is coming
    • consumerWg.Wait() waits for all consumers to finish processing remaining items
    • This ensures clean termination without goroutine leaks

Key Design Patterns:

  1. Closure Pattern: func(id int) { ... }(p) captures the loop variable p in each goroutine’s closure, ensuring each goroutine gets a unique ID.
  2. Dual WaitGroup Strategy: Separate wait groups for producers and consumers allow for coordinated shutdown – producers finish first, then buffer closes, then consumers finish.
  3. Buffered Channel as Queue: The buffer decouples production and consumption rates, allowing the system to handle temporary mismatches in speed.
  4. Range Loop for Consumers: for item := range buffer automatically exits when the buffer closes, providing natural termination.
  5. Simulated Work: Random delays make concurrency visible and demonstrate how the pattern handles varying processing speeds.

How It Works

  1. Producer Initialization: Multiple producer goroutines start generating data
  2. Data Production: Producers create data and send it to the shared buffer channel
  3. Buffer Management: The buffered channel holds data until consumers are ready
  4. Consumer Processing: Multiple consumer goroutines read from the buffer and process data
  5. Graceful Shutdown: When all producers finish, the buffer is closed, signaling consumers to stop

The pattern enables asynchronous processing where producers and consumers can operate at different rates.

Why This Implementation?

Buffered Channel as Buffer

  • Decoupling: Producers and consumers can operate independently
  • Burst Handling: Buffer can absorb temporary spikes in production
  • Flow Control: Natural backpressure when buffer is full

Multiple Producers

  • Parallel Production: Multiple sources can generate data simultaneously
  • Load Distribution: Work can be distributed across multiple producers
  • Fault Tolerance: If one producer fails, others continue

Multiple Consumers

  • Parallel Processing: Multiple consumers can process data simultaneously
  • Load Balancing: Work is automatically distributed among consumers
  • Scalability: Easy to add or remove consumers based on load

Separate WaitGroups

  • Producer Coordination: Ensures all producers finish before closing buffer
  • Consumer Coordination: Ensures all consumers finish before program exits
  • Clean Shutdown: Proper coordination prevents goroutine leaks

Channel Closing Strategy

  • Signal Completion: Closing the buffer signals consumers that no more data is coming
  • Graceful Termination: Consumers naturally exit when buffer is closed
  • Resource Cleanup: Prevents consumers from waiting indefinitely

Key Design Decisions

  1. Buffered Channel Size: The buffer size (5) determines how much data can be queued
  2. Producer Count: Multiple producers (2) demonstrate parallel data generation
  3. Consumer Count: Multiple consumers (3) demonstrate parallel data processing
  4. Simulated Work: Random delays simulate real processing time and make concurrency visible
  5. Structured Shutdown: Proper coordination ensures clean program termination

Performance Characteristics

Throughput

  • Limited by Slowest Component: Overall throughput is limited by the slowest producer or consumer
  • Buffer Impact: Larger buffers can handle bigger bursts but use more memory
  • Parallel Processing: Multiple consumers can increase processing throughput

Latency

  • Buffer Time: Data may wait in the buffer if consumers are slow
  • Processing Time: Individual item processing time depends on consumer speed
  • Fair Distribution: Items are consumed in FIFO order

Resource Usage

  • Memory: Buffer size determines memory usage
  • CPU: Multiple producers and consumers can utilize multiple CPU cores
  • Coordination Overhead: Minimal overhead from channel operations

Common Use Cases

Data Processing Pipelines

  • Log Processing: Multiple log sources → Buffer → Multiple log processors
  • Image Processing: Multiple cameras → Buffer → Multiple image analyzers
  • Sensor Data: Multiple sensors → Buffer → Multiple data processors

Message Queuing Systems

  • Event Processing: Multiple event sources → Buffer → Multiple event handlers
  • Notification Systems: Multiple notification sources → Buffer → Multiple delivery agents
  • Webhook Processing: Multiple webhook sources → Buffer → Multiple webhook handlers

Batch Processing

  • File Processing: Multiple file sources → Buffer → Multiple file processors
  • Report Generation: Multiple data sources → Buffer → Multiple report generators
  • Data Import: Multiple data sources → Buffer → Multiple import processors

Real-time Systems

  • Trading Systems: Multiple market data feeds → Buffer → Multiple trading algorithms
  • IoT Applications: Multiple device sensors → Buffer → Multiple data analyzers
  • Monitoring Systems: Multiple metric sources → Buffer → Multiple alert processors

API Rate Limiting

  • Request Processing: Multiple API clients → Buffer → Rate-limited API processors
  • Data Synchronization: Multiple sync sources → Buffer → Controlled sync processors
  • External Service Calls: Multiple call sources → Buffer → Rate-limited service callers

Content Processing

  • Video Processing: Multiple video sources → Buffer → Multiple video processors
  • Document Processing: Multiple document sources → Buffer → Multiple document analyzers
  • Audio Processing: Multiple audio sources → Buffer → Multiple audio processors

Database Operations

  • Write Operations: Multiple write sources → Buffer → Database writers
  • Read Operations: Multiple read requests → Buffer → Database readers
  • Migration Jobs: Multiple migration sources → Buffer → Migration processors

The producer-consumer pattern is particularly effective when you have:

  • Variable Production Rates: Producers that generate data at unpredictable rates
  • Variable Consumption Rates: Consumers that process data at different speeds
  • Bursty Workloads: Periods of high activity followed by low activity
  • Resource Constraints: Need to limit resource usage while maintaining throughput
  • Decoupled Systems: Components that should operate independently

This pattern provides a robust foundation for building scalable, responsive systems that can handle varying loads efficiently.

The final example implementation looks like this:

package examples

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

// RunProducerConsumer demonstrates the producer-consumer pattern with multiple producers and consumers.
func RunProducerConsumer() {
	fmt.Println("=== Producer-Consumer Pattern Example ===")

	bufferSize := 5
	numProducers := 2
	numConsumers := 3
	numItems := 10

	buffer := make(chan int, bufferSize)
	var wg sync.WaitGroup

	// Start producers
	for p := 1; p <= numProducers; p++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for i := 0; i < numItems; i++ {
				item := rand.Intn(100)
				buffer <- item
				fmt.Printf("Producer %d produced: %d\n", id, item)
				time.Sleep(time.Duration(rand.Intn(200)+100) * time.Millisecond)
			}
		}(p)
	}

	// Start consumers
	var consumerWg sync.WaitGroup
	for c := 1; c <= numConsumers; c++ {
		consumerWg.Add(1)
		go func(id int) {
			defer consumerWg.Done()
			for item := range buffer {
				fmt.Printf("Consumer %d consumed: %d\n", id, item)
				time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond)
			}
		}(c)
	}

	// Wait for all producers to finish, then close the buffer
	wg.Wait()
	close(buffer)

	// Wait for all consumers to finish
	consumerWg.Wait()

	fmt.Println("Producer-Consumer example completed!")
}

To run this example, and build the code yourself, check out this and other examples in the go-fluency-concurrency-model-patterns repo. That’s it for this topic, tomorrow I’ll post on the supervisor-restart pattern.

Go Concurrency Patterns(Worker Pools Pattern)

Overview

The Worker Pools pattern maintains a fixed number of workers that process jobs from a shared queue. This pattern is essential for controlling resource usage, handling bursty workloads, and providing predictable performance in concurrent systems.

NOTE: For other posts on concurrency patterns, check out the index post to this series of concurrency patterns.

Implementation Details

Structure

The worker pools implementation in examples/pools.go consists of three main components:

  1. Job Queue – A buffered channel that holds jobs to be processed
  2. Worker Pool – A fixed number of worker goroutines
  3. Result Collection – A channel for collecting processed results

Code Analysis

Let’s break down the main function and understand how each component works:

func RunPools() {
    // Configuration
    numWorkers := 3
    numJobs := 15

    // Create job channel
    jobs := make(chan int, numJobs)

    // Create result channel
    results := make(chan string, numJobs)

    // Start the worker pool
    var wg sync.WaitGroup

    // Launch workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go workerPool(i, jobs, results, &wg)
    }

    // Send jobs to the pool
    go func() {
        defer close(jobs)
        for i := 1; i <= numJobs; i++ {
            fmt.Printf("Sending job %d to pool\n", i)
            jobs <- i
            time.Sleep(100 * time.Millisecond) // Simulate job generation time
        }
    }()

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

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

Step-by-step breakdown:

  1. Configuration Setup:
    • numWorkers := 3 defines the fixed pool size – only 3 workers will ever be active
    • numJobs := 15 specifies how many jobs to process (more jobs than workers to demonstrate queuing)
    • These values control the concurrency level and workload
  2. Channel Creation:
    • jobs := make(chan int, numJobs) creates a buffered channel with capacity for all jobs
    • Buffering prevents job submission from blocking when workers are busy
    • results := make(chan string, numJobs) creates a buffered result channel
    • Both channels are buffered to handle the full workload without blocking
  3. Worker Pool Initialization:
    • var wg sync.WaitGroup tracks when all workers complete
    • The loop launches exactly numWorkers goroutines
    • Each worker gets a unique ID (1, 2, 3) for tracking and debugging
    • All workers share the same jobs and results channels
  4. Job Submission Goroutine:
    • Runs in background to send jobs asynchronously
    • defer close(jobs) ensures the job channel closes after all jobs are sent
    • The loop sends jobs one by one with a 100ms delay between submissions
    • This simulates real-world job generation (e.g., incoming HTTP requests)
  5. Cleanup Goroutine:
    • Waits for all workers to finish using wg.Wait()
    • Closes the results channel when all workers are done
    • Prevents deadlocks and signals that no more results are coming
  6. Result Collection:
    • The main goroutine consumes results as they become available
    • Results may arrive in any order (not necessarily job submission order)
    • The loop exits when the results channel closes

Worker Implementation

func workerPool(id int, jobs <-chan int, results chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()

    fmt.Printf("Worker %d started\n", id)

    for job := range jobs {
        // Simulate work processing
        processingTime := time.Duration(rand.Intn(300)+200) * time.Millisecond
        fmt.Printf("Worker %d processing job %d (will take %v)\n", id, job, processingTime)

        time.Sleep(processingTime)

        result := fmt.Sprintf("Job %d completed by worker %d in %v", job, id, processingTime)
        results <- result
    }

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

Worker function breakdown:

  1. Function Signature:
    • id int: Unique worker identifier (1, 2, 3) for tracking and debugging
    • jobs <-chan int: Read-only channel for receiving jobs
    • results chan<- string: Write-only channel for sending results
    • wg *sync.WaitGroup: For signaling when this worker completes
  2. Resource Management:
    • defer wg.Done() ensures the worker signals completion even if it panics
    • Automatic cleanup when the function exits
    • Prevents goroutine leaks and ensures proper shutdown
  3. Worker Lifecycle:
    • Prints startup message to track worker initialization
    • Enters the main processing loop
    • Prints completion message when the job channel closes
  4. Job Processing Loop:
    • for job := range jobs continuously processes jobs until the channel closes
    • Each worker competes for jobs from the shared channel (automatic load balancing)
    • When the jobs channel closes, all workers naturally exit the loop
  5. Simulated Work Processing:
    • processingTime := time.Duration(rand.Intn(300)+200) * time.Millisecond creates variable processing time
    • Random delay between 200-500ms simulates real-world processing variance
    • Makes concurrency visible – workers process at different speeds
    • Demonstrates how the pool handles varying job complexity
  6. Result Generation and Sending:
    • Creates a descriptive result string with job ID, worker ID, and processing time
    • results <- result sends the result to the shared result channel
    • Non-blocking because the result channel is buffered
    • Results can be sent in any order (not necessarily job submission order)

How It Works

  1. Pool Initialization: Creates a fixed number of worker goroutines that wait for jobs
  2. Job Distribution: Jobs are sent to the shared job channel
  3. Worker Processing: Workers compete for jobs from the channel, process them, and send results
  4. Result Collection: Results are collected from the result channel as they complete
  5. Graceful Shutdown: When the job channel is closed, workers finish processing remaining jobs and exit

The pattern ensures that only a fixed number of workers are active at any time, providing controlled concurrency and resource management.

Why This Implementation?

Fixed Worker Count

  • Resource Control: Limits the number of concurrent operations
  • Predictable Performance: Consistent resource usage regardless of job load
  • Stability: Prevents resource exhaustion under high load

Buffered Job Channel

  • Burst Handling: Can queue jobs when workers are busy
  • Non-blocking Job Submission: Job senders don’t block when workers are busy
  • Backpressure: Natural backpressure when queue is full

Shared Job Channel

  • Automatic Load Balancing: Workers naturally consume jobs as they become available
  • Fair Distribution: All workers have equal access to jobs
  • Simple Coordination: No complex job distribution logic needed

WaitGroup Synchronization

  • Proper Cleanup: Ensures all workers complete before closing result channel
  • Resource Management: Prevents goroutine leaks
  • Coordinated Shutdown: All workers finish before main function exits

Separate Result Channel

  • Asynchronous Results: Results can be collected independently of job submission
  • Order Independence: Results can be processed in completion order, not submission order
  • Non-blocking Collection: Result collection doesn’t block job processing

Key Design Decisions

  1. Fixed Worker Count: The pool size is determined at creation time and remains constant
  2. Buffered Job Channel: Allows queuing of jobs when workers are busy
  3. Unbuffered Result Channel: Provides natural synchronization for result collection
  4. Simulated Processing Time: Random delays simulate real work and demonstrate concurrency
  5. Graceful Shutdown: Proper channel closing and WaitGroup coordination

Performance Characteristics

Throughput

  • Limited by Worker Count: Maximum throughput is limited by the number of workers
  • Consistent Performance: Predictable performance regardless of job load
  • Optimal for I/O-bound Work: Workers can handle I/O operations efficiently

Latency

  • Queue Time: Jobs may wait in the queue if all workers are busy
  • Processing Time: Individual job processing time depends on the work being done
  • Fair Scheduling: Jobs are processed in FIFO order within the queue

Resource Usage

  • Memory: Fixed memory usage regardless of job load
  • CPU: Controlled CPU usage through fixed worker count
  • Network/File Handles: Limited resource usage through worker constraints

Common Use Cases

Web Servers

  • Request Processing: Handle HTTP requests with a fixed pool of workers
  • Database Operations: Process database queries with controlled concurrency
  • File Operations: Handle file uploads/downloads with limited workers

Background Job Processing

  • Email Sending: Process email sending with rate limiting
  • Image Processing: Resize, compress, or transform images
  • Report Generation: Generate reports with controlled resource usage

API Rate Limiting

  • External API Calls: Make API calls with controlled concurrency
  • Data Synchronization: Sync data with external systems
  • Webhook Processing: Process incoming webhooks with rate limiting

Resource-Intensive Operations

  • Machine Learning: Run ML models with limited GPU/CPU usage
  • Data Processing: Process large datasets with controlled memory usage
  • Encryption/Decryption: Handle cryptographic operations with limited CPU usage

System Administration

  • Backup Operations: Run backups with controlled I/O usage
  • Log Processing: Process logs with limited file handle usage
  • Monitoring: Collect metrics with controlled network usage

Microservices

  • Service Communication: Handle inter-service communication with rate limiting
  • Event Processing: Process events with controlled concurrency
  • Cache Management: Manage cache operations with limited memory usage

The worker pools pattern is particularly effective when you need to control resource usage, handle bursty workloads, or provide predictable performance in concurrent systems. It’s especially useful for I/O-bound operations where you want to limit the number of concurrent operations to prevent resource exhaustion.

The final example implementation looks like this:

package examples

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

// Pools demonstrates the worker pools pattern
func RunPools() {
	fmt.Println("=== Worker Pools Pattern Example ===")

	// Configuration
	numWorkers := 3
	numJobs := 15

	// Create job channel
	jobs := make(chan int, numJobs)

	// Create result channel
	results := make(chan string, numJobs)

	// Start the worker pool
	var wg sync.WaitGroup

	// Launch workers
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go workerPool(i, jobs, results, &wg)
	}

	// Send jobs to the pool
	go func() {
		defer close(jobs)
		for i := 1; i <= numJobs; i++ {
			fmt.Printf("Sending job %d to pool\n", i)
			jobs <- i
			time.Sleep(100 * time.Millisecond) // Simulate job generation time
		}
	}()

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

	// Collect results
	fmt.Printf("\nWorker pool with %d workers processing %d jobs:\n", numWorkers, numJobs)
	fmt.Println()

	count := 0
	for result := range results {
		fmt.Printf("Result: %s\n", result)
		count++
	}

	fmt.Printf("\nWorker pool completed! Processed %d jobs.\n", count)
}

// Worker function for the pool
func workerPool(id int, jobs <-chan int, results chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()

	fmt.Printf("Worker %d started\n", id)

	for job := range jobs {
		// Simulate work processing
		processingTime := time.Duration(rand.Intn(300)+200) * time.Millisecond
		fmt.Printf("Worker %d processing job %d (will take %v)\n", id, job, processingTime)

		time.Sleep(processingTime)

		result := fmt.Sprintf("Job %d completed by worker %d in %v", job, id, processingTime)
		results <- result
	}

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

To run this example, and build the code yourself, check out this and other examples in the go-fluency-concurrency-model-patterns repo. That’s it for this topic, tomorrow I’ll post on the producer consumer pattern.