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=