Advanced Go Channel Patterns: Mastering Communication in Concurrent Systems

ch := make(chan int)
go func() { ch <- 42 }()
value := <-ch

package main

import (
	"fmt"
	"time"
)

// produceValues demonstrates a function that only sends on a channel
func produceValues(ch chan<- int) {
	for i := 0; i < 5; i++ {
		fmt.Printf("Sending: %d\n", i)
		ch <- i
		time.Sleep(100 * time.Millisecond)
	}
	close(ch) // Producer is responsible for closing
}

// consumeValues demonstrates a function that only receives from a channel
func consumeValues(ch <-chan int) {
	// Using range loop automatically handles channel closure
	for val := range ch {
		fmt.Printf("Received: %d\n", val)
	}
}

func main() {
	// Create a bidirectional channel
	ch := make(chan int)
	
	// Start producer and consumer goroutines
	go produceValues(ch) // Channel converted to send-only
	go consumeValues(ch) // Channel converted to receive-only
	
	// Wait for completion
	time.Sleep(1 * time.Second)
	fmt.Println("Done")
}

package main

import (
	"fmt"
	"time"
)

func bufferingDemo() {
	fmt.Println("Unbuffered channel demonstration:")
	unbuffered := make(chan int)
	
	go func() {
		fmt.Println("Sender: Attempting to send")
		unbuffered <- 42
		fmt.Println("Sender: Send completed")
	}()
	
	// Give sender time to attempt sending
	time.Sleep(100 * time.Millisecond)
	fmt.Println("Receiver: About to receive")
	value := <-unbuffered
	fmt.Printf("Receiver: Received value %d\n", value)
	
	fmt.Println("\nBuffered channel demonstration:")
	buffered := make(chan int, 2)
	
	go func() {
		for i := 0; i < 3; i++ {
			fmt.Printf("Sender: Sending value %d\n", i)
			buffered <- i
			fmt.Printf("Sender: Sent value %d\n", i)
		}
	}()
	
	// Give sender time to send values
	time.Sleep(100 * time.Millisecond)
	
	for i := 0; i < 3; i++ {
		fmt.Println("Receiver: About to receive")
		value := <-buffered
		fmt.Printf("Receiver: Received value %d\n", value)
		time.Sleep(100 * time.Millisecond)
	}
}

func main() {
	bufferingDemo()
	time.Sleep(500 * time.Millisecond) // Ensure all output is printed
}

package main

import (
	"fmt"
	"time"
)

func channelClosureDemo() {
	ch := make(chan int)
	
	// Sender goroutine
	go func() {
		for i := 0; i < 5; i++ {
			ch <- i
		}
		close(ch)
		fmt.Println("Sender: Channel closed")
	}()
	
	// Receiver loop - continues until channel is closed
	for {
		value, ok := <-ch
		if !ok {
			fmt.Println("Receiver: Channel closed detected")
			break
		}
		fmt.Printf("Receiver: Got value %d\n", value)
	}
	
	// Demonstrate behavior of closed and nil channels
	closedCh := make(chan int)
	close(closedCh)
	
	// Reading from closed channel returns zero value
	val, ok := <-closedCh
	fmt.Printf("Reading from closed channel: value=%d, ok=%v\n", val, ok)
	
	// Writing to closed channel panics
	// closedCh <- 1 // This would panic
	
	// Nil channel operations block forever
	var nilCh chan int // nil channel
	
	// Demonstrating nil channel with timeout
	go func() {
		fmt.Println("Attempting to read from nil channel (will block forever)")
		// <-nilCh // This would block forever
	}()
}

func main() {
	channelClosureDemo()
	time.Sleep(500 * time.Millisecond) // Ensure all output is printed
}

package main

import (
	"fmt"
	"sync"
)

// ChannelOwner demonstrates the channel ownership pattern
type ChannelOwner struct {
	values chan int
	done   chan struct{}
}

// NewChannelOwner creates and returns a new ChannelOwner
// The constructor is the only place where the channels are created
func NewChannelOwner() *ChannelOwner {
	return &ChannelOwner{
		values: make(chan int),
		done:   make(chan struct{}),
	}
}

// Start begins producing values and returns a receive-only channel
// The owner starts the producer goroutine and manages its lifecycle
func (co *ChannelOwner) Start() <-chan int {
	go func() {
		defer close(co.values) // Owner ensures channel is closed
		
		for i := 0; i < 5; i++ {
			select {
			case co.values <- i:
				// Value sent successfully
			case <-co.done:
				fmt.Println("Producer received cancellation signal")
				return
			}
		}
	}()
	
	return co.values // Return receive-only channel to consumers
}

// Stop signals the producer to stop and cleans up resources
func (co *ChannelOwner) Stop() {
	close(co.done)
}

func main() {
	// Create owner and start production
	owner := NewChannelOwner()
	valuesCh := owner.Start()
	
	// Consume values
	var wg sync.WaitGroup
	wg.Add(1)
	
	go func() {
		defer wg.Done()
		for value := range valuesCh {
			fmt.Printf("Received: %d\n", value)
		}
		fmt.Println("Consumer finished")
	}()
	
	// Let it run for a bit, then stop
	// In a real application, this might be triggered by a timeout or user action
	for i := 0; i < 3; i++ {
		<-valuesCh
	}
	
	owner.Stop()
	wg.Wait()
}

package main

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

// timeoutOperation demonstrates using select for timeouts
func timeoutOperation() {
	ch := make(chan string)
	
	go func() {
		// Simulate work that takes random time
		delay := time.Duration(rand.Intn(500)) * time.Millisecond
		time.Sleep(delay)
		ch <- fmt.Sprintf("Operation completed in %v", delay)
	}()
	
	// Wait for result or timeout
	select {
	case result := <-ch:
		fmt.Println("Success:", result)
	case <-time.After(300 * time.Millisecond):
		fmt.Println("Operation timed out")
	}
}

