1# Concurrency Patterns
2 
3## Goroutine Lifecycle Management
4 
5```go
6package main
7 
8import (
9    "context"
10    "fmt"
11    "sync"
12    "time"
13)
14 
15// Worker pool with bounded concurrency
16type WorkerPool struct {
17    workers int
18    tasks   chan func()
19    wg      sync.WaitGroup
20}
21 
22func NewWorkerPool(workers int) *WorkerPool {
23    wp := &WorkerPool{
24        workers: workers,
25        tasks:   make(chan func(), workers*2), // Buffered channel
26    }
27    wp.start()
28    return wp
29}
30 
31func (wp *WorkerPool) start() {
32    for i := 0; i < wp.workers; i++ {
33        wp.wg.Add(1)
34        go func() {
35            defer wp.wg.Done()
36            for task := range wp.tasks {
37                task()
38            }
39        }()
40    }
41}
42 
43func (wp *WorkerPool) Submit(task func()) {
44    wp.tasks <- task
45}
46 
47func (wp *WorkerPool) Shutdown() {
48    close(wp.tasks)
49    wp.wg.Wait()
50}
51```
52 
53## Channel Patterns
54 
55```go
56// Generator pattern
57func generateNumbers(ctx context.Context, max int) <-chan int {
58    out := make(chan int)
59    go func() {
60        defer close(out)
61        for i := 0; i < max; i++ {
62            select {
63            case out <- i:
64            case <-ctx.Done():
65                return
66            }
67        }
68    }()
69    return out
70}
71 
72// Fan-out, fan-in pattern
73func fanOut(ctx context.Context, input <-chan int, workers int) []<-chan int {
74    channels := make([]<-chan int, workers)
75    for i := 0; i < workers; i++ {
76        channels[i] = process(ctx, input)
77    }
78    return channels
79}
80 
81func process(ctx context.Context, input <-chan int) <-chan int {
82    out := make(chan int)
83    go func() {
84        defer close(out)
85        for val := range input {
86            select {
87            case out <- val * 2:
88            case <-ctx.Done():
89                return
90            }
91        }
92    }()
93    return out
94}
95 
96func fanIn(ctx context.Context, channels ...<-chan int) <-chan int {
97    out := make(chan int)
98    var wg sync.WaitGroup
99 
100    for _, ch := range channels {
101        wg.Add(1)
102        go func(c <-chan int) {
103            defer wg.Done()
104            for val := range c {
105                select {
106                case out <- val:
107                case <-ctx.Done():
108                    return
109                }
110            }
111        }(ch)
112    }
113 
114    go func() {
115        wg.Wait()
116        close(out)
117    }()
118 
119    return out
120}
121```
122 
123## Select Statement Patterns
124 
125```go
126// Timeout pattern
127func fetchWithTimeout(ctx context.Context, url string) (string, error) {
128    result := make(chan string, 1)
129    errCh := make(chan error, 1)
130 
131    go func() {
132        // Simulate network call
133        time.Sleep(100 * time.Millisecond)
134        result <- "data from " + url
135    }()
136 
137    select {
138    case res := <-result:
139        return res, nil
140    case err := <-errCh:
141        return "", err
142    case <-time.After(50 * time.Millisecond):
143        return "", fmt.Errorf("timeout")
144    case <-ctx.Done():
145        return "", ctx.Err()
146    }
147}
148 
149// Done channel pattern for graceful shutdown
150type Server struct {
151    done chan struct{}
152}
153 
154func (s *Server) Shutdown() {
155    close(s.done)
156}
157 
158func (s *Server) Run(ctx context.Context) {
159    ticker := time.NewTicker(1 * time.Second)
160    defer ticker.Stop()
161 
162    for {
163        select {
164        case <-ticker.C:
165            fmt.Println("tick")
166        case <-s.done:
167            fmt.Println("shutting down")
168            return
169        case <-ctx.Done():
170            fmt.Println("context cancelled")
171            return
172        }
173    }
174}
175```
176 
177## Sync Primitives
178 
179```go
180import "sync"
181 
182// Mutex for protecting shared state
183type Counter struct {
184    mu    sync.Mutex
185    count int
186}
187 
188func (c *Counter) Increment() {
189    c.mu.Lock()
190    defer c.mu.Unlock()
191    c.count++
192}
193 
194func (c *Counter) Value() int {
195    c.mu.Lock()
196    defer c.mu.Unlock()
197    return c.count
198}
199 
200// RWMutex for read-heavy workloads
201type Cache struct {
202    mu    sync.RWMutex
203    items map[string]string
204}
205 
206func (c *Cache) Get(key string) (string, bool) {
207    c.mu.RLock()
208    defer c.mu.RUnlock()
209    val, ok := c.items[key]
210    return val, ok
211}
212 
213func (c *Cache) Set(key, value string) {
214    c.mu.Lock()
215    defer c.mu.Unlock()
216    c.items[key] = value
217}
218 
219// sync.Once for initialization
220type Service struct {
221    once   sync.Once
222    config *Config
223}
224 
225func (s *Service) getConfig() *Config {
226    s.once.Do(func() {
227        s.config = loadConfig() // Only called once
228    })
229    return s.config
230}
231```
232 
233## Rate Limiting and Backpressure
234 
235```go
236import "golang.org/x/time/rate"
237 
238// Token bucket rate limiter
239type RateLimiter struct {
240    limiter *rate.Limiter
241}
242 
243func NewRateLimiter(rps int) *RateLimiter {
244    return &RateLimiter{
245        limiter: rate.NewLimiter(rate.Limit(rps), rps),
246    }
247}
248 
249func (rl *RateLimiter) Process(ctx context.Context, item string) error {
250    if err := rl.limiter.Wait(ctx); err != nil {
251        return err
252    }
253    // Process item
254    return nil
255}
256 
257// Semaphore pattern for limiting concurrency
258type Semaphore struct {
259    slots chan struct{}
260}
261 
262func NewSemaphore(n int) *Semaphore {
263    return &Semaphore{
264        slots: make(chan struct{}, n),
265    }
266}
267 
268func (s *Semaphore) Acquire() {
269    s.slots <- struct{}{}
270}
271 
272func (s *Semaphore) Release() {
273    <-s.slots
274}
275 
276func (s *Semaphore) Do(fn func()) {
277    s.Acquire()
278    defer s.Release()
279    fn()
280}
281```
282 
283## Pipeline Pattern
284 
285```go
286// Stage-based processing pipeline
287func pipeline(ctx context.Context, input <-chan int) <-chan int {
288    // Stage 1: Square numbers
289    stage1 := make(chan int)
290    go func() {
291        defer close(stage1)
292        for num := range input {
293            select {
294            case stage1 <- num * num:
295            case <-ctx.Done():
296                return
297            }
298        }
299    }()
300 
301    // Stage 2: Filter even numbers
302    stage2 := make(chan int)
303    go func() {
304        defer close(stage2)
305        for num := range stage1 {
306            if num%2 == 0 {
307                select {
308                case stage2 <- num:
309                case <-ctx.Done():
310                    return
311                }
312            }
313        }
314    }()
315 
316    return stage2
317}
318```
319 
320## Quick Reference
321 
322| Pattern | Use Case | Key Points |
323|---------|----------|------------|
324| Worker Pool | Bounded concurrency | Limit goroutines, reuse workers |
325| Fan-out/Fan-in | Parallel processing | Distribute work, merge results |
326| Pipeline | Stream processing | Chain transformations |
327| Rate Limiter | API throttling | Control request rate |
328| Semaphore | Resource limits | Cap concurrent operations |
329| Done Channel | Graceful shutdown | Signal completion |
330