Overview

The Resource Pooling pattern manages a pool of reusable resources (like database connections, HTTP clients, or file handles) to avoid the overhead of creating and destroying resources frequently. This pattern is essential for improving performance, managing resource limits, reducing overhead, and ensuring efficient resource utilization.

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

Implementation Details

Structure

The resource pooling implementation in examples/resource_pooling.go demonstrates two main examples:

1. Database Connection Pool – Manages database connections with health checks
2. HTTP Client Pool – Manages HTTP clients for making requests

Code Analysis

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

func RunResourcePooling() {
    // Example 1: Database connection pool
    fmt.Println("1. Database Connection Pool Example:")
    dbPool := newDBConnectionPool(3, 5*time.Second)
    defer dbPool.Close()

    var wg sync.WaitGroup
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            conn := dbPool.Get()
            defer dbPool.Put(conn)

            fmt.Printf("Worker %d: Using connection %s\n", id, conn.ID)
            time.Sleep(time.Duration(rand.Intn(1000)+500) * time.Millisecond)
        }(i)
    }
    wg.Wait()

    // Example 2: HTTP client pool
    fmt.Println("\n2. HTTP Client Pool Example:")
    clientPool := newHTTPClientPool(2, 3*time.Second)
    defer clientPool.Close()

    for i := 1; i <= 4; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            client := clientPool.Get()
            defer clientPool.Put(client)

            fmt.Printf("Worker %d: Using HTTP client %s\n", id, client.ID)
            time.Sleep(time.Duration(rand.Intn(800)+300) * time.Millisecond)
        }(i)
    }
    wg.Wait()
}

Step-by-step breakdown:

1. Database Connection Pool Setup:
   • dbPool := newDBConnectionPool(3, 5*time.Second) creates a pool with 3 connections and 5-second timeout
   • Pool size of 3 means only 3 workers can use connections simultaneously
   • 5-second timeout prevents workers from waiting indefinitely for connections
   • defer dbPool.Close() ensures proper cleanup when the function exits
2. Database Pool Worker Launch:
   • var wg sync.WaitGroup tracks when all database workers complete
   • for i := 1; i <= 5; i++ launches 5 workers (more than pool size to demonstrate pooling)
   • Uses closure func(id int) { ... }(i) to capture the worker ID
   • wg.Add(1) increments the wait group before each worker
3. Database Pool Worker Implementation:
   • defer wg.Done() ensures the worker signals completion when it exits
   • conn := dbPool.Get() acquires a connection from the pool (may block if none available)
   • defer dbPool.Put(conn) ensures the connection is returned to the pool
   • time.Sleep(time.Duration(rand.Intn(1000)+500) * time.Millisecond) simulates database work (500-1500ms)
4. Database Pool Coordination:
   • wg.Wait() waits for all database workers to complete
   • Only 3 workers can use connections simultaneously due to pool size
   • Other workers will wait for connections to become available
5. HTTP Client Pool Setup:
   • clientPool := newHTTPClientPool(2, 3*time.Second) creates a pool with 2 clients and 3-second timeout
   • Smaller pool size (2) demonstrates different resource constraints
   • Shorter timeout (3 seconds) shows different timeout requirements
   • defer clientPool.Close() ensures proper cleanup
6. HTTP Client Pool Worker Launch:
   • for i := 1; i <= 4; i++ launches 4 workers (more than pool size)
   • Similar structure to database workers but with different resource type
   • wg.Add(1) tracks each HTTP client worker
7. HTTP Client Pool Worker Implementation:
   • defer wg.Done() ensures proper cleanup
   • client := clientPool.Get() acquires an HTTP client from the pool
   • defer clientPool.Put(client) ensures the client is returned to the pool
   • time.Sleep(time.Duration(rand.Intn(800)+300) * time.Millisecond) simulates HTTP work (300-1100ms)
8. HTTP Client Pool Coordination:
   • wg.Wait() waits for all HTTP client workers to complete
   • Only 2 workers can use clients simultaneously due to pool size
   • Demonstrates how pooling handles resource constraints

Database Connection Pool Implementation

type DBConnection struct {
    ID        string
    CreatedAt time.Time
    LastUsed  time.Time
}

type DBConnectionPool struct {
    connections chan *DBConnection
    maxSize     int
    timeout     time.Duration
    closed      bool
    mu          sync.Mutex
}

func newDBConnectionPool(maxSize int, timeout time.Duration) *DBConnectionPool {
    pool := &DBConnectionPool{
        connections: make(chan *DBConnection, maxSize),
        maxSize:     maxSize,
        timeout:     timeout,
    }

    // Pre-populate the pool
    for i := 0; i < maxSize; i++ {
        conn := &DBConnection{
            ID:        fmt.Sprintf("db-conn-%d", i+1),
            CreatedAt: time.Now(),
            LastUsed:  time.Now(),
        }
        pool.connections <- conn
    }

    return pool
}

func (p *DBConnectionPool) Get() *DBConnection {
    select {
    case conn := <-p.connections:
        conn.LastUsed = time.Now()
        fmt.Printf("  Acquired connection: %s\n", conn.ID)
        return conn
    case <-time.After(p.timeout):
        fmt.Println("  Timeout waiting for connection")
        return nil
    }
}

func (p *DBConnectionPool) Put(conn *DBConnection) {
    if conn == nil {
        return
    }

    p.mu.Lock()
    defer p.mu.Unlock()

    if p.closed {
        return
    }

    // Check if connection is still healthy
    if time.Since(conn.LastUsed) > p.timeout {
        fmt.Printf("  Discarding stale connection: %s\n", conn.ID)
        return
    }

    select {
    case p.connections <- conn:
        fmt.Printf("  Returned connection: %s\n", conn.ID)
    default:
        fmt.Printf("  Pool full, discarding connection: %s\n", conn.ID)
    }
}

func (p *DBConnectionPool) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()
    p.closed = true
    close(p.connections)
}

Database connection pool breakdown:

1. Connection Structure Design:
   • DBConnection struct represents a database connection
   • ID string provides unique identification for tracking and debugging
   • CreatedAt time.Time tracks when the connection was created
   • LastUsed time.Time tracks when the connection was last used for health checking
2. Pool Structure Design:
   • DBConnectionPool struct manages the connection pool
   • connections chan *DBConnection uses a buffered channel for thread-safe resource management
   • maxSize int defines the maximum number of connections in the pool
   • timeout time.Duration defines the timeout for connection acquisition
   • closed bool tracks pool state for graceful shutdown
   • mu sync.Mutex protects pool state during concurrent operations
3. Pool Constructor:
   • newDBConnectionPool(maxSize int, timeout time.Duration) creates a new pool
   • connections: make(chan *DBConnection, maxSize) creates a buffered channel with pool size
   • The channel buffer size matches the pool size for optimal performance
4. Pool Pre-population:
   • for i := 0; i < maxSize; i++ creates the initial set of connections
   • conn := &DBConnection{...} creates each connection with unique ID and timestamps
   • pool.connections <- conn adds each connection to the pool
   • Pre-population ensures connections are immediately available
5. Connection Acquisition (Get Method):
   • select statement handles both connection acquisition and timeout
   • case conn := <-p.connections: acquires a connection from the pool
   • conn.LastUsed = time.Now() updates the last used timestamp
   • case <-time.After(p.timeout): handles timeout when no connections are available
   • Returns nil on timeout to indicate failure
6. Connection Return (Put Method):
   • if conn == nil { return } handles nil connections gracefully
   • p.mu.Lock() and defer p.mu.Unlock() protect pool state during return
   • if p.closed { return } prevents returning connections to a closed pool
7. Health Checking:
   • if time.Since(conn.LastUsed) > p.timeout checks if connection is stale
   • Stale connections are discarded rather than returned to the pool
   • This ensures only healthy connections are reused
8. Connection Return Logic:
   • select { case p.connections <- conn: ... default: ... } attempts to return connection
   • If pool is full, connection is discarded (default case)
   • This prevents pool overflow and memory leaks
9. Pool Cleanup (Close Method):
   • p.mu.Lock() and defer p.mu.Unlock() protect pool state during shutdown
   • p.closed = true marks the pool as closed
   • close(p.connections) closes the channel to signal shutdown

HTTP Client Pool Implementation

type HTTPClient struct {
    ID        string
    CreatedAt time.Time
    LastUsed  time.Time
}

type HTTPClientPool struct {
    clients chan *HTTPClient
    maxSize int
    timeout time.Duration
    closed  bool
    mu      sync.Mutex
}

func newHTTPClientPool(maxSize int, timeout time.Duration) *HTTPClientPool {
    pool := &HTTPClientPool{
        clients: make(chan *HTTPClient, maxSize),
        maxSize: maxSize,
        timeout: timeout,
    }

    // Pre-populate the pool
    for i := 0; i < maxSize; i++ {
        client := &HTTPClient{
            ID:        fmt.Sprintf("http-client-%d", i+1),
            CreatedAt: time.Now(),
            LastUsed:  time.Now(),
        }
        pool.clients <- client
    }

    return pool
}

func (p *HTTPClientPool) Get() *HTTPClient {
    select {
    case client := <-p.clients:
        client.LastUsed = time.Now()
        fmt.Printf("  Acquired HTTP client: %s\n", client.ID)
        return client
    case <-time.After(p.timeout):
        fmt.Println("  Timeout waiting for HTTP client")
        return nil
    }
}

func (p *HTTPClientPool) Put(client *HTTPClient) {
    if client == nil {
        return
    }

    p.mu.Lock()
    defer p.mu.Unlock()

    if p.closed {
        return
    }

    // Check if client is still healthy
    if time.Since(client.LastUsed) > p.timeout {
        fmt.Printf("  Discarding stale HTTP client: %s\n", client.ID)
        return
    }

    select {
    case p.clients <- client:
        fmt.Printf("  Returned HTTP client: %s\n", client.ID)
    default:
        fmt.Printf("  Pool full, discarding HTTP client: %s\n", client.ID)
    }
}

func (p *HTTPClientPool) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()
    p.closed = true
    close(p.clients)
}

HTTP client pool breakdown:

1. Client Structure Design:
   • HTTPClient struct represents an HTTP client
   • Similar structure to database connections for consistency
   • ID string provides unique identification
   • CreatedAt and LastUsed timestamps for lifecycle tracking
2. Pool Structure Design:
   • HTTPClientPool struct manages the HTTP client pool
   • Identical structure to database pool for consistency
   • Uses the same channel-based approach for resource management
3. Pool Constructor and Pre-population:
   • newHTTPClientPool(maxSize int, timeout time.Duration) creates the pool
   • Pre-populates with maxSize HTTP clients
   • Each client gets a unique ID and initial timestamps
4. Client Acquisition and Return:
   • Get() method identical to database pool with timeout handling
   • Put() method includes health checking and graceful handling
   • Same mutex protection and closed pool handling
5. Health Checking and Cleanup:
   • Same health checking logic as database pool
   • Stale clients are discarded to maintain pool health
   • Close() method provides graceful shutdown

Key Design Patterns:

