Go Worker Pool Patterns: Scalable Concurrent Processing

// Don't do this - creates 10,000 goroutines
for _, image := range images {
    go processImage(image)
}

// Create a pool of 10 workers
jobs := make(chan Image, 100)
results := make(chan Result, 100)

// Start workers
for i := 0; i < 10; i++ {
    go worker(jobs, results)
}

// Send work
for _, image := range images {
    jobs <- image
}

package main

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

// Task represents a unit of work to be performed
type Task func() error

// WorkerPool manages a pool of workers that process tasks
type WorkerPool struct {
	tasks   chan Task
	wg      sync.WaitGroup
	workers int
}

// NewWorkerPool creates a new worker pool with the specified number of workers
// and buffer capacity for tasks
func NewWorkerPool(workers, capacity int) *WorkerPool {
	return &WorkerPool{
		tasks:   make(chan Task, capacity),
		workers: workers,
	}
}

// Start launches the worker pool
func (p *WorkerPool) Start() {
	// Launch workers
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			// Worker loop: continuously process tasks until channel is closed
			for task := range p.tasks {
				// Execute the task
				err := task()
				if err != nil {
					fmt.Printf("Worker %d encountered error: %v\n", workerID, err)
				}
			}
			fmt.Printf("Worker %d shutting down\n", workerID)
		}(i + 1) // Pass worker ID for logging
	}
}

// Submit adds a task to the pool
func (p *WorkerPool) Submit(task Task) {
	p.tasks <- task
}

// Stop gracefully shuts down the worker pool
func (p *WorkerPool) Stop() {
	close(p.tasks)
	p.wg.Wait()
	fmt.Println("All workers have completed their tasks")
}

func main() {
	// Create a worker pool with 3 workers and a task queue capacity of 10
	pool := NewWorkerPool(3, 10)
	
	// Start the worker pool
	pool.Start()
	
	// Submit tasks to the pool
	for i := 1; i <= 10; i++ {
		taskID := i // Capture loop variable
		pool.Submit(func() error {
			fmt.Printf("Processing task %d\n", taskID)
			// Simulate work
			time.Sleep(100 * time.Millisecond)
			return nil
		})
	}
	
	// Allow some time for tasks to be processed
	time.Sleep(500 * time.Millisecond)
	
	// Gracefully shut down the pool
	pool.Stop()
}

package main

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

func demonstrateChannelTypes() {
	// Unbuffered channel - synchronous communication
	unbufferedChan := make(chan int)
	
	// Buffered channel - asynchronous up to capacity
	bufferedChan := make(chan int, 5)
	
	var wg sync.WaitGroup
	wg.Add(2)
	
	// Producer for unbuffered channel
	go func() {
		defer wg.Done()
		for i := 1; i <= 3; i++ {
			fmt.Printf("Sending to unbuffered channel: %d\n", i)
			// This will block until a receiver is ready
			unbufferedChan <- i
			fmt.Printf("Sent to unbuffered channel: %d\n", i)
		}
		close(unbufferedChan)
	}()
	
	// Producer for buffered channel
	go func() {
		defer wg.Done()
		for i := 1; i <= 8; i++ {
			if i <= 5 {
				fmt.Printf("Sending to buffered channel: %d\n", i)
				// This won't block until buffer is full (after 5 items)
				bufferedChan <- i
				fmt.Printf("Sent to buffered channel: %d\n", i)
			} else {
				fmt.Printf("Sending to buffered channel: %d (will block)\n", i)
				// This will block because buffer is full
				bufferedChan <- i
				fmt.Printf("Sent to buffered channel: %d (unblocked)\n", i)
			}
		}
		close(bufferedChan)
	}()
	
	// Consumer for unbuffered channel
	go func() {
		for val := range unbufferedChan {
			fmt.Printf("Received from unbuffered channel: %d\n", val)
			time.Sleep(100 * time.Millisecond) // Simulate processing time
		}
	}()
	
	// Consumer for buffered channel
	go func() {
		// Wait a bit to let buffer fill up
		time.Sleep(200 * time.Millisecond)
		
		for val := range bufferedChan {
			fmt.Printf("Received from buffered channel: %d\n", val)
			time.Sleep(100 * time.Millisecond) // Simulate processing time
		}
	}()
	
	wg.Wait()
}

func main() {
	demonstrateChannelTypes()
}

package main

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

// Task represents a unit of work that returns a result and possibly an error
type Task[T any] func() (T, error)

// Result contains the output of a task execution
type Result[T any] struct {
	Value T
	Err   error
}

// WorkerPool manages a pool of workers that process tasks and return results
type WorkerPool[T any] struct {
	tasks   chan Task[T]
	results chan Result[T]
	wg      sync.WaitGroup
	workers int
}

// NewWorkerPool creates a new worker pool with the specified number of workers
// and buffer capacities for tasks and results
func NewWorkerPool[T any](workers, taskCapacity, resultCapacity int) *WorkerPool[T] {
	return &WorkerPool[T]{
		tasks:   make(chan Task[T], taskCapacity),
		results: make(chan Result[T], resultCapacity),
		workers: workers,
	}
}

// Start launches the worker pool
func (p *WorkerPool[T]) Start() {
	// Launch workers
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			// Worker loop: continuously process tasks until channel is closed
			for task := range p.tasks {
				// Execute the task and capture result
				value, err := task()
				p.results <- Result[T]{Value: value, Err: err}
			}
		}(i + 1)
	}
	
	// Close results channel when all workers are done
	go func() {
		p.wg.Wait()
		close(p.results)
	}()
}

// Submit adds a task to the pool
func (p *WorkerPool[T]) Submit(task Task[T]) {
	p.tasks <- task
}

// Results returns the channel of results
func (p *WorkerPool[T]) Results() <-chan Result[T] {
	return p.results
}

// Stop gracefully shuts down the worker pool
func (p *WorkerPool[T]) Stop() {
	close(p.tasks)
}