// nonBlockingReceive demonstrates non-blocking channel operations
func nonBlockingReceive(ch chan string) {
	select {
	case msg := <-ch:
		fmt.Println("Received message:", msg)
	default:
		fmt.Println("No message available")
	}
}

// nonBlockingSend demonstrates non-blocking send operation
func nonBlockingSend(ch chan string, msg string) {
	select {
	case ch <- msg:
		fmt.Println("Sent message:", msg)
	default:
		fmt.Println("Cannot send message, channel full or no receivers")
	}
}

// prioritySelect demonstrates channel priority using select
func prioritySelect(high, medium, low chan string) {
	for {
		select {
		case msg := <-high:
			fmt.Println("High priority:", msg)
			return
		default:
			// Continue to nested select
		}
		
		select {
		case msg := <-high:
			fmt.Println("High priority:", msg)
			return
		case msg := <-medium:
			fmt.Println("Medium priority:", msg)
			return
		default:
			// Continue to final select
		}
		
		select {
		case msg := <-high:
			fmt.Println("High priority:", msg)
		case msg := <-medium:
			fmt.Println("Medium priority:", msg)
		case msg := <-low:
			fmt.Println("Low priority:", msg)
		case <-time.After(500 * time.Millisecond):
			fmt.Println("Timed out waiting for messages")
		}
		return
	}
}

// dynamicSelectCases demonstrates how to handle a dynamic number of channels
func dynamicSelectCases(channels []chan int) {
	cases := make([]reflect.SelectCase, len(channels))
	for i, ch := range channels {
		cases[i] = reflect.SelectCase{
			Dir:  reflect.SelectRecv,
			Chan: reflect.ValueOf(ch),
		}
	}
	
	// Add timeout case
	cases = append(cases, reflect.SelectCase{
		Dir:  reflect.SelectRecv,
		Chan: reflect.ValueOf(time.After(1 * time.Second)),
	})
	
	// Wait for any channel to receive
	chosen, value, ok := reflect.Select(cases)
	if chosen == len(cases)-1 {
		fmt.Println("Timed out")
		return
	}
	
	if ok {
		fmt.Printf("Received value %v from channel %d\n", value.Int(), chosen)
	} else {
		fmt.Printf("Channel %d was closed\n", chosen)
	}
}

func main() {
	// Demonstrate timeout pattern
	fmt.Println("=== Timeout Pattern ===")
	timeoutOperation()
	
	// Demonstrate non-blocking operations
	fmt.Println("\n=== Non-blocking Operations ===")
	ch := make(chan string, 1)
	nonBlockingReceive(ch)
	nonBlockingSend(ch, "Hello")
	nonBlockingReceive(ch)
	nonBlockingSend(ch, "World") // Channel is now full
	
	// Demonstrate priority selection
	fmt.Println("\n=== Priority Selection ===")
	high := make(chan string, 1)
	medium := make(chan string, 1)
	low := make(chan string, 1)
	
	// Try different scenarios
	medium <- "Medium message"
	prioritySelect(high, medium, low)
	
	// For dynamic select case demo, we need the reflect package
	fmt.Println("\n=== Dynamic Select Cases ===")
	channels := make([]chan int, 3)
	for i := range channels {
		channels[i] = make(chan int)
		i := i // Create new variable to avoid closure problem
		go func() {
			time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
			channels[i] <- i
		}()
	}
	
	// Import reflect package at the top if running this code
	// dynamicSelectCases(channels)
	fmt.Println("(Dynamic select cases require the reflect package)")
}

package main

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

// worker processes items from input and sends results to output
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()
	
	for job := range jobs {
		fmt.Printf("Worker %d processing job %d\n", id, job)
		time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond) // Simulate work
		results <- job * 2                                           // Send result
	}
}

// fanOut distributes work across multiple workers
func fanOut(jobs <-chan int, numWorkers int) <-chan int {
	results := make(chan int)
	var wg sync.WaitGroup
	
	// Start workers
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go worker(i, jobs, results, &wg)
	}
	
	// Close results channel when all workers are done
	go func() {
		wg.Wait()
		close(results)
		fmt.Println("All workers completed")
	}()
	
	return results
}

// fanIn merges multiple channels into a single channel
func fanIn(channels ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	multiplexed := make(chan int)
	
	// Function to forward values from input channel to multiplexed channel
	forward := func(ch <-chan int) {
		defer wg.Done()
		for val := range ch {
			multiplexed <- val
		}
	}
	
	// Start a goroutine for each input channel
	wg.Add(len(channels))
	for _, ch := range channels {
		go forward(ch)
	}
	
	// Close multiplexed channel when all input channels are done
	go func() {
		wg.Wait()
		close(multiplexed)
	}()
	
	return multiplexed
}

// fanInReflect uses reflection to merge channels dynamically
func fanInReflect(channels ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	wg.Add(len(channels))
	
	for _, c := range channels {
		go func(c <-chan int) {
			defer wg.Done()
			for n := range c {
				out <- n
			}
		}(c)
	}
	
	go func() {
		wg.Wait()
		close(out)
	}()
	
	return out
}

func main() {
	// Create jobs channel and send jobs
	jobs := make(chan int, 10)
	for i := 1; i <= 10; i++ {
		jobs <- i
	}
	close(jobs)
	
	// Fan out to 3 workers
	fmt.Println("Starting fan-out with 3 workers")
	results := fanOut(jobs, 3)
	
	// Collect and print results
	for result := range results {
		fmt.Printf("Got result: %d\n", result)
	}
	
	// Demonstrate fan-in with multiple channels
	fmt.Println("\nDemonstrating fan-in pattern")
	ch1 := make(chan int)
	ch2 := make(chan int)
	ch3 := make(chan int)
	
	// Send values on each channel
	go func() {
		for i := 1; i <= 3; i++ {
			ch1 <- i * 10
			time.Sleep(100 * time.Millisecond)
		}
		close(ch1)
	}()
	
	go func() {
		for i := 1; i <= 3; i++ {
			ch2 <- i * 100
			time.Sleep(150 * time.Millisecond)
		}
		close(ch2)
	}()
	
	go func() {
		for i := 1; i <= 3; i++ {
			ch3 <- i * 1000
			time.Sleep(80 * time.Millisecond)
		}
		close(ch3)
	}()
	
	// Fan-in the channels
	merged := fanIn(ch1, ch2, ch3)
	
	// Print merged results
	for result := range merged {
		fmt.Printf("Merged result: %d\n", result)
	}
}