1. Channel-based Resource Management: Uses buffered channels for thread-safe resource allocation and return.
2. Pre-populated Pools: Creates all resources upfront for immediate availability and predictable performance.
3. Health Checking: Validates resource health before reuse to ensure reliability.
4. Timeout Handling: Prevents indefinite waiting with configurable timeouts.
5. Graceful Shutdown: Proper cleanup prevents resource leaks during shutdown.
6. Mutex Protection: Thread-safe state management for pool operations.
7. Resource Tracking: Comprehensive tracking of resource lifecycle for monitoring and debugging.

How It Works

Database Connection Pool

1. Pool Initialization: Creates a pool with pre-populated database connections
2. Connection Acquisition: Workers request connections from the pool
3. Connection Usage: Workers use connections for database operations
4. Connection Return: Workers return connections to the pool when done
5. Health Checking: Pool checks connection health and discards stale connections
6. Resource Management: Pool manages connection lifecycle and cleanup

HTTP Client Pool

1. Pool Initialization: Creates a pool with pre-populated HTTP clients
2. Client Acquisition: Workers request HTTP clients from the pool
3. Client Usage: Workers use clients for HTTP requests
4. Client Return: Workers return clients to the pool when done
5. Health Checking: Pool checks client health and discards stale clients
6. Resource Management: Pool manages client lifecycle and cleanup

Why This Implementation?

Channel-based Pool Management

• Thread Safety: Channels provide natural thread safety for pool operations
• Blocking Semantics: Workers block when no resources are available
• FIFO Order: Resources are distributed in first-in-first-out order
• Simple Implementation: Straightforward resource management

Pre-populated Pools

• Fast Startup: Resources are available immediately
• Predictable Performance: No delay for resource creation
• Resource Efficiency: Optimal resource utilization from start
• Warm-up Avoidance: No cold start performance issues

Health Checking

• Resource Validation: Ensures resources are still usable
• Stale Detection: Detects and removes stale resources
• Automatic Cleanup: Automatically manages resource lifecycle
• Reliability: Ensures only healthy resources are used

Timeout Handling

• Deadlock Prevention: Prevents workers from waiting indefinitely
• Graceful Degradation: Handles resource exhaustion gracefully
• User Experience: Provides feedback when resources are unavailable
• System Stability: Prevents system-wide resource starvation

Mutex Protection

• Thread Safety: Protects pool state during concurrent operations
• Cleanup Coordination: Ensures proper cleanup during shutdown
• State Consistency: Maintains consistent pool state
• Resource Safety: Prevents resource leaks during shutdown

Key Design Decisions

1. Two Different Pools: Demonstrates the pattern with different resource types
2. Health Checking: Implements resource health validation
3. Timeout Handling: Provides timeout mechanisms for resource acquisition
4. Graceful Shutdown: Implements proper cleanup and shutdown
5. Resource Tracking: Tracks resource usage and health metrics

Performance Characteristics

Throughput

• Resource Reuse: Eliminates resource creation/destruction overhead
• Connection Pooling: Reduces connection establishment time
• Parallel Processing: Multiple workers can use resources concurrently
• Efficient Allocation: Fast resource allocation from pool

Latency

• Immediate Availability: Pre-populated resources are immediately available
• Reduced Overhead: No resource creation overhead for each request
• Predictable Performance: Consistent latency regardless of load
• Connection Reuse: Reuses existing connections for better performance

Resource Usage

• Controlled Memory: Fixed memory usage regardless of load
• Efficient Utilization: Resources are used efficiently across workers
• Automatic Cleanup: Stale resources are automatically removed
• Scalability: Pool size can be tuned for optimal performance

Common Use Cases

Database Operations

• Connection Pooling: Manage database connections efficiently
• Query Optimization: Reuse connections for multiple queries
• Connection Limits: Respect database connection limits
• Health Monitoring: Monitor connection health and performance

HTTP Operations

• HTTP Client Pooling: Reuse HTTP clients for multiple requests
• API Calls: Efficiently manage API client connections
• Load Balancing: Distribute requests across multiple clients
• Connection Management: Handle connection limits and timeouts

File Operations

• File Handle Pooling: Manage file handles efficiently
• I/O Operations: Reuse file handles for multiple operations
• Resource Limits: Respect system file handle limits
• Performance Optimization: Reduce file open/close overhead

Network Operations

• Socket Pooling: Manage network socket connections
• Protocol Handlers: Reuse protocol handlers for multiple connections
• Connection Limits: Handle connection limits and timeouts
• Load Distribution: Distribute load across multiple connections

Memory Management

• Object Pooling: Reuse expensive objects
• Buffer Pooling: Manage memory buffers efficiently
• Allocation Optimization: Reduce memory allocation overhead
• Garbage Collection: Minimize garbage collection pressure

Thread Management

• Thread Pooling: Manage worker threads efficiently
• Task Distribution: Distribute tasks across worker threads
• Resource Limits: Control thread usage and limits
• Performance Optimization: Optimize thread creation/destruction

Advanced Patterns

Dynamic Pool Sizing

• Adaptive Sizing: Adjust pool size based on load
• Auto-scaling: Automatically scale pool size up or down
• Load-based Allocation: Allocate resources based on current load
• Performance Monitoring: Monitor and adjust based on performance metrics

Resource Health Monitoring

• Health Checks: Regular health checks for pooled resources
• Failure Detection: Detect and remove failed resources
• Recovery Mechanisms: Implement recovery for failed resources
• Metrics Collection: Collect health and performance metrics

Priority-based Pooling

• Priority Queues: Prioritize resource allocation
• Fair Distribution: Ensure fair resource distribution
• Resource Reservation: Reserve resources for high-priority operations
• Load Balancing: Balance load across different priority levels

Distributed Pooling

• Multi-node Pools: Coordinate pools across multiple nodes
• Shared Resources: Share resources across multiple nodes
• Load Distribution: Distribute load across multiple pools
• Fault Tolerance: Handle node failures gracefully

Best Practices

Pool Configuration

• Appropriate Size: Set pool size based on system capacity and requirements
• Timeout Values: Set appropriate timeouts for resource acquisition
• Health Check Intervals: Regular health checks for resource validation
• Monitoring: Monitor pool usage and performance metrics

Resource Management

• Proper Cleanup: Ensure resources are properly cleaned up
• Leak Prevention: Prevent resource leaks through proper management
• Health Monitoring: Monitor resource health and performance
• Error Handling: Handle resource failures gracefully

Performance Optimization

• Pool Sizing: Optimize pool size for your workload
• Resource Reuse: Maximize resource reuse for better performance
• Connection Limits: Respect system and service connection limits
• Load Testing: Test pool performance under various loads

Monitoring and Alerting

• Pool Metrics: Track pool usage, health, and performance
• Resource Metrics: Monitor individual resource health and performance
• Alerting: Alert on pool exhaustion or resource failures
• Capacity Planning: Use metrics for capacity planning

The resource pooling pattern is particularly effective when you have:

• Expensive Resources: Resources that are costly to create and destroy
• Resource Limits: Limited resources that need efficient management
• High Concurrency: Multiple concurrent operations that need resources
• Performance Requirements: Need for optimal performance and resource utilization
• Resource Constraints: Limited system resources that need careful management

This pattern provides essential tools for building efficient, scalable systems that can handle high concurrency while optimizing resource usage and improving performance.

The final example implementation looks like this:

package examples

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

// RunResourcePooling demonstrates the resource pooling pattern.
func RunResourcePooling() {
	fmt.Println("=== Resource Pooling Pattern Example ===")

	// Example 1: Database Connection Pool
	fmt.Println("\n1. Database Connection Pool Example:")
	dbPool := newDBConnectionPool(3, 5)

	var wg sync.WaitGroup
	for i := 1; i <= 8; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			conn := dbPool.getConnection()
			fmt.Printf("Worker %d: Got DB connection %d\n", id, conn.id)

			// Simulate database operation
			time.Sleep(time.Duration(rand.Intn(500)+200) * time.Millisecond)
			fmt.Printf("Worker %d: Executing query on connection %d\n", id, conn.id)

			dbPool.releaseConnection(conn)
			fmt.Printf("Worker %d: Released DB connection %d\n", id, conn.id)
		}(i)
	}
	wg.Wait()
	dbPool.close()

	// Example 2: HTTP Client Pool
	fmt.Println("\n2. HTTP Client Pool Example:")
	clientPool := newHTTPClientPool(2, 4)

	for i := 1; i <= 6; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			client := clientPool.getClient()
			fmt.Printf("Worker %d: Got HTTP client %d\n", id, client.id)

			// Simulate API request
			time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond)
			fmt.Printf("Worker %d: Making API request with client %d\n", id, client.id)

			clientPool.releaseClient(client)
			fmt.Printf("Worker %d: Released HTTP client %d\n", id, client.id)
		}(i)
	}
	wg.Wait()
	clientPool.close()

	fmt.Println("\nResource Pooling example completed!")
}

// Database Connection Pool
type dbConnection struct {
	id       int
	lastUsed time.Time
}

type dbConnectionPool struct {
	connections chan *dbConnection
	maxSize     int
	created     int
	mu          sync.Mutex
}

func newDBConnectionPool(initial, maxSize int) *dbConnectionPool {
	pool := &dbConnectionPool{
		connections: make(chan *dbConnection, maxSize),
		maxSize:     maxSize,
	}

	// Pre-populate with initial connections
	for i := 0; i < initial; i++ {
		pool.connections <- &dbConnection{
			id:       i + 1,
			lastUsed: time.Now(),
		}
		pool.created++
	}

	return pool
}

func (p *dbConnectionPool) getConnection() *dbConnection {
	select {
	case conn := <-p.connections:
		conn.lastUsed = time.Now()
		return conn
	default:
		// Create new connection if pool is empty and under max size
		p.mu.Lock()
		if p.created < p.maxSize {
			p.created++
			p.mu.Unlock()
			return &dbConnection{
				id:       p.created,
				lastUsed: time.Now(),
			}
		}
		p.mu.Unlock()

		// Wait for available connection
		conn := <-p.connections
		conn.lastUsed = time.Now()
		return conn
	}
}

func (p *dbConnectionPool) releaseConnection(conn *dbConnection) {
	select {
	case p.connections <- conn:
		// Successfully returned to pool
	default:
		// Pool is full, discard connection
		fmt.Printf("Pool full, discarding connection %d\n", conn.id)
	}
}

func (p *dbConnectionPool) close() {
	close(p.connections)
	fmt.Printf("DB pool closed. Total connections created: %d\n", p.created)
}

// HTTP Client Pool
type httpClient struct {
	id       int
	lastUsed time.Time
}

type httpClientPool struct {
	clients chan *httpClient
	maxSize int
	created int
	mu      sync.Mutex
}

func newHTTPClientPool(initial, maxSize int) *httpClientPool {
	pool := &httpClientPool{
		clients: make(chan *httpClient, maxSize),
		maxSize: maxSize,
	}

	// Pre-populate with initial clients
	for i := 0; i < initial; i++ {
		pool.clients <- &httpClient{
			id:       i + 1,
			lastUsed: time.Now(),
		}
		pool.created++
	}

	return pool
}