func main() {
	// Create a worker pool with 3 workers, task capacity 10, result capacity 10
	pool := NewWorkerPool[int](3, 10, 10)
	
	// Start the worker pool
	pool.Start()
	
	// Submit tasks to the pool
	for i := 1; i <= 5; i++ {
		taskID := i // Capture loop variable
		pool.Submit(func() (int, error) {
			// Simulate work
			time.Sleep(100 * time.Millisecond)
			
			// Return the square of the task ID
			return taskID * taskID, nil
		})
	}
	
	// Stop accepting new tasks
	pool.Stop()
	
	// Process results as they become available
	for result := range pool.Results() {
		if result.Err != nil {
			fmt.Printf("Task error: %v\n", result.Err)
		} else {
			fmt.Printf("Task result: %d\n", result.Value)
		}
	}
}

package main

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

// DynamicPool represents a worker pool that can scale based on load
type DynamicPool struct {
	tasks         chan func()
	workerCount   atomic.Int32
	minWorkers    int32
	maxWorkers    int32
	idleTimeout   time.Duration
	mutex         sync.Mutex
	wg            sync.WaitGroup
	quit          chan struct{}
	metricsUpdate chan struct{}
	queueDepth    atomic.Int32
	activeWorkers atomic.Int32
}

// NewDynamicPool creates a new dynamic worker pool
func NewDynamicPool(minWorkers, maxWorkers int32, queueSize int, idleTimeout time.Duration) *DynamicPool {
	if minWorkers <= 0 {
		minWorkers = 1
	}
	if maxWorkers < minWorkers {
		maxWorkers = minWorkers
	}
	
	pool := &DynamicPool{
		tasks:         make(chan func(), queueSize),
		minWorkers:    minWorkers,
		maxWorkers:    maxWorkers,
		idleTimeout:   idleTimeout,
		quit:          make(chan struct{}),
		metricsUpdate: make(chan struct{}, 1),
	}
	
	// Start the minimum number of workers
	for i := int32(0); i < minWorkers; i++ {
		pool.startWorker()
	}
	
	// Start the scaling manager
	go pool.scalingManager()
	
	return pool
}

// Submit adds a task to the pool
func (p *DynamicPool) Submit(task func()) {
	select {
	case p.tasks <- task:
		// Task submitted successfully
		p.queueDepth.Add(1)
		
		// Signal metrics update if not already pending
		select {
		case p.metricsUpdate <- struct{}{}:
		default:
			// Update already pending
		}
	case <-p.quit:
		// Pool is shutting down
	}
}

// startWorker launches a new worker goroutine
func (p *DynamicPool) startWorker() {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	
	p.workerCount.Add(1)
	p.wg.Add(1)
	
	go func() {
		defer p.wg.Done()
		defer p.workerCount.Add(-1)
		
		idleTimer := time.NewTimer(p.idleTimeout)
		defer idleTimer.Stop()
		
		for {
			select {
			case task, ok := <-p.tasks:
				if !ok {
					// Channel closed, worker should exit
					return
				}
				
				// Reset idle timer
				if !idleTimer.Stop() {
					select {
					case <-idleTimer.C:
					default:
					}
				}
				
				// Process task
				p.activeWorkers.Add(1)
				p.queueDepth.Add(-1)
				task()
				p.activeWorkers.Add(-1)
				
				// Restart idle timer
				idleTimer.Reset(p.idleTimeout)
				
				// Signal metrics update
				select {
				case p.metricsUpdate <- struct{}{}:
				default:
					// Update already pending
				}
				
			case <-idleTimer.C:
				// Worker has been idle for too long
				currentCount := p.workerCount.Load()
				if currentCount > p.minWorkers {
					// Only exit if we're above minimum worker count
					return
				}
				// Reset timer if we need to stay alive
				idleTimer.Reset(p.idleTimeout)
				
			case <-p.quit:
				// Pool is shutting down
				return
			}
		}
	}()
}

// scalingManager periodically checks if we need to scale up or down
func (p *DynamicPool) scalingManager() {
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	
	for {
		select {
		case <-p.metricsUpdate:
			p.evaluateScaling()
		case <-ticker.C:
			p.evaluateScaling()
		case <-p.quit:
			return
		}
	}
}

// evaluateScaling decides whether to scale up or down based on current metrics
func (p *DynamicPool) evaluateScaling() {
	queueDepth := p.queueDepth.Load()
	workerCount := p.workerCount.Load()
	activeWorkers := p.activeWorkers.Load()
	
	// Scale up if queue is building and we're below max workers
	if queueDepth > 0 && activeWorkers >= workerCount*80/100 && workerCount < p.maxWorkers {
		// Start new workers based on queue depth, but don't exceed max
		workersToAdd := min(queueDepth, p.maxWorkers-workerCount)
		for i := int32(0); i < workersToAdd; i++ {
			p.startWorker()
		}
		fmt.Printf("Scaled up to %d workers (added %d)\n", p.workerCount.Load(), workersToAdd)
	}
	
	// Note: We don't need to explicitly scale down as workers will exit after idle timeout
}

// min returns the minimum of two int32 values
func min(a, b int32) int32 {
	if a < b {
		return a
	}
	return b
}

// Stop gracefully shuts down the pool
func (p *DynamicPool) Stop() {
	close(p.quit)
	close(p.tasks)
	p.wg.Wait()
}

// Stats returns current pool statistics
func (p *DynamicPool) Stats() (workers, active, queueDepth int32) {
	return p.workerCount.Load(), p.activeWorkers.Load(), p.queueDepth.Load()
}

