Go Concurrency Optimization Best Practices¶

// internal/concurrency/optimized_pool.go
package concurrency

import (
    "context"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)

// OptimizedWorkerPool represents an optimized worker pool
type OptimizedWorkerPool struct {
    workers     int
    jobQueue    chan Job
    resultQueue chan Result
    wg          sync.WaitGroup
    ctx         context.Context
    cancel      context.CancelFunc

    // Performance metrics
    processedJobs int64
    activeWorkers int64
    queueSize     int64
}

// Job represents a unit of work
type Job struct {
    ID   string
    Data interface{}
    Priority int
}

// Result represents the result of a job
type Result struct {
    JobID string
    Data  interface{}
    Error error
    Duration time.Duration
}

// NewOptimizedWorkerPool creates a new optimized worker pool
func NewOptimizedWorkerPool(workers int, queueSize int) *OptimizedWorkerPool {
    ctx, cancel := context.WithCancel(context.Background())

    return &OptimizedWorkerPool{
        workers:     workers,
        jobQueue:    make(chan Job, queueSize),
        resultQueue: make(chan Result, queueSize),
        ctx:         ctx,
        cancel:      cancel,
    }
}

// Start starts the worker pool
func (owp *OptimizedWorkerPool) Start() {
    for i := 0; i < owp.workers; i++ {
        owp.wg.Add(1)
        go owp.worker(i)
    }
}

// worker processes jobs from the queue
func (owp *OptimizedWorkerPool) worker(id int) {
    defer owp.wg.Done()

    for {
        select {
        case job := <-owp.jobQueue:
            atomic.AddInt64(&owp.activeWorkers, 1)
            atomic.AddInt64(&owp.queueSize, -1)

            start := time.Now()
            result := owp.processJob(job)
            result.Duration = time.Since(start)

            owp.resultQueue <- result
            atomic.AddInt64(&owp.processedJobs, 1)
            atomic.AddInt64(&owp.activeWorkers, -1)

        case <-owp.ctx.Done():
            return
        }
    }
}

// processJob processes a single job
func (owp *OptimizedWorkerPool) processJob(job Job) Result {
    // Simulate work based on priority
    time.Sleep(time.Duration(job.Priority) * time.Millisecond)

    return Result{
        JobID: job.ID,
        Data:  fmt.Sprintf("Processed: %v", job.Data),
        Error: nil,
    }
}

// Submit submits a job to the pool
func (owp *OptimizedWorkerPool) Submit(job Job) error {
    select {
    case owp.jobQueue <- job:
        atomic.AddInt64(&owp.queueSize, 1)
        return nil
    case <-owp.ctx.Done():
        return context.Canceled
    default:
        return fmt.Errorf("job queue is full")
    }
}

// GetResults returns the result channel
func (owp *OptimizedWorkerPool) GetResults() <-chan Result {
    return owp.resultQueue
}

// GetMetrics returns performance metrics
func (owp *OptimizedWorkerPool) GetMetrics() PoolMetrics {
    return PoolMetrics{
        ProcessedJobs: atomic.LoadInt64(&owp.processedJobs),
        ActiveWorkers: atomic.LoadInt64(&owp.activeWorkers),
        QueueSize:     atomic.LoadInt64(&owp.queueSize),
        Workers:       owp.workers,
    }
}

// Stop stops the worker pool
func (owp *OptimizedWorkerPool) Stop() {
    owp.cancel()
    close(owp.jobQueue)
    owp.wg.Wait()
    close(owp.resultQueue)
}

// PoolMetrics represents pool performance metrics
type PoolMetrics struct {
    ProcessedJobs int64
    ActiveWorkers int64
    QueueSize     int64
    Workers       int
}

// internal/concurrency/work_stealing.go
package concurrency

import (
    "context"
    "sync"
    "sync/atomic"
)

// WorkStealingPool represents a work-stealing pool
type WorkStealingPool struct {
    workers    []*Worker
    numWorkers  int
    ctx         context.Context
    cancel      context.CancelFunc
    wg          sync.WaitGroup
}

// Worker represents a work-stealing worker
type Worker struct {
    id          int
    localQueue  []Job
    queueMutex  sync.Mutex
    pool        *WorkStealingPool
    processed   int64
}

