diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index c3d70df..70521de 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -158,6 +158,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err) } } + capacity, err := build.ParseCapacity(getenv("MESH_BUILDER_CAPACITY")) + if err != nil { + return fmt.Errorf("MESH_BUILDER_CAPACITY: %w", err) + } tools := &build.ToolchainSet{} presence := &build.PresenceSet{} b := &build.Builder{ @@ -165,6 +169,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Runner: build.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: bound}, Toolchains: tools, Presence: presence, + Places: &build.Places{Capacity: capacity}, Manifests: manifestReader{manifests}, Nodes: nodesPerFingerprint, Logf: logger.Printf, diff --git a/modules/mesh-delivery-builder/internal/build/choose.go b/modules/mesh-delivery-builder/internal/build/choose.go index 88e09d3..656684a 100644 --- a/modules/mesh-delivery-builder/internal/build/choose.go +++ b/modules/mesh-delivery-builder/internal/build/choose.go @@ -9,13 +9,14 @@ import ( ) // ChooseNodes picks up to want build nodes for a language (ADR 0478 clause 3): the nodes whose toolchains key reports the -// version the nodes agree on, the ones with the fewest steps the builder has running there first, a tie broken by the -// node's name. A node qualifies only when online is true of it (the watchman's presence of the node; nil means every +// version the nodes agree on, the ones with room under their build capacity first (places), then the fewest steps the +// builder has running there, a tie broken by the node's name. A node at its capacity is still chosen when the others are +// too (fewer than want have room): its steps wait for a place (Places.Acquire). A node qualifies only when online is true of it (the watchman's presence of the node; nil means every // node is). It answers the toolchain all the reporting nodes agree on, online or not, as the indexer computes it. When // nothing is reported, or the nodes report different versions (so the indexer agreed none), it fails naming the language // and why; when no reporting node is online it fails naming the nodes. A node whose build module refuses on its toolchain // refuses the step, and the build fails naming it. -func ChooseNodes(language string, entries []delivery.NodeToolchain, running map[string]int, online func(node string) bool, want int) ([]string, delivery.Toolchain2, error) { +func ChooseNodes(language string, entries []delivery.NodeToolchain, places *Places, online func(node string) bool, want int) ([]string, delivery.Toolchain2, error) { tc, none := delivery.AgreedToolchain(entries) if none != "" { return nil, delivery.Toolchain2{}, fmt.Errorf("no %s build node qualifies: %s", language, none) @@ -37,12 +38,10 @@ func ChooseNodes(language string, entries []delivery.NodeToolchain, running map[ sort.Strings(reporting) return nil, tc, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online", language, strings.Join(reporting, ", ")) } - sort.Slice(nodes, func(i, j int) bool { - if running[nodes[i]] != running[nodes[j]] { - return running[nodes[i]] < running[nodes[j]] - } - return nodes[i] < nodes[j] - }) + if places == nil { + places = &Places{} + } + places.order(nodes) // nodes with room first, then the fewest steps running, then by name if len(nodes) > want { nodes = nodes[:want] } diff --git a/modules/mesh-delivery-builder/internal/build/choose_test.go b/modules/mesh-delivery-builder/internal/build/choose_test.go index 7057f3f..88a6b53 100644 --- a/modules/mesh-delivery-builder/internal/build/choose_test.go +++ b/modules/mesh-delivery-builder/internal/build/choose_test.go @@ -17,7 +17,7 @@ func tcs(language, version string, nodes ...string) []delivery.NodeToolchain { func TestChooseTakesTheAgreedNodesWithFewestRunningStepsTiesByName(t *testing.T) { // 1607/AC4, ADR 0478 clause 3 entries := tcs("go", "go1.27.1", "novox", "ace", "g14", "shanks") - nodes, tc, err := ChooseNodes("go", entries, map[string]int{"novox": 3, "ace": 1, "g14": 1, "shanks": 0}, nil, 2) + nodes, tc, err := ChooseNodes("go", entries, placesRunning(map[string]int{"novox": 3, "ace": 1, "g14": 1, "shanks": 0}), nil, 2) if err != nil { t.Fatal(err) } @@ -65,7 +65,7 @@ func TestAgreeNeedsEqualDigestsAndNamesBothNodesWhenTheyDiffer(t *testing.T) { / func TestChooseSkipsNodesThatAreNotOnlineAndFailsNamingThemWhenNoneIs(t *testing.T) { // 1607 review 3, ADR 0478 clause 3 entries := tcs("go", "go1.27.1", "ace", "g14", "novox") online := func(n string) bool { return n != "g14" } // a sleeping laptop - nodes, tc, err := ChooseNodes("go", entries, map[string]int{"novox": 5}, online, 2) + nodes, tc, err := ChooseNodes("go", entries, placesRunning(map[string]int{"novox": 5}), online, 2) if err != nil || len(nodes) != 2 || nodes[0] != "ace" || nodes[1] != "novox" || tc.Version != "go1.27.1" { t.Errorf("nodes = %v, toolchain %+v, err %v, want [ace novox] with g14 left out although it runs nothing", nodes, tc, err) } @@ -77,3 +77,19 @@ func TestChooseSkipsNodesThatAreNotOnlineAndFailsNamingThemWhenNoneIs(t *testing t.Errorf("none online: the agreed toolchain %+v is still answered", tc) } } + +// placesRunning is Places with these steps counted as running, for a test of the choice. +func placesRunning(running map[string]int) *Places { + return &Places{running: running, changed: make(chan struct{})} +} + +func TestChoosePrefersNodesBelowTheirCapacity(t *testing.T) { // round 3 + entries := tcs("go", "go1.27.1", "ace", "novox", "shanks") + // shanks has capacity 3 with 2 running: room; ace has the default 1 with 1 running: none; novox has none running. + p := placesRunning(map[string]int{"shanks": 2, "ace": 1}) + p.Capacity = map[string]int{"shanks": 3} + nodes, _, err := ChooseNodes("go", entries, p, nil, 2) + if err != nil || len(nodes) != 2 || nodes[0] != "novox" || nodes[1] != "shanks" { + t.Errorf("nodes = %v, err %v, want [novox shanks]: the nodes with room, ace at capacity left out", nodes, err) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/places.go b/modules/mesh-delivery-builder/internal/build/places.go new file mode 100644 index 0000000..3ff6701 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/places.go @@ -0,0 +1,118 @@ +package build + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + "sync" +) + +// DefaultCapacity is the build capacity of a node that has no setting: one build step at a time. +const DefaultCapacity = 1 + +// ParseCapacity reads the builder's capacity setting: `node=steps` pairs separated by commas, such as +// `shanks=3,ace=2`. A node without a pair has DefaultCapacity. A pair with no node, a count that is not a whole number of +// at least 1, and a node named twice are refused. +func ParseCapacity(s string) (map[string]int, error) { + out := map[string]int{} + for _, pair := range strings.Split(s, ",") { + pair = strings.TrimSpace(pair) + if pair == "" { + continue + } + node, count, ok := strings.Cut(pair, "=") + node, count = strings.TrimSpace(node), strings.TrimSpace(count) + n, err := strconv.Atoi(count) + if !ok || node == "" || err != nil || n < 1 { + return nil, fmt.Errorf("the capacity %q is not =", pair) + } + if _, dup := out[node]; dup { + return nil, fmt.Errorf("the capacity names %s twice", node) + } + out[node] = n + } + return out, nil +} + +// Places counts the build steps the builder runs on each node and holds a step back while its node is at its +// capacity (the operator's decision of 2026-10-21; ADR 0478 clause 3). A step waits for a place by watching the ends of +// the other steps, not by a timer: ending a step wakes the waiting ones. +type Places struct { + // Capacity is the number of steps at once by node; a node not in it has DefaultCapacity. + Capacity map[string]int + + mu sync.Mutex + running map[string]int + changed chan struct{} // closed, and replaced, each time a step ends +} + +// CapacityOf is how many steps the builder runs on the node at once. +func (p *Places) CapacityOf(node string) int { + if n, ok := p.Capacity[node]; ok && n >= 1 { + return n + } + return DefaultCapacity +} + +// Running is the steps in flight now, by node. +func (p *Places) Running() map[string]int { + p.mu.Lock() + defer p.mu.Unlock() + out := make(map[string]int, len(p.running)) + for n, c := range p.running { + if c > 0 { + out[n] = c + } + } + return out +} + +// Acquire takes a place on the node, waiting while the node is at its capacity, and answers the function that gives it +// back. It returns an error when ctx ends first. +func (p *Places) Acquire(ctx context.Context, node string) (release func(), err error) { + for { + p.mu.Lock() + if p.running == nil { + p.running = map[string]int{} + p.changed = make(chan struct{}) + } + if p.running[node] < p.CapacityOf(node) { + p.running[node]++ + p.mu.Unlock() + var once sync.Once + return func() { + once.Do(func() { + p.mu.Lock() + p.running[node]-- + close(p.changed) + p.changed = make(chan struct{}) + p.mu.Unlock() + }) + }, nil + } + wait := p.changed + p.mu.Unlock() + select { + case <-wait: + case <-ctx.Done(): + return nil, fmt.Errorf("waiting for a place on %s: %w", node, ctx.Err()) + } + } +} + +// order orders the nodes for a choice: those under their capacity first, then fewest running, then by name. +func (p *Places) order(nodes []string) { + running := p.Running() + sort.SliceStable(nodes, func(i, j int) bool { + ri, rj := running[nodes[i]] < p.CapacityOf(nodes[i]), running[nodes[j]] < p.CapacityOf(nodes[j]) + if ri != rj { + return ri + } + if running[nodes[i]] != running[nodes[j]] { + return running[nodes[i]] < running[nodes[j]] + } + return nodes[i] < nodes[j] + }) +} diff --git a/modules/mesh-delivery-builder/internal/build/places_test.go b/modules/mesh-delivery-builder/internal/build/places_test.go new file mode 100644 index 0000000..5cb3d24 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/places_test.go @@ -0,0 +1,140 @@ +package build + +import ( + "context" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +func TestParseCapacityReadsNodeEqualsCountPairs(t *testing.T) { + c, err := ParseCapacity(" shanks=3, ace=2 ") + if err != nil || c["shanks"] != 3 || c["ace"] != 2 || len(c) != 2 { + t.Errorf("capacity = %v, %v", c, err) + } + if c, err := ParseCapacity(""); err != nil || len(c) != 0 { + t.Errorf("empty: %v, %v", c, err) + } + for _, bad := range []string{"shanks", "shanks=0", "shanks=x", "=2", "a=1,a=2", "shanks=-1"} { + if _, err := ParseCapacity(bad); err == nil { + t.Errorf("%q accepted", bad) + } + } +} + +func acquireAsync(p *Places, node string) chan func() { + got := make(chan func(), 1) + go func() { + release, err := p.Acquire(context.Background(), node) + if err != nil { + close(got) + return + } + got <- release + }() + return got +} + +func TestThreeStepsOnANodeOfCapacityTwoRunTwoAtOnceAndTheThirdStartsWhenOneEnds(t *testing.T) { // round 3 + p := &Places{Capacity: map[string]int{"ace": 2}} + a, b := acquireAsync(p, "ace"), acquireAsync(p, "ace") + ra, rb := <-a, <-b + third := acquireAsync(p, "ace") + select { + case <-third: + t.Fatal("a third step started on a node of capacity 2 with two running") + case <-time.After(50 * time.Millisecond): + } + if p.Running()["ace"] != 2 { + t.Errorf("running = %v, want 2 on ace", p.Running()) + } + ra() + select { + case r := <-third: + r() + case <-time.After(2 * time.Second): + t.Fatal("the third step did not start when one ended") + } + rb() +} + +func TestANodeWithNoSettingRunsOneStepAtATime(t *testing.T) { // round 3 + p := &Places{Capacity: map[string]int{"shanks": 3}} + first := <-acquireAsync(p, "g14") + second := acquireAsync(p, "g14") + select { + case <-second: + t.Fatal("two steps on a node without a setting") + case <-time.After(50 * time.Millisecond): + } + first() + r := <-second + r() + if p.CapacityOf("g14") != 1 || p.CapacityOf("shanks") != 3 { + t.Errorf("capacities = %d, %d", p.CapacityOf("g14"), p.CapacityOf("shanks")) + } +} + +func TestAWaitingStepStopsWhenItsContextEnds(t *testing.T) { // round 3 + p := &Places{} + r := <-acquireAsync(p, "ace") + defer r() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { _, err := p.Acquire(ctx, "ace"); done <- err }() + cancel() + select { + case err := <-done: + if err == nil { + t.Error("Acquire returned no error for a cancelled wait") + } + case <-time.After(2 * time.Second): + t.Fatal("a cancelled wait did not return") + } +} + +func TestStepsOfSeveralBuildsNeverExceedANodesCapacity(t *testing.T) { // round 3, through the builder + r := newRig(t, catalogue.Artifact{Name: "tool", Kind: "bundle", Language: "go"}) + r.b.Places = &Places{Capacity: map[string]int{"ace": 2}} // novox has none: one at a time + var now, most atomic.Int32 + var mu sync.Mutex + perNode := map[string]int{} + r.bm.steps.during = func(key string) func() { + node, _, _ := strings.Cut(key, ".") + mu.Lock() + perNode[node]++ + if int32(perNode[node]) > most.Load() && node == "ace" { + most.Store(int32(perNode[node])) + } + if node == "novox" && perNode[node] > 1 { + t.Errorf("%d steps at once on novox, which has capacity 1", perNode[node]) + } + mu.Unlock() + now.Add(1) + time.Sleep(15 * time.Millisecond) + return func() { mu.Lock(); perNode[node]--; mu.Unlock(); now.Add(-1) } + } + for _, m := range []string{"m1", "m2", "m3", "m4"} { + r.b.Module(context.Background(), m, moduleValue(t, "modules/"+m)) + } + r.b.Wait() + if most.Load() > 2 { + t.Errorf("%d steps at once on ace, which has capacity 2", most.Load()) + } + if most.Load() < 2 { + t.Errorf("at most %d steps at once on ace: the capacity was not used", most.Load()) + } +} + +func TestTwoNodeAgreementStillTakesTwoDifferentNodesWhenCapacitiesAreSet(t *testing.T) { // round 3, ADR 0478 clause 4 + r := newRig(t, catalogue.Artifact{Name: "tool", Kind: "bundle", Language: "go"}) + r.b.Places = &Places{Capacity: map[string]int{"ace": 2, "novox": 3}} + v := r.consider(t, "modules/m") + if v.State != "done" || !strings.Contains(v.Step, "ace and novox") { + t.Errorf("build = %+v, want done on ace and novox", v) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 4fad468..cd25c70 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -31,6 +31,8 @@ type Builder struct { // Presence says which nodes are online (the watchman's); a node that is not is not chosen. Presence interface{ Online(node string) bool } Manifests Manifests + // Places holds a step back while its node is at its build capacity; a Builder with none runs one step at a time on each node. + Places *Places // Nodes is how many build nodes build each fingerprint: two until release R4e (ADR 0478 clause 4). Nodes int // Limit is how many builds run at once; four when 0. @@ -44,7 +46,6 @@ type Builder struct { known map[string]delivery.Module // by modules key inflight map[string]bool // by build key again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built - running map[string]int // steps in flight, by `/` plans map[string]planned // by manifest digest } @@ -60,7 +61,10 @@ func (b *Builder) init() { limit = 4 } b.sem = make(chan struct{}, limit) - b.known, b.inflight, b.again, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]int{}, map[string]planned{} + b.known, b.inflight, b.again, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{} + if b.Places == nil { + b.Places = &Places{} + } }) } @@ -261,7 +265,7 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p var steps []string for _, bm := range plan.Modules() { language := LanguageOf[bm] - nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.runningOn(bm), b.online, b.Nodes) + nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.Places, b.online, b.Nodes) if err != nil { return failed("%s", err) } @@ -320,25 +324,13 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p func (b *Builder) online(node string) bool { return b.Presence == nil || b.Presence.Online(node) } -func (b *Builder) runningOn(buildModule string) map[string]int { - b.mu.Lock() - defer b.mu.Unlock() - out := map[string]int{} - for k, n := range b.running { - if mod, node, _ := strings.Cut(k, "/"); mod == buildModule { - out[node] = n - } - } - return out -} - -// step runs one step of a build module, counting it as running on the node while it does. +// step runs one step of a build module, once its node has a place for it. func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) { - k := bm + "/" + node - b.mu.Lock() - b.running[k]++ - b.mu.Unlock() - defer func() { b.mu.Lock(); b.running[k]--; b.mu.Unlock() }() + release, err := b.Places.Acquire(ctx, node) + if err != nil { + return nil, err + } + defer release() return b.Runner.Run(ctx, bm, node, toolPrefix[bm]+tool, args) } diff --git a/modules/mesh-delivery-builder/internal/build/steps_test.go b/modules/mesh-delivery-builder/internal/build/steps_test.go index 4a39fb9..4a0503b 100644 --- a/modules/mesh-delivery-builder/internal/build/steps_test.go +++ b/modules/mesh-delivery-builder/internal/build/steps_test.go @@ -36,13 +36,19 @@ func (f *fakeCaller) CallOn(node, key string, body any) (json.RawMessage, error) type fakeSteps struct { mu sync.Mutex steps map[string]delivery.Step + // during, when set, is called as a wait starts and the function it answers as the wait ends. + during func(key string) func() } func (f *fakeSteps) Wait(ctx context.Context, module, key string) (delivery.Step, error) { f.mu.Lock() s, ok := f.steps[module+"/"+key] + during := f.during f.mu.Unlock() if ok { + if during != nil { + defer during(key)() + } return s, nil } <-ctx.Done() diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index d957467..87781bc 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -123,7 +123,9 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1 t.Errorf("process = %v", p) } env, _ := p["env"].(map[string]any) - if len(env) != 1 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" { + // The build capacity of each node: how many build steps the builder runs on it at once (a node not named runs one). No + // part of delivery reads a per-node setting yet, so the value is the manifest's own; changing it is a change of this line. + if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_BUILDER_CAPACITY"] != "shanks=3,ace=2" { t.Errorf("env = %v", p["env"]) } if _, bus := p["bus"]; bus { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 1a902a2..a886fbe 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -53,7 +53,8 @@ "./mesh-delivery-builder" ], "env": { - "MESH_BROKER_FILE": "${dir:state}/broker" + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BUILDER_CAPACITY": "shanks=3,ace=2" }, "user": "mesh-delivery-builder", "health": {