func main() {
	// Create a dynamic pool with 2-10 workers, queue size 100, and 5s idle timeout
	pool := NewDynamicPool(2, 10, 100, 5*time.Second)
	
	// Submit a burst of tasks
	for i := 0; i < 20; i++ {
		taskID := i
		pool.Submit(func() {
			fmt.Printf("Processing task %d\n", taskID)
			// Simulate varying work durations
			time.Sleep(time.Duration(100+taskID*10) * time.Millisecond)
		})
	}
	
	// Monitor pool stats
	go func() {
		for i := 0; i < 10; i++ {
			workers, active, queue := pool.Stats()
			fmt.Printf("Stats: workers=%d, active=%d, queue=%d\n", workers, active, queue)
			time.Sleep(500 * time.Millisecond)
		}
	}()
	
	// Wait a bit for tasks to process
	time.Sleep(3 * time.Second)
	
	// Submit another burst of tasks
	for i := 20; i < 30; i++ {
		taskID := i
		pool.Submit(func() {
			fmt.Printf("Processing task %d\n", taskID)
			time.Sleep(time.Duration(50+taskID*5) * time.Millisecond)
		})
	}
	
	// Wait for all tasks to complete and workers to scale down
	time.Sleep(8 * time.Second)
	
	// Shut down the pool
	pool.Stop()
	fmt.Println("Pool shut down")
}

package main

import (
	"container/heap"
	"fmt"
	"sync"
	"time"
)

// PriorityTask represents a task with a priority level
type PriorityTask struct {
	execute  func()
	priority int
	index    int // Used by the heap implementation
}

// PriorityQueue implements heap.Interface for PriorityTask items
type PriorityQueue []*PriorityTask

// Len returns the length of the queue
func (pq PriorityQueue) Len() int { return len(pq) }

// Less defines the ordering of tasks (higher priority first)
func (pq PriorityQueue) Less(i, j int) bool {
	return pq[i].priority > pq[j].priority
}

// Swap swaps the elements with indexes i and j
func (pq PriorityQueue) Swap(i, j int) {
	pq[i], pq[j] = pq[j], pq[i]
	pq[i].index = i
	pq[j].index = j
}

// Push adds an element to the priority queue
func (pq *PriorityQueue) Push(x interface{}) {
	n := len(*pq)
	task := x.(*PriorityTask)
	task.index = n
	*pq = append(*pq, task)
}

// Pop removes and returns the highest priority element
func (pq *PriorityQueue) Pop() interface{} {
	old := *pq
	n := len(old)
	task := old[n-1]
	old[n-1] = nil  // Avoid memory leak
	task.index = -1 // For safety
	*pq = old[0 : n-1]
	return task
}

// PriorityWorkerPool manages a pool of workers that process tasks based on priority
type PriorityWorkerPool struct {
	workers    int
	taskQueue  PriorityQueue
	queueMutex sync.Mutex
	queueCond  *sync.Cond
	wg         sync.WaitGroup
	quit       chan struct{}
}

// NewPriorityWorkerPool creates a new priority-based worker pool
func NewPriorityWorkerPool(workers int) *PriorityWorkerPool {
	pool := &PriorityWorkerPool{
		workers:   workers,
		taskQueue: make(PriorityQueue, 0),
		quit:      make(chan struct{}),
	}
	pool.queueCond = sync.NewCond(&pool.queueMutex)
	return pool
}

// Start launches the worker pool
func (p *PriorityWorkerPool) Start() {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			for {
				// Get the next task from the priority queue
				p.queueMutex.Lock()
				for len(p.taskQueue) == 0 {
					// Check if we should quit
					select {
					case <-p.quit:
						p.queueMutex.Unlock()
						return
					default:
						// Wait for a task to be added
						p.queueCond.Wait()
						
						// Check again if we should quit after waking up
						select {
						case <-p.quit:
							p.queueMutex.Unlock()
							return
						default:
							// Continue to get task
						}
					}
				}
				
				// Get highest priority task
				task := heap.Pop(&p.taskQueue).(*PriorityTask)
				p.queueMutex.Unlock()
				
				// Execute the task
				fmt.Printf("Worker %d executing task with priority %d\n", workerID, task.priority)
				task.execute()
			}
		}(i + 1)
	}
}

// Submit adds a task to the pool with the specified priority
func (p *PriorityWorkerPool) Submit(task func(), priority int) {
	p.queueMutex.Lock()
	defer p.queueMutex.Unlock()
	
	// Create a new priority task
	priorityTask := &PriorityTask{
		execute:  task,
		priority: priority,
	}
	
	// Add to priority queue
	heap.Push(&p.taskQueue, priorityTask)
	
	// Signal that a task is available
	p.queueCond.Signal()
}

// Stop gracefully shuts down the pool
func (p *PriorityWorkerPool) Stop() {
	close(p.quit)
	p.queueCond.Broadcast() // Wake up all workers
	p.wg.Wait()
}

func main() {
	// Create a priority worker pool with 3 workers
	pool := NewPriorityWorkerPool(3)
	pool.Start()
	
	// Submit tasks with different priorities
	// Higher number = higher priority
	priorities := []int{1, 5, 2, 10, 3, 7, 1}
	
	for i, priority := range priorities {
		taskID := i
		taskPriority := priority
		
		pool.Submit(func() {
			fmt.Printf("Executing task %d with priority %d\n", taskID, taskPriority)
			// Simulate work
			time.Sleep(200 * time.Millisecond)
		}, taskPriority)
	}
	
	// Allow time for tasks to be processed
	time.Sleep(1 * time.Second)
	
	// Stop the pool
	pool.Stop()
	fmt.Println("Pool shut down")
}

package main

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

	"golang.org/x/time/rate"
)

// RateLimitedPool represents a worker pool with rate limiting
type RateLimitedPool struct {
	workers    int
	limiter    *rate.Limiter
	tasks      chan func()
	wg         sync.WaitGroup
	ctx        context.Context
	cancelFunc context.CancelFunc
}

// NewRateLimitedPool creates a new rate-limited worker pool
// rate is specified in tasks per second
func NewRateLimitedPool(workers int, tasksPerSecond float64, burstSize int, queueSize int) *RateLimitedPool {
	ctx, cancel := context.WithCancel(context.Background())
	
	return &RateLimitedPool{
		workers:    workers,
		limiter:    rate.NewLimiter(rate.Limit(tasksPerSecond), burstSize),
		tasks:      make(chan func(), queueSize),
		ctx:        ctx,
		cancelFunc: cancel,
	}
}