// NewWorkStealingPool creates a new work-stealing pool
func NewWorkStealingPool(numWorkers int) *WorkStealingPool {
    ctx, cancel := context.WithCancel(context.Background())

    pool := &WorkStealingPool{
        workers:    make([]*Worker, numWorkers),
        numWorkers: numWorkers,
        ctx:        ctx,
        cancel:     cancel,
    }

    for i := 0; i < numWorkers; i++ {
        pool.workers[i] = &Worker{
            id:   i,
            pool: pool,
        }
    }

    return pool
}

// Start starts the work-stealing pool
func (wsp *WorkStealingPool) Start() {
    for _, worker := range wsp.workers {
        wsp.wg.Add(1)
        go wsp.runWorker(worker)
    }
}

// runWorker runs a worker with work stealing
func (wsp *WorkStealingPool) runWorker(worker *Worker) {
    defer wsp.wg.Done()

    for {
        select {
        case <-wsp.ctx.Done():
            return
        default:
            if !wsp.processLocalJob(worker) {
                if !wsp.stealWork(worker) {
                    // No work available, yield
                    runtime.Gosched()
                }
            }
        }
    }
}

// processLocalJob processes a job from local queue
func (wsp *WorkStealingPool) processLocalJob(worker *Worker) bool {
    worker.queueMutex.Lock()
    defer worker.queueMutex.Unlock()

    if len(worker.localQueue) == 0 {
        return false
    }

    job := worker.localQueue[len(worker.localQueue)-1]
    worker.localQueue = worker.localQueue[:len(worker.localQueue)-1]

    // Process job
    wsp.executeJob(job)
    atomic.AddInt64(&worker.processed, 1)

    return true
}

// stealWork steals work from other workers
func (wsp *WorkStealingPool) stealWork(worker *Worker) bool {
    for i := 0; i < wsp.numWorkers; i++ {
        target := wsp.workers[(worker.id+i+1)%wsp.numWorkers]
        if target == worker {
            continue
        }

        if wsp.stealFromWorker(target, worker) {
            return true
        }
    }

    return false
}

// stealFromWorker steals work from a specific worker
func (wsp *WorkStealingPool) stealFromWorker(target, stealer *Worker) bool {
    target.queueMutex.Lock()
    defer target.queueMutex.Unlock()

    if len(target.localQueue) == 0 {
        return false
    }

    // Steal half of the work
    stealCount := len(target.localQueue) / 2
    if stealCount == 0 {
        stealCount = 1
    }

    stealer.queueMutex.Lock()
    defer stealer.queueMutex.Unlock()

    // Move work from target to stealer
    start := len(target.localQueue) - stealCount
    stealer.localQueue = append(stealer.localQueue, target.localQueue[start:]...)
    target.localQueue = target.localQueue[:start]

    return true
}

// executeJob executes a job
func (wsp *WorkStealingPool) executeJob(job Job) {
    // Implement job execution
    time.Sleep(time.Duration(job.Priority) * time.Millisecond)
}

// Submit submits a job to the pool
func (wsp *WorkStealingPool) Submit(job Job) {
    // Distribute work evenly
    target := wsp.workers[job.ID%wsp.numWorkers]
    target.queueMutex.Lock()
    target.localQueue = append(target.localQueue, job)
    target.queueMutex.Unlock()
}

// Stop stops the work-stealing pool
func (wsp *WorkStealingPool) Stop() {
    wsp.cancel()
    wsp.wg.Wait()
}

// GetMetrics returns pool metrics
func (wsp *WorkStealingPool) GetMetrics() []WorkerMetrics {
    metrics := make([]WorkerMetrics, len(wsp.workers))
    for i, worker := range wsp.workers {
        worker.queueMutex.Lock()
        metrics[i] = WorkerMetrics{
            ID:          worker.id,
            Processed:   atomic.LoadInt64(&worker.processed),
            QueueSize:   len(worker.localQueue),
        }
        worker.queueMutex.Unlock()
    }
    return metrics
}

// WorkerMetrics represents worker metrics
type WorkerMetrics struct {
    ID        int
    Processed int64
    QueueSize int
}

