mirror of
https://github.com/hanzoai/flow.git
synced 2026-10-17 01:29:18 +00:00
feat(core): the flow step/port contract — ONE parametric step-DAG engine
The seam the whole platform composes on. A stdlib-only Go module (core/) that
holds zero domain knowledge and is what every node-pack and execution backend
plugs into. flow is curried over two parameters:
Value type — what travels on an edge (Message / Tensor / connector I/O). The
engine never inspects it; a Step interprets it per its domain.
Mode — execution locality/durability (InProcess | GPUResident | Durable),
a PARAMETER not baked in: media flows keep latents pinned in VRAM
and serialize nothing; automations checkpoint every step on the
durable backend. Same scheduler, different Backend.
So every product is flow applied:
studio = Run(GPUResident, graph-of-Tensor-steps) // comfy nodes
automations = Run(Durable, graph-of-ConnectorIO) // triggers -> actions
llmFlow = Run(InProcess, graph-of-Message-steps) // langflow (lfx) nodes
Contract: Value (opaque, typed by PortType), Port/Step (typed in/out ports + Run),
Graph (Add/Wire/Validate — port typecheck + acyclicity), Mode + Backend (the
locality/durability seam), and Scheduler (the ONE topology walk, curried over
value-domain and Mode). Ships with the reference InProcess backend; GPUResident
(studio VRAM) and Durable (auto/tasks activities) register the same way.
Tests prove it: a domain-agnostic 2-step pipeline runs in order, port type
mismatch is rejected at wire time, AnyType bypasses, cycles are caught by
Validate, and Mode selects the backend over an identical scheduler walk.
Next: studio and langflow delete their private executors and become node-packs
on this scheduler; auto's tasks engine registers as the Durable backend.
This commit is contained in:
+25
@@ -0,0 +1,25 @@
|
||||
// Package flow is the ONE step-DAG engine the whole platform composes on.
|
||||
//
|
||||
// There is exactly one scheduler. Everything — LLM chains, media/diffusion
|
||||
// pipelines, IFTTT automations, agent tool-graphs — is this engine applied to
|
||||
// two parameters:
|
||||
//
|
||||
// Value type — what travels on an edge (a Message, a Tensor, connector I/O).
|
||||
// The engine NEVER inspects it; a Step interprets it per its domain.
|
||||
// Mode — execution locality/durability (InProcess | GPUResident | Durable).
|
||||
// A PARAMETER, never baked in: a media flow keeps latents pinned in
|
||||
// VRAM and serializes nothing; an automation checkpoints every step
|
||||
// on the durable backend. Same scheduler, different Backend.
|
||||
//
|
||||
// So a product is just flow curried:
|
||||
//
|
||||
// studio = Run(GPUResident, Graph-of-Tensor-steps) // comfy nodes
|
||||
// automations = Run(Durable, Graph-of-ConnectorIO) // triggers -> actions
|
||||
// llmFlow = Run(InProcess, Graph-of-Message-steps) // langflow (lfx) nodes
|
||||
//
|
||||
// A domain supplies Steps (typed input ports -> typed output ports + a Run fn) and,
|
||||
// if it has special execution needs, a Backend for a Mode. The engine only wires
|
||||
// matching ports and schedules them. This package is stdlib-only and holds no
|
||||
// domain knowledge — it is the seam every node-pack and every execution backend
|
||||
// plugs into.
|
||||
package flow
|
||||
+167
@@ -0,0 +1,167 @@
|
||||
package flow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// Value is an opaque typed value that travels along an edge. The engine never
|
||||
// inspects it; the receiving Step interprets it per its domain. PortType is the
|
||||
// tag the scheduler matches against a port's Type — the ONLY thing the engine
|
||||
// reads about a value.
|
||||
type Value interface {
|
||||
PortType() string
|
||||
}
|
||||
|
||||
// Port is a typed input or output slot on a Step. Two ports connect only when
|
||||
// their Type tags are equal (or one side is AnyType).
|
||||
type Port struct {
|
||||
Name string
|
||||
Type string
|
||||
}
|
||||
|
||||
// AnyType on either side of a wire disables the port type-check for that edge —
|
||||
// the escape hatch for a genuinely polymorphic step (a pass-through, a fan-in).
|
||||
const AnyType = "*"
|
||||
|
||||
// Inputs/Outputs carry a step's values keyed by port name.
|
||||
type (
|
||||
Inputs = map[string]Value
|
||||
Outputs = map[string]Value
|
||||
)
|
||||
|
||||
// Step is one node: typed input ports, typed output ports, and a Run that maps
|
||||
// inputs to outputs. Domains (LLM, tensor, connector, trigger) supply Steps; the
|
||||
// engine only schedules them. Run must be pure w.r.t. the graph — its only inputs
|
||||
// are `in` and ctx, its only outputs are the returned map and error. A Backend may
|
||||
// wrap Run (to pin VRAM, or to checkpoint for durability) — the Step is unaware.
|
||||
type Step interface {
|
||||
ID() string
|
||||
Ins() []Port
|
||||
Outs() []Port
|
||||
Run(ctx context.Context, in Inputs) (Outputs, error)
|
||||
}
|
||||
|
||||
// PortRef names one port on one step.
|
||||
type PortRef struct {
|
||||
Step string
|
||||
Port string
|
||||
}
|
||||
|
||||
// Edge wires an upstream output port to a downstream input port.
|
||||
type Edge struct {
|
||||
From PortRef // an Out port of From.Step
|
||||
To PortRef // an In port of To.Step
|
||||
}
|
||||
|
||||
// Graph is a DAG of steps wired output-port -> input-port. Build it with Add +
|
||||
// Wire, then Validate (port types + acyclicity) before handing it to a Scheduler.
|
||||
type Graph struct {
|
||||
steps map[string]Step
|
||||
edges []Edge
|
||||
}
|
||||
|
||||
// NewGraph returns an empty graph.
|
||||
func NewGraph() *Graph { return &Graph{steps: map[string]Step{}} }
|
||||
|
||||
// Add registers a step. A duplicate ID is an error — IDs are the graph's identity.
|
||||
func (g *Graph) Add(s Step) error {
|
||||
id := s.ID()
|
||||
if id == "" {
|
||||
return fmt.Errorf("flow: step has empty ID")
|
||||
}
|
||||
if _, dup := g.steps[id]; dup {
|
||||
return fmt.Errorf("flow: duplicate step ID %q", id)
|
||||
}
|
||||
g.steps[id] = s
|
||||
return nil
|
||||
}
|
||||
|
||||
// Wire connects an output port to an input port. Both endpoints must exist and
|
||||
// their port Types must be compatible (equal, or one side AnyType).
|
||||
func (g *Graph) Wire(from, to PortRef) error {
|
||||
out, err := g.port(from, false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
in, err := g.port(to, true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !typesMatch(out.Type, in.Type) {
|
||||
return fmt.Errorf("flow: type mismatch wiring %s.%s (%s) -> %s.%s (%s)",
|
||||
from.Step, from.Port, out.Type, to.Step, to.Port, in.Type)
|
||||
}
|
||||
g.edges = append(g.edges, Edge{From: from, To: to})
|
||||
return nil
|
||||
}
|
||||
|
||||
// Validate checks that every wired port exists and the graph is acyclic. A future
|
||||
// GPUResident/streaming mode may permit cycles; the base contract requires a DAG.
|
||||
func (g *Graph) Validate() error {
|
||||
for _, e := range g.edges {
|
||||
if _, err := g.port(e.From, false); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := g.port(e.To, true); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := g.topoOrder(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func typesMatch(a, b string) bool { return a == b || a == AnyType || b == AnyType }
|
||||
|
||||
// port resolves a PortRef to its Port, checking side (in vs out).
|
||||
func (g *Graph) port(r PortRef, input bool) (Port, error) {
|
||||
s, ok := g.steps[r.Step]
|
||||
if !ok {
|
||||
return Port{}, fmt.Errorf("flow: no step %q", r.Step)
|
||||
}
|
||||
ports := s.Outs()
|
||||
side := "output"
|
||||
if input {
|
||||
ports, side = s.Ins(), "input"
|
||||
}
|
||||
for _, p := range ports {
|
||||
if p.Name == r.Port {
|
||||
return p, nil
|
||||
}
|
||||
}
|
||||
return Port{}, fmt.Errorf("flow: step %q has no %s port %q", r.Step, side, r.Port)
|
||||
}
|
||||
|
||||
// topoOrder returns step IDs in dependency order (an edge From->To means From
|
||||
// runs before To), or an error naming the cycle. Kahn's algorithm — deterministic
|
||||
// by sorted ID so a graph always schedules the same way.
|
||||
func (g *Graph) topoOrder() ([]string, error) {
|
||||
indeg := make(map[string]int, len(g.steps))
|
||||
for id := range g.steps {
|
||||
indeg[id] = 0
|
||||
}
|
||||
succ := make(map[string][]string, len(g.steps))
|
||||
for _, e := range g.edges {
|
||||
succ[e.From.Step] = append(succ[e.From.Step], e.To.Step)
|
||||
indeg[e.To.Step]++
|
||||
}
|
||||
ready := sortedZero(indeg)
|
||||
order := make([]string, 0, len(g.steps))
|
||||
for len(ready) > 0 {
|
||||
id := ready[0]
|
||||
ready = ready[1:]
|
||||
order = append(order, id)
|
||||
for _, n := range succ[id] {
|
||||
indeg[n]--
|
||||
if indeg[n] == 0 {
|
||||
ready = insertSorted(ready, n)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(order) != len(g.steps) {
|
||||
return nil, fmt.Errorf("flow: graph has a cycle (%d of %d steps scheduled)", len(order), len(g.steps))
|
||||
}
|
||||
return order, nil
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package flow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A domain supplies its own Value type — the engine is agnostic to it. Here a
|
||||
// trivial "text" value stands in for a Message/Tensor/connector payload.
|
||||
type text string
|
||||
|
||||
func (text) PortType() string { return "text" }
|
||||
|
||||
// source emits a fixed text on port "out".
|
||||
type source struct {
|
||||
id string
|
||||
v text
|
||||
}
|
||||
|
||||
func (s source) ID() string { return s.id }
|
||||
func (s source) Ins() []Port { return nil }
|
||||
func (s source) Outs() []Port { return []Port{{Name: "out", Type: "text"}} }
|
||||
func (s source) Run(_ context.Context, _ Inputs) (Outputs, error) {
|
||||
return Outputs{"out": s.v}, nil
|
||||
}
|
||||
|
||||
// upper reads port "in" and emits its upper-cased text on port "out".
|
||||
type upper struct{ id string }
|
||||
|
||||
func (u upper) ID() string { return u.id }
|
||||
func (u upper) Ins() []Port { return []Port{{Name: "in", Type: "text"}} }
|
||||
func (u upper) Outs() []Port { return []Port{{Name: "out", Type: "text"}} }
|
||||
func (u upper) Run(_ context.Context, in Inputs) (Outputs, error) {
|
||||
return Outputs{"out": text(strings.ToUpper(string(in["in"].(text))))}, nil
|
||||
}
|
||||
|
||||
// wantNumber declares an input port of a DIFFERENT type, to exercise the wire typecheck.
|
||||
type wantNumber struct{ id string }
|
||||
|
||||
func (w wantNumber) ID() string { return w.id }
|
||||
func (w wantNumber) Ins() []Port { return []Port{{Name: "in", Type: "number"}} }
|
||||
func (w wantNumber) Outs() []Port { return nil }
|
||||
func (w wantNumber) Run(context.Context, Inputs) (Outputs, error) { return nil, nil }
|
||||
|
||||
// TestPipeline_RunsInProcess is the P0: two steps wired output->input run in
|
||||
// dependency order and the terminal step's output is returned — proving the ONE
|
||||
// scheduler drives a domain-agnostic graph with the reference backend.
|
||||
func TestPipeline_RunsInProcess(t *testing.T) {
|
||||
g := NewGraph()
|
||||
if err := g.Add(source{id: "src", v: "hello"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := g.Add(upper{id: "up"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := g.Wire(PortRef{"src", "out"}, PortRef{"up", "in"}); err != nil {
|
||||
t.Fatalf("wire: %v", err)
|
||||
}
|
||||
|
||||
out, err := New().Run(context.Background(), g, InProcess, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("run: %v", err)
|
||||
}
|
||||
// "up.out" is the only leaf (src.out feeds up, so it's not terminal).
|
||||
if got, ok := out["up.out"].(text); !ok || got != "HELLO" {
|
||||
t.Fatalf("leaf output = %v, want up.out=HELLO", out)
|
||||
}
|
||||
if _, leaked := out["src.out"]; leaked {
|
||||
t.Fatalf("non-terminal port src.out leaked into result: %v", out)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWire_TypeMismatchRejected: wiring text -> number fails at Wire, before any run.
|
||||
func TestWire_TypeMismatchRejected(t *testing.T) {
|
||||
g := NewGraph()
|
||||
_ = g.Add(source{id: "src", v: "x"})
|
||||
_ = g.Add(wantNumber{id: "n"})
|
||||
if err := g.Wire(PortRef{"src", "out"}, PortRef{"n", "in"}); err == nil {
|
||||
t.Fatal("expected type-mismatch error wiring text(out) -> number(in)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnyType_BypassesTypecheck: an AnyType port connects to anything.
|
||||
func TestAnyType_BypassesTypecheck(t *testing.T) {
|
||||
g := NewGraph()
|
||||
_ = g.Add(source{id: "src", v: "x"})
|
||||
_ = g.Add(passthrough{id: "p"})
|
||||
if err := g.Wire(PortRef{"src", "out"}, PortRef{"p", "in"}); err != nil {
|
||||
t.Fatalf("AnyType input should accept text: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
type passthrough struct{ id string }
|
||||
|
||||
func (p passthrough) ID() string { return p.id }
|
||||
func (p passthrough) Ins() []Port { return []Port{{Name: "in", Type: AnyType}} }
|
||||
func (p passthrough) Outs() []Port { return []Port{{Name: "out", Type: AnyType}} }
|
||||
func (p passthrough) Run(_ context.Context, in Inputs) (Outputs, error) {
|
||||
return Outputs{"out": in["in"]}, nil
|
||||
}
|
||||
|
||||
// TestValidate_RejectsCycle: a cycle is caught by Validate/topoOrder, not at runtime.
|
||||
func TestValidate_RejectsCycle(t *testing.T) {
|
||||
g := NewGraph()
|
||||
_ = g.Add(passthrough{id: "a"})
|
||||
_ = g.Add(passthrough{id: "b"})
|
||||
_ = g.Wire(PortRef{"a", "out"}, PortRef{"b", "in"})
|
||||
_ = g.Wire(PortRef{"b", "out"}, PortRef{"a", "in"})
|
||||
if err := g.Validate(); err == nil {
|
||||
t.Fatal("expected cycle detection error")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRun_ModeSelectsBackend: a custom backend for a mode is invoked, proving Mode
|
||||
// is just backend selection over the identical scheduler walk.
|
||||
func TestRun_ModeSelectsBackend(t *testing.T) {
|
||||
g := NewGraph()
|
||||
_ = g.Add(source{id: "src", v: "hi"})
|
||||
|
||||
spy := &countingBackend{mode: Durable}
|
||||
out, err := New(spy).Run(context.Background(), g, Durable, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("run: %v", err)
|
||||
}
|
||||
if spy.calls != 1 {
|
||||
t.Fatalf("durable backend Exec called %d times, want 1", spy.calls)
|
||||
}
|
||||
if got := out["src.out"].(text); got != "hi" {
|
||||
t.Fatalf("output = %q, want hi", got)
|
||||
}
|
||||
}
|
||||
|
||||
type countingBackend struct {
|
||||
mode Mode
|
||||
calls int
|
||||
}
|
||||
|
||||
func (c *countingBackend) Mode() Mode { return c.mode }
|
||||
func (c *countingBackend) Exec(ctx context.Context, s Step, in Inputs) (Outputs, error) {
|
||||
c.calls++
|
||||
return s.Run(ctx, in) // a real durable backend would checkpoint; the shape is identical
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
module github.com/hanzoai/flow/core
|
||||
|
||||
go 1.23
|
||||
@@ -0,0 +1,16 @@
|
||||
package flow
|
||||
|
||||
import "context"
|
||||
|
||||
// inProcess is the reference Backend: it runs a step by calling Step.Run directly,
|
||||
// values passed by reference in shared memory. No serialization, no checkpointing.
|
||||
// It is always registered (New seeds it) so the engine works out of the box; the
|
||||
// GPUResident and Durable backends live in their own packages (studio's VRAM
|
||||
// executor, auto/tasks' activity dispatcher) and register the same way.
|
||||
type inProcess struct{}
|
||||
|
||||
func (inProcess) Mode() Mode { return InProcess }
|
||||
|
||||
func (inProcess) Exec(ctx context.Context, s Step, in Inputs) (Outputs, error) {
|
||||
return s.Run(ctx, in)
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package flow
|
||||
|
||||
import "context"
|
||||
|
||||
// Mode is a graph's execution locality and durability. It is a PARAMETER of a run,
|
||||
// not a property of the engine — the single most important decomplection here.
|
||||
// The scheduler is identical across modes; Mode only selects the Backend that
|
||||
// realizes step execution and value hand-off.
|
||||
type Mode int
|
||||
|
||||
const (
|
||||
// InProcess: steps run in-process, values passed by reference in shared memory,
|
||||
// nothing serialized, no crash-resume. The default for LLM/agent chains.
|
||||
InProcess Mode = iota
|
||||
|
||||
// GPUResident: values (tensors/latents) stay pinned in device memory between
|
||||
// steps and are never serialized to disk. The mode that keeps a diffusion
|
||||
// pipeline fast — a Durable backend applied here would serialize a latent per
|
||||
// node and destroy throughput, which is exactly why Mode is a parameter.
|
||||
GPUResident
|
||||
|
||||
// Durable: every step is checkpointed so a run resumes after a crash; the
|
||||
// backend dispatches steps as activities on the durable workflow engine
|
||||
// (hanzoai/tasks). The mode that makes a triggered automation reliable.
|
||||
Durable
|
||||
)
|
||||
|
||||
func (m Mode) String() string {
|
||||
switch m {
|
||||
case InProcess:
|
||||
return "in-process"
|
||||
case GPUResident:
|
||||
return "gpu-resident"
|
||||
case Durable:
|
||||
return "durable"
|
||||
default:
|
||||
return "mode(?)"
|
||||
}
|
||||
}
|
||||
|
||||
// Backend realizes a Step's execution for one Mode. It is the seam where locality
|
||||
// and durability live: an InProcess backend just calls Step.Run; a GPUResident
|
||||
// backend pins device memory and reuses it across steps; a Durable backend records
|
||||
// the step as a checkpointed activity and replays on resume. A Backend holds NO
|
||||
// graph knowledge — the one Scheduler owns topology; a Backend owns only "how does
|
||||
// a single step actually run in this mode."
|
||||
type Backend interface {
|
||||
Mode() Mode
|
||||
Exec(ctx context.Context, s Step, in Inputs) (Outputs, error)
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package flow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// Scheduler is THE one scheduler. It runs any Graph in any Mode by delegating each
|
||||
// step's execution to the Mode's Backend. The topology walk — dependency order,
|
||||
// gathering each step's inputs from upstream output ports, propagating outputs — is
|
||||
// identical for every mode and every value domain. That single walk, curried over
|
||||
// (value-domain, Mode), is the whole platform's execution model.
|
||||
type Scheduler struct {
|
||||
backends map[Mode]Backend
|
||||
}
|
||||
|
||||
// New builds a scheduler over the given backends (one per Mode). InProcess is
|
||||
// always available as a fallback so a scheduler is never empty.
|
||||
func New(backends ...Backend) *Scheduler {
|
||||
s := &Scheduler{backends: map[Mode]Backend{InProcess: inProcess{}}}
|
||||
for _, b := range backends {
|
||||
s.backends[b.Mode()] = b
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// Run executes g in the given mode and returns the outputs of the graph's terminal
|
||||
// steps (steps whose output ports feed no edge), keyed "stepID.portName". The
|
||||
// caller supplies the graph's source inputs keyed the same way for unwired input
|
||||
// ports. Run validates first, so a malformed graph fails before any step executes.
|
||||
func (s *Scheduler) Run(ctx context.Context, g *Graph, mode Mode, in map[string]Value) (map[string]Value, error) {
|
||||
backend, ok := s.backends[mode]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("flow: no backend for %s", mode)
|
||||
}
|
||||
if err := g.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
order, err := g.topoOrder()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// wires[to] = the upstream (from) that feeds input port `to`.
|
||||
wires := make(map[PortRef]PortRef, len(g.edges))
|
||||
fed := make(map[PortRef]bool, len(g.edges)) // output ports that feed some edge
|
||||
for _, e := range g.edges {
|
||||
wires[e.To] = e.From
|
||||
fed[e.From] = true
|
||||
}
|
||||
|
||||
// produced[stepID.port] = value emitted by that output port.
|
||||
produced := make(map[string]Value)
|
||||
for _, id := range order {
|
||||
step := g.steps[id]
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stepIn := make(Inputs, len(step.Ins()))
|
||||
for _, p := range step.Ins() {
|
||||
ref := PortRef{Step: id, Port: p.Name}
|
||||
if up, wired := wires[ref]; wired {
|
||||
stepIn[p.Name] = produced[key(up)]
|
||||
} else if v, ok := in[key(ref)]; ok {
|
||||
stepIn[p.Name] = v // graph source input
|
||||
}
|
||||
}
|
||||
out, err := backend.Exec(ctx, step, stepIn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("flow: step %q: %w", id, err)
|
||||
}
|
||||
for _, p := range step.Outs() {
|
||||
if v, ok := out[p.Name]; ok {
|
||||
produced[key(PortRef{Step: id, Port: p.Name})] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Result = every produced output port that feeds no downstream edge (the leaves).
|
||||
result := make(map[string]Value)
|
||||
for _, id := range order {
|
||||
for _, p := range g.steps[id].Outs() {
|
||||
ref := PortRef{Step: id, Port: p.Name}
|
||||
if v, ok := produced[key(ref)]; ok && !fed[ref] {
|
||||
result[key(ref)] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func key(r PortRef) string { return r.Step + "." + r.Port }
|
||||
|
||||
// sortedZero returns the zero-indegree IDs in sorted order (deterministic scheduling).
|
||||
func sortedZero(indeg map[string]int) []string {
|
||||
var z []string
|
||||
for id, d := range indeg {
|
||||
if d == 0 {
|
||||
z = append(z, id)
|
||||
}
|
||||
}
|
||||
sort.Strings(z)
|
||||
return z
|
||||
}
|
||||
|
||||
// insertSorted keeps the ready queue sorted so topological order is deterministic.
|
||||
func insertSorted(q []string, id string) []string {
|
||||
i := sort.SearchStrings(q, id)
|
||||
q = append(q, "")
|
||||
copy(q[i+1:], q[i:])
|
||||
q[i] = id
|
||||
return q
|
||||
}
|
||||
Reference in New Issue
Block a user