// Start launches the worker pool
func (p *RateLimitedPool) Start() {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			for {
				select {
				case task, ok := <-p.tasks:
					if !ok {
						// Channel closed, worker should exit
						return
					}
					
					// Wait for rate limiter permission
					err := p.limiter.Wait(p.ctx)
					if err != nil {
						// Context canceled, worker should exit
						return
					}
					
					// Execute the task
					startTime := time.Now()
					task()
					duration := time.Since(startTime)
					fmt.Printf("Worker %d completed task in %v\n", workerID, duration)
					
				case <-p.ctx.Done():
					// Context canceled, worker should exit
					return
				}
			}
		}(i + 1)
	}
}

// Submit adds a task to the pool
func (p *RateLimitedPool) Submit(task func()) bool {
	select {
	case p.tasks <- task:
		return true
	case <-p.ctx.Done():
		return false
	}
}

// Stop gracefully shuts down the pool
func (p *RateLimitedPool) Stop() {
	p.cancelFunc() // Signal all workers to stop
	close(p.tasks) // Stop accepting new tasks
	p.wg.Wait()    // Wait for all workers to finish
}

func main() {
	// Create a rate-limited pool with 5 workers, 10 tasks per second,
	// burst size of 3, and queue size of 100
	pool := NewRateLimitedPool(5, 10, 3, 100)
	pool.Start()
	
	// Submit 30 tasks
	for i := 0; i < 30; i++ {
		taskID := i
		if !pool.Submit(func() {
			fmt.Printf("Processing task %d\n", taskID)
			// Simulate work
			time.Sleep(50 * time.Millisecond)
		}) {
			fmt.Printf("Failed to submit task %d\n", taskID)
		}
	}
	
	// Wait for tasks to complete (should take ~3 seconds due to rate limiting)
	time.Sleep(4 * time.Second)
	
	// Stop the pool
	pool.Stop()
	fmt.Println("Pool shut down")
}

package main

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

// BackpressurePool represents a worker pool with backpressure mechanisms
type BackpressurePool struct {
	workers       int
	tasks         chan func()
	wg            sync.WaitGroup
	quit          chan struct{}
	maxQueueDepth int
}

// NewBackpressurePool creates a new worker pool with backpressure
ressurePool {
	return &BackpressurePool{
		workers:       workers,
		tasks:         make(chan func(), maxQueueDepth),
		quit:          make(chan struct{}),
		maxQueueDepth: maxQueueDepth,
	}
}

// Start launches the worker pool
func (p *BackpressurePool) Start() {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			for {
				select {
				case task, ok := <-p.tasks:
					if !ok {
						return // Channel closed
					}
					
					// Execute the task
					task()
					
				case <-p.quit:
					return
				}
			}
		}(i + 1)
	}
}

// Submit adds a task to the pool with backpressure
// Returns true if the task was accepted, false if rejected due to backpressure
func (p *BackpressurePool) Submit(task func()) bool {
	select {
	case p.tasks <- task:
		return true
	default:
		// Channel buffer is full - apply backpressure by rejecting the task
		return false
	}
}

// SubmitWithTimeout adds a task to the pool with a timeout
// Returns true if the task was accepted, false if rejected due to backpressure or timeout
func (p *BackpressurePool) SubmitWithTimeout(task func(), timeout time.Duration) bool {
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	
	select {
	case p.tasks <- task:
		return true
	case <-timer.C:
		// Timeout occurred
		return false
	}
}

// QueueDepth returns the current number of tasks in the queue
func (p *BackpressurePool) QueueDepth() int {
	return len(p.tasks)
}

// Stop gracefully shuts down the pool
func (p *BackpressurePool) Stop() {
	close(p.quit)
	close(p.tasks)
	p.wg.Wait()
}

func main() {
	// Create a backpressure pool with 3 workers and max queue depth of 5
	pool := NewBackpressurePool(3, 5)
	pool.Start()
	
	// Submit tasks with backpressure
	for i := 0; i < 20; i++ {
		taskID := i
		success := pool.Submit(func() {
			fmt.Printf("Processing task %d\n", taskID)
			// Simulate work
			time.Sleep(200 * time.Millisecond)
		})
		
		if success {
			fmt.Printf("Task %d accepted\n", taskID)
		} else {
			fmt.Printf("Task %d rejected due to backpressure\n", taskID)
			// In a real system, you might:
			// 1. Retry after a delay
			// 2. Store the task for later processing
			// 3. Drop the task if it's no longer relevant
			// 4. Alert monitoring systems about rejection rate
			
			// For this example, we'll wait a bit before retrying
			time.Sleep(100 * time.Millisecond)
			i-- // Retry this task
		}
	}
	
	// Wait for tasks to complete
	time.Sleep(2 * time.Second)
	
	// Stop the pool
	pool.Stop()
	fmt.Println("Pool shut down")
}

package main

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

// Task represents a unit of work
type Task struct {
	ID       int
	Priority int
	Execute  func()
}

// AdaptivePool represents a worker pool that can adapt to changing workloads
type AdaptivePool struct {
	queues        []chan Task
	queueCount    int
	minWorkers    int
	maxWorkers    int
	workerCount   atomic.Int32
	activeWorkers atomic.Int32
	idleTimeout   time.Duration
	wg            sync.WaitGroup
	quit          chan struct{}
	metrics       struct {
		tasksProcessed atomic.Int64
		tasksRejected  atomic.Int64
	}
}

// NewAdaptivePool creates a new adaptive worker pool
func NewAdaptivePool(queueCount, queueSize, minWorkers, maxWorkers int, idleTimeout time.Duration) *AdaptivePool {
	if minWorkers <= 0 {
		minWorkers = 1
	}
	if maxWorkers < minWorkers {
		maxWorkers = minWorkers
	}
	if queueCount <= 0 {
		queueCount = 1
	}
	
	pool := &AdaptivePool{
		queues:      make([]chan Task, queueCount),
		queueCount:  queueCount,
		minWorkers:  minWorkers,
		maxWorkers:  maxWorkers,
		idleTimeout: idleTimeout,
		quit:        make(chan struct{}),
	}
	
	// Initialize task queues
	for i := 0; i < queueCount; i++ {
		pool.queues[i] = make(chan Task, queueSize)
	}
	
	return pool
}

