dag

Documentation

Overview

Package dag offers a simple, pure in-memory job scheduler based on Directed Acyclic graph. Most common use cases are to schedule a bunch of interconnected job in a cron or cli command.

Each vertex stands for an arbitrary function to be scheduled and the edge between them describes their dependency. The scheduler will run each vertex in an independent goroutine as soon as all its dependencies are finished. Vertexes with no direct dependency may be scheduled concurrently. The scheduler will not run any vertex twice.

If a vertex returns an error or if the dag context is canceled, the scheduler will prevent any subsequent vertexes from scheduling, cancel all vertex level contexts and return to the caller immediately.

Index

• func IsAcyclic(g graph) (bool, []int)
• type DAG
  • func New() *DAG
  • func (d *DAG) AddEdge(from, to VertexID) error
  • func (d *DAG) AddEdges(edges Edges) error
  • func (d *DAG) AddVertex(work func(ctx context.Context) error, option ...VertexOption) VertexID
  • func (d *DAG) Run(ctx context.Context) error
• type Edges
• type VertexID
• type VertexOption
  • func WithLogger(logger log.Logger) VertexOption
  • func WithName(name string) VertexOption

Examples

• DAG (Run)

Functions

func IsAcyclic

func IsAcyclic(g graph) (bool, []int)

IsAcyclic uses depth-first search to find cycles in a generic graph represented by graph interface. If a cycle is found, it returns a list of nodes that are in the cyclic path, identified by their orders.

Types

type DAG

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

DAG is a directed acyclic graph designed for job scheduling.

Example (Run)

This example shows how to pass results to the next vertex or the dag caller.

package main

import (
	"context"
	"fmt"
	"github.com/DoNewsCode/core/control/dag"

	"github.com/DoNewsCode/core/ctxmeta"
)

func main() {
	d := dag.New()
	v1 := d.AddVertex(func(ctx context.Context) error {
		ctxmeta.GetBaggage(ctx).Set("v1Result", "foo")
		return nil
	})
	v2 := d.AddVertex(func(ctx context.Context) error {
		v1Result, _ := ctxmeta.GetBaggage(ctx).Get("v1Result")
		fmt.Println(v1Result)
		return nil
	})
	d.AddEdge(v1, v2)

	_, ctx := ctxmeta.Inject(context.Background())
	d.Run(ctx)

}

Output:

foo

func New

func New() *DAG

New creates a new DAG instance.

func (*DAG) AddEdge

func (d *DAG) AddEdge(from, to VertexID) error

AddEdge adds an edge to the dag. AddEdge is not concurrent safe. All vertexes and edges are expected to be added synchronously before calling Run.

If the new edge leads to a cycle, AddEdge will return error.

func (*DAG) AddEdges

func (d *DAG) AddEdges(edges Edges) error

AddEdges adds multiple edges in one batch. See AddEdge for more details.

func (*DAG) AddVertex

func (d *DAG) AddVertex(work func(ctx context.Context) error, option ...VertexOption) VertexID

AddVertex adds a vertex to the dag. AddVertex is not concurrent safe. All vertexes and edges are expected to be added synchronously before calling Run.

func (*DAG) Run

func (d *DAG) Run(ctx context.Context) error

Run runs the dag. Vertexes with no dependency will be scheduled concurrently while the inked vertexes will be scheduled sequentially. The Scheduler optimizes the execution path so that the overall dag execution time is minimized.

If a vertex returns an error or if the dag context is canceled, the scheduler will prevent any subsequent vertexes from scheduling, cancel all vertex level contexts and return to the caller immediately.

One of the ways for parent vertexes to pass results to child vertexes (or the dag caller) is to store the results in context with the help of package ctxmeta. See example.

type Edges

type Edges [][]VertexID

Edges is a group of edges that are connected to a vertex.

type VertexID

type VertexID int

VertexID is the identifier of Vertex in a directed acyclic graph.

type VertexOption

type VertexOption func(*vertex)

VertexOption is the type of options that can be passed to the AddVertex function.

func WithLogger

func WithLogger(logger log.Logger) VertexOption

WithLogger sets the logger for the vertex. The logger can be set to arbitrary log level before passing in.

func WithName

func WithName(name string) VertexOption

WithName sets the name of the vertex. The name is useful for debugging.