Concurrency Patterns in Go: context, sync, errgroup, Worker Pools and Goroutine Leaks

Version note: The code in this post was built and run with Go 1.27.4; the module's go line is 1.26.0 and the code also compiles with Go 1.26.11. golang.org/x/sync v0.25.0 was used. WaitGroup.Go and testing/synctest arrived in Go 1.25; the goroutineleak profile became generally available in Go 1.27, and the lab that shows it is set apart with //go:build go1.27.

Guess First

Bookstore, the online bookstore used as the shared example across this site's series, has had an orders.Service, the counterpart of the Bookstore.Ordering module on the C# side, and an in-memory memstore since post 04. The comments on both said "not safe for concurrent use". Now 100 customers order Dune at the same time:

// Lab 08: 100 customers place an order at the same time.
package main

import (
    "context"
    "fmt"
    "sync"

    "bookstore/internal/catalog"
    "bookstore/internal/orders"
    "bookstore/internal/orders/memstore"
)

func main() {
    ctx := context.Background()
    repo := memstore.New()
    svc := orders.NewService(repo, catalog.Store{})

    const n = 100
    ids := make([]string, n)
    var wg sync.WaitGroup
    for i := range n {
        wg.Go(func() { // Go 1.25: Add(1), go, defer Done() in one call
            o, err := svc.Place(ctx, fmt.Sprintf("customer-%d", i),
                []orders.Line{{ISBN: "9780441172719", Quantity: 1}})
            if err != nil {
                fmt.Println("place:", err)
                return
            }
            ids[i] = o.ID // each goroutine writes only its own index
        })
    }
    wg.Wait()

    unique := map[string]bool{}
    for _, id := range ids {
        unique[id] = true
    }
    fmt.Printf("orders placed: %d, unique IDs: %d\n", n, len(unique))
}

Does the program print 100 orders and 100 different IDs, or fewer?

go run ./labs/08-orders

fatal error: concurrent map writes

goroutine 13 [running]:
internal/runtime/maps.fatal({0x10227bb1e?, 0x102247b0c?})
    /usr/local/go/src/runtime/panic.go:1197 +0x20
bookstore/internal/orders/memstore.(*Store).Save(0x0?, {0x6?, 0x1d0c56084e98?}, 0x1d0c560880e0)
    /home/user/bookstore/internal/orders/memstore/memstore.go:26 +0x30
bookstore/internal/orders.(*Service).Place(...)
    /home/user/bookstore/internal/orders/service.go:67 +0x5a4
main.main.func1()
    /home/user/bookstore/labs/08-orders/main.go:24 +0xa8
...
exit status 2

It printed nothing; the process crashed in five runs out of five. Go's map is not protected against concurrent writes, and when the runtime notices one it stops the program with a fatal error that cannot be recovered. This is the lucky case: the bug came loudly. Once we fix the map, a much quieter bug will remain. In this post we look at the sync package, the race detector, context.Context, a worker pool with errgroup and goroutine leak detection, while making Bookstore's order service safe for concurrent use.

sync.Mutex and RWMutex: Fixing memstore

If several goroutines access a map, you make them take turns with a mutex. I added a sync.RWMutex to memstore: Lock for writing, RLock for reading. With an RWMutex many readers can enter at once, while a writer enters alone. Its zero value is ready to use; there is nothing to initialise.

// Store is an in-memory order store, safe for concurrent use.
type Store struct {
    mu   sync.RWMutex // guards byID
    byID map[string]*orders.Order
}

// Compile-time check that *Store satisfies orders.Repository.
var _ orders.Repository = (*Store)(nil)

// New returns an empty store.
func New() *Store {
    return &Store{byID: make(map[string]*orders.Order)}
}

// Save stores the order, replacing any order with the same ID.
func (s *Store) Save(_ context.Context, o *orders.Order) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.byID[o.ID] = o
    return nil
}

// ByID returns the order with the given ID.
func (s *Store) ByID(_ context.Context, id string) (*orders.Order, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    o, ok := s.byID[id]
    if !ok {
        return nil, fmt.Errorf("id %s: %w", id, orders.ErrNotFound)
    }
    return o, nil
}

Putting the mutex right above the field it guards and saying in a comment what it guards is an established habit in Go. With defer s.mu.Unlock() the lock is released whichever way the function exits. This is the counterpart of the lock statement from the C# Asenkron Programlama series' "C# Senkronizasyon" post; the difference is that in Go the lock is a value and there is no syntax like lock.