// internal/concurrency/lockfree_queue.go
package concurrency

import (
    "sync/atomic"
    "unsafe"
)

// LockFreeQueue represents a lock-free queue
type LockFreeQueue struct {
    head unsafe.Pointer
    tail unsafe.Pointer
}

// QueueNode represents a queue node
type QueueNode struct {
    value interface{}
    next  unsafe.Pointer
}

// NewLockFreeQueue creates a new lock-free queue
func NewLockFreeQueue() *LockFreeQueue {
    node := &QueueNode{}
    return &LockFreeQueue{
        head: unsafe.Pointer(node),
        tail: unsafe.Pointer(node),
    }
}

// Enqueue enqueues a value
func (lfq *LockFreeQueue) Enqueue(value interface{}) {
    node := &QueueNode{value: value}

    for {
        tail := atomic.LoadPointer(&lfq.tail)
        tailNode := (*QueueNode)(tail)

        if atomic.CompareAndSwapPointer(&tailNode.next, nil, unsafe.Pointer(node)) {
            atomic.CompareAndSwapPointer(&lfq.tail, tail, unsafe.Pointer(node))
            break
        }
    }
}

// Dequeue dequeues a value
func (lfq *LockFreeQueue) Dequeue() (interface{}, bool) {
    for {
        head := atomic.LoadPointer(&lfq.head)
        headNode := (*QueueNode)(head)
        next := atomic.LoadPointer(&headNode.next)

        if next == nil {
            return nil, false
        }

        nextNode := (*QueueNode)(next)
        if atomic.CompareAndSwapPointer(&lfq.head, head, next) {
            return nextNode.value, true
        }
    }
}

// IsEmpty checks if the queue is empty
func (lfq *LockFreeQueue) IsEmpty() bool {
    head := atomic.LoadPointer(&lfq.head)
    headNode := (*QueueNode)(head)
    next := atomic.LoadPointer(&headNode.next)
    return next == nil
}

// internal/concurrency/lockfree_stack.go
package concurrency

import (
    "sync/atomic"
    "unsafe"
)

// LockFreeStack represents a lock-free stack
type LockFreeStack struct {
    head unsafe.Pointer
}

// StackNode represents a stack node
type StackNode struct {
    value interface{}
    next  unsafe.Pointer
}

// NewLockFreeStack creates a new lock-free stack
func NewLockFreeStack() *LockFreeStack {
    return &LockFreeStack{}
}

// Push pushes a value onto the stack
func (lfs *LockFreeStack) Push(value interface{}) {
    node := &StackNode{value: value}

    for {
        head := atomic.LoadPointer(&lfs.head)
        node.next = head

        if atomic.CompareAndSwapPointer(&lfs.head, head, unsafe.Pointer(node)) {
            break
        }
    }
}

// Pop pops a value from the stack
func (lfs *LockFreeStack) Pop() (interface{}, bool) {
    for {
        head := atomic.LoadPointer(&lfs.head)
        if head == nil {
            return nil, false
        }

        node := (*StackNode)(head)
        next := atomic.LoadPointer(&node.next)

        if atomic.CompareAndSwapPointer(&lfs.head, head, next) {
            return node.value, true
        }
    }
}

// IsEmpty checks if the stack is empty
func (lfs *LockFreeStack) IsEmpty() bool {
    head := atomic.LoadPointer(&lfs.head)
    return head == nil
}

// internal/concurrency/channel_pool.go
package concurrency

import (
    "sync"
)

// ChannelPool represents a pool of channels
type ChannelPool struct {
    channels []chan interface{}
    index    int64
    mutex    sync.Mutex
}

// NewChannelPool creates a new channel pool
func NewChannelPool(size, bufferSize int) *ChannelPool {
    channels := make([]chan interface{}, size)
    for i := 0; i < size; i++ {
        channels[i] = make(chan interface{}, bufferSize)
    }

    return &ChannelPool{
        channels: channels,
    }
}

// GetChannel gets a channel from the pool
func (cp *ChannelPool) GetChannel() chan interface{} {
    cp.mutex.Lock()
    defer cp.mutex.Unlock()

    channel := cp.channels[cp.index]
    cp.index = (cp.index + 1) % int64(len(cp.channels))
    return channel
}