// Start launches the worker pool
func (p *AdaptivePool) Start() {
	// Start minimum number of workers
	for i := 0; i < p.minWorkers; i++ {
		p.startWorker()
	}
	
	// Start the scaling manager
	go p.scalingManager()
}

// startWorker launches a new worker goroutine
func (p *AdaptivePool) startWorker() {
	p.workerCount.Add(1)
	p.wg.Add(1)
	
	go func() {
		defer p.wg.Done()
		defer p.workerCount.Add(-1)
		
		// Create a timer for idle timeout
		idleTimer := time.NewTimer(p.idleTimeout)
		defer idleTimer.Stop()
		
		// Create cases for select
		cases := make([]chan Task, len(p.queues))
		for i := range p.queues {
			cases[i] = p.queues[i]
		}
		
		for {
			// Reset idle timer
			if !idleTimer.Stop() {
				select {
				case <-idleTimer.C:
				default:
				}
			}
			idleTimer.Reset(p.idleTimeout)
			
			// Try to get a task from any queue
			var task Task
			var ok bool
			
			select {
			case task, ok = <-cases[rand.Intn(len(cases))]:
				if !ok {
					// Queue closed
					return
				}
				
			case <-idleTimer.C:
				// Worker has been idle
				if p.workerCount.Load() > int32(p.minWorkers) {
					// We can terminate this worker
					return
				}
				continue
				
			case <-p.quit:
				// Pool is shutting down
				return
			}
			
			// Process the task
			p.activeWorkers.Add(1)
			task.Execute()
			p.activeWorkers.Add(-1)
			p.metrics.tasksProcessed.Add(1)
		}
	}()
}

// scalingManager periodically checks if we need to scale up or down
func (p *AdaptivePool) scalingManager() {
	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()
	
	for {
		select {
		case <-ticker.C:
			p.evaluateScaling()
		case <-p.quit:
			return
		}
	}
}

// evaluateScaling decides whether to scale up or down
func (p *AdaptivePool) evaluateScaling() {
	// Calculate queue depths
	totalQueueDepth := 0
	for _, queue := range p.queues {
		totalQueueDepth += len(queue)
	}
	
	workerCount := p.workerCount.Load()
	activeWorkers := p.activeWorkers.Load()
	
	// Scale up if:
	// 1. Queue depth is growing
	// 2. Most workers are active
	// 3. We're below max workers
	if totalQueueDepth > 0 && activeWorkers >= workerCount*80/100 && workerCount < int32(p.maxWorkers) {
		// Start new workers based on queue depth
		workersToAdd := min(int32(totalQueueDepth), int32(p.maxWorkers)-workerCount)
		for i := int32(0); i < workersToAdd; i++ {
			p.startWorker()
		}
		fmt.Printf("Scaled up to %d workers (added %d)\n", p.workerCount.Load(), workersToAdd)
	}
	
	// Note: Workers self-terminate after idle timeout
}

// Submit adds a task to the pool
// Returns true if the task was accepted, false if rejected
func (p *AdaptivePool) Submit(task Task) bool {
	// Choose a queue based on task priority
	queueIndex := task.Priority % p.queueCount
	
	// Try to submit to the chosen queue
	select {
	case p.queues[queueIndex] <- task:
		return true
	default:
		// Queue is full, try other queues
		for i := 0; i < p.queueCount; i++ {
			if i == queueIndex {
				continue
			}
			
			select {
			case p.queues[i] <- task:
				return true
			default:
				// This queue is also full
			}
		}
		
		// All queues are full
		p.metrics.tasksRejected.Add(1)
		return false
	}
}

// Stop gracefully shuts down the pool
func (p *AdaptivePool) Stop() {
	close(p.quit)
	
	// Close all queues
	for _, queue := range p.queues {
		close(queue)
	}
	
	p.wg.Wait()
}

// Stats returns current pool statistics
func (p *AdaptivePool) Stats() (workers, active int32, processed, rejected int64) {
	return p.workerCount.Load(),
		p.activeWorkers.Load(),
		p.metrics.tasksProcessed.Load(),
		p.metrics.tasksRejected.Load()
}

func main() {
	// Seed random number generator
	rand.Seed(time.Now().UnixNano())
	
	// Create an adaptive pool with:
	// - 3 queues with capacity 10 each
	// - 2-10 workers
	// - 5s idle timeout
	pool := NewAdaptivePool(3, 10, 2, 10, 5*time.Second)
	pool.Start()
	
	// Submit tasks with varying priorities
	for i := 0; i < 50; i++ {
		taskID := i
		priority := rand.Intn(5) // Random priority 0-4
		
		success := pool.Submit(Task{
			ID:       taskID,
			Priority: priority,
			Execute: func() {
				fmt.Printf("Processing task %d with priority %d\n", taskID, priority)
				// Simulate varying work durations
				time.Sleep(time.Duration(50+rand.Intn(200)) * time.Millisecond)
			},
		})
		
		if success {
			fmt.Printf("Task %d accepted\n", taskID)
		} else {
			fmt.Printf("Task %d rejected\n", taskID)
		}
		
		// Submit tasks with varying rates
		time.Sleep(time.Duration(10+rand.Intn(50)) * time.Millisecond)
	}
	
	// Monitor pool stats
	for i := 0; i < 5; i++ {
		workers, active, processed, rejected := pool.Stats()
		fmt.Printf("Stats: workers=%d, active=%d, processed=%d, rejected=%d\n",
			workers, active, processed, rejected)
		time.Sleep(1 * time.Second)
	}
	
	// Stop the pool
	pool.Stop()
	fmt.Println("Pool shut down")
}

package main

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

// TaskFunc represents a function that can be executed by the worker pool
type TaskFunc func(ctx context.Context) (interface{}, error)

// TaskResult contains the result of a task execution
type TaskResult struct {
	Value interface{}
	Err   error
	Retry int
}