func (p *httpClientPool) getClient() *httpClient {
	select {
	case client := <-p.clients:
		client.lastUsed = time.Now()
		return client
	default:
		// Create new client if pool is empty and under max size
		p.mu.Lock()
		if p.created < p.maxSize {
			p.created++
			p.mu.Unlock()
			return &httpClient{
				id:       p.created,
				lastUsed: time.Now(),
			}
		}
		p.mu.Unlock()

		// Wait for available client
		client := <-p.clients
		client.lastUsed = time.Now()
		return client
	}
}

func (p *httpClientPool) releaseClient(client *httpClient) {
	select {
	case p.clients <- client:
		// Successfully returned to pool
	default:
		// Pool is full, discard client
		fmt.Printf("Pool full, discarding HTTP client %d\n", client.id)
	}
}

func (p *httpClientPool) close() {
	close(p.clients)
	fmt.Printf("HTTP client pool closed. Total clients created: %d\n", p.created)
}

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. A new series will begin in a day or few! 🤘🏻 Happy thrashing code!

Overview

The Event Loop pattern processes events from multiple sources in a single thread using a central event loop. This pattern is essential for handling multiple event sources efficiently, managing I/O operations, building reactive systems, and coordinating multiple concurrent operations in a controlled manner.

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

Implementation Details

Structure

The event loop implementation in examples/event_loop.go consists of three main components:

1. Event Loop – Central coordinator that processes events from multiple channels
2. Event Sources – Multiple goroutines that generate events
3. Event Processing – Centralized event handling logic

Code Analysis

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

func RunEventLoop() {
    // Create event channels
    userEvents := make(chan string, 10)
    systemEvents := make(chan string, 10)
    timerEvents := make(chan string, 10)
    stop := make(chan struct{})

    // Start event sources
    go userEventSource(userEvents)
    go systemEventSource(systemEvents)
    go timerEventSource(timerEvents)

    // Start the event loop
    go eventLoop(userEvents, systemEvents, timerEvents, stop)

    // Let the event loop run for a while
    time.Sleep(5 * time.Second)
    close(stop)
    fmt.Println("Event loop stopped.")
}

Step-by-step breakdown:

1. Event Channel Creation:
   • userEvents := make(chan string, 10) creates a buffered channel for user events
   • systemEvents := make(chan string, 10) creates a buffered channel for system events
   • timerEvents := make(chan string, 10) creates a buffered channel for timer events
   • stop := make(chan struct{}) creates a signal channel for graceful shutdown
   • Buffer size of 10 prevents blocking when event sources generate events faster than processing
2. Event Source Launch:
   • go userEventSource(userEvents) launches the user event generator in background
   • go systemEventSource(systemEvents) launches the system event generator in background
   • go timerEventSource(timerEvents) launches the timer event generator in background
   • Each event source runs independently and continuously generates events
3. Event Loop Launch:
   • go eventLoop(userEvents, systemEvents, timerEvents, stop) starts the central event loop
   • The event loop runs in its own goroutine to process events from all sources
   • This is the core coordinator that handles all events sequentially
4. Demonstration Period:
   • time.Sleep(5 * time.Second) lets the event loop run for 5 seconds
   • This provides enough time to see the pattern in action
   • During this time, all event sources are generating events and the loop is processing them
5. Graceful Shutdown:
   • close(stop) signals the event loop to stop processing
   • The event loop will receive this signal and exit gracefully
   • fmt.Println("Event loop stopped.") confirms the shutdown

Event Loop Implementation

func eventLoop(userEvents, systemEvents, timerEvents <-chan string, stop <-chan struct{}) {
    fmt.Println("Event loop started. Processing events...")

    for {
        select {
        case event := <-userEvents:
            fmt.Printf("Event Loop: Processing user event: %s\n", event)
            processUserEvent(event)

        case event := <-systemEvents:
            fmt.Printf("Event Loop: Processing system event: %s\n", event)
            processSystemEvent(event)

        case event := <-timerEvents:
            fmt.Printf("Event Loop: Processing timer event: %s\n", event)
            processTimerEvent(event)

        case <-stop:
            fmt.Println("Event Loop: Received stop signal, shutting down...")
            return
        }
    }
}

Event loop breakdown:

1. Function Signature:
   • Takes read-only channels (<-chan string) for each event type
   • Takes a read-only stop channel (<-chan struct{}) for shutdown
   • Directional channels provide encapsulation and prevent accidental misuse
2. Infinite Loop Structure:
   • for { ... } creates an infinite loop that continuously processes events
   • The loop will run until explicitly stopped by the stop signal
   • This ensures the event loop is always ready to handle events
3. Select Statement:
   • select statement waits for events from any of the four channels
   • Non-blocking – if no events are available, it waits efficiently
   • Fair selection – all event sources have equal priority
   • Only one case executes at a time, ensuring sequential processing
4. User Event Handling:
   • case event := <-userEvents: handles user events (clicks, keypresses, etc.)
   • fmt.Printf("Event Loop: Processing user event: %s\n", event) logs the event
   • processUserEvent(event) calls the appropriate handler function
   • User events typically need quick response for good UX
5. System Event Handling:
   • case event := <-systemEvents: handles system events (file changes, network status, etc.)
   • fmt.Printf("Event Loop: Processing system event: %s\n", event) logs the event
   • processSystemEvent(event) calls the system event handler
   • System events may require more processing time
6. Timer Event Handling:
   • case event := <-timerEvents: handles timer events (periodic ticks)
   • fmt.Printf("Event Loop: Processing timer event: %s\n", event) logs the event
   • processTimerEvent(event) calls the timer event handler
   • Timer events are typically lightweight and frequent
7. Stop Signal Handling:
   • case <-stop: handles the shutdown signal
   • fmt.Println("Event Loop: Received stop signal, shutting down...") logs shutdown
   • return exits the event loop gracefully
   • This ensures clean termination without data loss

Event Source Implementations

func userEventSource(events chan<- string) {
    eventTypes := []string{"click", "keypress", "scroll", "hover", "submit"}
    for {
        event := eventTypes[rand.Intn(len(eventTypes))]
        events <- fmt.Sprintf("User %s", event)
        time.Sleep(time.Duration(rand.Intn(1000)+500) * time.Millisecond)
    }
}

func systemEventSource(events chan<- string) {
    eventTypes := []string{"file_change", "network_status", "memory_usage", "cpu_load", "disk_space"}
    for {
        event := eventTypes[rand.Intn(len(eventTypes))]
        events <- fmt.Sprintf("System %s", event)
        time.Sleep(time.Duration(rand.Intn(1500)+1000) * time.Millisecond)
    }
}

func timerEventSource(events chan<- string) {
    ticker := time.NewTicker(800 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            events <- fmt.Sprintf("Timer tick at %v", time.Now().Format("15:04:05.000"))
        }
    }
}

Event source breakdown:

1. User Event Source:
   • eventTypes := []string{...} defines realistic user interaction events
   • for { ... } creates an infinite loop for continuous event generation
   • event := eventTypes[rand.Intn(len(eventTypes))] randomly selects an event type
   • events <- fmt.Sprintf("User %s", event) sends the formatted event to the channel
   • time.Sleep(time.Duration(rand.Intn(1000)+500) * time.Millisecond) simulates variable user interaction timing (500-1500ms)
2. System Event Source:
   • eventTypes := []string{...} defines realistic system monitoring events
   • Similar structure to user events but with system-specific event types
   • time.Sleep(time.Duration(rand.Intn(1500)+1000) * time.Millisecond) simulates slower system event generation (1000-2500ms)
   • System events typically occur less frequently than user events
3. Timer Event Source:
   • ticker := time.NewTicker(800 * time.Millisecond) creates a ticker for regular events
   • defer ticker.Stop() ensures the ticker is cleaned up when the function exits
   • for { select { case <-ticker.C: ... } } waits for ticker events
   • events <- fmt.Sprintf("Timer tick at %v", time.Now().Format("15:04:05.000")) sends timestamped timer events
   • Timer events occur at regular 800ms intervals, providing predictable timing

Event Processing Functions

func processUserEvent(event string) {
    // Simulate processing time
    time.Sleep(50 * time.Millisecond)
    fmt.Printf("  Processed user event: %s\n", event)
}

func processSystemEvent(event string) {
    // Simulate processing time
    time.Sleep(100 * time.Millisecond)
    fmt.Printf("  Processed system event: %s\n", event)
}

func processTimerEvent(event string) {
    // Simulate processing time
    time.Sleep(30 * time.Millisecond)
    fmt.Printf("  Processed timer event: %s\n", event)
}

Event processing breakdown:

1. User Event Processing:
   • time.Sleep(50 * time.Millisecond) simulates quick user event processing
   • User events need fast response for good user experience
   • fmt.Printf(" Processed user event: %s\n", event) confirms processing completion
2. System Event Processing:
   • time.Sleep(100 * time.Millisecond) simulates longer system event processing
   • System events may require more complex processing (file I/O, network calls, etc.)
   • fmt.Printf(" Processed system event: %s\n", event) confirms processing completion
3. Timer Event Processing:
   • time.Sleep(30 * time.Millisecond) simulates lightweight timer event processing
   • Timer events are typically simple and fast to process
   • fmt.Printf(" Processed timer event: %s\n", event) confirms processing completion

Key Design Patterns:

1. Single-threaded Event Processing: The event loop processes events sequentially in a single goroutine, preventing race conditions.
2. Channel-based Communication: All event sources communicate with the event loop through channels, providing natural synchronization.
3. Select Statement: The select statement provides non-blocking, fair event selection from multiple sources.
4. Buffered Channels: Buffer size of 10 prevents event sources from blocking when the event loop is busy processing.
5. Graceful Shutdown: The stop channel provides clean shutdown without data loss or resource leaks.
6. Structured Events: Each event type has a clear structure and meaning, making the system easy to understand and extend.
7. Simulated Processing: Random delays simulate real-world processing times and make the pattern’s behavior visible.

How It Works

1. Event Sources: Multiple goroutines generate events and send them to channels
2. Event Loop: A single goroutine runs the event loop, listening to all event channels
3. Event Selection: The select statement waits for events from any source
4. Event Processing: When an event arrives, it’s processed by the appropriate handler
5. Non-blocking: The loop continues to process events from other sources while one is being processed
6. Graceful Shutdown: The loop stops when it receives a stop signal

The pattern ensures that all events are processed in a controlled, sequential manner while maintaining responsiveness to multiple event sources.

Why This Implementation?

Single-threaded Event Processing

• Sequential Processing: Events are processed one at a time, preventing race conditions
• Predictable Behavior: Event processing order is deterministic
• Resource Efficiency: No need for complex synchronization between event handlers
• Debugging: Easier to debug and reason about event processing

Channel-based Event Sources

• Asynchronous Generation: Event sources can generate events independently
• Non-blocking: Event sources don’t block when the event loop is busy
• Buffering: Buffered channels prevent event loss during processing delays
• Scalability: Easy to add or remove event sources

Select Statement

• Non-blocking Selection: Waits for events from any source without blocking
• Fair Selection: All event sources have equal priority
• Efficient Waiting: Efficiently waits for multiple channels
• Clean Code: Provides clear, readable event handling logic

Centralized Event Handling

• Single Point of Control: All events flow through one place
• Consistent Processing: All events are processed with the same logic
• Easy Monitoring: Easy to monitor and log all events
• State Management: Centralized state management for event processing