// Close closes all channels in the pool
func (cp *ChannelPool) Close() {
    for _, ch := range cp.channels {
        close(ch)
    }
}

// internal/concurrency/channel_batching.go
package concurrency

import (
    "context"
    "time"
)

// BatchedChannel represents a batched channel
type BatchedChannel struct {
    input     chan interface{}
    output    chan []interface{}
    batchSize int
    timeout   time.Duration
    ctx       context.Context
}

// NewBatchedChannel creates a new batched channel
func NewBatchedChannel(batchSize int, timeout time.Duration) *BatchedChannel {
    return &BatchedChannel{
        input:     make(chan interface{}),
        output:    make(chan []interface{}),
        batchSize: batchSize,
        timeout:   timeout,
        ctx:       context.Background(),
    }
}

// Start starts the batched channel
func (bc *BatchedChannel) Start() {
    go bc.batchProcessor()
}

// batchProcessor processes batches
func (bc *BatchedChannel) batchProcessor() {
    batch := make([]interface{}, 0, bc.batchSize)
    timer := time.NewTimer(bc.timeout)
    timer.Stop()

    for {
        select {
        case item := <-bc.input:
            batch = append(batch, item)

            if len(batch) >= bc.batchSize {
                bc.output <- batch
                batch = make([]interface{}, 0, bc.batchSize)
                timer.Stop()
            } else if len(batch) == 1 {
                timer.Reset(bc.timeout)
            }

        case <-timer.C:
            if len(batch) > 0 {
                bc.output <- batch
                batch = make([]interface{}, 0, bc.batchSize)
            }

        case <-bc.ctx.Done():
            if len(batch) > 0 {
                bc.output <- batch
            }
            return
        }
    }
}

// Send sends an item to the batched channel
func (bc *BatchedChannel) Send(item interface{}) {
    bc.input <- item
}

// Receive receives a batch from the channel
func (bc *BatchedChannel) Receive() <-chan []interface{} {
    return bc.output
}

// Close closes the batched channel
func (bc *BatchedChannel) Close() {
    close(bc.input)
}

// internal/concurrency/atomic_counters.go
package concurrency

import (
    "sync/atomic"
    "time"
)

// AtomicCounter represents an atomic counter
type AtomicCounter struct {
    value int64
}

// NewAtomicCounter creates a new atomic counter
func NewAtomicCounter() *AtomicCounter {
    return &AtomicCounter{}
}

// Increment increments the counter
func (ac *AtomicCounter) Increment() int64 {
    return atomic.AddInt64(&ac.value, 1)
}

// Decrement decrements the counter
func (ac *AtomicCounter) Decrement() int64 {
    return atomic.AddInt64(&ac.value, -1)
}

// Add adds a value to the counter
func (ac *AtomicCounter) Add(delta int64) int64 {
    return atomic.AddInt64(&ac.value, delta)
}

// Get gets the current value
func (ac *AtomicCounter) Get() int64 {
    return atomic.LoadInt64(&ac.value)
}

// Set sets the counter value
func (ac *AtomicCounter) Set(value int64) {
    atomic.StoreInt64(&ac.value, value)
}

// CompareAndSwap compares and swaps the value
func (ac *AtomicCounter) CompareAndSwap(old, new int64) bool {
    return atomic.CompareAndSwapInt64(&ac.value, old, new)
}

// AtomicRateLimiter represents an atomic rate limiter
type AtomicRateLimiter struct {
    tokens    int64
    maxTokens int64
    lastRefill time.Time
    refillRate int64
    mutex     int32
}

// NewAtomicRateLimiter creates a new atomic rate limiter
func NewAtomicRateLimiter(maxTokens, refillRate int64) *AtomicRateLimiter {
    return &AtomicRateLimiter{
        tokens:     maxTokens,
        maxTokens:  maxTokens,
        lastRefill: time.Now(),
        refillRate: refillRate,
    }
}

