worker pool in golang - part 2

In Part 1 we
built a worker pool: a jobs channel, three workers, a results channel, and a
closer goroutine that calls wg.Wait() then close(results). We also walked
into a deadlock on purpose, because that is how you learn to read a concurrent
program.

Part 1 was easy in one specific way: the workers never shared anything. Each
worker read from a channel and wrote to a channel, and channels do the
synchronization for you. Channels are Go's "share memory by communicating"
answer.

In this part we do the opposite. We add a counter that every worker touches at
the same time, and we protect it with sync.Mutex. Then we prove it works with
go test, go test -race and go test -bench, and - the interesting part - we
delete the mutex and watch the race detector tear the program apart.

The full source is in
worker-pool-v2, and
the workflow diagram for this part is in diagram.md.

What we are adding

Four steps:

1. A Stats struct that counts processed and failed jobs.
2. A sync.Mutex inside it, so concurrent increments are safe.
3. Tests: unit, concurrent, race, benchmark.
4. Remove the mutex, run -race, and read the report.

Step 1: the Stats struct

// adding stats struct to follow the number of jobs successed and failed

type Stats struct {
    JobProcessed byte
    JobFailed    byte
    mu           sync.Mutex
}

func (s *Stats) processed() {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.JobProcessed++
}

func (s *Stats) failed() {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.JobFailed++
}

Three things worth saying out loud.

The mutex lives inside the struct it protects. Not next to it, not in a
global. When the lock and the data travel together, nobody can pass you the
counter without also passing you the thing that guards it.

The methods have pointer receivers. func (s *Stats), not func (s Stats).
A value receiver would copy the struct - including the mutex - and you would lock
a private copy that nobody else can see. go vet catches this one for you
(passes lock by value), and it is a mistake worth catching, because the code
still compiles and still runs and is simply wrong.

defer s.mu.Unlock() right after Lock(). The unlock runs even if the body
panics or returns early. In a two-line function this looks like ceremony. In a
twenty-line function with three return statements it is the difference between
working code and a permanently locked mutex.

Step 2: wiring it into the worker

func worker(id byte, jobs <-chan Job, results chan<- Result, stats *Stats, wg *sync.WaitGroup) {
    defer wg.Done()
    //iterate over incoming jobs from the channel and process them
    for job := range jobs {
        stats.processed()
        fmt.Printf("job %d is being processed by worker %d\n", job.ID, id)
        //simulate a failure
        //if the random value is eq to 3 then fail
        if rand.Intn(7) == 3 {
            stats.failed()
            continue
        }
        //sleep to simulate work
        time.Sleep((time.Duration(rand.Intn(3)) + 1) * time.Second)
        fmt.Printf("job %d is done by worker %d\n", job.ID, id)
        results <- Result{WorkerID: id, JobID: job.ID}
    }
    fmt.Println("worker ", id, " is done")
}

stats is a *Stats - one struct, shared by all three workers. That is the
whole point. If we passed Stats by value, each worker would count its own jobs
into its own copy and main would print zeros.

And in main:

stats := &Stats{}

for i := range NumOfWorkers {
    wg.Add(1)
    go worker(byte(i+1), jobs, results, stats, &wg)
}

go initJobs(jobs)

//using the clsoing go routines
go func() {
    wg.Wait()
    close(results)
}()

for result := range results {
    fmt.Printf("result of job %d is done by worker %d\n", result.JobID, result.WorkerID)
}
fmt.Printf("Stats - Processed: %d, Failed: %d\n", stats.JobProcessed, stats.JobFailed)

Notice the last line reads stats.JobProcessed without locking. Is that a
bug? No - and the reason is the closer goroutine we built in Part 1:

workers write stats ──► wg.Done() ──► wg.Wait() ──► close(results)
                                                         │
                                                         ▼
                                      range results ends ──► safe read

The for range results loop can only exit after close(results), which can only
happen after wg.Wait() returns, which can only happen after every worker has
returned. By the time we print, there is no other goroutine left to race with.
The Go memory model gives us that ordering for free. This is the same coordination
we built in Part 1, now paying a second dividend.

Step 3: testing it

Concurrency bugs do not show up reliably when you run the program by hand. They
show up when you write tests that try to break the code.

The unit tests

