Go Concurrency Patterns for Distributed Systems: Advanced Synchronization and Coordination

package main

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

// Task represents a unit of work to be processed
type Task struct {
	ID     int
	Input  string
	Result string
	Err    error
}

// fanOut distributes tasks to multiple worker goroutines
func fanOut(ctx context.Context, tasks []Task, workerCount int) <-chan Task {
	tasksCh := make(chan Task)
	
	go func() {
		defer close(tasksCh)
		
		for _, task := range tasks {
			select {
			case tasksCh <- task:
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return tasksCh
}

// processTask simulates processing a task with potential failures
func processTask(ctx context.Context, task Task) Task {
	// Simulate processing time
	select {
	case <-time.After(100 * time.Millisecond):
		task.Result = fmt.Sprintf("Processed: %s", task.Input)
		return task
	case <-ctx.Done():
		task.Err = ctx.Err()
		return task
	}
}

// worker processes tasks from the input channel and sends results to the output channel
func worker(ctx context.Context, id int, tasksCh <-chan Task, resultsCh chan<- Task) {
	for task := range tasksCh {
		select {
		case <-ctx.Done():
			return
		default:
			task.Result = fmt.Sprintf("Worker %d processed: %s", id, task.Input)
			
			// Simulate occasional failures
			if task.ID%7 == 0 {
				task.Err = fmt.Errorf("processing error on task %d", task.ID)
				task.Result = ""
			}
			
			// Send the result
			select {
			case resultsCh <- task:
			case <-ctx.Done():
				return
			}
		}
	}
}

// fanIn collects results from multiple workers into a single channel
func fanIn(ctx context.Context, workerCount int, resultsCh chan Task) <-chan Task {
	multiplexedCh := make(chan Task)
	var wg sync.WaitGroup
	wg.Add(workerCount)
	
	// Start a goroutine for each worker to collect results
	for i := 0; i < workerCount; i++ {
		go func() {
			defer wg.Done()
			for {
				select {
				case result, ok := <-resultsCh:
					if !ok {
						return
					}
					select {
					case multiplexedCh <- result:
					case <-ctx.Done():
						return
					}
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	
	// Close the multiplexed channel once all workers are done
	go func() {
		wg.Wait()
		close(multiplexedCh)
	}()
	
	return multiplexedCh
}

// distributeAndProcess implements the complete fan-out/fan-in pattern
func distributeAndProcess(ctx context.Context, tasks []Task, workerCount int) []Task {
	// Create a buffered channel for results to prevent blocking
	resultsCh := make(chan Task, len(tasks))
	
	// Fan out: distribute tasks to workers
	tasksCh := fanOut(ctx, tasks, workerCount)
	
	// Start workers
	var wg sync.WaitGroup
	wg.Add(workerCount)
	for i := 0; i < workerCount; i++ {
		go func(workerID int) {
			defer wg.Done()
			worker(ctx, workerID, tasksCh, resultsCh)
		}(i)
	}
	
	// Close results channel when all workers are done
	go func() {
		wg.Wait()
		close(resultsCh)
	}()
	
	// Fan in: collect results
	collectedResults := make([]Task, 0, len(tasks))
	for result := range resultsCh {
		collectedResults = append(collectedResults, result)
	}
	
	return collectedResults
}

func main() {
	// Create a context with timeout
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	
	// Generate sample tasks
	tasks := make([]Task, 20)
	for i := 0; i < 20; i++ {
		tasks[i] = Task{
			ID:    i,
			Input: fmt.Sprintf("Task %d input", i),
		}
	}
	
	// Process tasks using fan-out/fan-in pattern
	results := distributeAndProcess(ctx, tasks, 5)
	
	// Print results
	fmt.Println("Results:")
	successCount := 0
	failureCount := 0
	
	for _, result := range results {
		if result.Err != nil {
			fmt.Printf("Task %d failed: %v\n", result.ID, result.Err)
			failureCount++
		} else {
			fmt.Printf("Task %d succeeded: %s\n", result.ID, result.Result)
			successCount++
		}
	}
	
	fmt.Printf("\nSummary: %d succeeded, %d failed\n", successCount, failureCount)
}

package main

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

// Event represents a message in our system
type Event struct {
	Source  string
	Type    string
	Payload interface{}
	Time    time.Time
}

// EventSource generates events from different parts of a distributed system
func EventSource(ctx context.Context, name string, interval time.Duration, priority int) <-chan Event {
	ch := make(chan Event)
	
	go func() {
		defer close(ch)
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		
		for {
			select {
			case <-ticker.C:
				// Generate an event
				event := Event{
					Source:  name,
					Type:    fmt.Sprintf("event-type-%d", rand.Intn(3)),
					Payload: fmt.Sprintf("data from %s", name),
					Time:    time.Now(),
				}
				
				// Try to send the event
				select {
				case ch <- event:
					// Event sent successfully
				case <-ctx.Done():
					return
				case <-time.After(50 * time.Millisecond):
					// Couldn't send within timeout, log and continue
					fmt.Printf("Warning: Dropped event from %s due to backpressure\n", name)
				}
				
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return ch
}

// PriorityMultiplexer combines multiple event sources with priority handling
func PriorityMultiplexer(ctx context.Context, sources map[string]<-chan Event, priorities map[string]int) <-chan Event {
	multiplexed := make(chan Event)
	
	go func() {
		defer close(multiplexed)
		
		// Keep track of active sources
		remaining := len(sources)
		
		// Create a case for each source
		for remaining > 0 {
			// Find the highest priority source with available events
			var highestPriority int = -1
			var selectedEvent Event
			var selectedSource string
			
			for name, ch := range sources {
				if ch == nil {
					continue // This source is already closed
				}
				
				// Try to receive from this source with a short timeout
				select {
				case event, ok := <-ch:
					if !ok {
						// Source is closed, remove it
						sources[name] = nil
						remaining--
						continue
					}
					
					// Check if this source has higher priority than current selection
					priority := priorities[name]
					if highestPriority == -1 || priority > highestPriority {
						highestPriority = priority
						selectedEvent = event
						selectedSource = name
					}
				case <-time.After(1 * time.Millisecond):
					// No event available from this source right now
					continue
				}
			}
			
			// If we found an event, try to send it
			if highestPriority != -1 {
				select {
				case multiplexed <- selectedEvent:
					fmt.Printf("Forwarded event from %s (priority %d)\n", 
						selectedSource, highestPriority)
				case <-ctx.Done():
					return
				}
			} else {
				// No events available from any source, wait a bit
				select {
				case <-time.After(10 * time.Millisecond):
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	
	return multiplexed
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	
	// Create event sources with different priorities
	sources := make(map[string]<-chan Event)
	priorities := make(map[string]int)
	
	// High priority source (critical system events)
	sources["critical"] = EventSource(ctx, "critical", 500*time.Millisecond, 10)
	priorities["critical"] = 10
	
	// Medium priority source (user actions)
	sources["user"] = EventSource(ctx, "user", 200*time.Millisecond, 5)
	priorities["user"] = 5
	
	// Low priority source (metrics)
	sources["metrics"] = EventSource(ctx, "metrics", 100*time.Millisecond, 1)
	priorities["metrics"] = 1
	
	// Multiplex the sources with priority handling
	multiplexed := PriorityMultiplexer(ctx, sources, priorities)
	
	// Process the multiplexed events
	for event := range multiplexed {
		fmt.Printf("Processed: %s event from %s at %v\n", 
			event.Type, event.Source, event.Time.Format(time.RFC3339Nano))
	}
}

package main

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

// Result represents the outcome of a distributed operation
type Result struct {
	Value interface{}
	Err   error
}

// simulateDistributedOperation mimics a call to a remote service with variable latency
func simulateDistributedOperation(ctx context.Context, name string, minLatency, maxLatency time.Duration) <-chan Result {
	resultCh := make(chan Result, 1) // Buffered to prevent goroutine leak
	
	go func() {
		// Calculate a random latency between min and max
		latency := minLatency + time.Duration(rand.Int63n(int64(maxLatency-minLatency)))
		
		// Simulate processing
		select {
		case <-time.After(latency):
			// 10% chance of error
			if rand.Intn(10) == 0 {
				resultCh <- Result{nil, fmt.Errorf("%s operation failed", name)}
			} else {
				resultCh <- Result{fmt.Sprintf("%s result", name), nil}
			}
		case <-ctx.Done():
			resultCh <- Result{nil, ctx.Err()}
		}
		
		close(resultCh)
	}()
	
	return resultCh
}

// firstResponse returns the first successful result or the last error if all fail
func firstResponse(ctx context.Context, timeout time.Duration, operations ...func(context.Context) <-chan Result) Result {
	// Create a context with timeout
	opCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	
	// Create a channel for the first response
	firstCh := make(chan Result, 1)
	
	// Launch all operations
	for _, op := range operations {
		go func(operation func(context.Context) <-chan Result) {
			resultCh := operation(opCtx)
			result := <-resultCh
			
			// Only forward successful results or the last error
			if result.Err == nil {
				// Try to send the successful result, but don't block
				select {
				case firstCh <- result:
					// Successfully sent the result
					cancel() // Cancel other operations
				default:
					// Channel already has a result, do nothing
				}
			} else if ctx.Err() != nil {
				// Context was canceled, likely because another operation succeeded
				return
			} else {
				// This was an operation error, send it but don't cancel others
				select {
				case firstCh <- result:
					// Sent the error
				default:
					// Channel already has a result, do nothing
				}
			}
		}(op)
	}
	
	// Wait for the first result or timeout
	select {
	case result := <-firstCh:
		return result
	case <-ctx.Done():
		return Result{nil, ctx.Err()}
	}
}

func main() {
	// Seed the random number generator
	rand.Seed(time.Now().UnixNano())
	
	// Create a parent context
	ctx := context.Background()
	
	// Define operations with different latency profiles
	fastButUnreliable := func(ctx context.Context) <-chan Result {
		return simulateDistributedOperation(ctx, "fast-service", 50*time.Millisecond, 150*time.Millisecond)
	}
	
	mediumLatency := func(ctx context.Context) <-chan Result {
		return simulateDistributedOperation(ctx, "medium-service", 100*time.Millisecond, 300*time.Millisecond)
	}
	
	slowButReliable := func(ctx context.Context) <-chan Result {
		return simulateDistributedOperation(ctx, "slow-service", 200*time.Millisecond, 500*time.Millisecond)
	}
	
	// Try multiple operations with a timeout
	fmt.Println("Executing distributed operations with redundancy...")
	result := firstResponse(ctx, 400*time.Millisecond, fastButUnreliable, mediumLatency, slowButReliable)
	
	if result.Err != nil {
		fmt.Printf("All operations failed or timed out: %v\n", result.Err)
	} else {
		fmt.Printf("Operation succeeded with result: %v\n", result.Value)
	}
	
	// Demonstrate a more complex scenario with multiple attempts
	fmt.Println("\nExecuting with multiple attempts...")
	
	for attempt := 1; attempt <= 3; attempt++ {
		fmt.Printf("Attempt %d...\n", attempt)
		
		// Increase timeout with each attempt
		timeout := time.Duration(attempt) * 200 * time.Millisecond
		
		result = firstResponse(ctx, timeout, fastButUnreliable, mediumLatency, slowButReliable)
		
		if result.Err == nil {
			fmt.Printf("Success on attempt %d: %v\n", attempt, result.Value)
			break
		} else {
			fmt.Printf("Attempt %d failed: %v\n", attempt, result.Err)
		}
	}
}

package main

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

// Job represents a unit of work
type Job struct {
	ID       int
	Payload  interface{}
	Duration time.Duration // Simulated processing time
}

// Result represents the outcome of job processing
type Result struct {
	JobID   int
	Output  interface{}
	Err     error
	Latency time.Duration
}

// AdaptiveWorkerPool implements a worker pool that scales based on load
type AdaptiveWorkerPool struct {
	jobQueue       chan Job
	resultQueue    chan Result
	workerCount    int32
	activeWorkers  int32
	maxWorkers     int32
	minWorkers     int32
	pendingJobs    int32
	mu             sync.Mutex
	stopCh         chan struct{}
	workerWg       sync.WaitGroup
	metrics        *PoolMetrics
	scaleInterval  time.Duration
	scaleThreshold float64 // Threshold for scaling (0-1)
}

// PoolMetrics tracks performance metrics for the worker pool
type PoolMetrics struct {
	totalJobs      int64
	completedJobs  int64
	failedJobs     int64
	totalLatency   int64 // in nanoseconds
	queueHighWater int32
}

// NewAdaptiveWorkerPool creates a new adaptive worker pool
func NewAdaptiveWorkerPool(ctx context.Context, minWorkers, maxWorkers, queueSize int) *AdaptiveWorkerPool {
	pool := &AdaptiveWorkerPool{
		jobQueue:       make(chan Job, queueSize),
		resultQueue:    make(chan Result, queueSize),
		minWorkers:     int32(minWorkers),
		maxWorkers:     int32(maxWorkers),
		workerCount:    0,
		activeWorkers:  0,
		pendingJobs:    0,
		stopCh:         make(chan struct{}),
		metrics:        &PoolMetrics{},
		scaleInterval:  500 * time.Millisecond,
		scaleThreshold: 0.7, // Scale up when 70% of workers are busy
	}
	
	// Start the minimum number of workers
	for i := 0; i < minWorkers; i++ {
		pool.startWorker(ctx)
	}
	
	// Start the autoscaler
	go pool.autoscaler(ctx)
	
	return pool
}

// startWorker launches a new worker goroutine
func (p *AdaptiveWorkerPool) startWorker(ctx context.Context) {
	p.workerWg.Add(1)
	atomic.AddInt32(&p.workerCount, 1)
	
	go func() {
		defer p.workerWg.Done()
		defer atomic.AddInt32(&p.workerCount, -1)
		
		for {
			select {
			case job, ok := <-p.jobQueue:
				if !ok {
					return // Channel closed
				}
				
				// Mark worker as active
				atomic.AddInt32(&p.activeWorkers, 1)
				atomic.AddInt32(&p.pendingJobs, -1)
				
				// Process the job
				startTime := time.Now()
				var result Result
				
				// Simulate processing with potential failures
				time.Sleep(job.Duration)
				
				if rand.Intn(10) < 1 { // 10% failure rate
					result = Result{
						JobID:   job.ID,
						Output:  nil,
						Err:     fmt.Errorf("processing error on job %d", job.ID),
						Latency: time.Since(startTime),
					}
					atomic.AddInt64(&p.metrics.failedJobs, 1)
				} else {
					result = Result{
						JobID:   job.ID,
						Output:  fmt.Sprintf("Processed result for job %d", job.ID),
						Err:     nil,
						Latency: time.Since(startTime),
					}
				}
				
				// Update metrics
				atomic.AddInt64(&p.metrics.completedJobs, 1)
				atomic.AddInt64(&p.metrics.totalLatency, int64(result.Latency))
				
				// Send the result
				select {
				case p.resultQueue <- result:
				case <-ctx.Done():
					return
				}
				
				// Mark worker as inactive
				atomic.AddInt32(&p.activeWorkers, -1)
				
			case <-p.stopCh:
				return
				
			case <-ctx.Done():
				return
			}
		}
	}()
}

// autoscaler adjusts the number of workers based on load
func (p *AdaptiveWorkerPool) autoscaler(ctx context.Context) {
	ticker := time.NewTicker(p.scaleInterval)
	defer ticker.Stop()
	
	for {
		select {
		case <-ticker.C:
			p.adjustWorkerCount(ctx)
		case <-p.stopCh:
			return
		case <-ctx.Done():
			return
		}
	}
}

// adjustWorkerCount scales the worker pool up or down based on current load
func (p *AdaptiveWorkerPool) adjustWorkerCount(ctx context.Context) {
	currentWorkers := atomic.LoadInt32(&p.workerCount)
	activeWorkers := atomic.LoadInt32(&p.activeWorkers)
	pendingJobs := atomic.LoadInt32(&p.pendingJobs)
	
	// Calculate utilization
	var utilization float64
	if currentWorkers > 0 {
		utilization = float64(activeWorkers) / float64(currentWorkers)
	}
	
	// Scale up if utilization is high and there are pending jobs
	if utilization >= p.scaleThreshold && pendingJobs > 0 && currentWorkers < p.maxWorkers {
		// Calculate how many workers to add (up to 25% more, at least 1)
		toAdd := max(1, int(float64(currentWorkers)*0.25))
		
		// Don't exceed max workers
		if currentWorkers+int32(toAdd) > p.maxWorkers {
			toAdd = int(p.maxWorkers - currentWorkers)
		}
		
		fmt.Printf("Scaling up: Adding %d workers (utilization: %.2f, pending: %d)\n", 
			toAdd, utilization, pendingJobs)
		
		for i := 0; i < toAdd; i++ {
			p.startWorker(ctx)
		}
	}
	
	// Scale down if utilization is low and we have more than minimum workers
	if utilization < p.scaleThreshold*0.5 && currentWorkers > p.minWorkers && pendingJobs == 0 {
		// Calculate how many workers to remove (up to 15% fewer, at least 1)
		toRemove := max(1, int(float64(currentWorkers)*0.15))
		
		// Don't go below min workers
		if currentWorkers-int32(toRemove) < p.minWorkers {
			toRemove = int(currentWorkers - p.minWorkers)
		}
		
		fmt.Printf("Scaling down: Removing %d workers (utilization: %.2f)\n", 
			toRemove, utilization)
		
		// Signal workers to stop
		for i := 0; i < toRemove; i++ {
			select {
			case p.stopCh <- struct{}{}:
			default:
				// If channel is full, we've already signaled enough workers
				break
			}
		}
	}
}

// Submit adds a job to the pool
func (p *AdaptiveWorkerPool) Submit(ctx context.Context, job Job) error {
	select {
	case p.jobQueue <- job:
		atomic.AddInt64(&p.metrics.totalJobs, 1)
		atomic.AddInt32(&p.pendingJobs, 1)
		
		// Update high water mark for queue
		pending := atomic.LoadInt32(&p.pendingJobs)
		for {
			highWater := atomic.LoadInt32(&p.metrics.queueHighWater)
			if pending <= highWater || atomic.CompareAndSwapInt32(&p.metrics.queueHighWater, highWater, pending) {
				break
			}
		}
		
		return nil
	case <-ctx.Done():
		return ctx.Err()
	default:
		return fmt.Errorf("job queue is full")
	}
}

// Results returns the channel for receiving results
func (p *AdaptiveWorkerPool) Results() <-chan Result {
	return p.resultQueue
}

// Shutdown gracefully shuts down the worker pool
func (p *AdaptiveWorkerPool) Shutdown() {
	close(p.jobQueue)
	p.workerWg.Wait()
	close(p.resultQueue)
	close(p.stopCh)
}

// GetMetrics returns the current pool metrics
func (p *AdaptiveWorkerPool) GetMetrics() PoolMetrics {
	completed := atomic.LoadInt64(&p.metrics.completedJobs)
	metrics := PoolMetrics{
		totalJobs:      atomic.LoadInt64(&p.metrics.totalJobs),
		completedJobs:  completed,
		failedJobs:     atomic.LoadInt64(&p.metrics.failedJobs),
		queueHighWater: atomic.LoadInt32(&p.metrics.queueHighWater),
	}
	
	// Calculate average latency
	if completed > 0 {
		metrics.totalLatency = atomic.LoadInt64(&p.metrics.totalLatency) / completed
	}
	
	return metrics
}

// Helper function for max of two integers
func max(a, b int) int {
	if a > b {
		return a
	}
	return b
}

func main() {
	// Seed random number generator
	rand.Seed(time.Now().UnixNano())
	
	// Create a context with timeout
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	
	// Create an adaptive worker pool
	pool := NewAdaptiveWorkerPool(ctx, 5, 20, 100)
	defer pool.Shutdown()
	
	// Start a goroutine to collect results
	go func() {
		for result := range pool.Results() {
			if result.Err != nil {
				fmt.Printf("Job %d failed: %v (took %v)\n", 
					result.JobID, result.Err, result.Latency)
			} else {
				fmt.Printf("Job %d completed: %v (took %v)\n", 
					result.JobID, result.Output, result.Latency)
			}
		}
	}()
	
	// Submit jobs in waves to demonstrate scaling
	for wave := 0; wave < 3; wave++ {
		fmt.Printf("\n--- Starting job wave %d ---\n", wave+1)
		
		// Submit a batch of jobs
		jobCount := 50 + wave*25
		for i := 0; i < jobCount; i++ {
			// Create jobs with variable processing times
			duration := time.Duration(50+rand.Intn(200)) * time.Millisecond
			job := Job{
				ID:       wave*1000 + i,
				Payload:  fmt.Sprintf("Job data %d", i),
				Duration: duration,
			}
			
			if err := pool.Submit(ctx, job); err != nil {
				fmt.Printf("Failed to submit job: %v\n", err)
			}
		}
		
		// Wait between waves
		time.Sleep(2 * time.Second)
		
		// Print current metrics
		metrics := pool.GetMetrics()
		fmt.Printf("\nPool metrics after wave %d:\n", wave+1)
		fmt.Printf("- Total jobs: %d\n", metrics.totalJobs)
		fmt.Printf("- Completed jobs: %d\n", metrics.completedJobs)
		fmt.Printf("- Failed jobs: %d\n", metrics.failedJobs)
		fmt.Printf("- Average latency: %v\n", time.Duration(metrics.totalLatency))
		fmt.Printf("- Queue high water: %d\n", metrics.queueHighWater)
		fmt.Printf("- Current workers: %d\n", atomic.LoadInt32(&pool.workerCount))
		fmt.Printf("- Active workers: %d\n", atomic.LoadInt32(&pool.activeWorkers))
	}
	
	// Wait for all jobs to complete
	time.Sleep(1 * time.Second)
	
	// Final metrics
	metrics := pool.GetMetrics()
	fmt.Printf("\nFinal pool metrics:\n")
	fmt.Printf("- Total jobs: %d\n", metrics.totalJobs)
	fmt.Printf("- Completed jobs: %d\n", metrics.completedJobs)
	fmt.Printf("- Failed jobs: %d\n", metrics.failedJobs)
	fmt.Printf("- Average latency: %v\n", time.Duration(metrics.totalLatency))
	fmt.Printf("- Queue high water: %d\n", metrics.queueHighWater)
}

package main

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

// DataItem represents a piece of data flowing through the pipeline
type DataItem struct {
	ID      int
	Data    interface{}
	Metadata map[string]interface{}
	Error   error
}

// PipelineStage represents a processing stage in the pipeline
type PipelineStage func(ctx context.Context, in <-chan DataItem) <-chan DataItem

// Pipeline represents a multi-stage data processing pipeline
type Pipeline struct {
	stages []PipelineStage
}

// NewPipeline creates a new data processing pipeline
func NewPipeline(stages ...PipelineStage) *Pipeline {
	return &Pipeline{
		stages: stages,
	}
}

// Execute runs data through the pipeline
func (p *Pipeline) Execute(ctx context.Context, source <-chan DataItem) <-chan DataItem {
	// Start with the source channel
	current := source
	
	// Apply each stage in sequence
	for _, stage := range p.stages {
		current = stage(ctx, current)
	}
	
	return current
}

// Source creates a source channel for the pipeline
func Source(ctx context.Context, items []DataItem) <-chan DataItem {
	out := make(chan DataItem)
	
	go func() {
		defer close(out)
		
		for _, item := range items {
			select {
			case out <- item:
				// Item sent successfully
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return out
}

// Example pipeline stages

// Validate checks if data items are valid
func Validate(ctx context.Context, in <-chan DataItem) <-chan DataItem {
	out := make(chan DataItem)
	
	go func() {
		defer close(out)
		
		for item := range in {
			// Skip already failed items
			if item.Error != nil {
				select {
				case out <- item:
				case <-ctx.Done():
					return
				}
				continue
			}
			
			// Perform validation
			if item.Data == nil {
				item.Error = fmt.Errorf("invalid data: nil value")
			}
			
			// Forward the item
			select {
			case out <- item:
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return out
}

// Transform applies a transformation to data items
func Transform(ctx context.Context, in <-chan DataItem) <-chan DataItem {
	out := make(chan DataItem)
	
	go func() {
		defer close(out)
		
		for item := range in {
			// Skip already failed items
			if item.Error != nil {
				select {
				case out <- item:
				case <-ctx.Done():
					return
				}
				continue
			}
			
			// Apply transformation
			switch v := item.Data.(type) {
			case string:
				item.Data = fmt.Sprintf("Transformed: %s", v)
			case int:
				item.Data = v * 2
			default:
				item.Error = fmt.Errorf("unsupported data type")
			}
			
			// Add metadata
			if item.Metadata == nil {
				item.Metadata = make(map[string]interface{})
			}
			item.Metadata["transformed_at"] = time.Now()
			
			// Forward the item
			select {
			case out <- item:
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return out
}

// Enrich adds additional data to items
func Enrich(ctx context.Context, in <-chan DataItem) <-chan DataItem {
	out := make(chan DataItem)
	
	go func() {
		defer close(out)
		
		for item := range in {
			// Skip already failed items
			if item.Error != nil {
				select {
				case out <- item:
				case <-ctx.Done():
					return
				}
				continue
			}
			
			// Add enrichment data
			if item.Metadata == nil {
				item.Metadata = make(map[string]interface{})
			}
			item.Metadata["enriched"] = true
			item.Metadata["enriched_at"] = time.Now()
			
			// Forward the item
			select {
			case out <- item:
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return out
}

// ParallelStage processes items in parallel
func ParallelStage(workers int, processor func(DataItem) DataItem) PipelineStage {
	return func(ctx context.Context, in <-chan DataItem) <-chan DataItem {
		out := make(chan DataItem)
		
		// Start a fixed number of workers
		var wg sync.WaitGroup
		wg.Add(workers)
		
		for i := 0; i < workers; i++ {
			go func(workerID int) {
				defer wg.Done()
				
				for item := range in {
					// Skip already failed items
					if item.Error != nil {
						select {
						case out <- item:
						case <-ctx.Done():
							return
						}
						continue
					}
					
					// Process the item
					processedItem := processor(item)
					
					// Add worker metadata
					if processedItem.Metadata == nil {
						processedItem.Metadata = make(map[string]interface{})
					}
					processedItem.Metadata["worker_id"] = workerID
					
					// Forward the processed item
					select {
					case out <- processedItem:
					case <-ctx.Done():
						return
					}
				}
			}(i)
		}
		
		// Close the output channel when all workers are done
		go func() {
			wg.Wait()
			close(out)
		}()
		
		return out
	}
}

func main() {
	// Create a context with timeout
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	
	// Create sample data
	data := []DataItem{
		{ID: 1, Data: "item 1"},
		{ID: 2, Data: "item 2"},
		{ID: 3, Data: "item 3"},
		{ID: 4, Data: "item 4"},
		{ID: 5, Data: nil}, // This will fail validation
		{ID: 6, Data: "item 6"},
		{ID: 7, Data: 42}, // Different type
		{ID: 8, Data: "item 8"},
	}
	
	// Create a pipeline
	pipeline := NewPipeline(
		Validate,
		Transform,
		ParallelStage(3, func(item DataItem) DataItem {
			// Simulate processing time
			time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
			return item
		}),
		Enrich,
	)
	
	// Create a source channel
	source := Source(ctx, data)
	
	// Execute the pipeline
	results := pipeline.Execute(ctx, source)
	
	// Collect and print results
	for result := range results {
		if result.Error != nil {
			fmt.Printf("Item %d failed: %v\n", result.ID, result.Error)
		} else {
			fmt.Printf("Item %d processed: %v with metadata: %v\n", 
				result.ID, result.Data, result.Metadata)
		}
	}
}

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)

// DistributedMutex wraps etcd's mutex for distributed locking
type DistributedMutex struct {
	client    *clientv3.Client
	session   *concurrency.Session
	mutex     *concurrency.Mutex
	lockPath  string
	nodeID    string
	isLocked  bool
	lockCount int
}

// NewDistributedMutex creates a new distributed mutex
func NewDistributedMutex(endpoints []string, lockPath string) (*DistributedMutex, error) {
	// Create etcd client
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create etcd client: %w", err)
	}

	// Generate a unique node ID
	hostname, err := os.Hostname()
	if err != nil {
		hostname = "unknown-host"
	}
	nodeID := fmt.Sprintf("%s-%d", hostname, time.Now().UnixNano())

	// Create a session with keep-alive
	session, err := concurrency.NewSession(client, concurrency.WithTTL(10))
	if err != nil {
		client.Close()
		return nil, fmt.Errorf("failed to create etcd session: %w", err)
	}

	// Create the mutex
	mutex := concurrency.NewMutex(session, lockPath)

	return &DistributedMutex{
		client:   client,
		session:  session,
		mutex:    mutex,
		lockPath: lockPath,
		nodeID:   nodeID,
		isLocked: false,
	}, nil
}

// Lock acquires the distributed mutex
func (dm *DistributedMutex) Lock(ctx context.Context) error {
	if dm.isLocked {
		dm.lockCount++
		return nil // Already locked by this instance
	}

	// Try to acquire the lock
	if err := dm.mutex.Lock(ctx); err != nil {
		return fmt.Errorf("failed to acquire lock: %w", err)
	}

	dm.isLocked = true
	dm.lockCount = 1
	log.Printf("Node %s acquired lock on %s", dm.nodeID, dm.lockPath)
	return nil
}

// Unlock releases the distributed mutex
func (dm *DistributedMutex) Unlock(ctx context.Context) error {
	if !dm.isLocked {
		return fmt.Errorf("mutex is not locked")
	}

	dm.lockCount--
	if dm.lockCount > 0 {
		return nil // Still held by other operations
	}

	// Release the lock
	if err := dm.mutex.Unlock(ctx); err != nil {
		return fmt.Errorf("failed to release lock: %w", err)
	}

	dm.isLocked = false
	log.Printf("Node %s released lock on %s", dm.nodeID, dm.lockPath)
	return nil
}

// Close releases resources
func (dm *DistributedMutex) Close() {
	if dm.isLocked {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = dm.Unlock(ctx)
	}
	dm.session.Close()
	dm.client.Close()
}

// IsLocked returns whether this instance holds the lock
func (dm *DistributedMutex) IsLocked() bool {
	return dm.isLocked
}

// GetLockOwner returns the current owner of the lock
func (dm *DistributedMutex) GetLockOwner(ctx context.Context) (string, error) {
	resp, err := dm.client.Get(ctx, dm.lockPath, clientv3.WithPrefix())
	if err != nil {
		return "", fmt.Errorf("failed to get lock info: %w", err)
	}

	if len(resp.Kvs) == 0 {
		return "", nil // No lock owner
	}

	return string(resp.Kvs[0].Value), nil
}

func main() {
	// Connect to etcd
	endpoints := []string{"localhost:2379"}
	lockPath := "/locks/my-critical-section"

	// Create the distributed mutex
	mutex, err := NewDistributedMutex(endpoints, lockPath)
	if err != nil {
		log.Fatalf("Failed to create distributed mutex: %v", err)
	}
	defer mutex.Close()

	// Create a context with timeout
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Try to acquire the lock
	log.Printf("Attempting to acquire lock...")
	if err := mutex.Lock(ctx); err != nil {
		log.Fatalf("Failed to acquire lock: %v", err)
	}

	// Simulate doing work while holding the lock
	log.Printf("Lock acquired! Performing critical section work...")
	time.Sleep(5 * time.Second)

	// Release the lock
	if err := mutex.Unlock(ctx); err != nil {
		log.Fatalf("Failed to release lock: %v", err)
	}
	log.Printf("Lock released")
}

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)

// LeaderElection manages the leader election process
type LeaderElection struct {
	client     *clientv3.Client
	session    *concurrency.Session
	election   *concurrency.Election
	nodeID     string
	electionID string
	isLeader   bool
	leaderCh   chan bool
	stopCh     chan struct{}
}

// NewLeaderElection creates a new leader election instance
func NewLeaderElection(endpoints []string, electionID string) (*LeaderElection, error) {
	// Create etcd client
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create etcd client: %w", err)
	}

	// Generate a unique node ID
	hostname, err := os.Hostname()
	if err != nil {
		hostname = "unknown-host"
	}
	nodeID := fmt.Sprintf("%s-%d", hostname, time.Now().UnixNano())

	// Create a session with keep-alive
	session, err := concurrency.NewSession(client, concurrency.WithTTL(10))
	if err != nil {
		client.Close()
		return nil, fmt.Errorf("failed to create etcd session: %w", err)
	}

	// Create the election
	election := concurrency.NewElection(session, electionID)

	return &LeaderElection{
		client:     client,
		session:    session,
		election:   election,
		nodeID:     nodeID,
		electionID: electionID,
		isLeader:   false,
		leaderCh:   make(chan bool, 1),
		stopCh:     make(chan struct{}),
	}, nil
}

// Campaign starts the leader election process
func (le *LeaderElection) Campaign(ctx context.Context) error {
	// Start campaigning for leadership
	if err := le.election.Campaign(ctx, le.nodeID); err != nil {
		return fmt.Errorf("failed to campaign: %w", err)
	}

	le.isLeader = true
	le.leaderCh <- true
	log.Printf("Node %s became leader for %s", le.nodeID, le.electionID)
	return nil
}

// Resign gives up leadership
func (le *LeaderElection) Resign(ctx context.Context) error {
	if !le.isLeader {
		return nil // Not the leader
	}

	if err := le.election.Resign(ctx); err != nil {
		return fmt.Errorf("failed to resign: %w", err)
	}

	le.isLeader = false
	le.leaderCh <- false
	log.Printf("Node %s resigned leadership for %s", le.nodeID, le.electionID)
	return nil
}

// IsLeader returns whether this node is the leader
func (le *LeaderElection) IsLeader() bool {
	return le.isLeader
}

// LeaderChanges returns a channel that receives leadership change notifications
func (le *LeaderElection) LeaderChanges() <-chan bool {
	return le.leaderCh
}

// WatchLeader watches for leader changes
func (le *LeaderElection) WatchLeader(ctx context.Context) {
	go func() {
		defer close(le.leaderCh)

		for {
			select {
			case <-le.stopCh:
				return
			case <-ctx.Done():
				return
			default:
				// Get the current leader
				resp, err := le.election.Leader(ctx)
				if err != nil {
					log.Printf("Error getting leader: %v", err)
					time.Sleep(1 * time.Second)
					continue
				}

				currentLeader := string(resp.Kvs[0].Value)
				isLeader := currentLeader == le.nodeID

				// If leadership status changed, notify
				if isLeader != le.isLeader {
					le.isLeader = isLeader
					le.leaderCh <- isLeader
					if isLeader {
						log.Printf("Node %s became leader for %s", le.nodeID, le.electionID)
					} else {
						log.Printf("Node %s lost leadership for %s", le.nodeID, le.electionID)
					}
				}

				// Watch for changes
				watchCh := le.client.Watch(ctx, string(resp.Kvs[0].Key))
				select {
				case <-watchCh:
					// Leader changed, loop and check again
				case <-le.stopCh:
					return
				case <-ctx.Done():
					return
				}
			}
		}
	}()
}

// Close releases resources
func (le *LeaderElection) Close() {
	close(le.stopCh)
	if le.isLeader {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = le.Resign(ctx)
	}
	le.session.Close()
	le.client.Close()
}

// GetCurrentLeader returns the current leader's ID
func (le *LeaderElection) GetCurrentLeader(ctx context.Context) (string, error) {
	resp, err := le.election.Leader(ctx)
	if err != nil {
		return "", fmt.Errorf("failed to get leader: %w", err)
	}

	if len(resp.Kvs) == 0 {
		return "", nil // No leader
	}

	return string(resp.Kvs[0].Value), nil
}

func main() {
	// Connect to etcd
	endpoints := []string{"localhost:2379"}
	electionID := "/elections/my-service-leader"

	// Create the leader election
	election, err := NewLeaderElection(endpoints, electionID)
	if err != nil {
		log.Fatalf("Failed to create leader election: %v", err)
	}
	defer election.Close()

	// Create a context with timeout
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	// Start watching for leader changes
	election.WatchLeader(ctx)

	// Start campaigning for leadership
	log.Printf("Node %s starting campaign...", election.nodeID)
	go func() {
		if err := election.Campaign(ctx); err != nil {
			log.Printf("Campaign error: %v", err)
		}
	}()

	// Handle leadership changes
	for isLeader := range election.LeaderChanges() {
		if isLeader {
			log.Printf("I am now the leader! Starting leader tasks...")
			// Perform leader-specific work
			time.Sleep(10 * time.Second)
			
			// Simulate stepping down
			log.Printf("Resigning leadership...")
			if err := election.Resign(ctx); err != nil {
				log.Printf("Error resigning: %v", err)
			}
		} else {
			log.Printf("I am now a follower. Waiting for leadership...")
			// Perform follower-specific work
			
			// After some time, campaign again
			time.Sleep(5 * time.Second)
			log.Printf("Starting new campaign...")
			go func() {
				if err := election.Campaign(ctx); err != nil {
					log.Printf("Campaign error: %v", err)
				}
			}()
		}
	}
}

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"sync"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)

// DistributedSemaphore implements a semaphore that works across multiple nodes
type DistributedSemaphore struct {
	client    *clientv3.Client
	session   *concurrency.Session
	semPath   string
	nodeID    string
	count     int
	resources map[string]struct{}
	mu        sync.Mutex
}

// NewDistributedSemaphore creates a new distributed semaphore
func NewDistributedSemaphore(endpoints []string, semPath string, count int) (*DistributedSemaphore, error) {
	// Create etcd client
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create etcd client: %w", err)
	}

	// Generate a unique node ID
	hostname, err := os.Hostname()
	if err != nil {
		hostname = "unknown-host"
	}
	nodeID := fmt.Sprintf("%s-%d", hostname, time.Now().UnixNano())

	// Create a session with keep-alive
	session, err := concurrency.NewSession(client, concurrency.WithTTL(10))
	if err != nil {
		client.Close()
		return nil, fmt.Errorf("failed to create etcd session: %w", err)
	}

	return &DistributedSemaphore{
		client:    client,
		session:   session,
		semPath:   semPath,
		nodeID:    nodeID,
		count:     count,
		resources: make(map[string]struct{}),
	}, nil
}

// Acquire attempts to acquire a resource from the semaphore
func (ds *DistributedSemaphore) Acquire(ctx context.Context) (string, error) {
	ds.mu.Lock()
	defer ds.mu.Unlock()

	// Generate a unique resource ID
	resourceID := fmt.Sprintf("%s/%s-%d", ds.semPath, ds.nodeID, time.Now().UnixNano())

	// Try to create the resource key with a lease
	_, err := ds.client.Put(ctx, resourceID, ds.nodeID, clientv3.WithLease(ds.session.Lease()))
	if err != nil {
		return "", fmt.Errorf("failed to create resource: %w", err)
	}

	// Get all resources to check if we're within the limit
	resp, err := ds.client.Get(ctx, ds.semPath, clientv3.WithPrefix())
	if err != nil {
		// Try to clean up
		_, _ = ds.client.Delete(ctx, resourceID)
		return "", fmt.Errorf("failed to get resources: %w", err)
	}

	// Check if we're within the semaphore limit
	if len(resp.Kvs) > ds.count {
		// We need to release our resource and wait
		_, _ = ds.client.Delete(ctx, resourceID)
		return "", fmt.Errorf("semaphore limit reached")
	}

	// We successfully acquired a resource
	ds.resources[resourceID] = struct{}{}
	log.Printf("Node %s acquired resource %s", ds.nodeID, resourceID)
	return resourceID, nil
}

// Release releases a previously acquired resource
func (ds *DistributedSemaphore) Release(ctx context.Context, resourceID string) error {
	ds.mu.Lock()
	defer ds.mu.Unlock()

	// Check if we own this resource
	if _, exists := ds.resources[resourceID]; !exists {
		return fmt.Errorf("resource not owned by this semaphore instance")
	}

	// Delete the resource key
	_, err := ds.client.Delete(ctx, resourceID)
	if err != nil {
		return fmt.Errorf("failed to delete resource: %w", err)
	}

	// Remove from our local tracking
	delete(ds.resources, resourceID)
	log.Printf("Node %s released resource %s", ds.nodeID, resourceID)
	return nil
}

// TryAcquireWithTimeout attempts to acquire a resource with a timeout
func (ds *DistributedSemaphore) TryAcquireWithTimeout(ctx context.Context, timeout time.Duration) (string, error) {
	timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	// Try to acquire immediately first
	resourceID, err := ds.Acquire(ctx)
	if err == nil {
		return resourceID, nil
	}

	// If that fails, retry with exponential backoff
	backoff := 50 * time.Millisecond
	maxBackoff := 1 * time.Second

	for {
		select {
		case <-timeoutCtx.Done():
			return "", fmt.Errorf("timeout waiting for semaphore: %w", timeoutCtx.Err())
		case <-time.After(backoff):
			// Try again
			resourceID, err := ds.Acquire(ctx)
			if err == nil {
				return resourceID, nil
			}

			// Increase backoff for next attempt
			backoff *= 2
			if backoff > maxBackoff {
				backoff = maxBackoff
			}
		}
	}
}

// Close releases all resources and cleans up
func (ds *DistributedSemaphore) Close() {
	ds.mu.Lock()
	defer ds.mu.Unlock()

	// Release all acquired resources
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	for resourceID := range ds.resources {
		_, _ = ds.client.Delete(ctx, resourceID)
		delete(ds.resources, resourceID)
	}

	ds.session.Close()
	ds.client.Close()
}

// GetAvailable returns the number of available resources
func (ds *DistributedSemaphore) GetAvailable(ctx context.Context) (int, error) {
	resp, err := ds.client.Get(ctx, ds.semPath, clientv3.WithPrefix())
	if err != nil {
		return 0, fmt.Errorf("failed to get resources: %w", err)
	}

	return ds.count - len(resp.Kvs), nil
}

func main() {
	// Connect to etcd
	endpoints := []string{"localhost:2379"}
	semPath := "/semaphores/connection-limit"
	maxConnections := 3

	// Create the distributed semaphore
	sem, err := NewDistributedSemaphore(endpoints, semPath, maxConnections)
	if err != nil {
		log.Fatalf("Failed to create distributed semaphore: %v", err)
	}
	defer sem.Close()

	// Create a context with timeout
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Simulate multiple workers trying to acquire resources
	var wg sync.WaitGroup
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()

			log.Printf("Worker %d trying to acquire resource...", workerID)
			resourceID, err := sem.TryAcquireWithTimeout(ctx, 5*time.Second)
			if err != nil {
				log.Printf("Worker %d failed to acquire resource: %v", workerID, err)
				return
			}

			log.Printf("Worker %d acquired resource %s", workerID, resourceID)

			// Simulate doing work
			time.Sleep(2 * time.Second)

			// Release the resource
			if err := sem.Release(ctx, resourceID); err != nil {
				log.Printf("Worker %d failed to release resource: %v", workerID, err)
			} else {
				log.Printf("Worker %d released resource %s", workerID, resourceID)
			}
		}(i)
	}

	// Wait for all workers to finish
	wg.Wait()
}