// RetryablePool represents a worker pool with error handling and retry logic
type RetryablePool struct {
	workers     int
	maxRetries  int
	tasks       chan *taskWithMetadata
	results     chan TaskResult
	wg          sync.WaitGroup
	ctx         context.Context
	cancelFunc  context.CancelFunc
	errorPolicy ErrorPolicy
}

// taskWithMetadata wraps a task with its metadata
type taskWithMetadata struct {
	task      TaskFunc
	retries   int
	id        int
	createdAt time.Time
}

// ErrorPolicy defines how to handle task errors
type ErrorPolicy interface {
	// ShouldRetry determines if a failed task should be retried
	ShouldRetry(err error, retries int, maxRetries int) bool
	
	// RetryDelay returns how long to wait before retrying
	RetryDelay(retries int) time.Duration
}

// DefaultErrorPolicy implements a simple error policy
type DefaultErrorPolicy struct{}

// ShouldRetry implements ErrorPolicy.ShouldRetry
func (p DefaultErrorPolicy) ShouldRetry(err error, retries, maxRetries int) bool {
	// Don't retry if we've hit the max retries
	if retries >= maxRetries {
		return false
	}
	
	// Don't retry context cancellation errors
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return false
	}
	
	// Retry all other errors
	return true
}

// RetryDelay implements ErrorPolicy.RetryDelay
func (p DefaultErrorPolicy) RetryDelay(retries int) time.Duration {
	// Exponential backoff: 100ms, 200ms, 400ms, 800ms, etc.
	return time.Duration(100*(1<<retries)) * time.Millisecond
}

// NewRetryablePool creates a new worker pool with error handling and retry logic
func NewRetryablePool(workers, maxRetries, queueSize int) *RetryablePool {
	ctx, cancel := context.WithCancel(context.Background())
	
	return &RetryablePool{
		workers:     workers,
		maxRetries:  maxRetries,
		tasks:       make(chan *taskWithMetadata, queueSize),
		results:     make(chan TaskResult, queueSize),
		ctx:         ctx,
		cancelFunc:  cancel,
		errorPolicy: DefaultErrorPolicy{},
	}
}

// SetErrorPolicy sets a custom error policy
func (p *RetryablePool) SetErrorPolicy(policy ErrorPolicy) {
	p.errorPolicy = policy
}

// Start launches the worker pool
func (p *RetryablePool) Start() {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			for {
				select {
				case task, ok := <-p.tasks:
					if !ok {
						return // Channel closed
					}
					
					// Create a timeout context for this task
					taskCtx, cancel := context.WithTimeout(p.ctx, 10*time.Second)
					
					// Execute the task
					startTime := time.Now()
					value, err := task.task(taskCtx)
					duration := time.Since(startTime)
					
					// Clean up the timeout context
					cancel()
					
					if err != nil {
						fmt.Printf("Worker %d: Task %d failed after %v: %v (retry %d/%d)\n",
							workerID, task.id, duration, err, task.retries, p.maxRetries)
						
						// Check if we should retry
						if p.errorPolicy.ShouldRetry(err, task.retries, p.maxRetries) {
							// Increment retry counter
							task.retries++
							
							// Schedule retry after delay
							delay := p.errorPolicy.RetryDelay(task.retries)
							
							go func(t *taskWithMetadata) {
								select {
								case <-time.After(delay):
									// Resubmit the task
									select {
									case p.tasks <- t:
										fmt.Printf("Resubmitted task %d for retry %d/%d after %v\n",
											t.id, t.retries, p.maxRetries, delay)
									case <-p.ctx.Done():
										// Pool is shutting down
									}
								case <-p.ctx.Done():
									// Pool is shutting down
								}
							}(task)
							
							continue
						}
						
						// We're not retrying, send the error result
						p.results <- TaskResult{
							Value: nil,
							Err:   err,
							Retry: task.retries,
						}
						continue
					}
					
					// Task succeeded
					fmt.Printf("Worker %d: Task %d completed successfully in %v\n",
						workerID, task.id, duration)
					
					// Send the result
					p.results <- TaskResult{
						Value: value,
						Err:   nil,
						Retry: task.retries,
					}
					
				case <-p.ctx.Done():
					return // Context canceled
				}
			}
		}(i + 1)
	}
	
	// Close results channel when all workers are done
	go func() {
		p.wg.Wait()
		close(p.results)
	}()
}

// Submit adds a task to the pool
func (p *RetryablePool) Submit(id int, task TaskFunc) bool {
	select {
	case p.tasks <- &taskWithMetadata{
		task:      task,
		retries:   0,
		id:        id,
		createdAt: time.Now(),
	}:
		return true
	case <-p.ctx.Done():
		return false
	}
}

// Results returns the channel of results
func (p *RetryablePool) Results() <-chan TaskResult {
	return p.results
}

// Stop gracefully shuts down the pool
func (p *RetryablePool) Stop() {
	p.cancelFunc() // Signal all workers to stop
	close(p.tasks) // Stop accepting new tasks
	p.wg.Wait()    // Wait for all workers to finish
}

func main() {
	// Create a retryable pool with 3 workers, max 3 retries, and queue size 10
	pool := NewRetryablePool(3, 3, 10)
	
	// Start the pool
	pool.Start()
	
	// Process results in a separate goroutine
	go func() {
		for result := range pool.Results() {
			if result.Err != nil {
				fmt.Printf("Task failed after %d retries: %v\n", result.Retry, result.Err)
			} else {
				fmt.Printf("Task succeeded with result: %v (after %d retries)\n", result.Value, result.Retry)
			}
		}
	}()
	
	// Submit tasks with different error behaviors
	for i := 0; i < 10; i++ {
		taskID := i
		
		// Create different task behaviors
		var task TaskFunc
		
		switch i % 4 {
		case 0:
			// Task that always succeeds
			task = func(ctx context.Context) (interface{}, error) {
				time.Sleep(100 * time.Millisecond)
				return fmt.Sprintf("Result from task %d", taskID), nil
			}
		case 1:
			// Task that fails once then succeeds
			task = func(ctx context.Context) (interface{}, error) {
				time.Sleep(100 * time.Millisecond)
				
				if taskID%10 == 1 {
					// This will succeed on retry
					return fmt.Sprintf("Retry success from task %d", taskID), nil
				}
				
				return nil, fmt.Errorf("transient error in task %d", taskID)
			}
		case 2:
			// Task that always fails
			task = func(ctx context.Context) (interface{}, error) {
				time.Sleep(100 * time.Millisecond)
				return nil, fmt.Errorf("permanent error in task %d", taskID)
			}
		case 3:
			// Task that times out
			task = func(ctx context.Context) (interface{}, error) {
				select {
				case <-time.After(2 * time.Second):
					return fmt.Sprintf("Late result from task %d", taskID), nil
				case <-ctx.Done():
					return nil, ctx.Err()
				}
			}
		}
		
		pool.Submit(taskID, task)
	}
	
	// Wait for tasks to complete
	time.Sleep(5 * time.Second)
	
	// Stop the pool
	pool.Stop()
	fmt.Println("Pool shut down")
}