go run ./labs/08-orders

orders placed: 100, unique IDs: 94
orders placed: 100, unique IDs: 95
orders placed: 100, unique IDs: 93
orders placed: 100, unique IDs: 96

I ran the program four times. The crash is gone, but 4 to 7 of the 100 orders got the same ID as another order. Since an order saved with the same ID overwrites the previous one, those orders were silently lost. No error, no log.

The Race Detector: Finding the Silent Bug

Go's race detector is a tool built into the compiler: a program built with -race watches every memory access and reports when two goroutines access the same address without synchronisation and at least one of them writes.

go run -race ./labs/08-orders

==================
WARNING: DATA RACE
Read at 0x00c0000a2148 by goroutine 9:
  bookstore/internal/orders.(*Service).Place()
      /home/user/bookstore/internal/orders/service.go:61 +0x4a8
  main.main.func1()
      /home/user/bookstore/labs/08-orders/main.go:24 +0xe4

Previous write at 0x00c0000a2148 by goroutine 8:
  bookstore/internal/orders.(*Service).Place()
      /home/user/bookstore/internal/orders/service.go:61 +0x4bc
  main.main.func1()
      /home/user/bookstore/labs/08-orders/main.go:24 +0xe4
...
==================
orders placed: 100, unique IDs: 57
Found 3 data race(s)
exit status 66

service.go:61 is the s.nextID++ line left over from post 05. This operation is not one step: read the value, add one, write it back. When two goroutines read the same value and both write 1001, two orders get the same ID. This is exactly the read-modify-write case described in the Concurrency Teorisi series' "Race Condition ve Data Race: Check-Then-Act, Read-Modify-Write ve Atomicity" post. The fix is to guard the counter with its own mutex:

// Service places orders. It is safe for concurrent use.
type Service struct {
    repo  Repository
    books Catalog
    now   func() time.Time

    mu     sync.Mutex // guards nextID
    nextID int
}

// newID returns the next order ID. Only the counter is locked, not the
// catalog lookups or the repository call.
func (s *Service) newID() string {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.nextID++
    return fmt.Sprintf("ORD-%d", s.nextID)
}

go run -race ./labs/08-orders

orders placed: 100, unique IDs: 100

I kept the lock around the counter only. Had I locked the whole of Place, the catalog lookups and the repository call would queue up as well and the 100 orders would be processed one by one. The race detector has two limits. It only finds races that actually happen in that run; it says nothing about code paths that never run. It is also expensive: according to the Go documentation memory use can grow 5-10x and execution time 2-20x. That is why it is turned on in tests and CI, not in the production binary. On Windows -race needs cgo enabled and a C compiler based on mingw-w64 installed; on macOS no C compiler is needed.

context.Context: Cancellation, Deadlines and Values

A question was left open at the end of post 07: how do we tell goroutines that a request is no longer wanted? Go's answer is context.Context. A context can be cancelled, can carry a deadline, and can carry small values such as a request ID for the lifetime of a request. Think of it as the CancellationToken from the C# Asenkron Programlama series' "C# CancellationToken: İptal, Timeout ve Linked Token" post combined with ASP.NET Core's HttpContext.Items.

// Lab 08: cancellation, deadlines, causes and values with context.Context.
package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "bookstore/internal/catalog"
)

// requestIDKey is unexported, so no other package can collide with it.
type requestIDKey struct{}

// lookupStock stands in for an inventory call that takes delay.
// It gives up as soon as ctx is done.
func lookupStock(ctx context.Context, isbn string, delay time.Duration) (int, error) {
    rid, _ := ctx.Value(requestIDKey{}).(string)
    select {
    case <-time.After(delay):
        b, err := catalog.Store{}.ByISBN(isbn)
        return b.Stock, err
    case <-ctx.Done():
        return 0, fmt.Errorf("request %s: lookup %s: %w", rid, isbn, context.Cause(ctx))
    }
}

