mesh-build-go: start the integration level's store and bus beside its compile, and compile the go.mod directories side by side
The setup started the store, then the bus, then compiled each directory one after another: 132 s of setup in the L3 group check. Now the three start together and the directories compile up to the bound at once; the compile runs no test binary (-exec true), so it needs neither server and is given nothing a test reads. Ported from feat/try-speed S5 (issue 1921; ADR 0541 §1).
This commit is contained in:
@@ -12,6 +12,7 @@ import (
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
@@ -62,30 +63,73 @@ func (m *Module) integrationHere(ctx context.Context, workspace, tree string, pa
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
postgres, stopStore, err := m.startStore(ctx, run)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the integration level's store: %w", err)
|
||||
}
|
||||
defer stopStore()
|
||||
nats, stopBus, err := m.startBus(ctx, filepath.Join(run, "nats"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the integration level's bus: %w", err)
|
||||
}
|
||||
defer stopBus()
|
||||
env := append([]string{"MESH_TEST_POSTGRES=" + postgres, "MESH_TEST_NATS=" + nats}, m.checkEnv(workspace, nil)...)
|
||||
// Compiling counts as setup: the tests are built once, here, with the addresses a TestMain may read, and the run
|
||||
// below finds them built. Whether they compile is for the run to say; its failure is not read here.
|
||||
// Only the directories the selected packages lie in compile (ADR 0536 §3); no selection is every directory.
|
||||
byDir, err := packagesByDir(tree, dirs, packages)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// The store, the bus and the compile start together, and the directories compile side by side up to the bound
|
||||
// (issue 1921, try speed S5): the setup is the longest of them, not their sum. The compile runs no test binary
|
||||
// (-exec true), so it needs neither server and is given nothing a test reads; whether the tests compile is for the
|
||||
// run to say, its failure is not read here, and the run below finds them built.
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
postgres, nats string
|
||||
stopStore, stopBus func()
|
||||
storeErr, busErr error
|
||||
crash any
|
||||
crashMu sync.Mutex
|
||||
)
|
||||
guard := func() {
|
||||
if p := recover(); p != nil {
|
||||
crashMu.Lock()
|
||||
crash = p
|
||||
crashMu.Unlock()
|
||||
}
|
||||
}
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer guard()
|
||||
postgres, stopStore, storeErr = m.startStore(ctx, run)
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer guard()
|
||||
nats, stopBus, busErr = m.startBus(ctx, filepath.Join(run, "nats"))
|
||||
}()
|
||||
slots := make(chan struct{}, m.parallel())
|
||||
for _, d := range dirs {
|
||||
if len(byDir[d]) == 0 {
|
||||
continue
|
||||
}
|
||||
m.Run(ctx, append(m.goEnvFor(filepath.Join(tree, d)), env...), "go", append([]string{"-C", filepath.Join(tree, d), "test", "-tags", strings.Join(integrationTags, ","), "-run", "^$", "-count=1"}, byDir[d]...)...)
|
||||
wg.Add(1)
|
||||
slots <- struct{}{}
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer func() { <-slots }()
|
||||
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]...)...)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
// Whatever started is stopped however the level ends, a half-started store included.
|
||||
if stopStore != nil {
|
||||
defer stopStore()
|
||||
}
|
||||
if stopBus != nil {
|
||||
defer stopBus()
|
||||
}
|
||||
if crash != nil {
|
||||
panic(crash)
|
||||
}
|
||||
if storeErr != nil {
|
||||
return nil, fmt.Errorf("the integration level's store: %w", storeErr)
|
||||
}
|
||||
if busErr != nil {
|
||||
return nil, fmt.Errorf("the integration level's bus: %w", busErr)
|
||||
}
|
||||
env := append([]string{"MESH_TEST_POSTGRES=" + postgres, "MESH_TEST_NATS=" + nats}, m.checkEnv(workspace, nil)...)
|
||||
setup := m.clock().Sub(began)
|
||||
|
||||
return m.runLevel(ctx, tree, levelOpts{
|
||||
|
||||
@@ -35,6 +35,9 @@ type integFake struct {
|
||||
// command runs, with its line.
|
||||
gomods string
|
||||
onTest func(line string)
|
||||
// onCompile is called, unlocked, as each compile of the setup runs, and onStart as the store starts (issue 1921, S5).
|
||||
onCompile func(line string, env []string)
|
||||
onStart func()
|
||||
}
|
||||
|
||||
func newIntegFake(t *testing.T, captured string) *integFake {
|
||||
@@ -71,7 +74,10 @@ func (f *integFake) run(ctx context.Context, env []string, name string, args ...
|
||||
return forgekey.Ran{Stdout: f.gomods}
|
||||
}
|
||||
return forgekey.Ran{Stdout: "go.mod\n"}
|
||||
case name == "go" && strings.Contains(line, "-run ^$"):
|
||||
case name == "go" && strings.Contains(line, "-exec true"):
|
||||
if f.onCompile != nil {
|
||||
f.onCompile(line, env)
|
||||
}
|
||||
f.tick("compile")
|
||||
case name == "go":
|
||||
if f.onTest != nil {
|
||||
@@ -84,6 +90,9 @@ func (f *integFake) run(ctx context.Context, env []string, name string, args ...
|
||||
}
|
||||
return forgekey.Ran{Stdout: f.captured, Status: status}
|
||||
case name == "pg_ctl" && strings.HasSuffix(line, " start"):
|
||||
if f.onStart != nil {
|
||||
f.onStart()
|
||||
}
|
||||
f.tick("start")
|
||||
for i, a := range args {
|
||||
if a == "-D" {
|
||||
@@ -314,8 +323,8 @@ func TestAStoreThatDoesNotStartEndsTheStepFailedAndLeavesNothing(t *testing.T) {
|
||||
if !strings.Contains(f.lines(), " stop") {
|
||||
t.Errorf("the half-started store was not stopped:\n%s", f.lines())
|
||||
}
|
||||
if f.busDir != "" {
|
||||
t.Errorf("the bus was started after the store failed")
|
||||
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)
|
||||
|
||||
@@ -290,8 +290,8 @@ func TestTheIntegrationLevelIsGivenTheBesideDirectoryOnlyWhenSiblingsWerePlaced(
|
||||
}
|
||||
tested := 0
|
||||
for _, c := range f.calls {
|
||||
if !strings.HasPrefix(c.line, "go ") || !strings.Contains(c.line, "-tags integration") {
|
||||
continue
|
||||
if !strings.HasPrefix(c.line, "go ") || !strings.Contains(c.line, "-tags integration") || strings.Contains(c.line, "-exec true") {
|
||||
continue // the setup's compile runs no test binary, so it is given nothing a test reads (issue 1921, S5)
|
||||
}
|
||||
tested++
|
||||
got := ""
|
||||
|
||||
@@ -72,7 +72,7 @@ func TestOnlyTheSelectedDirectoryCompilesAndRuns(t *testing.T) {
|
||||
if !strings.HasPrefix(l, "go -C "+filepath.Join(tree, "b")+" test ") {
|
||||
t.Errorf("go test ran outside the selected directory: %s", l)
|
||||
}
|
||||
if strings.Contains(l, "-run ^$") {
|
||||
if strings.Contains(l, "-exec true") {
|
||||
compiled++
|
||||
} else {
|
||||
ran++
|
||||
@@ -91,7 +91,7 @@ func TestWithoutASelectionEveryDirectoryCompilesAndRuns(t *testing.T) {
|
||||
compiled, ran := 0, 0
|
||||
for _, l := range strings.Split(f.lines(), "\n") {
|
||||
if strings.HasPrefix(l, "go ") && strings.Contains(l, " test ") {
|
||||
if strings.Contains(l, "-run ^$") {
|
||||
if strings.Contains(l, "-exec true") {
|
||||
compiled++
|
||||
} else {
|
||||
ran++
|
||||
@@ -146,27 +146,85 @@ func TestSelectedDirectoriesRunAtOnceAndAnswerInDirectoryOrder(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The bound holds: with one directory at a time, no two go test commands run at once.
|
||||
// The bound holds for the setup's compiles and for the tests: no more of them run at once than the bound, and with a
|
||||
// bound of 3 the three directories compile at once, beside the store's start, given nothing a test reads, since the
|
||||
// compile runs no test binary (issue 1921, try speed S5).
|
||||
func TestTheDirectoriesAtOnceNeverPassTheBound(t *testing.T) {
|
||||
f := newIntegFake(t, "slow")
|
||||
var mu sync.Mutex
|
||||
running, most := 0, 0
|
||||
f.onTest = func(string) {
|
||||
mu.Lock()
|
||||
running++
|
||||
if running > most {
|
||||
most = running
|
||||
}
|
||||
mu.Unlock()
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
mu.Lock()
|
||||
running--
|
||||
mu.Unlock()
|
||||
}
|
||||
m, states := threeModules(t, f)
|
||||
m.Parallel = 1
|
||||
runIntegrationOf(t, m, states)
|
||||
if most != 1 {
|
||||
t.Errorf("%d go test commands ran at once with a bound of 1", most)
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
compile bool // the setup's compiles are counted, not the tests
|
||||
bound int
|
||||
want int // how many must run at once (at most the bound always)
|
||||
}{
|
||||
{"tests one at a time", false, 1, 1},
|
||||
{"compiles one at a time", true, 1, 1},
|
||||
{"compiles at once beside the store", true, 3, 3},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newIntegFake(t, "slow")
|
||||
var mu sync.Mutex
|
||||
running, most := 0, 0
|
||||
reached := make(chan struct{})
|
||||
storeStarting, beside := false, false
|
||||
count := func() {
|
||||
mu.Lock()
|
||||
running++
|
||||
most = max(most, running)
|
||||
if running == tc.want && tc.want > 1 {
|
||||
close(reached)
|
||||
}
|
||||
mu.Unlock()
|
||||
if tc.want > 1 {
|
||||
select { // held until all of them run, so "at once" does not hang on timing
|
||||
case <-reached:
|
||||
case <-time.After(2 * time.Second):
|
||||
}
|
||||
} else {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
mu.Lock()
|
||||
running--
|
||||
mu.Unlock()
|
||||
}
|
||||
if tc.compile {
|
||||
f.onCompile = func(_ string, env []string) {
|
||||
for _, e := range env {
|
||||
if strings.HasPrefix(e, "MESH_TEST_") {
|
||||
t.Errorf("the compile was given %s: it runs no test binary", e)
|
||||
}
|
||||
}
|
||||
mu.Lock()
|
||||
beside = beside || storeStarting
|
||||
mu.Unlock()
|
||||
count()
|
||||
}
|
||||
if tc.want > 1 {
|
||||
f.onStart = func() { // the store starts until the compiles have all run at once, or 2 s pass
|
||||
mu.Lock()
|
||||
storeStarting = true
|
||||
mu.Unlock()
|
||||
select {
|
||||
case <-reached:
|
||||
case <-time.After(2 * time.Second):
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
f.onTest = func(string) { count() }
|
||||
}
|
||||
m, states := threeModules(t, f)
|
||||
m.Parallel = tc.bound
|
||||
if _, st := runIntegrationOf(t, m, states); st.State != delivery.StepPassed {
|
||||
t.Fatalf("the step ended %s: %s", st.State, st.Reason)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if most > tc.bound || most < tc.want {
|
||||
t.Errorf("%d ran at once with a bound of %d, want %d", most, tc.bound, tc.want)
|
||||
}
|
||||
if tc.compile && tc.want > 1 && !beside {
|
||||
t.Errorf("no compile ran while the store was starting:\n%s", f.lines())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user