func TestStats(t *testing.T) {
    //testing stats struct methods
    t.Run("Test processed method", func(t *testing.T) {
        stats := &Stats{}
        stats.processed()
        if stats.JobProcessed != 1 {
            t.Errorf("expected JobProcessed to be 1, got %d", stats.JobProcessed)
        }
    })
    t.Run("Test failed method", func(t *testing.T) {
        stats := &Stats{}
        stats.failed()
        if stats.JobFailed != 1 {
            t.Errorf("expected JobFailed to be 1, got %d", stats.JobFailed)
        }
    })

Boring, and that is fine. They pin down the single-goroutine behaviour so that
when the concurrent test fails you know the arithmetic itself was never the
problem.

The concurrent test - the one that matters

    t.Run("Test concurrent access", func(t *testing.T) {
        stats := &Stats{}
        var wg sync.WaitGroup
        for range 100 {
            //waiting for two goroutines in each iteration
            wg.Add(2)
            go func() {
                defer wg.Done()
                stats.processed()
            }()
            go func() {
                defer wg.Done()
                stats.failed()
            }()
        }
        wg.Wait()
        if stats.JobProcessed != 100 {
            t.Errorf("expected JobProcessed to be 100, got %d", stats.JobProcessed)
        }
        if stats.JobFailed != 100 {
            t.Errorf("expected JobFailed to be 100, got %d", stats.JobFailed)
        }
    })
}

200 goroutines launched as fast as the runtime can spawn them, all hammering the
same two bytes. If the locking is wrong, the counters land below 100.

wg.Add(2) sits before the two go statements, not inside them. Calling
Add inside the goroutine is a classic race: wg.Wait() can return before the
goroutine has even started counting itself.

Also note the loop is for range 100 - a Go 1.22+ range-over-int, same as
for i := 0; i < 100; i++ without the unused variable.

The end-to-end test

func TestWorker(t *testing.T) {
    jobs := make(chan Job, NumOfJobs)
    results := make(chan Result, NumOfJobs)
    stats := &Stats{}
    var wg sync.WaitGroup
    //testing with one worker
    wg.Add(1)
    go worker(byte(1), jobs, results, stats, &wg)
    for i := range NumOfJobs {
        jobs <- Job{ID: byte(i)}
    }
    close(jobs)
    wg.Wait()
    close(results)
    //check if all jobs are processed
    if stats.JobProcessed != NumOfJobs {
        t.Errorf("expected %d jobs processed, got %d", NumOfJobs, stats.JobProcessed)
    }
}

One detail that is easy to miss: results is buffered here
(make(chan Result, NumOfJobs)), while in main it is unbuffered. In the test
there is no consumer ranging over results, so an unbuffered channel would block
the worker on its first send and wg.Wait() would hang forever - the exact
deadlock from Part 1, rebuilt by accident inside a test. The buffer gives the
worker somewhere to put its results so it can finish and call wg.Done().

We assert on JobProcessed only, not on JobFailed, because failures are
random (rand.Intn(7) == 3). Every job is counted as processed regardless, so
that number is deterministic. Asserting on random output is how you get a test
that fails once a week and teaches the team to ignore it.

Run it:

go test .

Step 4: go test -race

A passing test proves the result was right this time. It does not prove the
program is correct. Two goroutines can race and still, by luck, produce 100.

-race answers the stronger question. It instruments every memory access and
records which goroutine touched which address while holding which locks. If two
goroutines touch the same address, at least one writes, and nothing orders them -
it reports, even if the output happened to be correct.

go test -race .

Green. The mutex is doing its job.

The race detector is not free - roughly 5-10x slower and much more memory - so it
is a CI and development tool, not a production build flag. But every concurrent
package you write should have at least one test that runs under -race, and
that test needs enough concurrency to give the detector something to observe.
That is why TestStats/Test_concurrent_access exists.

Step 5: go test -bench

Locks cost something. Let's measure it rather than guess.

func BenchmarkStatsProcessed(b *testing.B) {
    //sequential cost of one Lock/Unlock pair
    stats := &Stats{}
    for b.Loop() {
        stats.processed()
    }
}

func BenchmarkStatsProcessedParallel(b *testing.B) {
    //same counter hammered from every available core -> lock contention
    stats := &Stats{}
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            stats.processed()
        }
    })
}

b.Loop() is the Go 1.24+ form of the benchmark loop. It replaces
for i := 0; i < b.N; i++ and keeps the compiler from optimizing away work whose
result you never use.

b.RunParallel is the one that tells you something you did not already know: it
runs the body on GOMAXPROCS goroutines at once, so you are measuring lock
contention, not just lock overhead.