func main() {
    base := context.WithValue(context.Background(), requestIDKey{}, "req-42")

    // 1) and 2) A 100ms deadline: fast enough, then too slow.
    for _, delay := range []time.Duration{50 * time.Millisecond, 300 * time.Millisecond} {
        ctx, cancel := context.WithTimeout(base, 100*time.Millisecond)
        start := time.Now()
        stock, err := lookupStock(ctx, "9780441172719", delay)
        cancel() // always release the timer, even on success
        fmt.Printf("delay %v: stock=%d err=%v (%v)\n", delay, stock, err,
            time.Since(start).Round(10*time.Millisecond))
        fmt.Println("  deadline exceeded?", errors.Is(err, context.DeadlineExceeded))
    }

    // 3) Cancelling with a cause.
    ctx, cancel := context.WithCancelCause(base)
    go func() {
        time.Sleep(20 * time.Millisecond)
        cancel(errors.New("client closed the connection"))
    }()
    _, err := lookupStock(ctx, "9780141439587", time.Second)
    fmt.Println(err)
    fmt.Println("  ctx.Err():", ctx.Err())

    // 4) Cancellation flows down: a child of a cancelled context is done at once.
    child, stop := context.WithTimeout(ctx, time.Hour)
    defer stop()
    <-child.Done()
    fmt.Println("child:", child.Err(), "| cause:", context.Cause(child))
}

go run ./labs/08-context

delay 50ms: stock=12 err=<nil> (50ms)
  deadline exceeded? false
delay 300ms: stock=0 err=request req-42: lookup 9780441172719: context deadline exceeded (100ms)
  deadline exceeded? true
request req-42: lookup 9780141439587: client closed the connection
  ctx.Err(): context canceled
child: context canceled | cause: client closed the connection

Three rules have become conventions in Go, and the context package's documentation states them too. A context is passed to every function that needs it as the first parameter, named ctx; Repository.Save(ctx, o) and Service.Place(ctx, ...) have looked like this since post 04. A context is not stored in a struct field. The cancel returned by WithTimeout and WithCancel is always called; if you do not, the timer and the link in the context tree are not released, and go vet's lostcancel check catches it.

The last line of the lab shows the direction of cancellation: it flows from parent to child, not the other way round. context.Cause, from Go 1.20, gives a better answer to "why was it cancelled?" than context canceled. When carrying values, use an unexported key type, and do not use the context in place of function parameters; it is only for request-scoped data that crosses boundaries, such as a request ID or identity information. In an HTTP server every request's context comes from r.Context() and is cancelled when the client's connection closes; in post 09 we will carry it from the handlers into the service layer.

WaitGroup.Go, sync.Once and atomic

The wg.Go(func() { ... }) in the lab arrived in Go 1.25 and folds the trio of wg.Add(1), go and defer wg.Done() into one call. The most common bug with the old pattern was calling Add inside the goroutine; the Go method makes that bug impossible. The waitgroupgo modernizer in Go 1.27's go fix rewrites the old pattern into this new call.

Two more small helpers: sync.OnceValue, from Go 1.21, returns a function that computes a value once, the first time it is needed; it is the counterpart of C#'s Lazy<T> for jobs such as loading the price list on the first request. Types such as atomic.Int64 in sync/atomic, from Go 1.19, are lighter than a mutex for a single counter; they are the counterpart of Interlocked.Increment. I could have written nextID with atomic.Int64 too; I chose the mutex because later we may want to guard another field next to the ID with the same lock. In the lab below I used atomic types for the counters.

A Worker Pool with errgroup: 10,000 Invoices

At the end of the month Bookstore has to produce invoice PDFs for 10,000 orders. Doing them one by one is slow, and starting them all at once loads the PDF library, the disk and the database all at the same time. golang.org/x/sync/errgroup covers both needs: SetLimit limits how many goroutines run at once, WithContext cancels a shared context on the first error, and Wait returns the first error. It is the job that C#'s Task.WhenAll and SemaphoreSlim do together.

go get golang.org/x/sync@v0.25.0

// Lab 08: rendering 10,000 invoices with errgroup, a limit and cancellation.
package main

import (
    "context"
    "fmt"
    "sync/atomic"
    "time"

    "golang.org/x/sync/errgroup"
)

// render stands in for producing one invoice PDF, which takes about 1ms.
func render(ctx context.Context, orderID, failAt int) error {
    if err := ctx.Err(); err != nil {
        return err // another invoice failed: do not start new work
    }
    time.Sleep(time.Millisecond)
    if orderID == failAt {
        return fmt.Errorf("render invoice ORD-%d: template not found", orderID)
    }
    return nil
}

