Running a DAG in Parallel in Go with gograph

Think about what happens when you run a build. You check out the code. Then you compile it and generate the docs. Then you run the tests and the linter. Some of these steps have to wait for others, and some can run at the same time.

Draw each step as a box, and each "has to wait for" as an arrow, and you get a directed acyclic graph, or DAG. Directed, because every arrow points one way. Acyclic, because no chain of arrows leads back to where it started. Builds, database migrations, service start-up and data pipelines all have this shape.

I started gograph, a graph library for Go, in 2022. For a long time it could sort a graph like this into one list, where every task comes after the tasks it needs. That list is all you need to run tasks one by one. Once you want to run them in parallel, you need other answers: which tasks can run together, and which ones can start right now.

Version 0.9 adds two small tools for that, dag.Levels and dag.Tracker. Go already has good ways to run work: goroutines, errgroup, worker pools. So I kept gograph's part small. It tells you what can start, and you run it however you like.

The pipeline

Here are the five steps as a graph. An edge from A to B means B waits for A.

g := gograph.New[string](gograph.Directed())
for _, e := range [][2]string{
	{"checkout", "build"},
	{"checkout", "docs"},
	{"build", "test"},
	{"build", "lint"},
} {
	_, _ = g.AddEdge(gograph.NewVertex(e[0]), gograph.NewVertex(e[1]))
}

build and docs need checkout. test and lint need build. Nothing needs docs.

If your edges come from a config file, create the graph with gograph.Acyclic() instead. Then AddEdge returns an error for any edge that would close a loop, and you find the mistake the moment you load the file.

The examples below print task names with this small helper:

func labels(vs []*gograph.Vertex[string]) []string {
	out := make([]string, len(vs))
	for i, v := range vs {
		out[i] = v.Label()
	}
	return out
}

Running in groups with Levels

dag.Levels splits the tasks into groups. The first group holds the tasks that wait for nothing. Each later group only waits for the groups before it, so everything in one group can run at the same time.

levels, err := dag.Levels(g)
if err != nil {
	return err
}
for i, level := range levels {
	fmt.Println(i, labels(level))
}

0 [checkout]
1 [build docs]
2 [test lint]

That's a correct schedule, and it's easy to run: start every task in a group, wait for all of them, then move on to the next group.

It has one weak spot. Say docs takes five minutes and build takes one. test and lint only need build, yet with groups they sit and wait four extra minutes for docs to finish.

Starting each task as soon as it can run, with Tracker

dag.Tracker fixes that. You ask it which tasks are ready, run them, and tell it when each one is done. A task becomes ready the moment every task it waits for is done, so a slow task only holds up the tasks that need it.

t, err := dag.NewTracker(g)
if err != nil {
	return err
}
fmt.Println(labels(t.Ready()))
_ = t.Done("checkout")
fmt.Println(labels(t.Ready()))
_ = t.Done("build")
fmt.Println(labels(t.Ready()))
fmt.Println(t.Remaining())

[checkout]
[build docs]
[test lint]
3

At first only checkout is ready. Once it's done, build and docs are ready. Once build is done, test and lint are ready, while docs may still be running. Remaining says three tasks are left: docs, test and lint.

Ready returns each task only once, so you can keep calling it and you only get new work.

A worker loop

Here's the whole thing with goroutines. One loop talks to the tracker, each task runs in its own goroutine, and the results come back on a channel. run is where your real work goes.

type result struct {
	label string
	err   error
}

t, err := dag.NewTracker(g)
if err != nil {
	return err
}

// buffered, so a worker never blocks if the loop returns early
results := make(chan result, g.Order())
for t.Remaining() > 0 {
	for _, v := range t.Ready() {
		go func() {
			results <- result{label: v.Label(), err: run(v.Label())}
		}()
	}

	r := <-results
	if r.err != nil {
		return r.err
	}
	if err := t.Done(r.label); err != nil {
		return err
	}
}

Each time around, the loop starts everything that's ready, waits for one result, and marks that task done. Marking a task done can make new tasks ready, and the next pass starts them.

A tracker is meant to be used from one goroutine, which is why only the loop calls Ready and Done. NewTracker copies what it needs from the graph, so you can change the graph later without affecting a tracker that's already running.

When a task fails

The loop above stops at the first error. Sometimes you'd rather keep going with the tasks that don't depend on the failed one. dag.Descendants gives you every task that depends on it, directly or further down. Mark the failed task done, and when Ready hands you one of its descendants, mark that one done too without running it.

On this graph, the descendants of build are test and lint. So if build fails, test and lint are skipped and docs still runs.

Try it

gograph.dev runs gograph on small examples you can step through, including a build pipeline like this one:

Topological sort: one order that runs every task after the tasks it needs
Ancestors: everything a task waits for
Descendants: everything that waits for a task

Wrapping up

Use Levels when running in groups is good enough, or when you want to show the plan to a person. Use Tracker when a slow task shouldn't hold up work that doesn't need it. Both take time in proportion to the number of tasks plus the number of edges. Both return gograph.ErrDAGHasCycle if the graph has a loop, because a loop has no valid order.

gograph has no dependencies outside the Go standard library. If you try it on your own pipeline, I'd like to hear how it goes.