Graceful Shutdown

• Controlled Termination: Event loop can be stopped gracefully
• Resource Cleanup: Proper cleanup of resources when stopping
• Coordinated Shutdown: All event sources can be coordinated during shutdown
• No Data Loss: Ensures no events are lost during shutdown

Key Design Decisions

1. Multiple Event Types: Demonstrates handling different types of events
2. Simulated Processing: Random delays simulate real event processing time
3. Buffered Channels: Prevent blocking of event sources
4. Structured Events: Events have clear structure and meaning
5. Proper Cleanup: Graceful shutdown with proper resource cleanup

Performance Characteristics

Throughput

• Sequential Processing: Throughput limited by processing time of individual events
• Event Queuing: Events can queue in channels during processing delays
• Fair Processing: All event sources are processed fairly
• Predictable Performance: Performance is predictable and consistent

Latency

• Processing Time: Latency depends on event processing time
• Queue Time: Events may wait in channels if event loop is busy
• Fair Scheduling: All event sources experience similar latency
• Responsiveness: System remains responsive to all event sources

Memory Usage

• Channel Buffering: Memory usage depends on channel buffer sizes
• Event Queuing: Queued events consume memory
• Efficient Storage: Minimal overhead for event storage
• Scalability: Memory usage scales with number of event sources

Common Use Cases

User Interface Systems

• GUI Applications: Handle mouse, keyboard, and window events
• Web Applications: Process user interactions and DOM events
• Game Engines: Handle input events and game state updates
• Mobile Apps: Process touch, gesture, and sensor events

Network Applications

• Web Servers: Handle HTTP requests and responses
• Chat Applications: Process messages and connection events
• API Gateways: Handle API requests and routing
• Load Balancers: Process connection and health check events

System Monitoring

• Process Monitoring: Handle process lifecycle events
• Resource Monitoring: Process CPU, memory, and disk events
• Network Monitoring: Handle network status and traffic events
• Log Processing: Process log file and system log events

IoT Applications

• Sensor Data: Process data from multiple sensors
• Device Control: Handle device status and control events
• Protocol Processing: Process various IoT protocol events
• Gateway Management: Handle device registration and communication events

Financial Systems

• Trading Systems: Process market data and order events
• Risk Management: Handle risk calculation and alert events
• Settlement Systems: Process transaction and settlement events
• Compliance Monitoring: Handle regulatory and compliance events

Real-time Systems

• Streaming Applications: Process media stream events
• Real-time Analytics: Handle analytics and reporting events
• Collaboration Tools: Process user collaboration events
• Live Broadcasting: Handle broadcast and viewer events

Embedded Systems

• Hardware Control: Process hardware interrupt events
• Sensor Networks: Handle sensor data and network events
• Control Systems: Process control and feedback events
• Automation Systems: Handle automation and scheduling events

Advanced Patterns

Priority-based Event Processing

• Event Priority: Process events based on priority levels
• Priority Queues: Use priority queues for event ordering
• Urgent Events: Handle urgent events with higher priority
• Fair Scheduling: Ensure fair processing across priority levels

Event Filtering and Routing

• Event Filtering: Filter events based on criteria
• Event Routing: Route events to appropriate handlers
• Event Transformation: Transform events before processing
• Event Aggregation: Aggregate multiple events into single events

Event Persistence

• Event Logging: Log all events for audit and debugging
• Event Replay: Replay events for testing and recovery
• Event Storage: Store events for later processing
• Event Archiving: Archive old events for compliance

Distributed Event Loops

• Multi-node Coordination: Coordinate event loops across multiple nodes
• Event Distribution: Distribute events across multiple event loops
• Load Balancing: Balance event processing load across nodes
• Fault Tolerance: Handle failures in distributed event processing

Best Practices

Event Design

• Structured Events: Use structured event types with clear meaning
• Event Size: Keep events reasonably sized for performance
• Event Immutability: Make events immutable to prevent side effects
• Event Validation: Validate events before processing

Performance Optimization

• Event Batching: Batch similar events for efficient processing
• Async Processing: Use async processing for long-running operations
• Event Pooling: Reuse event objects to reduce allocation overhead
• Memory Management: Monitor and manage memory usage

Error Handling

• Event Error Handling: Handle errors in event processing gracefully
• Error Recovery: Implement recovery mechanisms for failed events
• Error Reporting: Report errors for monitoring and debugging
• Circuit Breaker: Implement circuit breakers for external dependencies

Monitoring and Observability

• Event Metrics: Track event processing metrics
• Performance Monitoring: Monitor event loop performance
• Event Tracing: Trace events through the system
• Alerting: Alert on unusual event patterns or failures

The event loop pattern is particularly effective when you have:

• Multiple Event Sources: Multiple sources generating events
• Sequential Processing: Need for sequential, controlled event processing
• Resource Constraints: Limited resources that need efficient management
• Real-time Requirements: Need for responsive event handling
• State Management: Complex state that needs centralized management

This pattern provides a robust foundation for building responsive, efficient systems that can handle multiple event sources while maintaining control and predictability.

The final example implementation looks like this:

package examples

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

// RunEventLoop demonstrates the event loop pattern.
func RunEventLoop() {
	fmt.Println("=== Event Loop Pattern Example ===")

	// Create event channels
	userEvents := make(chan string, 10)
	systemEvents := make(chan string, 10)
	timerEvents := make(chan string, 10)
	shutdown := make(chan struct{})

	// Start event producers
	go userEventProducer(userEvents)
	go systemEventProducer(systemEvents)
	go timerEventProducer(timerEvents)

	// Start the event loop
	go eventLoop(userEvents, systemEvents, timerEvents, shutdown)

	// Let the system run for a while
	time.Sleep(5 * time.Second)

	// Shutdown
	fmt.Println("Shutting down event loop...")
	close(shutdown)

	// Wait a bit for cleanup
	time.Sleep(500 * time.Millisecond)
	fmt.Println("Event loop example completed!")
}

// Event loop that processes events from multiple sources
func eventLoop(userEvents, systemEvents, timerEvents <-chan string, shutdown <-chan struct{}) {
	fmt.Println("Event loop started...")

	for {
		select {
		case event := <-userEvents:
			fmt.Printf("Event Loop: Processing user event: %s\n", event)
			processUserEvent(event)

		case event := <-systemEvents:
			fmt.Printf("Event Loop: Processing system event: %s\n", event)
			processSystemEvent(event)

		case event := <-timerEvents:
			fmt.Printf("Event Loop: Processing timer event: %s\n", event)
			processTimerEvent(event)

		case <-shutdown:
			fmt.Println("Event Loop: Shutdown signal received, cleaning up...")
			return
		}
	}
}

// Event producers
func userEventProducer(events chan<- string) {
	userActions := []string{"login", "logout", "click", "scroll", "submit"}
	for i := 0; i < 8; i++ {
		time.Sleep(time.Duration(rand.Intn(800)+200) * time.Millisecond)
		action := userActions[rand.Intn(len(userActions))]
		events <- fmt.Sprintf("%s (user_%d)", action, i+1)
	}
}

func systemEventProducer(events chan<- string) {
	systemEvents := []string{"backup", "update", "maintenance", "alert", "sync"}
	for i := 0; i < 6; i++ {
		time.Sleep(time.Duration(rand.Intn(1000)+500) * time.Millisecond)
		event := systemEvents[rand.Intn(len(systemEvents))]
		events <- fmt.Sprintf("%s (system_%d)", event, i+1)
	}
}

func timerEventProducer(events chan<- string) {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	count := 0
	for range ticker.C {
		if count >= 5 {
			break
		}
		events <- fmt.Sprintf("heartbeat (timer_%d)", count+1)
		count++
	}
}

// Event processors
func processUserEvent(event string) {
	// Simulate processing time
	time.Sleep(100 * time.Millisecond)
	fmt.Printf("  -> User event processed: %s\n", event)
}

func processSystemEvent(event string) {
	// Simulate processing time
	time.Sleep(150 * time.Millisecond)
	fmt.Printf("  -> System event processed: %s\n", event)
}

func processTimerEvent(event string) {
	// Simulate processing time
	time.Sleep(50 * time.Millisecond)
	fmt.Printf("  -> Timer event processed: %s\n", event)
}

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 resource pooling pattern.

Overview

The Singleflight pattern ensures that only one execution of a function is in-flight for a given key at a time. Duplicate callers wait for the result instead of executing the function again. This pattern is essential for preventing duplicate expensive operations, caching with concurrent access, API call deduplication, and database query optimization.

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

Implementation Details

Structure

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

1. Singleflight Group – Manages in-flight calls and their results
2. Call Tracking – Tracks active calls for each key
3. Result Sharing – Shares results among duplicate callers

Code Analysis

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

func RunSingleflight() {
    // Create a singleflight group
    sf := newSingleflight()

    // Simulate multiple concurrent requests for the same key
    key := "user:123"
    numRequests := 5

    var wg sync.WaitGroup
    results := make([]string, numRequests)

    fmt.Printf("Making %d concurrent requests for key: %s\n", numRequests, key)

    // Launch concurrent requests
    for i := 0; i < numRequests; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("Request %d: Starting...\n", id)

            result := sf.Do(key, func() (interface{}, error) {
                // Simulate expensive operation (e.g., database query, API call)
                fmt.Printf("Request %d: Executing expensive operation...\n", id)
                time.Sleep(2 * time.Second)
                return fmt.Sprintf("Data for %s (processed by request %d)", key, id), nil
            })

            results[id] = result.(string)
            fmt.Printf("Request %d: Completed with result: %s\n", id, result)
        }(i)
    }

    wg.Wait()

    // Show that all results are the same (same execution)
    fmt.Println("\nAll results should be identical:")
    for i, result := range results {
        fmt.Printf("  Request %d: %s\n", i, result)
    }
}

Step-by-step breakdown:

1. Singleflight Group Creation:
   • sf := newSingleflight() creates a new singleflight group instance
   • The group manages all in-flight calls and their results
   • This is the central coordinator that prevents duplicate executions
2. Request Configuration:
   • key := "user:123" defines the key that all requests will use
   • numRequests := 5 specifies how many concurrent requests to simulate
   • Using the same key ensures that the singleflight pattern will be triggered
3. Result Storage Setup:
   • var wg sync.WaitGroup tracks when all request goroutines complete
   • results := make([]string, numRequests) creates a slice to store results
   • This allows us to verify that all requests get the same result
4. Concurrent Request Launch:
   • for i := 0; i < numRequests; i++ launches 5 concurrent requests
   • Each request gets a unique ID (0-4) for tracking and debugging
   • Uses closure func(id int) { ... }(i) to capture the request ID
5. Request Goroutine Implementation:
   • defer wg.Done() ensures the request signals completion when it exits
   • fmt.Printf("Request %d: Starting...\n", id) shows when each request starts
   • This helps demonstrate the concurrent nature of the requests
6. Singleflight Call Execution:
   • result := sf.Do(key, func() (interface{}, error) { ... }) calls the singleflight group
   • The function parameter defines the expensive operation to deduplicate
   • Only one execution of this function will occur, regardless of how many requests call it
