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)

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)
 	}

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)
 	}
 }
 

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))
 }

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()

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
+}

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)
+	}
+}

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

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

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=