package main

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

// Metrics collects performance data about the worker pool
type Metrics struct {
	TasksSubmitted   atomic.Int64
	TasksCompleted   atomic.Int64
	TasksRejected    atomic.Int64
	TasksFailed      atomic.Int64
	QueueTimeNanos   atomic.Int64
	ProcessTimeNanos atomic.Int64
	TaskCount        atomic.Int64 // Current tasks in queue or being processed
}

// AvgQueueTime returns the average time tasks spend in the queue
func (m *Metrics) AvgQueueTime() time.Duration {
	completed := m.TasksCompleted.Load()
	if completed == 0 {
		return 0
	}
	avgNanos := m.QueueTimeNanos.Load() / completed
	return time.Duration(avgNanos) * time.Nanosecond
}

// AvgProcessTime returns the average time to process tasks
func (m *Metrics) AvgProcessTime() time.Duration {
	completed := m.TasksCompleted.Load()
	if completed == 0 {
		return 0
	}
	avgNanos := m.ProcessTimeNanos.Load() / completed
	return time.Duration(avgNanos) * time.Nanosecond
}

// InstrumentedTask represents a task with timing information
type InstrumentedTask struct {
	Execute    func() error
	EnqueuedAt time.Time
	StartedAt  time.Time
	FinishedAt time.Time
}

// InstrumentedPool represents a worker pool with performance instrumentation
type InstrumentedPool struct {
	workers    int
	tasks      chan *InstrumentedTask
	wg         sync.WaitGroup
	quit       chan struct{}
	metrics    Metrics
	sampleRate int // Sample rate for detailed metrics (1 = all tasks, 10 = 10% of tasks)
}

// NewInstrumentedPool creates a new instrumented worker pool
func NewInstrumentedPool(workers, queueSize, sampleRate int) *InstrumentedPool {
	return &InstrumentedPool{
		workers:    workers,
		tasks:      make(chan *InstrumentedTask, queueSize),
		quit:       make(chan struct{}),
		sampleRate: sampleRate,
	}
}

// Start launches the worker pool
func (p *InstrumentedPool) Start() {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			for {
				select {
				case task, ok := <-p.tasks:
					if !ok {
						return // Channel closed
					}
					
					// Record start time
					task.StartedAt = time.Now()
					
					// Calculate and record queue time
					queueTime := task.StartedAt.Sub(task.EnqueuedAt)
					p.metrics.QueueTimeNanos.Add(int64(queueTime))
					
					// Execute the task
					err := task.Execute()
					
					// Record finish time
					task.FinishedAt = time.Now()
					
					// Calculate and record process time
					processTime := task.FinishedAt.Sub(task.StartedAt)
					p.metrics.ProcessTimeNanos.Add(int64(processTime))
					
					// Update metrics
					p.metrics.TasksCompleted.Add(1)
					p.metrics.TaskCount.Add(-1)
					
					if err != nil {
						p.metrics.TasksFailed.Add(1)
					}
					
					// Detailed logging for sampled tasks
					if p.metrics.TasksCompleted.Load()%int64(p.sampleRate) == 0 {
						fmt.Printf("Task completed: queue_time=%v process_time=%v error=%v\n",
							queueTime, processTime, err)
					}
					
				case <-p.quit:
					return
				}
			}
		}(i + 1)
	}
	
	// Start metrics reporter
	go p.reportMetrics()
}

// Submit adds a task to the pool
func (p *InstrumentedPool) Submit(execute func() error) bool {
	task := &InstrumentedTask{
		Execute:    execute,
		EnqueuedAt: time.Now(),
	}
	
	select {
	case p.tasks <- task:
		p.metrics.TasksSubmitted.Add(1)
		p.metrics.TaskCount.Add(1)
		return true
	default:
		p.metrics.TasksRejected.Add(1)
		return false
	}
}

// reportMetrics periodically reports pool metrics
func (p *InstrumentedPool) reportMetrics() {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	
	for {
		select {
		case <-ticker.C:
			p.logMetrics()
		case <-p.quit:
			return
		}
	}
}

// logMetrics logs the current metrics
func (p *InstrumentedPool) logMetrics() {
	fmt.Printf("\n--- Worker Pool Metrics ---\n")
	fmt.Printf("Tasks Submitted: %d\n", p.metrics.TasksSubmitted.Load())
	fmt.Printf("Tasks Completed: %d\n", p.metrics.TasksCompleted.Load())
	fmt.Printf("Tasks Rejected: %d\n", p.metrics.TasksRejected.Load())
	fmt.Printf("Tasks Failed: %d\n", p.metrics.TasksFailed.Load())
	fmt.Printf("Current Tasks: %d\n", p.metrics.TaskCount.Load())
	fmt.Printf("Avg Queue Time: %v\n", p.metrics.AvgQueueTime())
	fmt.Printf("Avg Process Time: %v\n", p.metrics.AvgProcessTime())
	fmt.Printf("---------------------------\n")
}

// GetMetrics returns a copy of the current metrics
func (p *InstrumentedPool) GetMetrics() Metrics {
	return p.metrics
}