func run(name string, limit, failAt int) {
    g, ctx := errgroup.WithContext(context.Background())
    g.SetLimit(limit) // -1 means no limit

    var rendered, inFlight, peak atomic.Int64
    start := time.Now()
    for id := 1; id <= 10_000; id++ {
        g.Go(func() error { // blocks while limit goroutines are running
            n := inFlight.Add(1)
            defer inFlight.Add(-1)
            for { // record the highest number of invoices rendered at once
                p := peak.Load()
                if n <= p || peak.CompareAndSwap(p, n) {
                    break
                }
            }
            if err := render(ctx, id, failAt); err != nil {
                return err
            }
            rendered.Add(1)
            return nil
        })
    }
    err := g.Wait() // the first error, after every goroutine has returned
    fmt.Printf("%-12s rendered=%5d peak=%5d time=%-7v err=%v\n", name, rendered.Load(), peak.Load(),
        time.Since(start).Round(time.Millisecond), err)
}

func main() {
    run("limit 8", 8, 0)
    run("limit 8+err", 8, 5000)
    run("no limit", -1, 0)
}

go run ./labs/08-invoices

limit 8      rendered=10000 peak=    8 time=1.575s  err=<nil>
limit 8+err  rendered= 5006 peak=    8 time=788ms   err=render invoice ORD-5000: template not found
no limit     rendered=10000 peak= 5313 time=5ms     err=<nil>

With a limit of 8, at most 8 invoices were produced at once. In the second run invoice 5000 failed; the context was cancelled, the few invoices in progress at that moment finished, and the rest returned at the ctx.Err() check without starting. The run without a limit took 5 ms, but 5313 invoices were being produced at once. The time.Sleep in this lab consumes no resources; a real PDF library or a database connection pool would not survive that load. If you need a rate limit such as at most N requests per second, the golang.org/x/time/rate package provides one with a token bucket algorithm.

Finding a Goroutine Leak: the goroutineleak Profile

firstMissing from post 07 leaked three goroutines on every call, and we noticed it from runtime.NumGoroutine going up. To find which line leaks in a real service, Go 1.27 has the goroutineleak profile in runtime/pprof. According to the release notes, the runtime uses the garbage collector's reachability information: if the channel or mutex a goroutine waits on cannot be reached from any goroutine that can still run, that goroutine can never wake up again, so it has leaked.

func main() {
    cart := []string{"9780000000000", "9780441172719", "9780547928227", "9780141439549"}
    for range 3 {
        _ = firstMissing(cart)
    }
    time.Sleep(100 * time.Millisecond) // let the senders reach their blocked send

    if err := pprof.Lookup("goroutineleak").WriteTo(os.Stdout, 1); err != nil {
        fmt.Println(err)
    }
}

go run ./labs/08-leakprofile

goroutineleak profile: total 9
9 @ 0x102447f78 0x1023e2224 0x1023e1e38 0x10249bae4 0x10244e084
#    0x10249bae3    main.firstMissing.func1+0x63    /home/user/bookstore/labs/08-leakprofile/main.go:25

All nine goroutines leaked by the three calls are waiting on the same line, the errs <- err on line 25. The release notes also name a limit: if the channel can be reached from a global variable or from a local variable of a running goroutine, the runtime may not see it as a leak. In post 13 we will serve the profiles from a separate port with net/http/pprof; the same profile is there at /debug/pprof/goroutineleak.

Time-Dependent Tests with testing/synctest

In Bookstore an unpaid order is cancelled after 15 minutes. Testing this rule with the real clock takes 15 minutes. testing/synctest, generally available since Go 1.25, runs the test inside a "bubble": the time functions inside the bubble use a fake clock, and when every goroutine in the bubble is blocked the clock jumps forward at once. We will look at writing tests in detail in post 12; here is only the test for this lab.

// awaitPayment waits until paid is closed or timeout has passed.
func awaitPayment(ctx context.Context, paid <-chan struct{}, timeout time.Duration) error {
    ctx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()
    select {
    case <-paid:
        return nil
    case <-ctx.Done():
        return fmt.Errorf("await payment: %w", ctx.Err())
    }
}

package main

import (
    "context"
    "errors"
    "testing"
    "testing/synctest"
    "time"
)