package main

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

// generator creates a channel that emits the provided integers
func generator(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}

// square receives integers, squares them, and sends them to a returned channel
func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			out <- n * n
		}
	}()
	return out
}

// filter receives integers and sends only those that satisfy the predicate function
func filter(in <-chan int, predicate func(int) bool) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			if predicate(n) {
				out <- n
			}
		}
	}()
	return out
}

// merge combines multiple channels into a single channel
func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)
	
	// Start an output goroutine for each input channel
	output := func(c <-chan int) {
		defer wg.Done()
		for n := range c {
			out <- n
		}
	}
	
	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}
	
	// Close the output channel when all input channels are done
	go func() {
		wg.Wait()
		close(out)
	}()
	
	return out
}

// batchProcessor demonstrates processing data in batches
func batchProcessor(in <-chan int, batchSize int) <-chan []int {
	out := make(chan []int)
	
	go func() {
		defer close(out)
		
		batch := make([]int, 0, batchSize)
		for n := range in {
			batch = append(batch, n)
			
			// When batch is full, send it and create a new one
			if len(batch) == batchSize {
				out <- batch
				batch = make([]int, 0, batchSize)
			}
		}
		
		// Send any remaining items in the last batch
		if len(batch) > 0 {
			out <- batch
		}
	}()
	
	return out
}

func main() {
	// Create a pipeline: generate -> square -> filter -> print
	fmt.Println("=== Basic Pipeline ===")
	c := generator(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
	c = square(c)
	c = filter(c, func(n int) bool {
		return n%2 == 0 // Only even numbers
	})
	
	// Consume the output
	for n := range c {
		fmt.Println(n)
	}
	
	// Demonstrate a more complex pipeline with multiple sources and fan-in
	fmt.Println("\n=== Multi-source Pipeline with Fan-in ===")
	
	// Create multiple sources
	c1 := generator(1, 2, 3)
	c2 := generator(4, 5, 6)
	c3 := generator(7, 8, 9)
	
	// Process each source
	c1 = square(c1)
	c2 = square(c2)
	c3 = square(c3)
	
	// Merge the results
	merged := merge(c1, c2, c3)
	
	// Consume the merged output
	for n := range merged {
		fmt.Println(n)
	}
	
	// Demonstrate batch processing
	fmt.Println("\n=== Batch Processing Pipeline ===")
	
	// Generate 10 random numbers
	source := make(chan int)
	go func() {
		defer close(source)
		for i := 0; i < 10; i++ {
			source <- rand.Intn(100)
		}
	}()
	
	// Process in batches of 3
	batches := batchProcessor(source, 3)
	
	// Consume and process batches
	for batch := range batches {
		fmt.Printf("Processing batch: %v\n", batch)
		sum := 0
		for _, n := range batch {
			sum += n
		}
		fmt.Printf("Batch sum: %d\n", sum)
	}
}

package main

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

// Task represents a unit of work
type Task struct {
	ID     int
	Data   interface{}
	Result interface{}
	Err    error
}

// WorkerPool manages a pool of workers
type WorkerPool struct {
	tasks       chan Task
	results     chan Task
	concurrency int
	wg          sync.WaitGroup
	ctx         context.Context
	cancel      context.CancelFunc
}

// NewWorkerPool creates a new worker pool with the specified concurrency
func NewWorkerPool(concurrency int) *WorkerPool {
	ctx, cancel := context.WithCancel(context.Background())
	return &WorkerPool{
		tasks:       make(chan Task),
		results:     make(chan Task),
		concurrency: concurrency,
		ctx:         ctx,
		cancel:      cancel,
	}
}

// Start launches the worker pool
func (p *WorkerPool) Start() {
	// Start workers
	for i := 0; i < p.concurrency; i++ {
		p.wg.Add(1)
		go p.worker(i)
	}
	
	// Start result collector
	go func() {
		p.wg.Wait()
		close(p.results)
	}()
}

// worker processes tasks from the task channel
func (p *WorkerPool) worker(id int) {
	defer p.wg.Done()
	
	for {
		select {
		case task, ok := <-p.tasks:
			if !ok {
				return // Task channel closed
			}
			
			// Process the task
			fmt.Printf("Worker %d processing task %d\n", id, task.ID)
			time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond) // Simulate work
			
			// Generate result (in a real application, this would be actual processing)
			task.Result = task.Data.(int) * 2
			
			// Send result
			select {
			case p.results <- task:
				// Result sent successfully
			case <-p.ctx.Done():
				return // Context cancelled
			}
			
		case <-p.ctx.Done():
			return // Context cancelled
		}
	}
}

// Submit adds a task to the worker pool
func (p *WorkerPool) Submit(task Task) {
	select {
	case p.tasks <- task:
		// Task submitted successfully
	case <-p.ctx.Done():
		// Pool is shutting down
	}
}

// Results returns the channel of completed tasks
func (p *WorkerPool) Results() <-chan Task {
	return p.results
}

// Stop gracefully shuts down the worker pool
func (p *WorkerPool) Stop() {
	p.cancel()  // Signal all workers to stop
	close(p.tasks) // Close task channel
}

// ThrottledWorkerPool extends WorkerPool with rate limiting
type ThrottledWorkerPool struct {
	*WorkerPool
	rate  time.Duration // Minimum time between task submissions
	limit chan struct{} // Semaphore for limiting concurrent tasks
}