7. Expensive Operation Simulation:
   • fmt.Printf("Request %d: Executing expensive operation...\n", id) shows which request actually executes
   • time.Sleep(2 * time.Second) simulates a 2-second expensive operation
   • Only the first request will see this message – others will wait for the result
8. Result Processing:
   • results[id] = result.(string) stores the result for this request
   • fmt.Printf("Request %d: Completed with result: %s\n", id, result) shows completion
   • All requests should complete with the same result
9. Verification and Display:
   • wg.Wait() waits for all requests to complete
   • The final loop displays all results to verify they’re identical
   • This demonstrates that only one execution occurred, but all requests got the result

Singleflight Implementation

type singleflight struct {
    mu    sync.Mutex
    calls map[string]*call
}

type call struct {
    wg    sync.WaitGroup
    val   interface{}
    err   error
    dups  int
}

func newSingleflight() *singleflight {
    return &singleflight{
        calls: make(map[string]*call),
    }
}

func (sf *singleflight) Do(key string, fn func() (interface{}, error)) interface{} {
    sf.mu.Lock()

    if c, exists := sf.calls[key]; exists {
        // Another call is in progress for this key
        c.dups++
        sf.mu.Unlock()
        fmt.Printf("Duplicate call for key %s, waiting for result...\n", key)
        c.wg.Wait()
        return c.val
    }

    // Create new call
    c := &call{}
    c.wg.Add(1)
    sf.calls[key] = c
    sf.mu.Unlock()

    // Execute the function
    c.val, c.err = fn()
    c.wg.Done()

    // Clean up
    sf.mu.Lock()
    delete(sf.calls, key)
    sf.mu.Unlock()

    return c.val
}

Singleflight implementation breakdown:

1. Data Structure Design:
   • singleflight struct contains the main singleflight group
   • mu sync.Mutex protects concurrent access to the calls map
   • calls map[string]*call tracks active calls by key
   • Simple structure with minimal memory footprint
2. Call Structure Design:
   • call struct represents an in-flight function call
   • wg sync.WaitGroup coordinates between executing and waiting requests
   • val interface{} stores the result of the function execution
   • err error stores any error from the function execution
   • dups int tracks the number of duplicate requests (for monitoring)
3. Constructor Function:
   • newSingleflight() creates a new singleflight group
   • calls: make(map[string]*call) initializes the empty calls map
   • Returns a pointer to the singleflight group for method calls
4. Do Method – Initial Lock and Duplicate Check:
   • sf.mu.Lock() acquires the mutex to protect the calls map
   • if c, exists := sf.calls[key]; exists checks if a call is already in progress
   • If a call exists, this is a duplicate request
5. Duplicate Request Handling:
   • c.dups++ increments the duplicate counter for monitoring
   • sf.mu.Unlock() releases the mutex immediately (no need to hold it while waiting)
   • fmt.Printf("Duplicate call for key %s, waiting for result...\n", key) shows duplicate detection
   • c.wg.Wait() blocks until the original call completes
   • return c.val returns the shared result to the duplicate request
6. New Call Creation:
   • c := &call{} creates a new call structure for this key
   • c.wg.Add(1) increments the WaitGroup (will be decremented when execution completes)
   • sf.calls[key] = c stores the call in the map to track it
   • sf.mu.Unlock() releases the mutex before executing the function
7. Function Execution:
   • c.val, c.err = fn() executes the provided function
   • This is the actual expensive operation that we want to deduplicate
   • Only one request will reach this point for each key
8. Result Sharing and Cleanup:
   • c.wg.Done() signals that execution is complete
   • This unblocks all waiting duplicate requests
   • sf.mu.Lock() acquires the mutex for cleanup
   • delete(sf.calls, key) removes the call from the map to free memory
   • sf.mu.Unlock() releases the mutex
   • return c.val returns the result to the original request

Key Design Patterns:

1. Mutex Protection: sync.Mutex ensures thread-safe access to the shared calls map during concurrent operations.
2. WaitGroup Coordination: sync.WaitGroup coordinates between the executing request and all waiting duplicate requests.
3. Map-based Call Tracking: Uses a map to efficiently track active calls by key with O(1) lookup time.
4. Function-based Interface: Generic interface func() (interface{}, error) supports any expensive operation.
5. Automatic Cleanup: Call entries are automatically removed from the map after completion to prevent memory leaks.
6. Duplicate Monitoring: Tracks the number of duplicate calls for observability and debugging.
7. Closure Pattern: func(id int) { ... }(i) captures loop variables in request goroutines for proper ID tracking.

How It Works

1. Request Arrival: Multiple requests arrive for the same key
2. First Request: The first request creates a call entry and executes the function
3. Duplicate Detection: Subsequent requests detect an existing call for the key
4. Result Waiting: Duplicate requests wait for the first request to complete
5. Result Sharing: All requests receive the same result from the single execution
6. Cleanup: The call entry is removed after completion

The pattern ensures that expensive operations are executed only once, even under high concurrency.

Why This Implementation?

Mutex-based Synchronization

• Thread Safety: Protects the calls map during concurrent access
• Simple Implementation: Straightforward synchronization for the call tracking
• Performance: Minimal overhead for the typical singleflight use case
• Reliability: Ensures data consistency during concurrent operations

WaitGroup for Result Coordination

• Synchronization: WaitGroup coordinates between the executing request and waiting requests
• Non-blocking: Executing request can complete without blocking
• Clean Coordination: All requests are properly synchronized
• Resource Efficiency: Minimal overhead for coordination

Map-based Call Tracking

• Key-based Lookup: Efficient lookup of active calls by key
• Memory Management: Calls are removed after completion to prevent memory leaks
• Scalability: Can handle multiple different keys simultaneously
• Flexibility: Easy to extend for additional metadata

Function-based Interface

• Generic Design: Can work with any function that returns (interface{}, error)
• Type Safety: Caller specifies the function to execute
• Flexibility: Supports different types of expensive operations
• Error Handling: Proper error propagation from the executed function

Duplicate Counting

• Monitoring: Tracks the number of duplicate calls for observability
• Debugging: Helps identify when singleflight is being used effectively
• Metrics: Can be used for performance monitoring
• Optimization: Helps identify opportunities for caching

Key Design Decisions

1. Map-based Storage: Uses a map to track active calls by key
2. WaitGroup Coordination: Uses WaitGroup to synchronize result sharing
3. Mutex Protection: Protects the calls map during concurrent access
4. Automatic Cleanup: Removes call entries after completion
5. Generic Interface: Supports any function with (interface{}, error) signature

Performance Characteristics

Throughput

• Deduplication: Eliminates duplicate expensive operations
• Resource Efficiency: Reduces resource usage for duplicate requests
• Scalability: Scales well with multiple concurrent users
• Memory Usage: Minimal memory overhead for call tracking

Latency

• First Request: Latency is the same as the original operation
• Duplicate Requests: Latency is reduced to just coordination overhead
• Coordination Overhead: Minimal overhead for result sharing
• Predictable: Latency is predictable and consistent

Memory Usage

• Call Tracking: Memory usage proportional to number of active calls
• Automatic Cleanup: Memory is freed when calls complete
• Efficient Storage: Minimal memory overhead per call
• Scalability: Memory usage doesn’t grow with duplicate requests

Common Use Cases

Caching Systems

• Cache Miss Handling: Prevent multiple cache misses for the same key
• Cache Population: Ensure only one request populates the cache
• Cache Warming: Prevent duplicate cache warming operations
• Distributed Caching: Coordinate cache operations across multiple nodes

Database Operations

• Query Deduplication: Prevent duplicate database queries
• Connection Pooling: Ensure only one connection is established per key
• Transaction Management: Prevent duplicate transaction operations
• Data Loading: Prevent duplicate data loading operations

API Calls

• External API Calls: Prevent duplicate calls to external APIs
• Rate Limiting: Ensure rate limits are respected across concurrent requests
• Authentication: Prevent duplicate authentication requests
• Data Fetching: Prevent duplicate data fetching operations

File Operations

• File Reading: Prevent duplicate file read operations
• File Processing: Ensure only one process handles a file
• Configuration Loading: Prevent duplicate configuration loading
• Resource Loading: Prevent duplicate resource loading

Background Jobs

• Job Deduplication: Prevent duplicate background job execution
• Scheduled Tasks: Ensure scheduled tasks run only once
• Data Processing: Prevent duplicate data processing operations
• Report Generation: Prevent duplicate report generation

System Operations

• Service Discovery: Prevent duplicate service discovery requests
• Health Checks: Prevent duplicate health check operations
• Configuration Updates: Prevent duplicate configuration updates
• Maintenance Operations: Prevent duplicate maintenance operations

User Interface

• Data Loading: Prevent duplicate data loading in UI components
• Form Submission: Prevent duplicate form submissions
• Search Operations: Prevent duplicate search requests
• Navigation: Prevent duplicate navigation operations

Advanced Patterns

Time-based Expiration

• Call Expiration: Expire calls after a certain time
• Stale Results: Handle cases where results become stale
• Refresh Logic: Automatically refresh expired results
• TTL Management: Manage time-to-live for cached results

Error Handling

• Error Propagation: Properly propagate errors to all callers
• Retry Logic: Implement retry logic for failed operations
• Circuit Breaker: Integrate with circuit breaker patterns
• Fallback Mechanisms: Provide fallback mechanisms for failures

Distributed Singleflight

• Multi-node Coordination: Coordinate across multiple nodes
• Shared State: Use Redis or similar for shared state
• Consistency: Ensure consistent behavior across nodes
• Network Overhead: Handle additional network communication

Metrics and Monitoring

• Call Tracking: Track the number of calls and duplicates
• Performance Metrics: Monitor performance impact
• Hit Rates: Track cache hit rates and effectiveness
• Alerting: Alert on unusual patterns or failures

Best Practices

Key Design

• Appropriate Keys: Use keys that properly identify the operation
• Key Uniqueness: Ensure keys are unique for different operations
• Key Stability: Use stable keys that don’t change frequently
• Key Size: Keep keys reasonably sized for performance

Function Design

• Idempotent Functions: Ensure functions are idempotent
• Error Handling: Properly handle and propagate errors
• Resource Management: Ensure proper resource cleanup
• Timeout Handling: Implement appropriate timeouts

Memory Management

• Call Cleanup: Ensure calls are properly cleaned up
• Memory Monitoring: Monitor memory usage of call tracking
• Leak Prevention: Prevent memory leaks from abandoned calls
• Size Limits: Implement limits on the number of active calls

Performance Optimization

• Key Hashing: Use efficient key hashing for large key sets
• Call Pooling: Reuse call objects to reduce allocation overhead
• Lazy Cleanup: Implement lazy cleanup to reduce lock contention
• Metrics Collection: Collect metrics for performance tuning

The singleflight pattern is particularly effective when you have:

• Expensive Operations: Operations that are costly to execute
• High Concurrency: Multiple concurrent requests for the same data
• Cache Misses: Frequent cache misses for the same keys
• External Dependencies: Operations that depend on external services
• Resource Constraints: Limited resources that need to be conserved

This pattern provides essential tools for building efficient, scalable systems that can handle high concurrency while minimizing resource usage and improving performance.