go test -bench . -run XXX .

(-run XXX matches no test name, so only benchmarks run.)

On my M1 the two numbers are roughly:

BenchmarkStatsProcessed-8           54193095    22.13 ns/op
BenchmarkStatsProcessedParallel-8    8557812   157.4  ns/op

Uncontended, a lock/unlock pair is ~22 ns - cheap. Contended across 8 goroutines
it is ~157 ns, about 6.5x worse, because the goroutines are now queuing behind
each other and paying for cache-line ping-pong between cores.

That is the real lesson of mutex performance: the lock itself is cheap, waiting
for it is not. Our workers hold the lock for one increment and spend seconds
sleeping outside it, so contention is irrelevant here. If you ever find yourself
doing I/O inside a critical section, this is the number that will bite you.

(For a plain counter like this, sync/atomic would be faster still - but that is for later chapter)

Now the fun part: what if we drop the mutex?

Let's break it deliberately, the same way Part 1 broke on a deadlock. Comment out
the locks:

func (s *Stats) processed() {
    // s.mu.Lock()
    // defer s.mu.Unlock()
    s.JobProcessed++
}

func (s *Stats) failed() {
    // s.mu.Lock()
    // defer s.mu.Unlock()
    s.JobFailed++
}

Why this is broken

s.JobProcessed++ looks atomic. It is not. On any real CPU it is three steps:

        read  JobProcessed   ->  register
        add   1              ->  register
        write register       ->  JobProcessed

Two goroutines can interleave those steps:

 time ──────────────────────────────────────────────────►

 Worker 1   read(5) ──── add ──── write(6)
 Worker 2        read(5) ──── add ──── write(6)
                                          ▲
                                two increments, counter moved by 1
                                          = lost update

Both read 5, both compute 6, both write 6. One increment vanished. This is a
lost update, and with 200 goroutines it does not happen once - it happens
constantly.

It gets worse than lost counts. Without synchronization there is no
happens-before relationship between the goroutines at all, so the Go memory
model gives the compiler and the CPU permission to reorder and cache these
accesses. A goroutine can keep JobProcessed in a register and never publish it.
The behaviour is undefined, not merely "slightly off" - which is exactly why you
cannot debug this by adding fmt.Println and staring at it.

Running it

go test .

This may well pass. That is the trap. With only 100 iterations and a fast
machine, the goroutines often serialize by accident and you get 100 and 100.
Data races are not deterministic; a green test here proves nothing.

Now the real check:

go test -race .

==================
WARNING: DATA RACE
Read at 0x00c0000123d5 by goroutine 13:
  worker-pool.(*Stats).failed()
      .../worker-pool-series/golang/worker-pool-v2/main.go:46 +0x78
  worker-pool.TestStats.func3.2()
      .../worker-pool-series/golang/worker-pool-v2/main_test.go:52 +0x74

Previous write at 0x00c0000123d5 by goroutine 15:
  worker-pool.(*Stats).failed()
      .../worker-pool-series/golang/worker-pool-v2/main.go:46 +0x8c
  worker-pool.TestStats.func3.2()
      .../worker-pool-series/golang/worker-pool-v2/main_test.go:52 +0x74

Goroutine 13 (running) created at:
  worker-pool.TestStats.func3()
      .../worker-pool-series/golang/worker-pool-v2/main_test.go:50 +0x78
  testing.tRunner()
...
==================
--- FAIL: TestStats/Test_concurrent_access
    expected JobProcessed to be 100, got 97
FAIL

How to read that report, because it is genuinely useful once you know the shape:

• Read at 0x... / Previous write at 0x... - the same memory address, hit by two different goroutines. Same address is the whole finding.
• The two stacks - where each access happened. Here both point at main.go:42, the bare s.JobFailed++. The detector puts your bug on a line number.
• Goroutine N created at - where each goroutine was spawned, so you can trace it back to the go statement in the test.
• Then the assertion failure underneath: got 97 instead of 100. Three lost updates in a single run.

Put the two s.mu lines back, rerun go test -race ., and it is green again.
That round trip - break it, see the report, fix it, see it clear - is the fastest
way I know to build a real intuition for what a mutex is actually buying you.

Next

In Part 3 we keep going through sync: sync/atomic for counters like this one
(no lock at all), sync.RWMutex for read-heavy state.

• GitHub: https://github.com/aymane-belassiria/worker-pool-series/tree/main/golang/worker-pool-v2