// NewThrottledWorkerPool creates a new rate-limited worker pool
func NewThrottledWorkerPool(concurrency int, rate time.Duration, maxQueued int) *ThrottledWorkerPool {
	return &ThrottledWorkerPool{
		WorkerPool: NewWorkerPool(concurrency),
		rate:       rate,
		limit:      make(chan struct{}, maxQueued),
	}
}

// Submit adds a task to the throttled worker pool, respecting rate limits
func (p *ThrottledWorkerPool) Submit(task Task) {
	// Add to limit before submitting
	p.limit <- struct{}{}
	
	go func() {
		defer func() { <-p.limit }() // Release limit when done
		
		// Submit to underlying pool
		p.WorkerPool.Submit(task)
		
		// Enforce rate limit
		time.Sleep(p.rate)
	}()
}

func main() {
	// Create a worker pool with 3 workers
	fmt.Println("=== Basic Worker Pool ===")
	pool := NewWorkerPool(3)
	pool.Start()
	
	// Submit 10 tasks
	for i := 0; i < 10; i++ {
		pool.Submit(Task{
			ID:   i,
			Data: i,
		})
	}
	
	// Close the task channel to signal no more tasks
	pool.Stop()
	
	// Collect results
	for task := range pool.Results() {
		fmt.Printf("Task %d result: %v\n", task.ID, task.Result)
	}
	
	// Demonstrate throttled worker pool
	fmt.Println("\n=== Throttled Worker Pool ===")
	throttled := NewThrottledWorkerPool(3, 200*time.Millisecond, 5)
	throttled.Start()
	
	// Submit tasks rapidly,
	// Submit tasks rapidly, but they'll be rate-limited
	for i := 0; i < 10; i++ {
		throttled.Submit(Task{
			ID:   i,
			Data: i,
		})
		fmt.Printf("Submitted task %d\n", i)
	}
	
	// Wait for a bit to see the rate limiting in action
	time.Sleep(1 * time.Second)
	
	// Stop the pool
	throttled.Stop()
	
	// Collect results
	for task := range throttled.Results() {
		fmt.Printf("Throttled task %d result: %v\n", task.ID, task.Result)
	}
}

package main

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

// Semaphore represents a counting semaphore
type Semaphore chan struct{}

// Acquire n resources from the semaphore
func (s Semaphore) Acquire(n int) {
	for i := 0; i < n; i++ {
		s <- struct{}{}
	}
}

// Release n resources back to the semaphore
func (s Semaphore) Release(n int) {
	for i := 0; i < n; i++ {
		<-s
	}
}

// TryAcquire attempts to acquire n resources without blocking
// Returns true if successful, false if would block
func (s Semaphore) TryAcquire(n int) bool {
	select {
	case s <- struct{}{}:
		if n > 1 {
			// For multiple resources, we need to be careful
			// If we can't acquire all, release what we've acquired
			if !s.TryAcquire(n - 1) {
				<-s // Release the one we just acquired
				return false
			}
		}
		return true
	default:
		return false
	}
}

// NewSemaphore creates a new semaphore with the given capacity
func NewSemaphore(capacity int) Semaphore {
	return make(Semaphore, capacity)
}

// simulateResourceUsage demonstrates using a semaphore to limit concurrent access
func simulateResourceUsage(id int, sem Semaphore, wg *sync.WaitGroup) {
	defer wg.Done()
	
	// Try to acquire the resource
	fmt.Printf("Client %d: Attempting to acquire resource\n", id)
	
	// Try non-blocking acquire first
	if sem.TryAcquire(1) {
		fmt.Printf("Client %d: Immediately acquired resource\n", id)
	} else {
		fmt.Printf("Client %d: Waiting for resource...\n", id)
		sem.Acquire(1)
		fmt.Printf("Client %d: Eventually acquired resource\n", id)
	}
	
	// Use the resource
	fmt.Printf("Client %d: Using resource\n", id)
	time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
	
	// Release the resource
	fmt.Printf("Client %d: Releasing resource\n", id)
	sem.Release(1)
}

func main() {
	// Create a semaphore with capacity 3
	sem := NewSemaphore(3)
	
	// Create 10 clients that will try to use the resource
	var wg sync.WaitGroup
	for i := 1; i <= 10; i++ {
		wg.Add(1)
		go simulateResourceUsage(i, sem, &wg)
		time.Sleep(50 * time.Millisecond) // Stagger client starts
	}
	
	// Wait for all clients to finish
	wg.Wait()
	fmt.Println("All clients finished")
}

package main

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

// timeoutOperation demonstrates a simple timeout pattern
func timeoutOperation(timeout time.Duration) (result string, err error) {
	// Create a channel for the operation result
	resultCh := make(chan string, 1)
	
	// Start the operation in a goroutine
	go func() {
		// Simulate work
		time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
		resultCh <- "Operation completed successfully"
	}()
	
	// Wait for result or timeout
	select {
	case result = <-resultCh:
		return result, nil
	case <-time.After(timeout):
		return "", fmt.Errorf("operation timed out after %v", timeout)
	}
}

