From 545a172bbf8ae81de4c59a3c730226e4e5f3691a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 01:26:50 +0200 Subject: [PATCH] mesh-build-go: bound the processes of an ask's levels together, the processors divided by the build capacity, through lock files in its workspace, now that the levels run side by side (issue 1921, try speed S10) --- .../cmd/mesh-build-go/integration.go | 14 +++- .../mesh-build-go/integration_level_test.go | 11 ++- .../mesh-build-go/cmd/mesh-build-go/level.go | 6 +- .../mesh-build-go/cmd/mesh-build-go/lint.go | 27 +++++-- .../mesh-build-go/cmd/mesh-build-go/slots.go | 74 +++++++++++++++++++ .../cmd/mesh-build-go/slots_test.go | 61 +++++++++++++++ .../mesh-build-go/cmd/mesh-build-go/test.go | 32 ++++++-- modules/mesh-build-go/go.mod | 2 +- modules/mesh-build-go/go.sum | 2 + 9 files changed, 206 insertions(+), 23 deletions(-) create mode 100644 modules/mesh-build-go/cmd/mesh-build-go/slots.go create mode 100644 modules/mesh-build-go/cmd/mesh-build-go/slots_test.go diff --git a/modules/mesh-build-go/cmd/mesh-build-go/integration.go b/modules/mesh-build-go/cmd/mesh-build-go/integration.go index 0edcfb6a..116797c9 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/integration.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/integration.go @@ -114,16 +114,21 @@ func (m *Module) integrationHere(ctx context.Context, workspace, tree string, pa defer guard() nats, stopBus, busErr = m.startBus(ctx, filepath.Join(run, "nats")) }() - slots := make(chan struct{}, m.parallel()) + slots := m.slots(workspace) // shared with the ask's unit level (issue 1921, try speed S10) + var slotErr error for _, d := range dirs { if len(byDir[d]) == 0 { continue } + release, err := slots.acquire(ctx) + if err != nil { + slotErr = err + break + } wg.Add(1) - slots <- struct{}{} go func() { defer wg.Done() - defer func() { <-slots }() + defer release() defer guard() m.Run(ctx, m.goEnvFor(filepath.Join(tree, d)), "go", append([]string{"-C", filepath.Join(tree, d), "test", "-tags", strings.Join(integrationTags, ","), "-exec", "true", "-count=1"}, byDir[d]...)...) }() @@ -138,6 +143,9 @@ func (m *Module) integrationHere(ctx context.Context, workspace, tree string, pa if crash != nil { panic(crash) } + if slotErr != nil { + return nil, slotErr + } if storeErr != nil { return nil, fmt.Errorf("the integration level's store: %w", storeErr) } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go b/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go index 1c231392..8a791627 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go @@ -345,8 +345,15 @@ func TestAStoreThatDoesNotStartEndsTheStepFailedAndLeavesNothing(t *testing.T) { if f.busDir != "" && !f.busDown { t.Errorf("the bus, started beside the store, was not stopped after the store failed") } - if entries, _ := os.ReadDir(filepath.Join(m.Work, "ws1")); len(entries) != 1 || entries[0].Name() != "tree" { - t.Errorf("the workspace holds %v, want only its tree", entries) + var left []string + entries, _ := os.ReadDir(filepath.Join(m.Work, "ws1")) + for _, e := range entries { + if e.Name() != "slots" { // the ask's bound on processes, gone with the workspace (issue 1921, S10) + left = append(left, e.Name()) + } + } + if len(left) != 1 || left[0] != "tree" { + t.Errorf("the workspace holds %v, want only its tree", left) } } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/level.go b/modules/mesh-build-go/cmd/mesh-build-go/level.go index 6ff18aa6..2d545259 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/level.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/level.go @@ -41,6 +41,8 @@ type levelSpec struct { Run string `json:"run,omitempty"` Setup time.Duration `json:"setup"` Bound time.Duration `json:"bound"` + // Parallel is the ask's bound on processes, which the process shares with the ask's other level (issue 1921, S10). + Parallel int `json:"parallel,omitempty"` } // levelAnswer is the one line the process prints: the level's result or the digest, or why it could not run. @@ -83,7 +85,7 @@ func (m *Module) integrationShutOff(ctx context.Context, workspace string, packa // workspace read-write, the program alone read-only, the shared caches beneath a throwaway layer, the Node build // module's toolchains read-only for the integration level. func (m *Module) shutOff(ctx context.Context, exe string, spec levelSpec) (levelAnswer, error) { - spec.Work, spec.Cache = m.Work, m.Cache + spec.Work, spec.Cache, spec.Parallel = m.Work, m.Cache, m.parallel() raw, err := json.Marshal(spec) if err != nil { return levelAnswer{}, err @@ -165,7 +167,7 @@ func runLevelProcess(args []string) int { return say(levelAnswer{Error: fmt.Sprintf("%q is not a workspace id", spec.Workspace)}) } m := &Module{Inside: true, Recipe: Current(), Work: spec.Work, Cache: spec.Cache, Home: os.Getenv("HOME"), UnitBound: spec.Bound, - IntegrationBounds: Bounds{Setup: spec.Setup, Run: spec.Bound}, Say: func(string, ...any) {}} + IntegrationBounds: Bounds{Setup: spec.Setup, Run: spec.Bound}, Parallel: spec.Parallel, Say: func(string, ...any) {}} m.Run = StepRunner(m.Home) return say(m.answer(context.Background(), spec)) } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/lint.go b/modules/mesh-build-go/cmd/mesh-build-go/lint.go index 75240ec5..3b5ec2b1 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/lint.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/lint.go @@ -107,6 +107,15 @@ func unformatted(tree, d string, only map[string]bool) (files, tags []string, ne return files, tags, negated, err } +// boundOf is a bound of n processes of this process alone. +func boundOf(n int) func(context.Context) (func(), error) { + slots := make(chan struct{}, n) + return func(context.Context) (func(), error) { + slots <- struct{}{} + return func() { <-slots }, nil + } +} + // lintScope is what of a go.mod directory d the lint covers, from the patterns the level runs there: every package // (./..., and the directories nil) when a pattern is relative or none is given, else the import paths given and the // directories of their files, relative to d. @@ -188,26 +197,30 @@ func (m *Module) lint(ctx context.Context, run forgekey.Runner, tree, d string, // lintAll lints each directory, up to m.parallel() at once like the level's tests; the failures join in directory order // and the error is the first directory's, so the answer does not depend on which ended first (issue 1930). byDir names -// the packages the level runs in each directory (packagesByDir); nil lints each directory whole. slots is the bound the -// lint shares with the level's tests; nil gives it one of its own. -func (m *Module) lintAll(ctx context.Context, run forgekey.Runner, tree string, dirs []string, env func(d string) []string, byDir map[string][]string, slots chan struct{}) ([]Failure, error) { +// the packages the level runs in each directory (packagesByDir); nil lints each directory whole. acquire takes a slot of +// the bound the lint shares with the ask's tests; nil gives it one of its own. +func (m *Module) lintAll(ctx context.Context, run forgekey.Runner, tree string, dirs []string, env func(d string) []string, byDir map[string][]string, acquire func(context.Context) (func(), error)) ([]Failure, error) { type answer struct { failed []Failure err error } answers := make([]answer, len(dirs)) - if slots == nil { - slots = make(chan struct{}, m.parallel()) + if acquire == nil { + acquire = boundOf(m.parallel()) } var wg sync.WaitGroup var crash any // a panic of a lint goes on in the caller, as a panic of a test run does var crashMu sync.Mutex for i, d := range dirs { + release, err := acquire(ctx) + if err != nil { + answers[i].err = err + break + } wg.Add(1) - slots <- struct{}{} go func() { defer wg.Done() - defer func() { <-slots }() + defer release() defer func() { if p := recover(); p != nil { crashMu.Lock() diff --git a/modules/mesh-build-go/cmd/mesh-build-go/slots.go b/modules/mesh-build-go/cmd/mesh-build-go/slots.go new file mode 100644 index 00000000..4c44d730 --- /dev/null +++ b/modules/mesh-build-go/cmd/mesh-build-go/slots.go @@ -0,0 +1,74 @@ +package main + +// The bound on the processes one ask starts (issue 1921, try speed S10; ADR 0541): the levels of an ask run side by side, +// the unit level in this process and the integration level in a shut-off process of its own, and they share one bound, +// the processors divided by the node's build capacity. The bound lives in the ask's workspace as lock files, one per +// slot, so both processes count against it. + +import ( + "context" + "fmt" + "os" + "path/filepath" + "runtime" + "strconv" + "sync/atomic" + "syscall" +) + +// numCPU is the node's processor count. +func numCPU() int { return runtime.NumCPU() } + +// slotPool is the bound of one workspace: n slots, each a lock file under dir. +type slotPool struct { + dir string + n int + next *atomic.Uint32 +} + +// slotTurns spreads the waits of the commands that find every slot taken over the slots. +var slotTurns atomic.Uint32 + +// slots is the bound of the workspace's processes, m.parallel() slots. +func (m *Module) slots(workspace string) slotPool { + return slotPool{dir: filepath.Join(m.Work, workspace, "slots"), n: m.parallel(), next: &slotTurns} +} + +// acquire takes a slot of the pool, waiting for one when every slot is taken, and answers what frees it. A free slot is +// taken at once; when none is, the wait is on one slot, taken in turn, until its holder frees it. +func (p slotPool) acquire(ctx context.Context) (func(), error) { + if err := os.MkdirAll(p.dir, 0o700); err != nil { + return nil, fmt.Errorf("the workspace's bound on processes: %w", err) + } + take := func(i int, how int) (func(), bool, error) { + f, err := os.OpenFile(filepath.Join(p.dir, "slot-"+strconv.Itoa(i)), os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, false, err + } + if err := syscall.Flock(int(f.Fd()), how); err != nil { + f.Close() + if err == syscall.EWOULDBLOCK { + return nil, false, nil + } + return nil, false, err + } + return func() { syscall.Flock(int(f.Fd()), syscall.LOCK_UN); f.Close() }, true, nil + } + for i := 0; i < p.n; i++ { + release, ok, err := take(i, syscall.LOCK_EX|syscall.LOCK_NB) + if err != nil { + return nil, fmt.Errorf("the workspace's bound on processes: %w", err) + } + if ok { + return release, nil + } + } + if err := ctx.Err(); err != nil { + return nil, err + } + release, _, err := take(int(p.next.Add(1))%p.n, syscall.LOCK_EX) + if err != nil { + return nil, fmt.Errorf("the workspace's bound on processes: %w", err) + } + return release, nil +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/slots_test.go b/modules/mesh-build-go/cmd/mesh-build-go/slots_test.go new file mode 100644 index 00000000..ffc102a6 --- /dev/null +++ b/modules/mesh-build-go/cmd/mesh-build-go/slots_test.go @@ -0,0 +1,61 @@ +package main + +import ( + "context" + "strings" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/forgekey" +) + +// 1921/S10 (ADR 0541): the levels of one ask, running side by side, share one bound on the processes they start: the +// bound is the workspace's, so a slot another level of the ask holds is one this level waits for. +func TestTheLevelsOfAnAskShareOneBoundOnProcesses(t *testing.T) { + m, states, ws, _ := workspaceFor(t, "failing") + m.Parallel = 1 + held, err := m.slots(ws).acquire(context.Background()) // the other level holds the one slot + if err != nil { + t.Fatal(err) + } + var ran atomic.Int32 + inner := m.Run + m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran { + if strings.Contains(strings.Join(args, " "), " go ") || name == "go" { + ran.Add(1) + } + return inner(ctx, env, name, args...) + } + done := make(chan delivery.Step, 1) + go func() { _, st := runUnit(t, m, states, ws); done <- st }() + time.Sleep(150 * time.Millisecond) + if n := ran.Load(); n != 0 { + t.Errorf("%d go commands ran while the ask's one slot was held by its other level", n) + } + held() + select { + case st := <-done: + if st.State != delivery.StepPassed { + t.Errorf("the step ended %s: %s", st.State, st.Reason) + } + case <-time.After(3 * time.Second): + t.Fatal("the level never ran after the slot was freed") + } + if ran.Load() == 0 { + t.Error("no go command ran") + } +} + +// 1921/S10 (ADR 0541): the bound is the processors divided by the node's build capacity, not each level half of them. +func TestTheBoundIsTheProcessorsDividedByTheBuildCapacity(t *testing.T) { + m := &Module{Capacity: func() int { return 3 }} + if got, want := m.parallel(), max(1, numCPU()/3); got != want { + t.Errorf("parallel = %d, want %d", got, want) + } + m = &Module{} + if got, want := m.parallel(), max(1, numCPU()/2); got != want { + t.Errorf("with no capacity known parallel = %d, want %d", got, want) + } +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/test.go b/modules/mesh-build-go/cmd/mesh-build-go/test.go index 72386f0f..abc1194f 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/test.go @@ -10,7 +10,6 @@ import ( "os" "path/filepath" "regexp" - "runtime" "strings" "sync" "time" @@ -208,8 +207,9 @@ func (m *Module) runLevel(ctx context.Context, tree string, o levelOpts) (*TestR todo = append(todo, d) } } - // The level's commands share one bound on processes: the lint's and the tests' (issue 1921, try speed S4). - slots := make(chan struct{}, m.parallel()) + // The level's commands share one bound on processes with each other and with the ask's other level: the lint's and + // the tests' (issue 1921, try speed S4, S10). + slots := m.slots(o.workspace) var lintFailed []Failure var lintErr error var lintDone chan struct{} @@ -231,18 +231,23 @@ func (m *Module) runLevel(ctx context.Context, tree string, o levelOpts) (*TestR linted := m.clock() lintFailed, lintErr = m.lintAll(ctx, run, tree, todo, func(d string) []string { return append(m.goEnvFor(filepath.Join(tree, d)), o.env...) - }, byDir, slots) + }, byDir, slots.acquire) res.Lint = m.clock().Sub(linted).Seconds() }() } runs := make([]dirRun, len(todo)) var wg sync.WaitGroup + var slotErr error for i, d := range todo { + release, err := slots.acquire(ctx) + if err != nil { + slotErr = err + break + } wg.Add(1) - slots <- struct{}{} go func() { defer wg.Done() - defer func() { <-slots }() + defer release() defer func() { if p := recover(); p != nil { crashMu.Lock() @@ -278,6 +283,9 @@ func (m *Module) runLevel(ctx context.Context, tree string, o levelOpts) (*TestR if lintErr != nil { return nil, lintErr } + if slotErr != nil { + return nil, slotErr + } res.Failed = append(res.Failed, lintFailed...) // the lint's failures first, then the tests' for i, d := range todo { dr := runs[i] @@ -304,12 +312,20 @@ func (m *Module) runLevel(ctx context.Context, tree string, o levelOpts) (*TestR // runAnswer is what a step runner answers. type runAnswer = forgekey.Ran -// parallel is how many directories a level runs at once. +// parallel is how many processes the levels of one ask start at once, together (ADR 0541; issue 1921, try speed S10): +// the processors divided by the node's build capacity, half of them when the capacity is not known. func (m *Module) parallel() int { if m.Parallel > 0 { return m.Parallel } - return max(1, runtime.NumCPU()/2) + capacity := 0 + if m.Capacity != nil { + capacity = m.Capacity() + } + if capacity <= 0 { + capacity = 2 + } + return max(1, numCPU()/capacity) } // packagesByDir says which patterns run in which go.mod directory. A relative pattern (./...) runs in every directory, as diff --git a/modules/mesh-build-go/go.mod b/modules/mesh-build-go/go.mod index 339169cc..3d085fb4 100644 --- a/modules/mesh-build-go/go.mod +++ b/modules/mesh-build-go/go.mod @@ -3,7 +3,7 @@ module meshbuildgo go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024230839-4a12ccb78d19 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024232250-9b408db24a58 github.com/distribution/distribution/v3 v3.1.2 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/mesh-build-go/go.sum b/modules/mesh-build-go/go.sum index 3eb57f08..4809d7f0 100644 --- a/modules/mesh-build-go/go.sum +++ b/modules/mesh-build-go/go.sum @@ -4,6 +4,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024225846-cb548a6a588c h1:GWHP9ozf git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024225846-cb548a6a588c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024230839-4a12ccb78d19 h1:NqU4hkk/jHdTEcGFnnXvdOJwD8dH5u+qWaCdQQ+JYMc= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024230839-4a12ccb78d19/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024232250-9b408db24a58 h1:4Damtdqlf6u6nJhvMA99peDy1jmjixtcwFXJ4bWEkLk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024232250-9b408db24a58/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI= github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=