dag

README

dag

dag.Runner is a mechanism to orchestrate goroutines and the order in which they run, using the semantics of a directed acyclic graph.

Create a zero value dag.Runner, add vertices (functions) to it, add edges or dependencies between vertices, and finally invoke Run. Run will run all of the vertices in parallel topological order returning after either all vertices complete, or an error gets returned.

Example

var r dag.Runner

r.AddVertex("one", func() error {
    fmt.Println("one and two will run in parallel before three")
    return nil
})
r.AddVertex("two", func() error {
    fmt.Println("one and two will run in parallel before three")
    return nil
})
r.AddVertex("three", func() error {
    fmt.Println("three will run before four")
    return errors.New("three is broken")
})
r.AddVertex("four", func() error {
    fmt.Println("four will never run")
    return nil
})

r.AddEdge("one", "three")
r.AddEdge("two", "three")

r.AddEdge("three", "four")

fmt.Printf("the runner terminated with: %v\n", r.Run())

Documentation

Overview

Package dag implements a directed acyclic graph task runner with deterministic teardown. it is similar to package errgroup, in that it runs multiple tasks in parallel and returns the first error it encounters. Users define a Runner as a set vertices (functions) and edges between them. During Run, the directed acyclec graph will be validated and each vertex will run in parallel as soon as it's dependencies have been resolved. The Runner will only return after all running goroutines have stopped.

Index

• type Runner
  • func (d *Runner) AddEdge(from, to string)
  • func (d *Runner) AddVertex(name string, fn func() error)
  • func (d *Runner) Run() error

Examples

• Runner

Types

type Runner

type Runner struct {
	// contains filtered or unexported fields
}

Runner collects functions and arranges them as vertices and edges of a directed acyclic graph. Upon validation of the graph, functions are run in parallel topological order. The zero value is useful.

Example

package main

import (
	"errors"
	"fmt"

	"github.com/natessilva/dag"
)

func main() {
	var r dag.Runner

	r.AddVertex("one", func() error {
		fmt.Println("one and two will run in parallel before three")
		return nil
	})
	r.AddVertex("two", func() error {
		fmt.Println("one and two will run in parallel before three")
		return nil
	})
	r.AddVertex("three", func() error {
		fmt.Println("three will run before four")
		return errors.New("three is broken")
	})
	r.AddVertex("four", func() error {
		fmt.Println("four will never run")
		return nil
	})

	r.AddEdge("one", "three")
	r.AddEdge("two", "three")

	r.AddEdge("three", "four")

	fmt.Printf("the runner terminated with: %v\n", r.Run())
}

Output:

one and two will run in parallel before three
one and two will run in parallel before three
three will run before four
the runner terminated with: three is broken

func (*Runner) AddEdge

func (d *Runner) AddEdge(from, to string)

AddEdge establishes a dependency between two vertices in the graph. Both from and to must exist in the graph, or Run will err. The vertex at from will execute before the vertex at to.

func (*Runner) AddVertex

func (d *Runner) AddVertex(name string, fn func() error)

AddVertex adds a function as a vertex in the graph. Only functions which have been added in this way will be executed during Run.

func (*Runner) Run

func (d *Runner) Run() error

Run will validate that all edges in the graph point to existing vertices, and that there are no dependency cycles. After validation, each vertex will be run, deterministically, in parallel topological order. If any vertex returns an error, no more vertices will be scheduled and Run will exit and return that error once all in-flight functions finish execution.