// contextAwareOperation demonstrates using context for cancellation
func contextAwareOperation(ctx context.Context) (string, error) {
	// Create a channel for the operation result
	resultCh := make(chan string, 1)
	
	// Start the operation in a goroutine
	go func() {
		// Simulate work with periodic cancellation checks
		for i := 0; i < 5; i++ {
			select {
			case <-ctx.Done():
				// Context was cancelled, abort operation
				return
			case <-time.After(200 * time.Millisecond):
				// Continue working
				fmt.Println("Operation in progress...")
			}
		}
		
		resultCh <- "Operation completed successfully"
	}()
	
	// Wait for result or cancellation
	select {
	case result := <-resultCh:
		return result, nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

// multiStageTimeout demonstrates different timeouts for different stages
func multiStageTimeout() error {
	// Stage 1: Connect (short timeout)
	connectCtx, cancelConnect := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancelConnect()
	
	fmt.Println("Stage 1: Connecting...")
	if err := stageOperation(connectCtx, "connect", 300); err != nil {
		return fmt.Errorf("connect failed: %w", err)
	}
	
	// Stage 2: Process (longer timeout)
	processCtx, cancelProcess := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancelProcess()
	
	fmt.Println("Stage 2: Processing...")
	if err := stageOperation(processCtx, "process", 1000); err != nil {
		return fmt.Errorf("process failed: %w", err)
	}
	
	// Stage 3: Finalize (medium timeout)
	finalizeCtx, cancelFinalize := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancelFinalize()
	
	fmt.Println("Stage 3: Finalizing...")
	if err := stageOperation(finalizeCtx, "finalize", 500); err != nil {
		return fmt.Errorf("finalize failed: %w", err)
	}
	
	fmt.Println("All stages completed successfully")
	return nil
}

// stageOperation simulates a stage in a multi-stage operation
func stageOperation(ctx context.Context, name string, maxDuration int) error {
	// Simulate work with random duration
	duration := time.Duration(rand.Intn(maxDuration)) * time.Millisecond
	
	select {
	case <-time.After(duration):
		fmt.Printf("Stage '%s' completed in %v\n", name, duration)
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// propagatingCancellation demonstrates how cancellation propagates through a pipeline
func propagatingCancellation() {
	// Create a cancellable context
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel() // Ensure cancellation in all cases
	
	// Create a pipeline with this context
	nums := generateNumbers(ctx, 1, 100)
	squares := squareNumbers(ctx, nums)
	results := filterNumbers(ctx, squares, func(n int) bool {
		return n%10 == 0 // Only numbers divisible by 10
	})
	
	// Process results for a while, then cancel
	go func() {
		time.Sleep(500 * time.Millisecond)
		fmt.Println("Cancelling pipeline...")
		cancel()
	}()
	
	// Consume results until cancellation
	for result := range results {
		fmt.Printf("Result: %d\n", result)
		time.Sleep(100 * time.Millisecond) // Slow consumer
	}
	
	fmt.Println("Pipeline terminated")
}

// generateNumbers produces a sequence of integers, respecting context cancellation
func generateNumbers(ctx context.Context, start, end int) <-chan int {
	out := make(chan int)
	
	go func() {
		defer close(out)
		for i := start; i <= end; i++ {
			select {
			case <-ctx.Done():
				fmt.Println("Generator cancelled")
				return
			case out <- i:
				// Successfully sent value
			}
		}
	}()
	
	return out
}

// squareNumbers transforms input numbers by squaring them
func squareNumbers(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	
	go func() {
		defer close(out)
		for n := range in {
			select {
			case <-ctx.Done():
				fmt.Println("Square operation cancelled")
				return
			case out <- n * n:
				// Successfully sent value
			}
		}
	}()
	
	return out
}

// filterNumbers filters numbers based on a predicate function
func filterNumbers(ctx context.Context, in <-chan int, predicate func(int) bool) <-chan int {
	out := make(chan int)
	
	go func() {
		defer close(out)
		for n := range in {
			if predicate(n) {
				select {
				case <-ctx.Done():
					fmt.Println("Filter operation cancelled")
					return
				case out <- n:
					// Successfully sent value
				}
			}
		}
	}()
	
	return out
}

func main() {
	// Demonstrate simple timeout
	fmt.Println("=== Simple Timeout Pattern ===")
	result, err := timeoutOperation(500 * time.Millisecond)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
	} else {
		fmt.Printf("Result: %s\n", result)
	}
	
	// Demonstrate context cancellation
	fmt.Println("\n=== Context Cancellation Pattern ===")
	ctx, cancel := context.WithTimeout(context.Background(), 700*time.Millisecond)
	defer cancel()
	
	result, err = contextAwareOperation(ctx)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
	} else {
		fmt.Printf("Result: %s\n", result)
	}
	
	// Demonstrate multi-stage timeout
	fmt.Println("\n=== Multi-stage Timeout Pattern ===")
	if err := multiStageTimeout(); err != nil {
		fmt.Printf("Error: %v\n", err)
	}
	
	// Demonstrate cancellation propagation
	fmt.Println("\n=== Cancellation Propagation Pattern ===")
	propagatingCancellation()
}

package main

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

// Barrier synchronizes multiple goroutines to a common point
type Barrier struct {
	count    int
	mutex    sync.Mutex
	cond     *sync.Cond
	required int
}

// NewBarrier creates a new barrier that waits for the required number of goroutines
func NewBarrier(required int) *Barrier {
	b := &Barrier{required: required}
	b.cond = sync.NewCond(&b.mutex)
	return b
}

// Wait blocks until all goroutines have reached the barrier
func (b *Barrier) Wait() {
	b.mutex.Lock()
	defer b.mutex.Unlock()
	
	b.count++
	if b.count == b.required {
		// Last goroutine to arrive, reset and broadcast
		b.count = 0
		b.cond.Broadcast()
	} else {
		// Wait for all goroutines to arrive
		b.cond.Wait()
	}
}

// WorkGroup coordinates multiple goroutines with phases
type WorkGroup struct {
	size      int
	wg        sync.WaitGroup
	startChan chan struct{}
	doneChan  chan struct{}
	barriers  []*Barrier
}

// NewWorkGroup creates a new work group for coordinating goroutines
func NewWorkGroup(size int, phases int) *WorkGroup {
	wg := &WorkGroup{
		size:      size,
		startChan: make(chan struct{}),
		doneChan:  make(chan struct{}),
		barriers:  make([]*Barrier, phases),
	}
	
	// Create barriers for each phase
	for i := 0; i < phases; i++ {
		wg.barriers[i] = NewBarrier(size)
	}
	
	return wg
}

// Start launches the work group
func (wg *WorkGroup) Start(work func(id, phase int)) {
	// Start all workers
	wg.wg.Add(wg.size)
	for i := 0; i < wg.size; i++ {
		go func(id int) {
			defer wg.wg.Done()
			
			// Wait for start signal
			<-wg.startChan
			
			// Execute work through all phases
			for phase := 0; phase < len(wg.barriers); phase++ {
				work(id, phase)
				wg.barriers[phase].Wait() // Synchronize at barrier
			}
		}(i)
	}
	
	// Signal all goroutines to start
	close(wg.startChan)
	
	// Wait for all to complete in a separate goroutine
	go func() {
		wg.wg.Wait()
		close(wg.doneChan)
	}()
}

// Wait blocks until all work is complete
func (wg *WorkGroup) Wait() {
	<-wg.doneChan
}

// Rendezvous coordinates pairs of goroutines
type Rendezvous struct {
	firstArrived  chan int
	secondArrived chan struct{}
}

// NewRendezvous creates a new rendezvous point
func NewRendezvous() *Rendezvous {
	return &Rendezvous{
		firstArrived:  make(chan int),
		secondArrived: make(chan struct{}),
	}
}

// Arrive waits for a pair of goroutines to meet
// Returns the ID of the first to arrive if this is the second arrival
func (r *Rendezvous) Arrive(id int) (int, bool) {
	select {
	case r.firstArrived <- id:
		// We're first to arrive, wait for second
		<-r.secondArrived
		return 0, false
	case firstID := <-r.firstArrived:
		// We're second to arrive, signal first
		close(r.secondArrived)
		return firstID, true
	}
}

func main() {
	// Demonstrate barrier synchronization
	fmt.Println("=== Barrier Synchronization ===")
	barrier := NewBarrier(3)
	
	for i := 0; i < 3; i++ {
		go func(id int) {
			fmt.Printf("Goroutine %d starting\n", id)
			time.Sleep(time.Duration(id*300) * time.Millisecond) // Different work times
			fmt.Printf("Goroutine %d arriving at barrier\n", id)
			barrier.Wait()
			fmt.Printf("Goroutine %d continuing after barrier\n", id)
		}(i)
	}
	
	// Demonstrate phased work group
	fmt.Println("\n=== Phased Work Group ===")
	workGroup := NewWorkGroup(4, 3)
	
	workGroup.Start(func(id, phase int) {
		fmt.Printf("Worker %d executing phase %d\n", id, phase)
		time.Sleep(time.Duration(100*(id+phase)) * time.Millisecond)
		fmt.Printf("Worker %d completed phase %d\n", id, phase)
	})
	
	workGroup.Wait()
	fmt.Println("All workers completed all phases")
	
	// Demonstrate rendezvous pattern
	fmt.Println("\n=== Rendezvous Pattern ===")
	rendezvous := NewRendezvous()
	
	go func() {
		fmt.Println("Goroutine A starting")
		time.Sleep(300 * time.Millisecond)
		fmt.Println("Goroutine A arriving at rendezvous")
		otherID, isSecond := rendezvous.Arrive(1)
		if isSecond {
			fmt.Printf("Goroutine A met with Goroutine %d\n", otherID)
		} else {
			fmt.Println("Goroutine A continuing after rendezvous")
		}
	}()
	
	go func() {
		fmt.Println("Goroutine B starting")
		time.Sleep(100 * time.Millisecond)
		fmt.Println("Goroutine B arriving at rendezvous")
		otherID, isSecond := rendezvous.Arrive(2)
		if isSecond {
			fmt.Printf("Goroutine B met with Goroutine %d\n", otherID)
		} else {
			fmt.Println("Goroutine B continuing after rendezvous")
		}
	}()
	
	// Wait for demonstration to complete
	time.Sleep(1 * time.Second)
}

package main

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

// multiplexChannels combines multiple input channels into a single output channel
func multiplexChannels(inputs ...<-chan int) <-chan int {
	output := make(chan int)
	var wg sync.WaitGroup
	
	// Start a goroutine for each input channel
	for i, ch := range inputs {
		wg.Add(1)
		go func(id int, ch <-chan int) {
			defer wg.Done()
			for val := range ch {
				fmt.Printf("Multiplexer receiving from input %d: %d\n", id, val)
				output <- val
			}
		}(i, ch)
	}
	
	// Close the output channel when all input channels are done
	go func() {
		wg.Wait()
		close(output)
		fmt.Println("All input channels closed, multiplexer shutting down")
	}()
	
	return output
}

// demultiplexChannel splits a single input channel into multiple output channels
func demultiplexChannel(input <-chan int, n int) []<-chan int {
	outputs := make([]chan int, n)
	for i := range outputs {
		outputs[i] = make(chan int)
	}
	
	// Convert to read-only channels for return
	readOnlyOutputs := make([]<-chan int, n)
	for i, ch := range outputs {
		readOnlyOutputs[i] = ch
	}
	
	// Start the demultiplexer
	go func() {
		defer func() {
			for _, ch := range outputs {
				close(ch)
			}
			fmt.Println("Demultiplexer shutting down, all output channels closed")
		}()
		
		// Round-robin distribution
		for val := range input {
			// Determine which output channel to use based on the value
			outChan := val % n
			fmt.Printf("Demultiplexer sending %d to output %d\n", val, outChan)
			outputs[outChan] <- val
		}
	}()
	
	return readOnlyOutputs
}

// contentBasedDemux routes messages based on their content
func contentBasedDemux(input <-chan int, predicates ...func(int) bool) []<-chan int {
	outputs := make([]chan int, len(predicates))
	for i := range outputs {
		outputs[i] = make(chan int)
	}
	
	// Convert to read-only channels for return
	readOnlyOutputs := make([]<-chan int, len(predicates))
	for i, ch := range outputs {
		readOnlyOutputs[i] = ch
	}
	
	// Start the content-based router
	go func() {
		defer func() {
			for _, ch := range outputs {
				close(ch)
			}
			fmt.Println("Content router shutting down, all output channels closed")
		}()
		
		for val := range input {
			// Send to all matching outputs
			for i, predicate := range predicates {
				if predicate(val) {
					fmt.Printf("Router sending %d to output %d\n", val, i)
					outputs[i] <- val
				}
			}
		}
	}()
	
	return readOnlyOutputs
}

// broadcastChannel sends each input value to all output channels
func broadcastChannel(input <-chan int, n int) []<-chan int {
	outputs := make([]chan int, n)
	for i := range outputs {
		outputs[i] = make(chan int, 5) // Buffer to prevent blocking
	}
	
	// Convert to read-only channels for return
	readOnlyOutputs := make([]<-chan int, n)
	for i, ch := range outputs {
		readOnlyOutputs[i] = ch
	}
	
	// Start the broadcaster
	go func() {
		defer func() {
			for _, ch := range outputs {
				close(ch)
			}
			fmt.Println("Broadcaster shutting down, all output channels closed")
		}()
		
		for val := range input {
			fmt.Printf("Broadcasting value: %d\n", val)
			for i, ch := range outputs {
				select {
				case ch <- val:
					// Value sent successfully
				default:
					// Channel buffer full, log and continue
					fmt.Printf("Warning: Output %d buffer full, dropping value %d\n", i, val)
				}
			}
		}
	}()
	
	return readOnlyOutputs
}

func main() {
	// Create input channels for multiplexing demo
	fmt.Println("=== Channel Multiplexing ===")
	inputs := make([]chan int, 3)
	for i := range inputs {
		inputs[i] = make(chan int)
		i := i // Create new variable to avoid closure problem
		
		// Start producer for each input channel
		go func() {
			defer close(inputs[i])
			for j := 0; j < 3; j++ {
				val := i*10 + j
				fmt.Printf("Producer %d sending: %d\n", i, val)
				inputs[i] <- val
				time.Sleep(time.Duration(rand.Intn(300)) * time.Millisecond)
			}
		}()
	}
	
	// Convert to read-only channels for multiplexing
	readOnlyInputs := make([]<-chan int, len(inputs))
	for i, ch := range inputs {
		readOnlyInputs[i] = ch
	}
	
	// Multiplex the channels
	multiplexed := multiplexChannels(readOnlyInputs...)
	
	// Consume multiplexed output
	go func() {
		for val := range multiplexed {
			fmt.Printf("Consumer received from multiplexed channel: %d\n", val)
		}
	}()
	
	// Wait for multiplexing demo to complete
	time.Sleep(2 * time.Second)
	
	// Demonstrate demultiplexing
	fmt.Println("\n=== Channel Demultiplexing ===")
	input := make(chan int)
	
	// Start producer for input channel
	go func() {
		defer close(input)
		for i := 0; i < 10; i++ {
			fmt.Printf("Demux producer sending: %d\n", i)
			input <- i
			time.Sleep(100 * time.Millisecond)
		}
	}()
	
	// Demultiplex into 3 output channels
	outputs := demultiplexChannel(input, 3)
	
	// Start consumers for each output channel
	var wg sync.WaitGroup
	for i, ch := range outputs {
		wg.Add(1)
		i := i // Create new variable to avoid closure problem
		ch := ch
		
		go func() {
			defer wg.Done()
			for val := range ch {
				fmt.Printf("Demux consumer %d received: %d\n", i, val)
			}
		}()
	}
	
	// Wait for demultiplexing demo to complete
	time.Sleep(2 * time.Second)
	
	// Demonstrate content-based routing
	fmt.Println("\n=== Content-Based Routing ===")
	routerInput := make(chan int)
	
	// Define predicates for routing
	isEven := func(n int) bool { return n%2 == 0 }
	isDivisibleBy3 := func(n int) bool { return n%3 == 0 }
	isGreaterThan5 := func(n int) bool { return n > 5 }
	
	// Create router
	routedOutputs := contentBasedDemux(routerInput, isEven, isDivisibleBy3, isGreaterThan5)
	
	// Start consumers for each routed output
	for i, ch := range routedOutputs {
		wg.Add(1)
		i := i // Create new variable to avoid closure problem
		ch := ch
		
		go func() {
			defer wg.Done()
			for val := range ch {
				var condition string
				switch i {
				case 0:
					condition = "even"
				case 1:
					condition = "divisible by 3"
				case 2:
					condition = "greater than 5"
				}
				fmt.Printf("Router consumer %d received %d (%s)\n", i, val, condition)
			}
		}()
	}
	
	// Send values to the router
	go func() {
		defer close(routerInput)
		for i := 0; i < 10; i++ {
			fmt.Printf("Router producer sending: %d\n", i)
			routerInput <- i
			time.Sleep(100 * time.Millisecond)
		}
	}()
	
	// Wait for routing demo to complete
	time.Sleep(2 * time.Second)
}

package main

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

// Result represents a computation result or error
type Result struct {
	Value int
	Err   error
}

// errorPropagationBasic demonstrates the basic error propagation pattern
func errorPropagationBasic() {
	// Create a channel for results
	results := make(chan Result)
	
	// Start a worker that might encounter errors
	go func() {
		defer close(results)
		
		// Simulate work with possible errors
		for i := 0; i < 5; i++ {
			// 30% chance of error
			if rand.Float32() < 0.3 {
				results <- Result{Err: fmt.Errorf("error processing item %d", i)}
				continue
			}
			
			// Successful computation
			results <- Result{Value: i * 10}
		}
	}()
	
	// Process results, handling errors
	for result := range results {
		if result.Err != nil {
			fmt.Printf("Error: %v\n", result.Err)
		} else {
			fmt.Printf("Success: %d\n", result.Value)
		}
	}
}

// errorPropagationPipeline demonstrates error propagation through a pipeline
func errorPropagationPipeline() {
	// Create a pipeline with error propagation
	source := generateWithErrors(1, 10)
	processed := processWithErrors(source)
	
	// Consume results
	for result := range processed {
		if result.Err != nil {
			fmt.Printf("Pipeline error: %v\n", result.Err)
		} else {
			fmt.Printf("Pipeline result: %d\n", result.Value)
		}
	}
}

// generateWithErrors produces integers with possible errors
func generateWithErrors(start, end int) <-chan Result {
	out := make(chan Result)
	
	go func() {
		defer close(out)
		for i := start; i <= end; i++ {
			// 20% chance of error
			if rand.Float32() < 0.2 {
				out <- Result{Err: fmt.Errorf("failed to generate item %d", i)}
				continue
			}
			
			// Simulate work
			time.Sleep(50 * time.Millisecond)
			out <- Result{Value: i}
		}
	}()
	
	return out
}

// processWithErrors transforms values and propagates errors
func processWithErrors(in <-chan Result) <-chan Result {
	out := make(chan Result)
	
	go func() {
		defer close(out)
		for result := range in {
			// If input already has an error, propagate it
			if result.Err != nil {
				out <- result
				continue
			}
			
			// 10% chance of new error during processing
			if rand.Float32() < 0.1 {
				out <- Result{Err: fmt.Errorf("failed to process value %d", result.Value)}
				continue
			}
			
			// Successful processing
			time.Sleep(30 * time.Millisecond)
			out <- Result{Value: result.Value * result.Value}
		}
	}()
	
	return out
}

// errorAggregation demonstrates collecting and aggregating errors
func errorAggregation() {
	// Create multiple workers that might produce errors
	numWorkers := 5
	results := make(chan Result)
	var wg sync.WaitGroup
	
	// Start workers
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			
			// 40% chance of error
			if rand.Float32() < 0.4 {
				results <- Result{Err: fmt.Errorf("worker %d failed", id)}
				return
			}
			
			// Successful computation
			results <- Result{Value: id * 10}
		}(i)
	}
	
	// Close results channel when all workers are done
	go func() {
		wg.Wait()
		close(results)
	}()
	
	// Collect and aggregate results
	var successCount, errorCount int
	var errors []error
	var values []int
	
	for result := range results {
		if result.Err != nil {
			errorCount++
			errors = append(errors, result.Err)
		} else {
			successCount++
			values = append(values, result.Value)
		}
	}
	
	// Report aggregated results
	fmt.Printf("Successful operations: %d\n", successCount)
	fmt.Printf("Failed operations: %d\n", errorCount)
	
	if errorCount > 0 {
		fmt.Println("Errors encountered:")
		for _, err := range errors {
			fmt.Printf("  - %v\n", err)
		}
	}
	
	if successCount > 0 {
		fmt.Printf("Values: %v\n", values)
	}
}

