Part 3 · 2 chapters · ~12 min

Concurrency

Goroutines, unbuffered and buffered channels, select with timeouts, the sync package (Mutex, RWMutex, WaitGroup, Once), context for cancellation and deadlines, errgroup, worker pools and pipelines, the race detector, and common goroutine leaks.

Channels, select and context

code

// a worker pool with a bounded queue, cancellation and results
func process(ctx context.Context, transfers []Transfer) error {
	g, ctx := errgroup.WithContext(ctx)
	jobs := make(chan Transfer)
	g.Go(func() error { defer close(jobs)
		for _, t := range transfers { select { case jobs <- t: case <-ctx.Done(): return ctx.Err() } }
		return nil })
	for w := 0; w < 8; w++ {
		g.Go(func() error { for t := range jobs { if err := post(ctx, t); err != nil { return err } }; return nil })
	}
	return g.Wait()
}

// a timeout with select
select {
case res := <-resultCh: use(res)
case <-time.After(2 * time.Second): return ErrTimeout
case <-ctx.Done(): return ctx.Err()
}

CONTEXT: CANCELLATION AND DEADLINES DOWN THE CALL TREE

one request, one context, every goroutine and call underneath it

swipe the figure sideways, or tap expand for full screen

1/5

one context per request

net/http gives each request a context that is cancelled when the client disconnects. Add a deadline at the top: ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second).

request context + deadlinealways defer cancel()

The race detector and goroutine leaks

code

go test -race ./...          # instruments memory accesses; fails tests on data races
go run -race main.go

// a goroutine leak: nobody ever reads from ch after a timeout, so the sender blocks forever
func leaky() string {
	ch := make(chan string)                       // fix: make(chan string, 1) so the send never blocks
	go func() { ch <- slowCall() }()
	select { case s := <-ch: return s; case <-time.After(time.Second): return "timeout" }
}
// detect leaks: runtime.NumGoroutine() over time, /debug/pprof/goroutine, go.uber.org/goleak in tests