The final example implementation looks like this:

package examples

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

// RunSingleflight demonstrates the singleflight (spaceflight) pattern.
func RunSingleflight() {
	fmt.Println("=== Singleflight (Spaceflight) Pattern Example ===")

	// Create a singleflight group
	sf := newSingleflight()

	// Simulate multiple concurrent requests for the same key
	key := "user:123"
	numRequests := 5

	var wg sync.WaitGroup
	results := make([]string, numRequests)

	fmt.Printf("Making %d concurrent requests for key: %s\n", numRequests, key)

	// Launch concurrent requests
	for i := 0; i < numRequests; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			fmt.Printf("Request %d: Starting...\n", id)

			result := sf.Do(key, func() (interface{}, error) {
				// Simulate expensive operation (e.g., database query, API call)
				fmt.Printf("Request %d: Executing expensive operation...\n", id)
				time.Sleep(2 * time.Second)
				return fmt.Sprintf("Data for %s (processed by request %d)", key, id), nil
			})

			results[id] = result.(string)
			fmt.Printf("Request %d: Completed with result: %s\n", id, result)
		}(i)
	}

	wg.Wait()

	// Show that all results are the same (same execution)
	fmt.Println("\nAll results should be identical:")
	for i, result := range results {
		fmt.Printf("  Request %d: %s\n", i, result)
	}

	// Test with different keys
	fmt.Println("\nTesting with different keys:")
	keys := []string{"user:123", "user:456", "user:123"}

	for i, key := range keys {
		wg.Add(1)
		go func(id int, k string) {
			defer wg.Done()
			result := sf.Do(k, func() (interface{}, error) {
				fmt.Printf("Request %d: Executing for key %s...\n", id, k)
				time.Sleep(1 * time.Second)
				return fmt.Sprintf("Data for %s", k), nil
			})
			fmt.Printf("Request %d: Key %s -> %s\n", id, k, result)
		}(i, key)
	}

	wg.Wait()
	fmt.Println("\nSingleflight example completed!")
}

// Singleflight ensures only one execution per key
type singleflight struct {
	mu    sync.Mutex
	calls map[string]*call
}

type call struct {
	wg   sync.WaitGroup
	val  interface{}
	err  error
	dups int
}

func newSingleflight() *singleflight {
	return &singleflight{
		calls: make(map[string]*call),
	}
}

func (sf *singleflight) Do(key string, fn func() (interface{}, error)) interface{} {
	sf.mu.Lock()

	if c, exists := sf.calls[key]; exists {
		// Another call is in progress for this key
		c.dups++
		sf.mu.Unlock()
		fmt.Printf("Duplicate call for key %s, waiting for result...\n", key)
		c.wg.Wait()
		return c.val
	}

	// Create new call
	c := &call{}
	c.wg.Add(1)
	sf.calls[key] = c
	sf.mu.Unlock()

	// Execute the function
	c.val, c.err = fn()
	c.wg.Done()

	// Clean up
	sf.mu.Lock()
	delete(sf.calls, key)
	sf.mu.Unlock()

	return c.val
}

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 resource pooling pattern.

Overview

The MapReduce pattern processes large datasets by breaking the work into two phases: Map (process data in parallel) and Reduce (aggregate results). This pattern is essential for processing large datasets in parallel, distributed computing, data analytics and aggregation, and batch processing jobs.

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

Implementation Details

Structure

The MapReduce implementation in examples/mapreduce.go consists of three main phases:

1. Map Phase – Splits text into words and emits (word, 1) pairs
2. Shuffle Phase – Groups key-value pairs by key
3. Reduce Phase – Counts occurrences of each word

Code Analysis

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

func RunMapReduce() {
    // Sample data: words to count
    data := []string{
        "hello world",
        "hello go",
        "world of concurrency",
        "go programming",
        "concurrency patterns",
        "hello concurrency",
        "go world",
        "patterns in go",
    }

    fmt.Printf("Input data: %v\n", data)

    // Map phase: split words and emit (word, 1) pairs
    mapped := mapPhase(data)

    // Shuffle phase: group by key
    grouped := shufflePhase(mapped)

    // Reduce phase: count occurrences
    result := reducePhase(grouped)

    // Display results
    fmt.Println("\nWord count results:")
    for word, count := range result {
        fmt.Printf("  %s: %d\n", word, count)
    }
}

Step-by-step breakdown:

1. Input Data Preparation:
   • data := []string{...} creates a slice of text lines to process
   • Each line contains multiple words that will be counted
   • The data is designed to demonstrate word frequency (e.g., “hello” appears multiple times)
   • This simulates a real-world scenario where you have text documents to analyze
2. Map Phase Execution:
   • mapped := mapPhase(data) processes the input data in parallel
   • Returns a channel of KeyValue pairs where each word is paired with a count of 1
   • This is the first phase of the MapReduce pattern – transforming raw data into key-value pairs
   • The channel-based approach allows for streaming data between phases
3. Shuffle Phase Execution:
   • grouped := shufflePhase(mapped) groups the key-value pairs by key
   • Takes the channel output from the map phase
   • Groups all occurrences of the same word together
   • Returns a map where keys are words and values are slices of counts
4. Reduce Phase Execution:
   • result := reducePhase(grouped) aggregates the grouped data
   • Sums up all the counts for each word
   • Produces the final word count results
   • This is the final phase that produces the desired output
5. Result Display:
   • Iterates through the final results map
   • Displays each word and its total count
   • Shows the complete word frequency analysis

Map Phase Implementation

func mapPhase(data []string) <-chan KeyValue {
    out := make(chan KeyValue, len(data)*10) // Buffer for multiple words per line

    var wg sync.WaitGroup
    for _, line := range data {
        wg.Add(1)
        go func(text string) {
            defer wg.Done()
            words := strings.Fields(strings.ToLower(text))
            for _, word := range words {
                // Simulate some processing time
                time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond)
                out <- KeyValue{Key: word, Value: 1}
                fmt.Printf("Map: emitted (%s, 1)\n", word)
            }
        }(line)
    }

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

    return out
}

Map phase breakdown:

1. Channel Setup:
   • out := make(chan KeyValue, len(data)*10) creates a buffered channel for output
   • Buffer size of len(data)*10 accounts for multiple words per line
   • Buffering prevents blocking when multiple goroutines emit simultaneously
   • Returns a read-only channel (<-chan KeyValue) for encapsulation
2. Parallel Processing Setup:
   • var wg sync.WaitGroup tracks when all map goroutines complete
   • Each line of text gets its own goroutine for parallel processing
   • This enables true parallelism – all lines can be processed simultaneously
3. Goroutine Launch:
   • for _, line := range data iterates through each line of input
   • go func(text string) { ... }(line) launches a goroutine for each line
   • Uses closure to capture the current line in each goroutine
   • wg.Add(1) increments the wait group before each goroutine
4. Word Processing:
   • defer wg.Done() ensures the goroutine signals completion when it exits
   • strings.Fields(strings.ToLower(text)) splits the line into words and converts to lowercase
   • strings.Fields() splits on whitespace, handling multiple spaces correctly
   • strings.ToLower() ensures consistent word matching (case-insensitive)
5. Key-Value Emission:
   • for _, word := range words processes each word in the line
   • time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond) simulates processing time
   • out <- KeyValue{Key: word, Value: 1} emits a key-value pair for each word
   • Each word gets a value of 1 (will be summed in reduce phase)
6. Channel Cleanup:
   • go func() { wg.Wait(); close(out) }() runs in background
   • Waits for all map goroutines to complete
   • Closes the output channel when all processing is done
   • This signals to downstream phases that no more data is coming

Shuffle Phase Implementation

func shufflePhase(mapped <-chan KeyValue) map[string][]int {
    grouped := make(map[string][]int)
    var mu sync.Mutex

    var wg sync.WaitGroup
    for kv := range mapped {
        wg.Add(1)
        go func(kv KeyValue) {
            defer wg.Done()
            mu.Lock()
            grouped[kv.Key] = append(grouped[kv.Key], kv.Value)
            mu.Unlock()
            fmt.Printf("Shuffle: grouped %s -> %v\n", kv.Key, grouped[kv.Key])
        }(kv)
    }

    wg.Wait()
    return grouped
}

Shuffle phase breakdown:

1. Data Structure Setup:
   • grouped := make(map[string][]int) creates the output map
   • Keys are words (strings), values are slices of counts ([]int)
   • This structure groups all occurrences of each word together
   • var mu sync.Mutex provides thread safety for concurrent map access
2. WaitGroup Setup:
   • var wg sync.WaitGroup tracks when all shuffle goroutines complete
   • Each key-value pair gets its own goroutine for parallel processing
   • This maintains parallelism from the map phase
3. Channel Consumption:
   • for kv := range mapped consumes key-value pairs from the map phase
   • The loop continues until the map phase closes the channel
   • Each key-value pair is processed as it arrives
4. Parallel Grouping:
   • go func(kv KeyValue) { ... }(kv) launches a goroutine for each key-value pair
   • Uses closure to capture the current key-value pair
   • wg.Add(1) increments the wait group before each goroutine
5. Thread-Safe Grouping:
   • defer wg.Done() ensures proper cleanup
   • mu.Lock() and mu.Unlock() protect the shared map during concurrent access
   • grouped[kv.Key] = append(grouped[kv.Key], kv.Value) adds the value to the appropriate group
   • This groups all values for the same key together
6. Completion and Return:
   • wg.Wait() waits for all shuffle goroutines to complete
   • Returns the grouped map with all words and their associated counts
   • The map is now ready for the reduce phase

Reduce Phase Implementation

func reducePhase(grouped map[string][]int) map[string]int {
    result := make(map[string]int)
    var mu sync.Mutex

    var wg sync.WaitGroup
    for word, counts := range grouped {
        wg.Add(1)
        go func(word string, counts []int) {
            defer wg.Done()
            // Simulate some processing time
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)

            total := 0
            for _, count := range counts {
                total += count
            }

            mu.Lock()
            result[word] = total
            mu.Unlock()
            fmt.Printf("Reduce: %s -> %d\n", word, total)
        }(word, counts)
    }

    wg.Wait()
    return result
}

Reduce phase breakdown:

1. Result Structure Setup:
   • result := make(map[string]int) creates the final output map
   • Keys are words (strings), values are total counts (int)
   • This is the final result structure after aggregation
   • var mu sync.Mutex provides thread safety for concurrent result writing
2. WaitGroup Setup:
   • var wg sync.WaitGroup tracks when all reduce goroutines complete
   • Each word group gets its own goroutine for parallel processing
   • This maintains parallelism for the final aggregation phase
3. Parallel Processing Setup:
   • for word, counts := range grouped iterates through each word and its counts
   • Each word group is processed independently in parallel
   • This allows multiple words to be aggregated simultaneously
4. Goroutine Launch:
   • go func(word string, counts []int) { ... }(word, counts) launches a goroutine for each word
   • Uses closure to capture the current word and its counts
   • wg.Add(1) increments the wait group before each goroutine
5. Aggregation Processing:
   • defer wg.Done() ensures proper cleanup
   • time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) simulates processing time
   • total := 0 initializes the sum for this word
   • for _, count := range counts { total += count } sums all counts for the word