// Stop gracefully shuts down the pool
func (p *InstrumentedPool) Stop() {
	close(p.quit)
	close(p.tasks)
	p.wg.Wait()
	
	// Final metrics report
	p.logMetrics()
}

func main() {
	// Create an instrumented pool with 5 workers, queue size 20, sample rate 10
	pool := NewInstrumentedPool(5, 20, 10)
	pool.Start()
	
	// Submit tasks with varying processing times
	for i := 0; i < 100; i++ {
		taskID := i
		
		// Simulate varying submission rates
		if i%10 == 0 {
			time.Sleep(500 * time.Millisecond)
		}
		
		success := pool.Submit(func() error {
			// Simulate varying processing times
			processingTime := time.Duration(50+i%200) * time.Millisecond
			time.Sleep(processingTime)
			
			// Simulate occasional errors
			if i%20 == 0 {
				return fmt.Errorf("simulated error in task %d", taskID)
			}
			
			return nil
		})
		
		if !success {
			fmt.Printf("Task %d rejected\n", taskID)
		}
	}
	
	// Wait for tasks to complete
	time.Sleep(10 * time.Second)
	
	// Stop the pool
	pool.Stop()
}

package main

import (
func NewBackpressurePool(workers, maxQueueDepth int) *Backp
	"context"
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

// GracefulPool represents a worker pool with graceful shutdown capabilities
type GracefulPool struct {
	workers      int
	tasks        chan func(ctx context.Context) error
	wg           sync.WaitGroup
	ctx          context.Context
	cancel       context.CancelFunc
	shutdownCh   chan struct{}
	drainTimeout time.Duration
}

// NewGracefulPool creates a new worker pool with graceful shutdown
func NewGracefulPool(workers, queueSize int, drainTimeout time.Duration) *GracefulPool {
	ctx, cancel := context.WithCancel(context.Background())
	
	return &GracefulPool{
		workers:      workers,
		tasks:        make(chan func(ctx context.Context) error, queueSize),
		ctx:          ctx,
		cancel:       cancel,
		shutdownCh:   make(chan struct{}),
		drainTimeout: drainTimeout,
	}
}

// Start launches the worker pool
func (p *GracefulPool) Start() {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			
			for {
				select {
				case task, ok := <-p.tasks:
					if !ok {
						// Channel closed, worker should exit
						fmt.Printf("Worker %d shutting down (channel closed)\n", workerID)
						return
					}
					
					// Execute the task with the pool's context
					err := task(p.ctx)
					if err != nil {
						fmt.Printf("Worker %d task error: %v\n", workerID, err)
					}
					
				case <-p.ctx.Done():
					// Context canceled, worker should exit
					fmt.Printf("Worker %d shutting down (context canceled)\n", workerID)
					return
				}
			}
		}(i + 1)
	}
	
	// Handle OS signals for graceful shutdown
	go p.handleSignals()
}

// Submit adds a task to the pool
func (p *GracefulPool) Submit(task func(ctx context.Context) error) bool {
	select {
	case p.tasks <- task:
		return true
	case <-p.ctx.Done():
		return false
	default:
		// Queue is full
		return false
	}
}

// handleSignals sets up signal handling for graceful shutdown
func (p *GracefulPool) handleSignals() {
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
	
	// Wait for termination signal
	sig := <-signals
	fmt.Printf("Received signal %v, initiating graceful shutdown\n", sig)
	
	// Trigger graceful shutdown
	p.Shutdown()
}

// Shutdown initiates a graceful shutdown of the pool
func (p *GracefulPool) Shutdown() {
	fmt.Println("Starting graceful shutdown...")
	
	// Phase 1: Stop accepting new tasks
	fmt.Println("Phase 1: Stopping task acceptance")
	p.cancel() // Cancel context to prevent new submissions
	
	// Phase 2: Process remaining tasks with timeout
	fmt.Println("Phase 2: Processing remaining tasks")
	
	// Create a timeout for draining the queue
	drainCtx, drainCancel := context.WithTimeout(context.Background(), p.drainTimeout)
	defer drainCancel()
	
	// Create a goroutine to signal when all workers are done
	workersDone := make(chan struct{})
	go func() {
		p.wg.Wait()
		close(workersDone)
	}()
	
	// Wait for either all tasks to be processed or timeout
	select {
	case <-workersDone:
		fmt.Println("All tasks processed successfully")
	case <-drainCtx.Done():
		fmt.Println("Drain timeout reached, some tasks may not be processed")
	}
	
	// Phase 3: Close task channel to signal workers to exit
	fmt.Println("Phase 3: Closing task channel")
	close(p.tasks)
	
	// Phase 4: Wait for all workers to exit
	fmt.Println("Phase 4: Waiting for workers to exit")
	p.wg.Wait()
	
	fmt.Println("Graceful shutdown complete")
	close(p.shutdownCh)
}

// Wait blocks until the pool has shut down
func (p *GracefulPool) Wait() {
	<-p.shutdownCh
}

func main() {
	// Create a graceful pool with 5 workers, queue size 10, and 5s drain timeout
	pool := NewGracefulPool(5, 10, 5*time.Second)
	pool.Start()
	
	// Submit some tasks
	for i := 0; i < 20; i++ {
		taskID := i
		pool.Submit(func(ctx context.Context) error {
			fmt.Printf("Processing task %d\n", taskID)
			
			// Simulate work with context awareness
			select {
			case <-time.After(500 * time.Millisecond):
				fmt.Printf("Completed task %d\n", taskID)
				return nil
			case <-ctx.Done():
				fmt.Printf("Task %d canceled\n", taskID)
				return ctx.Err()
			}
		})
	}
	
	// Simulate running for a while
	time.Sleep(2 * time.Second)
	
	// Manually trigger shutdown (in a real app, this would be triggered by OS signals)
	go func() {
		time.Sleep(1 * time.Second)
		fmt.Println("Manually triggering shutdown")
		pool.Shutdown()
	}()
	
	// Wait for shutdown to complete
	pool.Wait()
	fmt.Println("Application exiting")
}