Concurrency patterns

1var loadConfig = sync.OnceValue(func() map[string]string {
2	fmt.Println("loading config")       // runs once, however many callers
3	return map[string]string{"env": "prod"}
4})
5
6cfg := loadConfig()                     // first call loads
7cfg = loadConfig()                      // later calls get the same map

1var wg sync.WaitGroup
2for i := range 3 {
3	wg.Go(func() {                     // Go 1.25+
4		fmt.Println("task", i)          // own i per iteration since Go 1.22
5	})
6}
7wg.Wait()

1func fetchAll(ctx context.Context, urls []string) ([]string, error) {
2	g, ctx := errgroup.WithContext(ctx)   // ctx is cancelled on the first error
3	g.SetLimit(4)                         // at most 4 fetches at a time
4
5	bodies := make([]string, len(urls))
6	for i, url := range urls {
7		g.Go(func() error {
8			body, err := fetch(ctx, url)  // fetch must honour ctx
9			if err != nil {
10				return err
11			}
12			bodies[i] = body             // each goroutine writes its own index: no race
13			return nil
14		})
15	}
16	if err := g.Wait(); err != nil {
17		return nil, err
18	}
19	return bodies, nil
20}

1type group struct {
2	wg     sync.WaitGroup
3	once   sync.Once
4	err    error
5	cancel context.CancelCauseFunc
6}
7
8func withContext(ctx context.Context) (*group, context.Context) {
9	ctx, cancel := context.WithCancelCause(ctx)
10	return &group{cancel: cancel}, ctx
11}
12
13func (g *group) Go(f func() error) {
14	g.wg.Add(1)
15	go func() {
16		defer g.wg.Done()
17		if err := f(); err != nil {
18			g.once.Do(func() { // keep only the first error
19				g.err = err
20				g.cancel(err)
21			})
22		}
23	}()
24}
25
26func (g *group) Wait() error {
27	g.wg.Wait()
28	g.cancel(g.err) // release the context's resources
29	return g.err
30}

1package main
2
3import (
4	"fmt"
5	"sync"
6)
7
8func main() {
9	jobs := make(chan int)
10	results := make(chan int)
11
12	var wg sync.WaitGroup
13	for range 3 { // 3 workers: at most 3 jobs run at once
14		wg.Add(1)
15		go func() {
16			defer wg.Done()
17			for n := range jobs { // ends when jobs is closed and empty
18				results <- n * n
19			}
20		}()
21	}
22
23	go func() { // the producer: sends, then closes
24		for n := 1; n <= 5; n++ {
25			jobs <- n
26		}
27		close(jobs)
28	}()
29
30	go func() { // close results only after every worker is done
31		wg.Wait()
32		close(results)
33	}()
34
35	sum := 0
36	for r := range results {
37		sum += r
38	}
39	fmt.Println("sum of squares 1..5:", sum) // always 55
40}

1package main
2
3import (
4	"fmt"
5	"runtime"
6	"time"
7)
8
9// slowLookup gives up after 10ms, but the work takes 50ms.
10func slowLookup(buffered bool) (int, bool) {
11	var ch chan int
12	if buffered {
13		ch = make(chan int, 1) // room for the one send: it never blocks
14	} else {
15		ch = make(chan int) // the send waits for a receiver that has left
16	}
17	go func() {
18		time.Sleep(50 * time.Millisecond)
19		ch <- 42
20	}()
21	select {
22	case v := <-ch:
23		return v, true
24	case <-time.After(10 * time.Millisecond):
25		return 0, false // we stop listening here
26	}
27}
28
29func main() {
30	before := runtime.NumGoroutine()
31
32	for range 3 {
33		slowLookup(false)
34	}
35	time.Sleep(100 * time.Millisecond) // let the workers finish if they can
36	fmt.Println("unbuffered, leaked:", runtime.NumGoroutine()-before)
37
38	before = runtime.NumGoroutine()
39	for range 3 {
40		slowLookup(true)
41	}
42	time.Sleep(100 * time.Millisecond)
43	fmt.Println("buffered, leaked:", runtime.NumGoroutine()-before)
44}