6. Thread-Safe Result Storage:
   • mu.Lock() and mu.Unlock() protect the shared result map
   • result[word] = total stores the final count for the word
   • This ensures thread-safe writing to the final result map
7. Completion and Return:
   • wg.Wait() waits for all reduce goroutines to complete
   • Returns the final result map with word counts
   • This completes the MapReduce pattern

Key Design Patterns:

1. Channel-based Data Flow: Channels provide natural streaming between phases with automatic backpressure.
2. Goroutine-per-Item Processing: Each data item gets its own goroutine for true parallelism.
3. Buffered Channels: Appropriate buffering prevents blocking and improves performance.
4. Mutex Protection: Thread-safe access to shared data structures during concurrent operations.
5. WaitGroup Coordination: Proper synchronization ensures all goroutines complete before proceeding.
6. Closure Pattern: func(text string) { ... }(line) captures loop variables in goroutines.
7. Structured Data Types: KeyValue struct provides type safety and clarity.

How It Works

Map Phase

1. Data Input: Receives input data (lines of text)
2. Parallel Processing: Each line is processed in its own goroutine
3. Word Extraction: Splits each line into individual words
4. Key-Value Emission: Emits (word, 1) pairs for each word
5. Channel Output: Sends key-value pairs to the shuffle phase

Shuffle Phase

1. Key-Value Reception: Receives key-value pairs from map phase
2. Parallel Grouping: Each key-value pair is processed in its own goroutine
3. Grouping by Key: Groups values by their keys
4. Thread Safety: Uses mutex to protect the shared grouped map
5. Output Preparation: Prepares grouped data for reduce phase

Reduce Phase

1. Grouped Data Input: Receives grouped data from shuffle phase
2. Parallel Aggregation: Each word group is processed in its own goroutine
3. Value Summation: Sums all values for each key
4. Result Storage: Stores final results in the result map
5. Final Output: Returns the final word count results

Why This Implementation?

Channel-based Communication

• Natural Flow: Channels provide natural data flow between phases
• Backpressure: Automatic backpressure when downstream phases are slow
• Synchronization: Channels handle synchronization between phases
• Composability: Easy to modify or extend individual phases

Goroutine-per-Item Processing

• True Parallelism: Each item is processed independently
• Scalability: Can utilize multiple CPU cores effectively
• Fault Isolation: Failure of one item doesn’t affect others
• Load Distribution: Work is naturally distributed across workers

Buffered Channels

• Performance: Buffered channels reduce blocking between phases
• Memory Management: Appropriate buffer sizes prevent memory issues
• Flow Control: Buffers can handle temporary processing delays
• Efficiency: Reduces context switching overhead

Mutex Protection

• Thread Safety: Protects shared data structures during concurrent access
• Simple Implementation: Straightforward synchronization for grouped data
• Performance: Minimal overhead for the typical MapReduce use case
• Reliability: Ensures data consistency during concurrent operations

Structured Data Types

• Type Safety: KeyValue struct provides type safety
• Clarity: Clear structure makes the code easy to understand
• Extensibility: Easy to extend for different data types
• Debugging: Structured data makes debugging easier

Key Design Decisions

1. Word Count Example: Simple example that clearly demonstrates the pattern
2. Simulated Processing Time: Random delays make concurrency visible
3. Detailed Logging: Output shows the flow through each phase
4. Error Handling: Simple implementation focuses on the core pattern
5. Memory Management: Appropriate buffer sizes prevent memory issues

Performance Characteristics

Throughput

• Parallel Processing: Throughput scales with the number of CPU cores
• Phase Overlap: Phases can overlap in processing (pipelining)
• Memory Usage: Memory usage depends on data size and buffer sizes
• Network Overhead: In distributed systems, network communication adds overhead

Latency

• Processing Time: Latency depends on the slowest phase
• Data Size: Larger datasets increase processing time
• Parallelism: More parallelism reduces latency
• I/O Operations: File I/O can be a significant bottleneck

Scalability

• Horizontal Scaling: Can distribute across multiple machines
• Vertical Scaling: Can utilize multiple CPU cores on single machine
• Data Partitioning: Data can be partitioned for parallel processing
• Load Balancing: Work is naturally distributed across workers

Common Use Cases

Data Processing

• Log Analysis: Process large log files to extract insights
• Text Processing: Analyze text documents for patterns and statistics
• Data Cleaning: Clean and validate large datasets
• ETL Operations: Extract, transform, and load data

Analytics and Reporting

• Business Intelligence: Generate reports from large datasets
• User Behavior Analysis: Analyze user activity patterns
• Performance Metrics: Calculate performance metrics from logs
• Trend Analysis: Identify trends in time-series data

Machine Learning

• Feature Engineering: Extract features from raw data
• Model Training: Process training data for machine learning models
• Data Preprocessing: Prepare data for machine learning algorithms
• Model Evaluation: Evaluate models on large datasets

Search and Indexing

• Web Crawling: Process web pages for search indexing
• Document Indexing: Index large document collections
• Content Analysis: Analyze content for search relevance
• Inverted Index: Build inverted indexes for search engines

Financial Data Processing

• Risk Analysis: Analyze financial data for risk assessment
• Trading Analysis: Process trading data for market analysis
• Fraud Detection: Analyze transactions for fraud patterns
• Portfolio Optimization: Process portfolio data for optimization

Scientific Computing

• Simulation Data: Process simulation results
• Sensor Data: Analyze data from scientific instruments
• Image Processing: Process large image datasets
• Genomic Analysis: Analyze genetic data

Social Media Analysis

• Sentiment Analysis: Analyze social media posts for sentiment
• Trend Detection: Identify trending topics and hashtags
• Network Analysis: Analyze social network connections
• Content Recommendation: Process user behavior for recommendations

Advanced Patterns

Distributed MapReduce

• Multi-node Processing: Distribute processing across multiple machines
• Fault Tolerance: Handle machine failures gracefully
• Load Balancing: Balance load across multiple nodes
• Data Locality: Process data close to where it’s stored

Streaming MapReduce

• Real-time Processing: Process data as it arrives
• Window-based Processing: Process data in time windows
• Incremental Updates: Update results incrementally
• Low Latency: Provide results with minimal delay

Iterative MapReduce

• Multiple Passes: Process data through multiple MapReduce cycles
• Iterative Algorithms: Support for iterative algorithms like PageRank
• Convergence: Continue until convergence criteria are met
• State Management: Maintain state across iterations

The MapReduce pattern is particularly effective when you have:

• Large Datasets: Datasets too large to process on a single machine
• Parallelizable Work: Work that can be divided into independent tasks
• Batch Processing: Operations that can be processed in batches
• Data Analytics: Need to extract insights from large datasets
• Distributed Computing: Need to utilize multiple machines or cores

This pattern provides a powerful framework for processing large datasets efficiently by leveraging parallel processing and distributed computing capabilities.

The final example implementation looks like this:

package examples

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

// RunMapReduce demonstrates the MapReduce pattern.
func RunMapReduce() {
	fmt.Println("=== MapReduce Pattern Example ===")

	// Sample data: words to count
	data := []string{
		"hello world",
		"hello go",
		"world of concurrency",
		"go programming",
		"concurrency patterns",
		"hello concurrency",
		"go world",
		"patterns in go",
	}

	fmt.Printf("Input data: %v\n", data)

	// Map phase: split words and emit (word, 1) pairs
	mapped := mapPhase(data)

	// Shuffle phase: group by key
	grouped := shufflePhase(mapped)

	// Reduce phase: count occurrences
	result := reducePhase(grouped)

	// Display results
	fmt.Println("\nWord count results:")
	for word, count := range result {
		fmt.Printf("  %s: %d\n", word, count)
	}

	fmt.Println("\nMapReduce example completed!")
}

// MapPhase splits text into words and emits (word, 1) pairs
func mapPhase(data []string) <-chan KeyValue {
	out := make(chan KeyValue, len(data)*10) // Buffer for multiple words per line

	var wg sync.WaitGroup
	for _, line := range data {
		wg.Add(1)
		go func(text string) {
			defer wg.Done()
			words := strings.Fields(strings.ToLower(text))
			for _, word := range words {
				// Simulate some processing time
				time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond)
				out <- KeyValue{Key: word, Value: 1}
				fmt.Printf("Map: emitted (%s, 1)\n", word)
			}
		}(line)
	}

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

	return out
}

// ShufflePhase groups key-value pairs by key
func shufflePhase(mapped <-chan KeyValue) map[string][]int {
	grouped := make(map[string][]int)
	var mu sync.Mutex

	var wg sync.WaitGroup
	for kv := range mapped {
		wg.Add(1)
		go func(kv KeyValue) {
			defer wg.Done()
			mu.Lock()
			grouped[kv.Key] = append(grouped[kv.Key], kv.Value)
			mu.Unlock()
			fmt.Printf("Shuffle: grouped %s -> %v\n", kv.Key, grouped[kv.Key])
		}(kv)
	}

	wg.Wait()
	return grouped
}

// ReducePhase counts occurrences of each word
func reducePhase(grouped map[string][]int) map[string]int {
	result := make(map[string]int)
	var mu sync.Mutex

	var wg sync.WaitGroup
	for word, counts := range grouped {
		wg.Add(1)
		go func(word string, counts []int) {
			defer wg.Done()
			// Simulate some processing time
			time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)

			total := 0
			for _, count := range counts {
				total += count
			}

			mu.Lock()
			result[word] = total
			mu.Unlock()
			fmt.Printf("Reduce: %s -> %d\n", word, total)
		}(word, counts)
	}

	wg.Wait()
	return result
}

// KeyValue represents a key-value pair
type KeyValue struct {
	Key   string
	Value int
}

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 singleflight pattern.

Overview

The Rate Limiting pattern controls the frequency of operations to prevent resource exhaustion and ensure fair usage. This pattern is essential for API request throttling, resource protection, preventing DoS attacks, and ensuring system stability under load.

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

Implementation Details

Structure

The rate limiting implementation in examples/rate_limiting.go demonstrates two main techniques:

1. Fixed Rate Limiting – Using time.Ticker for consistent rate control
2. Token Bucket Rate Limiting – Using a token bucket algorithm for burst handling

Code Analysis

func RunRateLimiting() {
    // Example 1: Fixed rate limiting
    fmt.Println("\n1. Fixed rate limiting (2 requests per second):")
    limiter := newFixedRateLimiter(2, time.Second)
    var wg sync.WaitGroup

    for i := 1; i <= 6; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            limiter.Wait()
            fmt.Printf("Request %d processed at %v\n", id, time.Now().Format("15:04:05.000"))
        }(i)
    }

    wg.Wait()

    // Example 2: Token bucket rate limiting
    fmt.Println("\n2. Token bucket rate limiting (3 tokens per second, burst of 5):")
    tokenLimiter := newTokenBucketLimiter(3, 5)
    var wg2 sync.WaitGroup

    for i := 1; i <= 10; i++ {
        wg2.Add(1)
        go func(id int) {
            defer wg2.Done()
            if tokenLimiter.Allow() {
                fmt.Printf("Token request %d granted at %v\n", id, time.Now().Format("15:04:05.000"))
            } else {
                fmt.Printf("Token request %d denied at %v\n", id, time.Now().Format("15:04:05.000"))
            }
        }(i)
    }

    wg2.Wait()
}