// earlyTerminationOnError demonstrates stopping a pipeline on first error
func earlyTerminationOnError() {
	// Create a context for cancellation
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	
	// Create a pipeline with early termination
	source := generateWithContext(ctx, 1, 20)
	processed := processWithContext(ctx, source)
	
	// Consume results until first error
	for result := range processed {
		if result.Err != nil {
			fmt.Printf("Pipeline error: %v\n", result.Err)
			fmt.Println("Cancelling pipeline...")
			cancel() // Signal all stages to stop
			break
		}
		
		fmt.Printf("Pipeline result: %d\n", result.Value)
	}
	
	// Wait for pipeline to fully terminate
	time.Sleep(100 * time.Millisecond)
	fmt.Println("Pipeline terminated")
}

// generateWithContext produces integers with possible errors and respects context
func generateWithContext(ctx context.Context, start, end int) <-chan Result {
	out := make(chan Result)
	
	go func() {
		defer close(out)
		for i := start; i <= end; i++ {
			// Check for cancellation
			select {
			case <-ctx.Done():
				fmt.Println("Generator cancelled")
				return
			default:
				// Continue processing
			}
			
			// 10% chance of error
			if rand.Float32() < 0.1 {
				select {
				case out <- Result{Err: fmt.Errorf("failed to generate item %d", i)}:
					// Error sent
				case <-ctx.Done():
					return
				}
				continue
			}
			
			// Simulate work
			time.Sleep(50 * time.Millisecond)
			
			// Send result
			select {
			case out <- Result{Value: i}:
				// Value sent
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return out
}

// processWithContext transforms values and respects context
func processWithContext(ctx context.Context, in <-chan Result) <-chan Result {
	out := make(chan Result)
	
	go func() {
		defer close(out)
		for {
			// Check for cancellation between items
			select {
			case <-ctx.Done():
				fmt.Println("Processor cancelled")
				return
			default:
				// Continue processing
			}
			
			// Try to receive next item
			var result Result
			var ok bool
			select {
			case result, ok = <-in:
				if !ok {
					return // Input channel closed
				}
			case <-ctx.Done():
				return // Context cancelled
			}
			
			// If input already has an error, propagate it
			if result.Err != nil {
				select {
				case out <- result:
					// Error propagated
				case <-ctx.Done():
					return
				}
				continue
			}
			
			// 15% chance of new error during processing
			if rand.Float32() < 0.15 {
				select {
				case out <- Result{Err: fmt.Errorf("failed to process value %d", result.Value)}:
					// Error sent
				case <-ctx.Done():
					return
				}
				continue
			}
			
			// Successful processing
			time.Sleep(30 * time.Millisecond)
			select {
			case out <- Result{Value: result.Value * result.Value}:
				// Result sent
			case <-ctx.Done():
				return
			}
		}
	}()
	
	return out
}

func main() {
	// Seed random number generator
	rand.Seed(time.Now().UnixNano())
	
	// Demonstrate basic error propagation
	fmt.Println("=== Basic Error Propagation ===")
	errorPropagationBasic()
	
	// Demonstrate error propagation through a pipeline
	fmt.Println("\n=== Pipeline Error Propagation ===")
	errorPropagationPipeline()
	
	// Demonstrate error aggregation
	fmt.Println("\n=== Error Aggregation ===")
	errorAggregation()
	
	// Demonstrate early termination on error
	fmt.Println("\n=== Early Termination on Error ===")
	earlyTerminationOnError()
}