Add deterministic dependency scheduler
This commit is contained in:
200
internal/engine/dependency_order.go
Normal file
200
internal/engine/dependency_order.go
Normal file
@@ -0,0 +1,200 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"container/heap"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// dependencyKey is an immutable tuple used to order dependency-graph nodes.
|
||||
// Callers provide unique keys and valid edge endpoints.
|
||||
type dependencyKey []string
|
||||
|
||||
// dependencyEdge points from a prerequisite to the node that depends on it.
|
||||
type dependencyEdge struct {
|
||||
prerequisite int
|
||||
dependent int
|
||||
}
|
||||
|
||||
type dependencyReadyHeap struct {
|
||||
nodes []int
|
||||
keys []dependencyKey
|
||||
}
|
||||
|
||||
func (h dependencyReadyHeap) Len() int { return len(h.nodes) }
|
||||
|
||||
func (h dependencyReadyHeap) Less(i, j int) bool {
|
||||
return dependencyKeyLess(h.keys[h.nodes[i]], h.keys[h.nodes[j]])
|
||||
}
|
||||
|
||||
func (h dependencyReadyHeap) Swap(i, j int) {
|
||||
h.nodes[i], h.nodes[j] = h.nodes[j], h.nodes[i]
|
||||
}
|
||||
|
||||
func (h *dependencyReadyHeap) Push(value any) {
|
||||
h.nodes = append(h.nodes, value.(int))
|
||||
}
|
||||
|
||||
func (h *dependencyReadyHeap) Pop() any {
|
||||
last := len(h.nodes) - 1
|
||||
node := h.nodes[last]
|
||||
h.nodes = h.nodes[:last]
|
||||
return node
|
||||
}
|
||||
|
||||
// dependencyKeyLess compares tuple components bytewise, with a shorter tuple
|
||||
// sorting first when all of its components match the corresponding prefix.
|
||||
func dependencyKeyLess(left, right dependencyKey) bool {
|
||||
for i := 0; i < len(left) && i < len(right); i++ {
|
||||
if left[i] != right[i] {
|
||||
return left[i] < right[i]
|
||||
}
|
||||
}
|
||||
return len(left) < len(right)
|
||||
}
|
||||
|
||||
// scheduleDependencies applies Kahn's algorithm, always selecting the
|
||||
// smallest currently-ready key. Duplicate edges count only once. If scheduling
|
||||
// stalls, an explicit depth-first traversal of the residual graph returns an
|
||||
// actual directed cycle, rotated to its smallest key and closed by repeating
|
||||
// its first node; blocked non-cycle dependents are excluded.
|
||||
//
|
||||
// Nodes are visited by precomputed key rank, which also makes each adjacency
|
||||
// list deterministic without sorting edges. The work is O(E + V log V), with
|
||||
// O(V + E) additional memory, apart from key copying and tuple comparisons.
|
||||
func scheduleDependencies(keys []dependencyKey, edges []dependencyEdge) (order, cycle []int) {
|
||||
ownedKeys := make([]dependencyKey, len(keys))
|
||||
for i, key := range keys {
|
||||
ownedKeys[i] = append(dependencyKey(nil), key...)
|
||||
}
|
||||
|
||||
nodeOrder := make([]int, len(ownedKeys))
|
||||
for i := range nodeOrder {
|
||||
nodeOrder[i] = i
|
||||
}
|
||||
sort.Slice(nodeOrder, func(i, j int) bool {
|
||||
return dependencyKeyLess(ownedKeys[nodeOrder[i]], ownedKeys[nodeOrder[j]])
|
||||
})
|
||||
|
||||
incoming := make([][]int, len(ownedKeys))
|
||||
for _, edge := range edges {
|
||||
incoming[edge.dependent] = append(incoming[edge.dependent], edge.prerequisite)
|
||||
}
|
||||
|
||||
outgoing := make([][]int, len(ownedKeys))
|
||||
indegree := make([]int, len(ownedKeys))
|
||||
for _, dependent := range nodeOrder {
|
||||
seen := make(map[int]struct{}, len(incoming[dependent]))
|
||||
for _, prerequisite := range incoming[dependent] {
|
||||
if _, duplicate := seen[prerequisite]; duplicate {
|
||||
continue
|
||||
}
|
||||
seen[prerequisite] = struct{}{}
|
||||
indegree[dependent]++
|
||||
// Dependents are visited by key rank, so each outgoing list is
|
||||
// already ordered by key.
|
||||
outgoing[prerequisite] = append(outgoing[prerequisite], dependent)
|
||||
}
|
||||
}
|
||||
|
||||
ready := dependencyReadyHeap{keys: ownedKeys}
|
||||
for _, node := range nodeOrder {
|
||||
if indegree[node] == 0 {
|
||||
ready.nodes = append(ready.nodes, node)
|
||||
}
|
||||
}
|
||||
heap.Init(&ready)
|
||||
|
||||
order = make([]int, 0, len(ownedKeys))
|
||||
for ready.Len() != 0 {
|
||||
node := heap.Pop(&ready).(int)
|
||||
order = append(order, node)
|
||||
for _, dependent := range outgoing[node] {
|
||||
indegree[dependent]--
|
||||
if indegree[dependent] == 0 {
|
||||
heap.Push(&ready, dependent)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(order) == len(ownedKeys) {
|
||||
return order, nil
|
||||
}
|
||||
return nil, dependencyCycle(ownedKeys, nodeOrder, outgoing, indegree)
|
||||
}
|
||||
|
||||
type dependencyDFSFrame struct {
|
||||
node int
|
||||
next int
|
||||
}
|
||||
|
||||
// dependencyCycle searches the residual graph iteratively to avoid recursion
|
||||
// depth depending on input size.
|
||||
func dependencyCycle(keys []dependencyKey, nodeOrder []int, outgoing [][]int, indegree []int) []int {
|
||||
const (
|
||||
unvisited = uint8(0)
|
||||
active = uint8(1)
|
||||
finished = uint8(2)
|
||||
)
|
||||
|
||||
state := make([]uint8, len(keys))
|
||||
position := make([]int, len(keys))
|
||||
for i := range position {
|
||||
position[i] = -1
|
||||
}
|
||||
path := make([]int, 0, len(keys))
|
||||
frames := make([]dependencyDFSFrame, 0, len(keys))
|
||||
|
||||
for _, root := range nodeOrder {
|
||||
if indegree[root] == 0 || state[root] != unvisited {
|
||||
continue
|
||||
}
|
||||
state[root] = active
|
||||
position[root] = len(path)
|
||||
path = append(path, root)
|
||||
frames = append(frames, dependencyDFSFrame{node: root})
|
||||
|
||||
for len(frames) != 0 {
|
||||
frame := &frames[len(frames)-1]
|
||||
if frame.next == len(outgoing[frame.node]) {
|
||||
state[frame.node] = finished
|
||||
position[frame.node] = -1
|
||||
path = path[:len(path)-1]
|
||||
frames = frames[:len(frames)-1]
|
||||
continue
|
||||
}
|
||||
|
||||
next := outgoing[frame.node][frame.next]
|
||||
frame.next++
|
||||
if indegree[next] == 0 {
|
||||
continue
|
||||
}
|
||||
switch state[next] {
|
||||
case unvisited:
|
||||
state[next] = active
|
||||
position[next] = len(path)
|
||||
path = append(path, next)
|
||||
frames = append(frames, dependencyDFSFrame{node: next})
|
||||
case active:
|
||||
return canonicalDependencyCycle(keys, path[position[next]:])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A stalled Kahn traversal always leaves a cycle in the residual graph.
|
||||
return nil
|
||||
}
|
||||
|
||||
func canonicalDependencyCycle(keys []dependencyKey, path []int) []int {
|
||||
start := 0
|
||||
for i := 1; i < len(path); i++ {
|
||||
if dependencyKeyLess(keys[path[i]], keys[path[start]]) {
|
||||
start = i
|
||||
}
|
||||
}
|
||||
|
||||
cycle := make([]int, len(path)+1)
|
||||
for i := range path {
|
||||
cycle[i] = path[(start+i)%len(path)]
|
||||
}
|
||||
cycle[len(path)] = cycle[0]
|
||||
return cycle
|
||||
}
|
||||
215
internal/engine/dependency_order_test.go
Normal file
215
internal/engine/dependency_order_test.go
Normal file
@@ -0,0 +1,215 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestScheduleDependenciesOrdersReadyNodes(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
keys []string
|
||||
edges [][2]string
|
||||
want []string
|
||||
}{
|
||||
{
|
||||
name: "independent branches",
|
||||
keys: []string{"branch-c", "branch-a", "branch-b"},
|
||||
want: []string{"branch-a", "branch-b", "branch-c"},
|
||||
},
|
||||
{
|
||||
name: "unequal chains",
|
||||
keys: []string{"chain-d", "chain-c", "chain-b", "chain-a"},
|
||||
edges: [][2]string{
|
||||
{"chain-a", "chain-b"},
|
||||
{"chain-b", "chain-c"},
|
||||
{"chain-c", "chain-d"},
|
||||
},
|
||||
want: []string{"chain-a", "chain-b", "chain-c", "chain-d"},
|
||||
},
|
||||
{
|
||||
name: "multiple dependencies",
|
||||
keys: []string{"application", "right", "base", "left"},
|
||||
edges: [][2]string{
|
||||
{"base", "left"}, {"base", "right"},
|
||||
{"left", "application"}, {"right", "application"},
|
||||
},
|
||||
want: []string{"base", "left", "right", "application"},
|
||||
},
|
||||
{
|
||||
name: "shared dependency at equal depths",
|
||||
keys: []string{"sink", "right", "root", "left"},
|
||||
edges: [][2]string{
|
||||
{"root", "left"}, {"root", "right"},
|
||||
{"left", "sink"}, {"right", "sink"},
|
||||
},
|
||||
want: []string{"root", "left", "right", "sink"},
|
||||
},
|
||||
{
|
||||
name: "shared dependency at unequal depths",
|
||||
keys: []string{"sink", "deep", "side", "middle", "root"},
|
||||
edges: [][2]string{
|
||||
{"root", "middle"}, {"middle", "deep"}, {"root", "side"},
|
||||
{"deep", "sink"}, {"side", "sink"},
|
||||
},
|
||||
want: []string{"root", "middle", "deep", "side", "sink"},
|
||||
},
|
||||
{
|
||||
name: "newly ready small key precedes older large key",
|
||||
keys: []string{"z-ready", "a-new", "b-ready"},
|
||||
edges: [][2]string{{"b-ready", "a-new"}},
|
||||
want: []string{"b-ready", "a-new", "z-ready"},
|
||||
},
|
||||
{
|
||||
name: "duplicate edges count once",
|
||||
keys: []string{"dependent", "prerequisite"},
|
||||
edges: [][2]string{
|
||||
{"prerequisite", "dependent"},
|
||||
{"prerequisite", "dependent"},
|
||||
},
|
||||
want: []string{"prerequisite", "dependent"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
keys, edges, indexes := dependencyGraph(test.keys, test.edges)
|
||||
order, cycle := scheduleDependencies(keys, edges)
|
||||
if len(cycle) != 0 {
|
||||
t.Fatalf("unexpected cycle: %v", dependencyNames(cycle, indexes))
|
||||
}
|
||||
if got := dependencyNames(order, indexes); !reflect.DeepEqual(got, test.want) {
|
||||
t.Fatalf("schedule = %v, want %v", got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduleDependenciesUsesTupleKeys(t *testing.T) {
|
||||
keys := []dependencyKey{
|
||||
{"a", "z"},
|
||||
{"a"},
|
||||
{"a", "aa"},
|
||||
{"A"},
|
||||
}
|
||||
order, cycle := scheduleDependencies(keys, nil)
|
||||
if len(cycle) != 0 {
|
||||
t.Fatalf("unexpected cycle: %v", cycle)
|
||||
}
|
||||
want := []int{3, 1, 2, 0}
|
||||
if !reflect.DeepEqual(order, want) {
|
||||
t.Fatalf("schedule = %v, want %v", order, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduleDependenciesReportsActualCanonicalCycles(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
keys []string
|
||||
edges [][2]string
|
||||
want []string
|
||||
}{
|
||||
{
|
||||
name: "self cycle",
|
||||
keys: []string{"self"},
|
||||
edges: [][2]string{{"self", "self"}},
|
||||
want: []string{"self", "self"},
|
||||
},
|
||||
{
|
||||
name: "multi-node cycle excludes blocked dependent",
|
||||
keys: []string{"blocked", "cycle-b", "cycle-a"},
|
||||
edges: [][2]string{
|
||||
{"cycle-a", "cycle-b"},
|
||||
{"cycle-b", "cycle-a"},
|
||||
{"cycle-b", "blocked"},
|
||||
},
|
||||
want: []string{"cycle-a", "cycle-b", "cycle-a"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
keys, edges, indexes := dependencyGraph(test.keys, test.edges)
|
||||
order, cycle := scheduleDependencies(keys, edges)
|
||||
if len(order) != 0 {
|
||||
t.Fatalf("unexpected schedule: %v", dependencyNames(order, indexes))
|
||||
}
|
||||
if got := dependencyNames(cycle, indexes); !reflect.DeepEqual(got, test.want) {
|
||||
t.Fatalf("cycle = %v, want %v", got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestScheduleDependenciesIsStableAcrossInsertionPermutations(t *testing.T) {
|
||||
edges := [][2]string{
|
||||
{"base", "left"}, {"base", "right"},
|
||||
{"left", "sink"}, {"right", "sink"},
|
||||
}
|
||||
permutations := [][]string{
|
||||
{"sink", "right", "base", "left"},
|
||||
{"left", "base", "sink", "right"},
|
||||
{"right", "left", "base", "sink"},
|
||||
}
|
||||
for i, permutation := range permutations {
|
||||
keys, permutedEdges, indexes := dependencyGraph(permutation, edges)
|
||||
if i%2 != 0 {
|
||||
reverseDependencyEdges(permutedEdges)
|
||||
}
|
||||
order, cycle := scheduleDependencies(keys, permutedEdges)
|
||||
if len(cycle) != 0 {
|
||||
t.Fatalf("permutation %d returned cycle %v", i, dependencyNames(cycle, indexes))
|
||||
}
|
||||
if got, want := dependencyNames(order, indexes), []string{"base", "left", "right", "sink"}; !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("permutation %d schedule = %v, want %v", i, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
cycleEdges := [][2]string{{"b", "a"}, {"a", "b"}, {"b", "blocked"}}
|
||||
cyclePermutations := [][]string{{"blocked", "b", "a"}, {"a", "blocked", "b"}, {"b", "a", "blocked"}}
|
||||
for i, permutation := range cyclePermutations {
|
||||
keys, permutedEdges, indexes := dependencyGraph(permutation, cycleEdges)
|
||||
if i%2 != 0 {
|
||||
reverseDependencyEdges(permutedEdges)
|
||||
}
|
||||
order, cycle := scheduleDependencies(keys, permutedEdges)
|
||||
if len(order) != 0 {
|
||||
t.Fatalf("cycle permutation %d returned schedule %v", i, dependencyNames(order, indexes))
|
||||
}
|
||||
if got, want := dependencyNames(cycle, indexes), []string{"a", "b", "a"}; !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("cycle permutation %d returned %v, want %v", i, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func dependencyGraph(nodeNames []string, namedEdges [][2]string) ([]dependencyKey, []dependencyEdge, map[int]string) {
|
||||
keys := make([]dependencyKey, len(nodeNames))
|
||||
indexes := make(map[string]int, len(nodeNames))
|
||||
for i, name := range nodeNames {
|
||||
keys[i] = dependencyKey{name}
|
||||
indexes[name] = i
|
||||
}
|
||||
edges := make([]dependencyEdge, len(namedEdges))
|
||||
for i, edge := range namedEdges {
|
||||
edges[i] = dependencyEdge{prerequisite: indexes[edge[0]], dependent: indexes[edge[1]]}
|
||||
}
|
||||
byIndex := make(map[int]string, len(nodeNames))
|
||||
for name, index := range indexes {
|
||||
byIndex[index] = name
|
||||
}
|
||||
return keys, edges, byIndex
|
||||
}
|
||||
|
||||
func dependencyNames(nodes []int, names map[int]string) []string {
|
||||
result := make([]string, len(nodes))
|
||||
for i, node := range nodes {
|
||||
result[i] = names[node]
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func reverseDependencyEdges(edges []dependencyEdge) {
|
||||
for i, j := 0, len(edges)-1; i < j; i, j = i+1, j-1 {
|
||||
edges[i], edges[j] = edges[j], edges[i]
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user