Fixed Rate Limiter Implementation

type fixedRateLimiter struct {
    ticker *time.Ticker
    stop   chan struct{}
}

func newFixedRateLimiter(rate int, interval time.Duration) *fixedRateLimiter {
    limiter := &fixedRateLimiter{
        ticker: time.NewTicker(interval / time.Duration(rate)),
        stop:   make(chan struct{}),
    }
    return limiter
}

func (r *fixedRateLimiter) Wait() {
    <-r.ticker.C
}

func (r *fixedRateLimiter) Stop() {
    r.ticker.Stop()
    close(r.stop)
}

Token Bucket Limiter Implementation

type tokenBucketLimiter struct {
    tokens    chan struct{}
    rate      time.Duration
    burst     int
    mu        sync.Mutex
    lastRefill time.Time
}

func newTokenBucketLimiter(rate int, burst int) *tokenBucketLimiter {
    limiter := &tokenBucketLimiter{
        tokens:     make(chan struct{}, burst),
        rate:       time.Second / time.Duration(rate),
        burst:      burst,
        lastRefill: time.Now(),
    }

    // Fill the bucket initially
    for i := 0; i < burst; i++ {
        limiter.tokens <- struct{}{}
    }

    // Start refilling tokens
    go limiter.refill()

    return limiter
}

func (t *tokenBucketLimiter) refill() {
    ticker := time.NewTicker(t.rate)
    defer ticker.Stop()

    for range ticker.C {
        select {
        case t.tokens <- struct{}{}:
            // Token added successfully
        default:
            // Bucket is full, skip
        }
    }
}

func (t *tokenBucketLimiter) Allow() bool {
    select {
    case <-t.tokens:
        return true
    default:
        return false
    }
}

func (t *tokenBucketLimiter) Wait() {
    <-t.tokens
}

How It Works

Fixed Rate Limiting

1. Ticker Creation: Creates a ticker that fires at the desired rate
2. Request Processing: Each request waits for the next tick
3. Rate Control: Requests are processed at exactly the specified rate
4. Synchronization: All requests are synchronized to the ticker

Token Bucket Rate Limiting

1. Bucket Initialization: Creates a bucket filled with initial tokens
2. Token Consumption: Requests consume tokens from the bucket
3. Token Refilling: Tokens are refilled at a constant rate
4. Burst Handling: Can handle bursts up to the bucket capacity

Why This Implementation?

Fixed Rate Limiter

• Predictable Rate: Ensures exactly the specified rate of operations
• Simple Implementation: Straightforward using Go’s time.Ticker
• Synchronization: All operations are naturally synchronized
• Resource Efficiency: Minimal memory and CPU overhead

Token Bucket Limiter

• Burst Handling: Can handle bursts of requests up to bucket capacity
• Flexible Rate: Allows for variable request patterns
• Non-blocking: Requests can be denied immediately if no tokens available
• Fair Distribution: Tokens are consumed in FIFO order

Channel-based Token Storage

• Thread Safety: Channels provide natural thread safety
• FIFO Order: Tokens are consumed in the order they were added
• Non-blocking Operations: Can check for tokens without blocking
• Efficient Implementation: Minimal overhead for token management

Goroutine-based Refilling

• Continuous Refilling: Tokens are refilled continuously in the background
• Independent Operation: Refilling doesn’t block request processing
• Automatic Management: No manual intervention required
• Resource Efficiency: Single goroutine handles all refilling

Key Design Decisions

1. Two Different Approaches: Demonstrates both fixed-rate and token-bucket techniques
2. Simulated Requests: Multiple concurrent requests show rate limiting in action
3. Timing Information: Output includes timestamps to show rate control
4. Non-blocking Token Check: Token bucket allows immediate denial of requests
5. Proper Cleanup: Both limiters provide cleanup methods

Performance Characteristics

Fixed Rate Limiter

• Consistent Rate: Exactly the specified rate, no variation
• Low Overhead: Minimal CPU and memory usage
• Synchronization: All operations are synchronized to the ticker
• Predictable: Behavior is completely predictable

Token Bucket Limiter

• Burst Handling: Can handle bursts up to bucket capacity
• Variable Rate: Average rate is maintained over time
• Immediate Response: Requests are processed immediately if tokens available
• Fair Distribution: Requests are processed in order

Memory Usage

• Fixed Limiter: Minimal memory usage (just ticker and channel)
• Token Bucket: Memory usage proportional to burst capacity
• Scalability: Both scale well with multiple concurrent users

Common Use Cases

API Rate Limiting

• REST APIs: Limit requests per user or IP address
• GraphQL APIs: Control query complexity and frequency
• Webhook Processing: Limit incoming webhook frequency
• Third-party APIs: Respect external API rate limits

Resource Protection

• Database Connections: Limit concurrent database connections
• File Operations: Control file system access frequency
• Network Requests: Limit outgoing network requests
• CPU-intensive Operations: Control CPU usage

User Experience

• UI Interactions: Limit button clicks or form submissions
• Search Queries: Control search request frequency
• File Uploads: Limit upload frequency and size
• Real-time Updates: Control update frequency

Security

• Brute Force Protection: Prevent password guessing attacks
• DDoS Mitigation: Limit requests from suspicious sources
• Account Protection: Prevent account takeover attempts
• API Abuse Prevention: Prevent API abuse and scraping

System Stability

• Load Balancing: Control load on backend services
• Cache Management: Limit cache invalidation frequency
• Background Jobs: Control job execution frequency
• Monitoring: Limit metric collection frequency

Microservices

• Service Communication: Limit inter-service request frequency
• Circuit Breaker: Control circuit breaker state changes
• Health Checks: Limit health check frequency
• Configuration Updates: Control configuration change frequency

Advanced Patterns

Sliding Window Rate Limiting

• Time-based Windows: Rate limits based on sliding time windows
• Precise Control: More precise than fixed windows
• Memory Efficient: Uses less memory than token buckets
• Complex Implementation: More complex to implement correctly

Leaky Bucket Rate Limiting

• Constant Rate: Processes requests at a constant rate
• Queue Management: Queues requests when rate is exceeded
• Smooth Processing: Provides smooth, predictable processing
• Memory Usage: Can use significant memory during bursts

Distributed Rate Limiting

• Multi-node Coordination: Coordinate rate limits across multiple nodes
• Shared State: Use Redis or similar for shared rate limit state
• Consistency: Ensure consistent rate limiting across all nodes
• Network Overhead: Additional network communication required

Adaptive Rate Limiting

• Dynamic Adjustment: Adjust rate limits based on system load
• Health-based: Reduce rates when system is unhealthy
• User-based: Different limits for different user types
• Time-based: Different limits for different times of day

Best Practices

Rate Limit Configuration

• Appropriate Limits: Set limits based on system capacity and requirements
• User Experience: Consider impact on user experience
• Monitoring: Monitor rate limit effectiveness and adjust as needed
• Documentation: Clearly document rate limits for API users

Error Handling

• Graceful Degradation: Provide meaningful error messages when limits are exceeded
• Retry Logic: Implement appropriate retry logic for rate-limited requests
• Backoff Strategies: Use exponential backoff for retries
• User Feedback: Inform users when they’re approaching limits

Monitoring and Alerting

• Rate Limit Metrics: Track rate limit usage and violations
• Alerting: Alert when rate limits are frequently exceeded
• Trend Analysis: Analyze rate limit patterns over time
• Capacity Planning: Use rate limit data for capacity planning

The rate limiting pattern is particularly effective when you have:

• Resource Constraints: Limited system resources that need protection
• Fair Usage: Need to ensure fair usage across multiple users
• Security Requirements: Need to prevent abuse and attacks
• API Management: Need to control API usage and prevent abuse
• System Stability: Need to maintain system stability under varying load

This pattern provides essential tools for building robust, scalable systems that can handle varying loads while protecting system resources and ensuring fair usage.

The final example implementation looks like this:

package examples

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

// RunRateLimiting demonstrates rate limiting patterns.
func RunRateLimiting() {
	fmt.Println("=== Rate Limiting Pattern Example ===")

	// Example 1: Fixed rate limiting
	fmt.Println("\n1. Fixed rate limiting (2 requests per second):")
	limiter := newFixedRateLimiter(2, time.Second)
	var wg sync.WaitGroup

	for i := 1; i <= 6; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			limiter.Wait()
			fmt.Printf("Request %d processed at %v\n", id, time.Now().Format("15:04:05.000"))
		}(i)
	}

	wg.Wait()

	// Example 2: Token bucket rate limiting
	fmt.Println("\n2. Token bucket rate limiting (3 tokens per second, burst of 5):")
	tokenLimiter := newTokenBucketLimiter(3, 5)
	var wg2 sync.WaitGroup

	for i := 1; i <= 10; i++ {
		wg2.Add(1)
		go func(id int) {
			defer wg2.Done()
			if tokenLimiter.Allow() {
				fmt.Printf("Token request %d granted at %v\n", id, time.Now().Format("15:04:05.000"))
			} else {
				fmt.Printf("Token request %d denied at %v\n", id, time.Now().Format("15:04:05.000"))
			}
		}(i)
	}

	wg2.Wait()

	fmt.Println("\nRate Limiting example completed!")
}

// Fixed rate limiter using time.Ticker
type fixedRateLimiter struct {
	ticker *time.Ticker
	stop   chan struct{}
}

func newFixedRateLimiter(rate int, interval time.Duration) *fixedRateLimiter {
	limiter := &fixedRateLimiter{
		ticker: time.NewTicker(interval / time.Duration(rate)),
		stop:   make(chan struct{}),
	}
	return limiter
}

func (r *fixedRateLimiter) Wait() {
	<-r.ticker.C
}

func (r *fixedRateLimiter) Stop() {
	r.ticker.Stop()
	close(r.stop)
}

// Token bucket rate limiter
type tokenBucketLimiter struct {
	tokens     chan struct{}
	rate       time.Duration
	burst      int
	mu         sync.Mutex
	lastRefill time.Time
}

func newTokenBucketLimiter(rate int, burst int) *tokenBucketLimiter {
	limiter := &tokenBucketLimiter{
		tokens:     make(chan struct{}, burst),
		rate:       time.Second / time.Duration(rate),
		burst:      burst,
		lastRefill: time.Now(),
	}

	// Fill the bucket initially
	for i := 0; i < burst; i++ {
		limiter.tokens <- struct{}{}
	}

	// Start refilling tokens
	go limiter.refill()

	return limiter
}

func (t *tokenBucketLimiter) refill() {
	ticker := time.NewTicker(t.rate)
	defer ticker.Stop()

	for range ticker.C {
		select {
		case t.tokens <- struct{}{}:
			// Token added successfully
		default:
			// Bucket is full, skip
		}
	}
}

func (t *tokenBucketLimiter) Allow() bool {
	select {
	case <-t.tokens:
		return true
	default:
		return false
	}
}

func (t *tokenBucketLimiter) Wait() {
	<-t.tokens
}

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 mapreduce pattern.