// Allow checks if a request is allowed
func (arl *AtomicRateLimiter) Allow() bool {
    now := time.Now()

    // Try to acquire lock
    if !atomic.CompareAndSwapInt32(&arl.mutex, 0, 1) {
        return false
    }
    defer atomic.StoreInt32(&arl.mutex, 0)

    // Refill tokens
    elapsed := now.Sub(arl.lastRefill)
    tokensToAdd := int64(elapsed.Seconds()) * arl.refillRate

    if tokensToAdd > 0 {
        arl.tokens = atomic.AddInt64(&arl.tokens, tokensToAdd)
        if arl.tokens > arl.maxTokens {
            arl.tokens = arl.maxTokens
        }
        arl.lastRefill = now
    }

    // Check if we have tokens
    if arl.tokens > 0 {
        atomic.AddInt64(&arl.tokens, -1)
        return true
    }

    return false
}

// internal/concurrency/context_pool.go
package concurrency

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

// ContextPool represents a pool of contexts
type ContextPool struct {
    pool sync.Pool
}

// NewContextPool creates a new context pool
func NewContextPool() *ContextPool {
    return &ContextPool{
        pool: sync.Pool{
            New: func() interface{} {
                return context.Background()
            },
        },
    }
}

// Get gets a context from the pool
func (cp *ContextPool) Get() context.Context {
    return cp.pool.Get().(context.Context)
}

// Put puts a context back into the pool
func (cp *ContextPool) Put(ctx context.Context) {
    cp.pool.Put(ctx)
}

// WithTimeout creates a context with timeout
func (cp *ContextPool) WithTimeout(timeout time.Duration) context.Context {
    ctx := cp.Get()
    return context.WithTimeout(ctx, timeout)
}

// WithCancel creates a context with cancellation
func (cp *ContextPool) WithCancel() (context.Context, context.CancelFunc) {
    ctx := cp.Get()
    return context.WithCancel(ctx)
}

// internal/concurrency/metrics.go
package concurrency

import (
    "sync/atomic"
    "time"
)

// ConcurrencyMetrics tracks concurrency metrics
type ConcurrencyMetrics struct {
    goroutinesCreated int64
    goroutinesActive  int64
    channelsCreated   int64
    channelsActive    int64
    locksAcquired     int64
    locksContended    int64
    startTime         time.Time
}

// NewConcurrencyMetrics creates new concurrency metrics
func NewConcurrencyMetrics() *ConcurrencyMetrics {
    return &ConcurrencyMetrics{
        startTime: time.Now(),
    }
}

// RecordGoroutineCreated records a goroutine creation
func (cm *ConcurrencyMetrics) RecordGoroutineCreated() {
    atomic.AddInt64(&cm.goroutinesCreated, 1)
    atomic.AddInt64(&cm.goroutinesActive, 1)
}

// RecordGoroutineFinished records a goroutine completion
func (cm *ConcurrencyMetrics) RecordGoroutineFinished() {
    atomic.AddInt64(&cm.goroutinesActive, -1)
}

// RecordChannelCreated records a channel creation
func (cm *ConcurrencyMetrics) RecordChannelCreated() {
    atomic.AddInt64(&cm.channelsCreated, 1)
    atomic.AddInt64(&cm.channelsActive, 1)
}

// RecordChannelClosed records a channel closure
func (cm *ConcurrencyMetrics) RecordChannelClosed() {
    atomic.AddInt64(&cm.channelsActive, -1)
}

// RecordLockAcquired records a lock acquisition
func (cm *ConcurrencyMetrics) RecordLockAcquired() {
    atomic.AddInt64(&cm.locksAcquired, 1)
}

// RecordLockContended records lock contention
func (cm *ConcurrencyMetrics) RecordLockContended() {
    atomic.AddInt64(&cm.locksContended, 1)
}

// GetStats returns current statistics
func (cm *ConcurrencyMetrics) GetStats() ConcurrencyStats {
    return ConcurrencyStats{
        GoroutinesCreated: atomic.LoadInt64(&cm.goroutinesCreated),
        GoroutinesActive:  atomic.LoadInt64(&cm.goroutinesActive),
        ChannelsCreated:   atomic.LoadInt64(&cm.channelsCreated),
        ChannelsActive:    atomic.LoadInt64(&cm.channelsActive),
        LocksAcquired:     atomic.LoadInt64(&cm.locksAcquired),
        LocksContended:    atomic.LoadInt64(&cm.locksContended),
        Uptime:           time.Since(cm.startTime),
    }
}

