{adamcoding}
← The Machine and the Market
Part IV
20
Chapter 20

Fan-Out

The problem

Now the real thing. 500 parameter combinations, one backtest each, all eight cores.

The worker pool

The pattern: a channel of jobs, a channel of results, and N goroutines pulling from one and pushing to the other.

go
package main

import (
	"fmt"
	"runtime"
	"sort"
	"sync"
)

type Params struct {
	Fast, Slow int
}

type Outcome struct {
	Params     Params
	Expectancy float64
	Trades     int
}

func worker(id int, jobs <-chan Params, results chan<- Outcome,
	bars []Bar, wg *sync.WaitGroup) {

	defer wg.Done()
	for p := range jobs {
		strategy := NewEmaCross(p.Fast, p.Slow)   // fresh instance per job
		r := RunBacktest(strategy, bars)
		results <- Outcome{
			Params: p, Expectancy: r.Expectancy, Trades: r.TradeCount,
		}
	}
}

func Sweep(bars []Bar, combos []Params, numWorkers int) []Outcome {
	jobs := make(chan Params, len(combos))
	results := make(chan Outcome, len(combos))

	var wg sync.WaitGroup
	for w := 1; w <= numWorkers; w++ {
		wg.Add(1)
		go worker(w, jobs, results, bars, &wg)
	}

	for _, c := range combos {
		jobs <- c
	}
	close(jobs)          // tells workers there's nothing more coming

	wg.Wait()
	close(results)

	var out []Outcome
	for r := range results {
		out = append(out, r)
	}

	// Results arrive in completion order, which is not deterministic.
	sort.Slice(out, func(i, j int) bool {
		if out[i].Params.Fast != out[j].Params.Fast {
			return out[i].Params.Fast < out[j].Params.Fast
		}
		return out[i].Params.Slow < out[j].Params.Slow
	})
	return out
}

Several things here are load-bearing.

`jobs <-chan Params` and `results chan<- Outcome`. The arrows in the type make these directional: the worker can only receive from jobs and only send to results. The compiler enforces it. Free documentation and a real safety net.

`close(jobs)` after queueing. That's what ends the workers' range loops. Forget it and they block forever, wg.Wait() never returns, and you get a deadlock.

A fresh strategy per job. NewEmaCross inside the loop, not shared. A strategy holds state - ring buffers, positions - and two workers sharing one would corrupt each other's results. This is the bug Chapter 21 is about, and it's the single easiest way to get a parameter sweep subtly wrong.

Sorting at the end. Results come back in whatever order the workers finish, which varies between runs. Any output you'll compare across runs needs sorting. Chapter 12, earning its place.

See It Work: measure the speedup

This is the experiment worth running, because the result is not what most people expect.

This one depends on Sweep above plus your own Bar, RunBacktest, and helpers to build the inputs - so it's a sketch of the driver, not a standalone file. Drop it into the module you've been building and supply loadTestBars and makeCombos from your own code:

go
package main

import (
	"fmt"
	"runtime"
	"time"
)

// Supply these from your own packages:
//   loadTestBars(n int) []Bar   - e.g. market.LoadCSV, truncated to n
//   makeCombos() []Params       - e.g. every fast in 5..20, slow in 21..60

func main() {
	bars := loadTestBars(5000)
	combos := makeCombos() // say 200 parameter pairs

	fmt.Printf("CPUs available: %d\n", runtime.NumCPU())
	fmt.Printf("%-10s %-14s %-10s\n", "workers", "duration", "speedup")

	var baseline time.Duration

	for _, n := range []int{1, 2, 4, 8, 16, 32} {
		start := time.Now()
		Sweep(bars, combos, n)
		elapsed := time.Since(start)

		if n == 1 {
			baseline = elapsed
		}
		fmt.Printf("%-10d %-14v %.2fx\n", n, elapsed.Round(time.Millisecond),
			float64(baseline)/float64(elapsed))
	}
}

Typical output on an 8-core machine:

CPUs available: 8
workers    duration       speedup
1          12.4s          1.00x
2          6.3s           1.97x
4          3.2s           3.88x
8          1.8s           6.89x
16         1.7s           7.29x
32         1.7s           7.29x

Read that carefully, because it contains three lessons:

Doubling workers roughly halves the time - up to a point. Two workers gave 1.97×, close to perfect.

The gain is never quite linear. Eight workers gave 6.89×, not 8×. There's coordination overhead, and some of the program (queueing jobs, collecting results) can't be parallelised. That ceiling has a name - Amdahl's law - and it says your speedup is capped by whatever fraction of the work is inherently serial.

Past the core count, nothing improves. 16 and 32 workers are the same as 8. You have eight cores; more goroutines just means more of them waiting. Beyond that the extra scheduling can make things slightly worse.

So the sensible default:

go
numWorkers := runtime.NumCPU()

Run this on your own machine before you believe the table. The shape will match; the numbers won't.

When not to bother

Concurrency has a cost - in complexity, in bugs, in the tests you now need. It's worth it when:

  • The work is genuinely parallel (independent backtests: perfect)
  • Each unit is big enough to dwarf the coordination overhead (milliseconds, not microseconds)
  • You're actually CPU-bound

It's not worth it when the work is trivially fast, when the tasks depend on each other, or when you're waiting on one file on one disk. Parallelising something that takes 3ms serially will make it slower, and you can prove that with the same harness above.

Cancellation

If one job fails or the user hits Ctrl-C, stop the rest. That's what context is for:

go
import "context"

func worker(ctx context.Context, jobs <-chan Params, results chan<- Outcome,
	bars []Bar, wg *sync.WaitGroup) {

	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			return                    // cancelled - stop cleanly
		case p, ok := <-jobs:
			if !ok {
				return                // no more jobs
			}
			results <- runOne(p, bars)
		}
	}
}

The caller creates a cancellable context, and cancel() makes every worker return promptly. This is how every long-running Go program handles shutdown, and it's worth adopting the habit early.

Exercises

20.1 Implement Sweep and run a 200-combination parameter sweep over prices.csv.

20.2 Run the speedup experiment on your machine. Tabulate and plot. Where does it flatten, and does that match runtime.NumCPU()?

20.3 Deliberately remove close(jobs) and observe the deadlock. Read the message and explain it.

20.4 Deliberately share one strategy instance across all workers. Run the sweep three times and compare results. Do they match each other? Do they match the serial version?

20.5 Add context cancellation and a flag stopping the sweep after the first result exceeding some expectancy.

20.6 Parallelise something trivially fast - squaring a million integers - and measure. Is it faster or slower than the serial version? Explain.

20.7 Harder. Add a progress reporter: a goroutine printing "142/500 complete" once a second, using select with time.Ticker.

Solutions

20.3 Workers block forever in range jobs, so wg.Wait() never returns. Go detects that every goroutine is blocked and reports the deadlock. close is not cleanup - it's the signal that ends the loop.

20.4 The results differ between runs and none of them match the serial version. Ring buffers get values from two parameter sets mixed together. This is the chapter's most important exercise: a concurrency bug doesn't crash, it quietly produces wrong numbers, and in a parameter sweep those numbers look entirely plausible.

20.6 Slower. The work per item is nanoseconds and the channel coordination is far more expensive. This is the counterexample to "concurrency makes things faster."