func TestAwaitPaymentTimesOut(t *testing.T) {
    synctest.Test(t, func(t *testing.T) {
        start := time.Now() // fake clock inside the bubble
        err := awaitPayment(t.Context(), make(chan struct{}), 15*time.Minute)
        if !errors.Is(err, context.DeadlineExceeded) {
            t.Fatalf("err = %v, want DeadlineExceeded", err)
        }
        if waited := time.Since(start); waited != 15*time.Minute {
            t.Fatalf("waited %v, want exactly 15m", waited)
        }
    })
}

func TestAwaitPaymentPaidInTime(t *testing.T) {
    synctest.Test(t, func(t *testing.T) {
        paid := make(chan struct{})
        go func() {
            time.Sleep(14 * time.Minute)
            close(paid)
        }()
        if err := awaitPayment(t.Context(), paid, 15*time.Minute); err != nil {
            t.Fatalf("err = %v, want nil", err)
        }
    })
}

go test -race -v ./labs/08-payment

=== RUN   TestAwaitPaymentTimesOut
--- PASS: TestAwaitPaymentTimesOut (0.00s)
=== RUN   TestAwaitPaymentPaidInTime
--- PASS: TestAwaitPaymentPaidInTime (0.00s)
PASS
ok      bookstore/labs/08-payment    1.228s

Both tests passed in 0.00 seconds, and the first one verified that exactly 15 minutes passed on the fake clock. The 1.2 seconds in total come from building and starting the race-enabled binary. Adding -race to go test is cheap insurance for every test that involves concurrency.

In Production

Do not do I/O inside a lock. If you make an HTTP call or a database query while holding a mutex, every other request waits for as long as that call takes. That is why newID locks only the counter.

memstore still returns the pointer it stores. The map is safe now, but if two goroutines call MarkPaid on the same *Order there is still a data race. An in-memory store has to either store and return copies or make state changes under the store's lock. The sqlstore in post 11 builds a new struct on every read, so it does not have this problem.

Put the race detector in CI. Running go test -race ./... on every pull request catches silent ID collisions like the one in this post before they reach production. But it only checks code that tests actually run concurrently; do not expect anything from the race detector without a concurrent test.

Common Pitfalls

Copying a mutex and calling Add inside the goroutine. Both compile and usually seem to work; go vet catches both:

package main

import (
    "fmt"
    "sync"
)

type Counter struct {
    mu sync.Mutex
    n  int
}

// Value has a value receiver: every call works on a copy of c, mutex included.
func (c Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.n
}

func main() {
    var c Counter
    var wg sync.WaitGroup
    for range 10 {
        go func() {
            wg.Add(1) // too late: Wait may already have returned
            defer wg.Done()
            c.mu.Lock()
            c.n++
            c.mu.Unlock()
        }()
    }
    wg.Wait()
    fmt.Println(c.Value())
}

go vet .
go run .

main.go:14:9: Value passes lock by value: vetdemo.Counter contains sync.Mutex
main.go:25:10: WaitGroup.Add called from inside new goroutine
2

The program printed 2 instead of 10: Wait returned before most goroutines reached Add. Use pointer receivers on the methods of a struct that contains a mutex, and prefer WaitGroup.Go.

Storing a context in a struct. A context put in a field of a service carries the lifetime of the first request into all requests; once it is cancelled, every later call fails at once. A context arrives as a parameter on every call.

Not calling cancel. Write defer cancel() even if WithTimeout finishes successfully; otherwise the timer lives until the deadline.

A goroutine that does not listen to its context. A function that takes ctx as a parameter and never looks at ctx.Done() does not see cancellation. Every long-running loop or wait should listen to the context, with select or a ctx.Err() check.

A Small Experiment

★☆☆ Delete the mutex from Service, make nextID an atomic.Int64, and confirm with go run -race ./labs/08-orders that the race detector stays silent.

★★☆ Remove the ctx.Err() check from render in labs/08-invoices. How many invoices were produced after the error, and how long did the run take? Explain the result.

★★★ Change memstore.Store so that it returns a copy of the order it stores (do not forget to copy the Items slice too). Then write a lab in which two goroutines read the same order and call MarkPaid, and compare the state before and after the change with -race.

The concurrency block ends here. In the next post we return to the HTTP layer: pattern routing and the precedence rules of net/http's ServeMux, building a middleware chain, request ID and logging middleware, an http.Server with its timeouts set instead of http.ListenAndServe, and http.CrossOriginProtection.