// ConcurrencyStats represents concurrency statistics
type ConcurrencyStats struct {
    GoroutinesCreated int64
    GoroutinesActive  int64
    ChannelsCreated   int64
    ChannelsActive    int64
    LocksAcquired     int64
    LocksContended    int64
    Uptime           time.Duration
}

// internal/concurrency/concurrency_test.go
package concurrency

import (
    "testing"
    "time"
)

func TestOptimizedWorkerPool(t *testing.T) {
    pool := NewOptimizedWorkerPool(4, 100)
    pool.Start()
    defer pool.Stop()

    // Submit jobs
    for i := 0; i < 100; i++ {
        job := Job{
            ID:       fmt.Sprintf("job-%d", i),
            Data:     fmt.Sprintf("data-%d", i),
            Priority: i % 5,
        }

        err := pool.Submit(job)
        if err != nil {
            t.Fatalf("Failed to submit job: %v", err)
        }
    }

    // Collect results
    results := make([]Result, 0, 100)
    for i := 0; i < 100; i++ {
        select {
        case result := <-pool.GetResults():
            results = append(results, result)
        case <-time.After(5 * time.Second):
            t.Fatal("Timeout waiting for results")
        }
    }

    if len(results) != 100 {
        t.Errorf("Expected 100 results, got %d", len(results))
    }
}

func TestWorkStealingPool(t *testing.T) {
    pool := NewWorkStealingPool(4)
    pool.Start()
    defer pool.Stop()

    // Submit jobs
    for i := 0; i < 100; i++ {
        job := Job{
            ID:       fmt.Sprintf("job-%d", i),
            Data:     fmt.Sprintf("data-%d", i),
            Priority: i % 5,
        }
        pool.Submit(job)
    }

    // Wait for processing
    time.Sleep(100 * time.Millisecond)

    metrics := pool.GetMetrics()
    if len(metrics) != 4 {
        t.Errorf("Expected 4 workers, got %d", len(metrics))
    }
}

func TestLockFreeQueue(t *testing.T) {
    queue := NewLockFreeQueue()

    // Test enqueue/dequeue
    for i := 0; i < 100; i++ {
        queue.Enqueue(i)
    }

    for i := 0; i < 100; i++ {
        value, ok := queue.Dequeue()
        if !ok {
            t.Fatal("Failed to dequeue")
        }
        if value != i {
            t.Errorf("Expected %d, got %v", i, value)
        }
    }

    if !queue.IsEmpty() {
        t.Error("Queue should be empty")
    }
}

func BenchmarkOptimizedWorkerPool(b *testing.B) {
    pool := NewOptimizedWorkerPool(4, 1000)
    pool.Start()
    defer pool.Stop()

    b.ResetTimer()
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            job := Job{
                ID:       "benchmark-job",
                Data:     "benchmark-data",
                Priority: 1,
            }
            pool.Submit(job)
        }
    })
}

func BenchmarkLockFreeQueue(b *testing.B) {
    queue := NewLockFreeQueue()

    b.ResetTimer()
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            queue.Enqueue("test")
            queue.Dequeue()
        }
    })
}

// 1. Optimized worker pool
pool := NewOptimizedWorkerPool(4, 1000)
pool.Start()
defer pool.Stop()

// 2. Work stealing
stealingPool := NewWorkStealingPool(4)
stealingPool.Start()
defer stealingPool.Stop()

// 3. Lock-free data structures
queue := NewLockFreeQueue()
stack := NewLockFreeStack()

// 4. Atomic operations
counter := NewAtomicCounter()
counter.Increment()

// Channel optimization
channelPool := NewChannelPool(10, 100)
ch := channelPool.GetChannel()

// Batched processing
batchedCh := NewBatchedChannel(100, 100*time.Millisecond)
batchedCh.Start()

// Context pooling
ctxPool := NewContextPool()
ctx := ctxPool.WithTimeout(5 * time.Second)

// Performance monitoring
metrics := NewConcurrencyMetrics()
stats := metrics.GetStats()