diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index acc2bfe..7e38f3b 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -57,7 +57,13 @@ func read(s stdio.KeptState) build.StateRead { if err != nil || e == nil { return nil, time.Time{}, err } - return e.Value, written(s, key, e.Revision), nil + return e.Value, time.Time{}, nil + }, Written: func(key string) time.Time { + e, err := s.Get(key) + if err != nil || e == nil { + return time.Time{} + } + return written(s, key, e.Revision) }} } @@ -81,7 +87,7 @@ func main() { steps[bm] = read(stdio.State(bm + ".steps")) } reasks := stdio.State(build.ReAsksState) - r := build.StuckReader{Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps} + r := build.StuckReader{Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] "+f+"\n", a...) }, Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps} put := func(key string, value []byte) error { _, err := reasks.Put(key, json.RawMessage(value)) return err diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 17d5cf4..0420512 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -8,6 +8,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "strings" "sync" + "sync/atomic" "testing" "time" @@ -807,6 +808,52 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { a[1].Asking != 1 || !strings.HasPrefix(a[1].Step, "running on g14") { t.Fatalf("asks = %+v, want go at asking 2 with no step and image at asking 1 running on g14", a) } + + // 1827: on the live shape (3000 build keys over 100 modules, 1 ms a Get, three keys of each module building) a module reads only its + // own keys, the whole answer reads MaxReads values in parallel and says it was cut, and the history (Written) is asked + // only for the rows kept. + var gets, written, inFlight, maxFlight atomic.Int64 + vals := map[string]string{} + for i := 0; i < 3000; i++ { + k, _ := delivery.BuilderBuildKey(fmt.Sprintf("mod%02d", i%100), fmt.Sprintf("src2:%064x", i)) + vals[k] = string(finished) + if (i/100)%10 == 0 { + vals[k] = string(building) + } + } + slow := state(vals) + get := slow.Get + slow.Get = func(k string) (json.RawMessage, time.Time, error) { + gets.Add(1) + n := inFlight.Add(1) + for m := maxFlight.Load(); n > m && !maxFlight.CompareAndSwap(m, n); m = maxFlight.Load() { + } + time.Sleep(time.Millisecond) + inFlight.Add(-1) + raw, _, err := get(k) + return raw, time.Time{}, err + } + slow.Written = func(string) time.Time { written.Add(1); return at } + for _, row := range []struct { + module string + stuck int + gets int64 + cut bool + }{{"mod10", 3, 30, false}, {"", 100, MaxReads, true}} { + gets.Store(0) + written.Store(0) + out, err := StuckReader{Builds: slow, ReAsks: state(map[string]string{}), Steps: map[string]StateRead{}, Now: func() time.Time { return at }}.Stuck(row.module) + if err != nil { + t.Fatal(err) + } + _, cut := out["cut"] + if n := len(out["stuck"].([]Stuck)); n != row.stuck || cut != row.cut || written.Load() != int64(n) || gets.Load() != row.gets { + t.Errorf("%q: %d stuck, %d gets, %d written, cut %v; want %d stuck and gets, written for each, cut %v", row.module, n, gets.Load(), written.Load(), cut, row.stuck, row.cut) + } + } + if maxFlight.Load() < 2 { + t.Errorf("the values were read one at a time") + } }) } diff --git a/modules/mesh-delivery-builder/internal/build/stuck.go b/modules/mesh-delivery-builder/internal/build/stuck.go index 5bcaf0c..41184c4 100644 --- a/modules/mesh-delivery-builder/internal/build/stuck.go +++ b/modules/mesh-delivery-builder/internal/build/stuck.go @@ -7,6 +7,8 @@ import ( "sort" "strconv" "strings" + "sync" + "sync/atomic" "time" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -65,6 +67,9 @@ func NewReAsk(module, fp, why, by string, at time.Time) (string, []byte, error) type StateRead struct { Keys func() ([]string, error) Get func(key string) (json.RawMessage, time.Time, error) + // Written is when the key's current value was written, from the state's history: one more round trip, so the tools ask it + // only for a row they keep, and only when Get did not say. + Written func(key string) time.Time } // StuckAsk is one ask a key waits on. @@ -102,10 +107,56 @@ type StuckReader struct { // Steps are the build modules' steps states, by build module. Steps map[string]StateRead Now func() time.Time + // Logf, when set, gets the one line a call logs: the values it read and the time it took. + Logf func(format string, args ...any) +} + +// ReadWorkers is how many values builder_stuck asks the bus for at once, and MaxReads how many it reads in a call (issue 1827: +// every value is a round trip through the runtime, so one after the other a few thousand keys outlast the call's 30 s). +// A copy of the delivery view's and the checker's: a few lines are not worth a group delivery through mesh-sdk. +const ( + ReadWorkers = 16 + MaxReads = 1000 +) + +// parallel runs read(i) for i in [0, n), at most ReadWorkers at a time, and answers the first error. +func parallel(n int, read func(i int) error) error { + var ( + next atomic.Int64 + failed atomic.Bool + once sync.Once + first error + wg sync.WaitGroup + ) + for w := 0; w < min(ReadWorkers, n); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for !failed.Load() { + i := int(next.Add(1) - 1) + if i >= n { + return + } + if err := read(i); err != nil { + once.Do(func() { first = err }) + failed.Store(true) + } + } + }() + } + wg.Wait() + return first } // Stuck is every key building or queued with its asks, oldest first; with module, that module's only. func (r StuckReader) Stuck(module string) (map[string]any, error) { + started := time.Now() + reads := 0 + defer func() { + if r.Logf != nil { + r.Logf("builder_stuck read %d values in %s", reads, time.Since(started).Round(time.Millisecond)) + } + }() keys, err := r.Builds.Keys() if err != nil { return nil, err @@ -114,28 +165,64 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) { if r.Now != nil { now = r.Now } - stepKeys := map[string][]string{} + // Steps are found by the ask's id, which their key ends in: one index per build module, no value read until an ask names it. + stepKeys := map[string]map[string]string{} for bm, s := range r.Steps { ks, err := s.Keys() if err != nil { return nil, fmt.Errorf("the steps of %s: %w", bm, err) } - stepKeys[bm] = ks + byID := make(map[string]string, len(ks)) + for _, k := range ks { + if _, id, err := delivery.ParseStepKey(k); err == nil { + byID[id] = k + } + } + stepKeys[bm] = byID } - out := []Stuck{} + // Narrowed by the key (the module) before any value is read; the state is in the value, so every matching key is read, in + // parallel, at most MaxReads of them. + type parsed struct{ key, mod, fp string } + var matching []parsed for _, key := range keys { mod, fp, err := delivery.ParseBuilderBuildKey(key) if err != nil || (module != "" && mod != module) { continue } - raw, at, err := r.Builds.Get(key) - if err != nil || raw == nil { + matching = append(matching, parsed{key, mod, fp}) + } + sort.Slice(matching, func(i, j int) bool { return matching[i].key < matching[j].key }) + cut := len(matching) > MaxReads + total := len(matching) + if cut { + matching = matching[:MaxReads] + } + values := make([]json.RawMessage, len(matching)) + ats := make([]time.Time, len(matching)) + if err := parallel(len(matching), func(i int) error { + raw, at, err := r.Builds.Get(matching[i].key) + if err == nil { + values[i], ats[i] = raw, at + } + return nil + }); err != nil { + return nil, err + } + reads += len(matching) + out := []Stuck{} + for i, m := range matching { + key, mod, fp := m.key, m.mod, m.fp + raw, at := values[i], ats[i] + if raw == nil { continue } v, err := delivery.DecodeBuilderBuild(raw) if err != nil || (v.State != delivery.BuildBuilding && v.State != delivery.BuildQueued) { continue } + if at.IsZero() && r.Builds.Written != nil { + at = r.Builds.Written(key) + } s := Stuck{Key: key, Module: mod, FP: fp, State: v.State, Step: v.Step, By: v.By, Since: at, Asks: []StuckAsk{}} if !at.IsZero() { s.Age = now().Sub(at).Round(time.Second).String() @@ -147,10 +234,8 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) { a.Asked = t } if st, ok := r.Steps[bm]; ok { - for _, k := range stepKeys[bm] { - if !strings.HasSuffix(k, "."+id) { - continue - } + if k, ok := stepKeys[bm][id]; ok { + reads++ if raw, _, err := st.Get(k); err == nil && raw != nil { if step, err := delivery.DecodeStep(raw); err == nil { a.Step = fmt.Sprintf("%s on %s since %s", step.State, step.Node, step.Started.UTC().Format(time.RFC3339)) @@ -166,6 +251,7 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) { sort.Slice(s.Asks, func(i, j int) bool { return s.Asks[i].Language < s.Asks[j].Language }) s.Unasked = v.State == delivery.BuildBuilding && len(v.Asks) == 0 if r.ReAsks.Get != nil { + reads++ if raw, _, err := r.ReAsks.Get(key); err == nil && raw != nil { if ra, err := DecodeReAsk(raw); err == nil { s.ReAsk = &ra @@ -180,8 +266,12 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) { } return out[i].Key < out[j].Key }) - return map[string]any{"stuck": out, "count": len(out), - "delivery_counts": "not readable by the builder: the build seats' workers' information is granted to the delivery view alone (its backlog); an ask given up on is told by the builder at its check bound and asked again"}, nil + answer := map[string]any{"stuck": out, "count": len(out), + "delivery_counts": "not readable by the builder: the build seats' workers' information is granted to the delivery view alone (its backlog); an ask given up on is told by the builder at its check bound and asked again"} + if cut { + answer["cut"], answer["note"] = true, fmt.Sprintf("read %d of %d build keys in key order and stopped; give a module to read its keys alone", MaxReads, total) + } + return answer, nil } // StuckBuildModules are the build modules whose steps builder_stuck reads. diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index c43f8cc..da0aabe 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "os" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -15,8 +16,35 @@ import ( "mesh-delivery-checker/internal/check" ) +// reads counts the values this call read from the bus: the runtime launches the tools for one call, so the count is the call's. +var reads atomic.Int64 + +// cutAnswer says in the answer that it was cut, and how to narrow it. +func cutAnswer(answer map[string]any, cut string) map[string]any { + if cut != "" { + answer["cut"], answer["note"] = true, cut + } + return answer +} + +// logged runs every tool with one line to stderr per call: the tool, the values it read and the time it took. +func logged(tools []stdio.Tool) []stdio.Tool { + for i := range tools { + name, run := tools[i].Name, tools[i].Run + tools[i].Run = func(args map[string]any) (any, error) { + reads.Store(0) + start := time.Now() + out, err := run(args) + fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %s read %d values in %s (error: %v)\n", name, reads.Load(), time.Since(start).Round(time.Millisecond), err) + return out, err + } + } + return tools +} + func read(s stdio.KeptState) check.StateRead { return check.StateRead{Keys: s.Keys, Get: func(key string) (json.RawMessage, error) { + reads.Add(1) e, err := s.Get(key) if err != nil || e == nil { return nil, err @@ -100,8 +128,8 @@ func Tools(r check.Reader) []stdio.Tool { "group": map[string]any{"type": "string", "description": "the delivery group's branch name"}}}, Run: func(args map[string]any) (any, error) { group, _ := args["group"].(string) - vs, err := r.Verdicts(group) - return map[string]any{"verdicts": vs}, err + vs, cut, err := r.Verdicts(group) + return cutAnswer(map[string]any{"verdicts": vs}, cut), err }}, {Name: check.PullReadTool, Description: "The verdict that judges one pull request and the full report of its status mesh/check. Read-only.", @@ -122,8 +150,8 @@ func Tools(r check.Reader) []stdio.Tool { Run: func(args map[string]any) (any, error) { level, _ := args["level"].(string) digest, _ := args["digest"].(string) - kept, err := r.Kept(level, digest) - return map[string]any{"kept": kept}, err + kept, cut, err := r.Kept(level, digest) + return cutAnswer(map[string]any{"kept": kept}, cut), err }}, {Name: check.PreviewsReadTool, Description: "The merge previews the checker holds, each as stored under previews...: the merge commit, per module directory " + @@ -138,8 +166,8 @@ func Tools(r check.Reader) []stdio.Tool { if head != "" && len(head) < 7 { return nil, fmt.Errorf("the head %q is too short: give at least 7 characters", head) } - ps, bad, err := r.Previews(repository, head) - return map[string]any{"previews": ps, "unreadable": bad}, err + ps, bad, cut, err := r.Previews(repository, head) + return cutAnswer(map[string]any{"previews": ps, "unreadable": bad}, cut), err }}, {Name: check.StatusReadTool, Description: "What runs: the steps in flight on the build modules by node, the candidates without a verdict for their heads, " + @@ -184,7 +212,7 @@ func main() { }}, Trunks: trunks, } - if err := stdio.Serve("", append(Tools(r), TryTools(trying)...)); err != nil { + if err := stdio.Serve("", logged(append(Tools(r), TryTools(trying)...))); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-checker/internal/check/bounded.go b/modules/mesh-delivery-checker/internal/check/bounded.go new file mode 100644 index 0000000..3d432fe --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/bounded.go @@ -0,0 +1,79 @@ +package check + +import ( + "fmt" + "sort" + "sync" + "sync/atomic" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ReadWorkers is how many values a read tool asks the bus for at once. Each value is a round trip from the tool through the +// runtime to the bus, so reading a bucket of thousands of keys one after the other outlasts the 30 s a tool call is given +// (issue 1827, as the delivery view's issue 1945). It is a copy of the view's: a few lines are not worth a group delivery +// through mesh-sdk (rules-are-weighed). +const ReadWorkers = 16 + +// MaxReads bounds the values one call of a read tool reads; past it the answer says it was cut and how to narrow it. +const MaxReads = 1000 + +// parallel runs read(i) for i in [0, n), at most ReadWorkers at a time, and answers the first error; once one failed no +// further read starts. +func parallel(n int, read func(i int) error) error { + var ( + next atomic.Int64 + failed atomic.Bool + once sync.Once + first error + wg sync.WaitGroup + ) + for w := 0; w < min(ReadWorkers, n); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for !failed.Load() { + i := int(next.Add(1) - 1) + if i >= n { + return + } + if err := read(i); err != nil { + once.Do(func() { first = err }) + failed.Store(true) + } + } + }() + } + wg.Wait() + return first +} + +// capKeys keeps the first MaxReads keys and answers a note when it cut the rest, "" otherwise. +func capKeys(keys []string, narrow string) ([]string, string) { + if len(keys) <= MaxReads { + return keys, "" + } + return keys[:MaxReads], fmt.Sprintf("read %d of %d matching keys and stopped; %s", MaxReads, len(keys), narrow) +} + +// newestFirst sorts step keys (.) by id, newest first: an id is the time in base 36, then six hex digits, so the +// longer id is the later and equal lengths compare as text. A key that is no step key goes last. +func newestFirst(keys []string) { + id := func(k string) (string, bool) { + _, id, err := delivery.ParseStepKey(k) + return id, err == nil + } + sort.Slice(keys, func(i, j int) bool { + a, aok := id(keys[i]) + b, bok := id(keys[j]) + switch { + case aok != bok: + return aok + case len(a) != len(b): + return len(a) > len(b) + case a != b: + return a > b + } + return keys[i] > keys[j] + }) +} diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index 02d6686..f5953f9 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -89,37 +89,46 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) { return out, nil } -// Verdicts are the verdicts the checker holds, by key; with a group only that group's. -func (r Reader) Verdicts(group string) ([]VerdictRead, error) { +// Verdicts are the verdicts the checker holds, by key; with a group only that group's. The keys are narrowed by the group +// before any value is read, at most MaxReads are read (sixteen at a time), and cut says when it stopped there. +func (r Reader) Verdicts(group string) (out []VerdictRead, cut string, err error) { if r.VerdictState.Keys == nil || r.VerdictState.Get == nil { - return nil, fmt.Errorf("the verdicts state cannot be read here") + return nil, "", fmt.Errorf("the verdicts state cannot be read here") } keys, err := r.VerdictState.Keys() if err != nil { - return nil, err + return nil, "", err } sort.Strings(keys) - var out []VerdictRead + var matching []string for _, k := range keys { if group != "" { if g, _, err := delivery.ParseCheckerVerdictKey(k); err != nil || g != group { continue } } - raw, err := r.VerdictState.Get(k) - if err != nil { - return nil, err - } - if len(raw) == 0 { - continue - } - vr, err := r.read(k, raw) - if err != nil { - return nil, err - } - out = append(out, vr) + matching = append(matching, k) } - return out, nil + matching, cut = capKeys(matching, "give a group to narrow the answer") + rows := make([]VerdictRead, len(matching)) + found := make([]bool, len(matching)) + if err := parallel(len(matching), func(i int) error { + raw, err := r.VerdictState.Get(matching[i]) + if err != nil || len(raw) == 0 { + return err + } + rows[i], err = r.read(matching[i], raw) + found[i] = err == nil + return err + }); err != nil { + return nil, "", err + } + for i, row := range rows { + if found[i] { + out = append(out, row) + } + } + return out, cut, nil } // Pull answers for one pull request: the verdict of the candidate that holds it, with its report; for one the indexer holds @@ -133,74 +142,95 @@ func (r Reader) Pull(repository string, number int) (VerdictRead, error) { return VerdictRead{}, err } sort.Strings(keys) - for _, k := range keys { - raw, err := r.CandidateState.Get(k) - if err != nil || len(raw) == 0 { - continue - } - c, err := delivery.DecodeCandidate(raw) - if err != nil { - continue - } - for _, p := range c.Pulls { - if p.Repository != repository || p.Number != number { + keys, cut := capKeys(keys, "") + // The candidates are read in batches, each in parallel and taken in key order, so the read stops at the batch that holds it. + for from := 0; from < len(keys); from += ReadWorkers * 4 { + batch := keys[from:min(from+ReadWorkers*4, len(keys))] + cands := make([]delivery.Candidate, len(batch)) + ok := make([]bool, len(batch)) + _ = parallel(len(batch), func(i int) error { + raw, err := r.CandidateState.Get(batch[i]) + if err != nil || len(raw) == 0 { + return nil + } + c, err := delivery.DecodeCandidate(raw) + cands[i], ok[i] = c, err == nil + return nil + }) + for i, c := range cands { + if !ok[i] { continue } - vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) - if err != nil { - return VerdictRead{}, err + for _, p := range c.Pulls { + if p.Repository != repository || p.Number != number { + continue + } + vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) + if err != nil { + return VerdictRead{}, err + } + vraw, err := r.VerdictState.Get(vkey) + if err != nil { + return VerdictRead{}, err + } + if len(vraw) == 0 { + return VerdictRead{Key: vkey, Note: "no verdict yet for the heads of its candidate"}, nil + } + return r.read(vkey, vraw) } - vraw, err := r.VerdictState.Get(vkey) - if err != nil { - return VerdictRead{}, err - } - if len(vraw) == 0 { - return VerdictRead{Key: vkey, Note: "no verdict yet for the heads of its candidate"}, nil - } - return r.read(vkey, vraw) } } + if cut != "" { + return VerdictRead{}, fmt.Errorf("no candidate among the first %d holds %s#%d", MaxReads, repository, number) + } return VerdictRead{}, fmt.Errorf("no candidate holds %s#%d", repository, number) } // Kept is the kept verdicts: one key's value as stored, or with an empty key the keys of a level (or every level) and their values. -func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, error) { +func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, string, error) { if r.KeptState.Keys == nil || r.KeptState.Get == nil { - return nil, fmt.Errorf("the kept state cannot be read here") + return nil, "", fmt.Errorf("the kept state cannot be read here") } if digest != "" { key, err := delivery.KeptKey(level, digest) if err != nil { - return nil, err + return nil, "", err } raw, err := r.KeptState.Get(key) if err != nil { - return nil, err + return nil, "", err } if len(raw) == 0 { - return nil, fmt.Errorf("no kept verdict %s", key) + return nil, "", fmt.Errorf("no kept verdict %s", key) } - return map[string]json.RawMessage{key: raw}, nil + return map[string]json.RawMessage{key: raw}, "", nil } keys, err := r.KeptState.Keys() if err != nil { - return nil, err + return nil, "", err } sort.Strings(keys) - out := map[string]json.RawMessage{} + var matching []string for _, k := range keys { - if l, _, err := delivery.ParseKeptKey(k); err != nil || (level != "" && l != level) { - continue - } - raw, err := r.KeptState.Get(k) - if err != nil { - return nil, err - } - if len(raw) > 0 { - out[k] = raw + if l, _, err := delivery.ParseKeptKey(k); err == nil && (level == "" || l == level) { + matching = append(matching, k) } } - return out, nil + matching, cut := capKeys(matching, "give a level and a digest to read one") + values := make([]json.RawMessage, len(matching)) + if err := parallel(len(matching), func(i int) (err error) { + values[i], err = r.KeptState.Get(matching[i]) + return err + }); err != nil { + return nil, "", err + } + out := map[string]json.RawMessage{} + for i, k := range matching { + if len(values[i]) > 0 { + out[k] = values[i] + } + } + return out, cut, nil } // PreviewRead is one preview as stored, with the parts of its key. @@ -214,17 +244,18 @@ type PreviewRead struct { // Previews are the previews the checker holds, each as stored. With a repository only that repository's; with a head (a // prefix of at least 7 characters is enough) only previews of that head. A key that is no preview key is listed as unreadable. -func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, error) { +func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, string, error) { if r.PreviewState.Keys == nil || r.PreviewState.Get == nil { - return nil, nil, fmt.Errorf("the previews state cannot be read here") + return nil, nil, "", fmt.Errorf("the previews state cannot be read here") } keys, err := r.PreviewState.Keys() if err != nil { - return nil, nil, err + return nil, nil, "", err } sort.Strings(keys) - out := []PreviewRead{} - var bad []string + type part struct{ repo, head, trunk string } + parts := map[string]part{} + var matching, bad []string for _, k := range keys { repo, h, trunk, err := delivery.ParsePreviewKey(k) if err != nil { @@ -234,15 +265,25 @@ func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, erro if (repository != "" && repo != repository) || (head != "" && !strings.HasPrefix(h, head)) { continue } - raw, err := r.PreviewState.Get(k) - if err != nil { - return nil, nil, err - } - if len(raw) > 0 { - out = append(out, PreviewRead{Key: k, Repository: repo, Head: h, Trunk: trunk, Preview: raw}) + parts[k] = part{repo, h, trunk} + matching = append(matching, k) + } + matching, cut := capKeys(matching, "give a repository and a head to narrow the answer") + values := make([]json.RawMessage, len(matching)) + if err := parallel(len(matching), func(i int) (err error) { + values[i], err = r.PreviewState.Get(matching[i]) + return err + }); err != nil { + return nil, nil, "", err + } + out := []PreviewRead{} + for i, k := range matching { + if len(values[i]) > 0 { + p := parts[k] + out = append(out, PreviewRead{Key: k, Repository: p.repo, Head: p.head, Trunk: p.trunk, Preview: values[i]}) } } - return out, bad, nil + return out, bad, cut, nil } // RunningStep is a step in flight on a build module. @@ -300,16 +341,28 @@ func (r Reader) Status() (Status, error) { st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error()) continue } - sort.Strings(keys) - for _, k := range keys { - raw, err := sr.Get(k) + // A step that runs started lately: the newest ids are read first, at most MaxReads of them. + newestFirst(keys) + keys, cut := capKeys(keys, "") + if cut != "" { + st.Notes = append(st.Notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown") + } + steps := make([]delivery.Step, len(keys)) + isRunning := make([]bool, len(keys)) + _ = parallel(len(keys), func(i int) error { + raw, err := sr.Get(keys[i]) if err != nil || len(raw) == 0 { - continue + return nil } step, err := delivery.DecodeStep(raw) - if err != nil || step.State != delivery.StepRunning { + steps[i], isRunning[i] = step, err == nil && step.State == delivery.StepRunning + return nil + }) + for i := range keys { + if !isRunning[i] { continue } + step := steps[i] rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool} if !step.Started.IsZero() { rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z") @@ -344,23 +397,41 @@ func (r Reader) Status() (Status, error) { return st, err } sort.Strings(keys) - for _, k := range keys { - raw, err := r.CandidateState.Get(k) + total := len(keys) + keys, cut := capKeys(keys, "") + if cut != "" { + st.Notes = append(st.Notes, fmt.Sprintf("read the first %d candidates of %d", MaxReads, total)) + } + type judged struct { + c delivery.Candidate + raw json.RawMessage + ok bool + } + rows := make([]judged, len(keys)) + _ = parallel(len(keys), func(i int) error { + raw, err := r.CandidateState.Get(keys[i]) if err != nil || len(raw) == 0 { - continue + return nil } c, err := delivery.DecodeCandidate(raw) if err != nil { - continue + return nil } vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) if err != nil { + return nil + } + rows[i].c = c + if vraw, err := r.VerdictState.Get(vkey); err == nil { + rows[i].raw, rows[i].ok = vraw, true + } + return nil + }) + for _, row := range rows { + if !row.ok { continue } - raw, err = r.VerdictState.Get(vkey) - if err != nil { - continue - } + c, raw := row.c, row.raw if len(raw) == 0 { st.Unjudged = append(st.Unjudged, c.Group) st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group) diff --git a/modules/mesh-delivery-checker/internal/check/read_test.go b/modules/mesh-delivery-checker/internal/check/read_test.go index ecd4109..4d620fd 100644 --- a/modules/mesh-delivery-checker/internal/check/read_test.go +++ b/modules/mesh-delivery-checker/internal/check/read_test.go @@ -3,7 +3,9 @@ package check import ( "context" "encoding/json" + "fmt" "strings" + "sync/atomic" "testing" "time" @@ -51,7 +53,7 @@ func readerOf(t *testing.T) (Reader, string) { func TestTheVerdictToolsAnswerTheStoredVerdictWithTheFullReportOfEachPullRequest(t *testing.T) { // N9/AC1, 1686 rd, key := readerOf(t) - vs, err := rd.Verdicts("") + vs, _, err := rd.Verdicts("") if err != nil || len(vs) != 1 || vs[0].Key != key { t.Fatalf("verdicts = %+v, %v", vs, err) } @@ -61,7 +63,7 @@ func TestTheVerdictToolsAnswerTheStoredVerdictWithTheFullReportOfEachPullRequest if got := vs[0].Reports["novox/mesh-delivery#7"]; !strings.HasPrefix(got, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(got, "composition not yet run") { t.Errorf("reports = %v", vs[0].Reports) } - if only, _ := rd.Verdicts("feat/other"); len(only) != 0 { + if only, _, _ := rd.Verdicts("feat/other"); len(only) != 0 { t.Errorf("another group's verdicts = %+v", only) } } @@ -85,11 +87,11 @@ func TestAnHqPullRequestIsAnsweredAsAnyOther(t *testing.T) { func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { // N9/AC1 rd, _ := readerOf(t) - all, err := rd.Kept("", "") + all, _, err := rd.Kept("", "") if err != nil || len(all) != 1 { t.Fatalf("kept = %v, %v", all, err) } - one, err := rd.Kept("unit", digA) + one, _, err := rd.Kept("unit", digA) if err != nil || len(one) != 1 { t.Fatalf("one kept = %v, %v", one, err) } @@ -98,10 +100,10 @@ func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { // t.Errorf("kept value = %+v, %v", k, err) } } - if none, _ := rd.Kept("integration", ""); len(none) != 0 { + if none, _, _ := rd.Kept("integration", ""); len(none) != 0 { t.Errorf("integration = %v", none) } - if _, err := rd.Kept("unit", digB); err == nil { + if _, _, err := rd.Kept("unit", digB); err == nil { t.Error("a digest nothing kept was answered") } } @@ -115,17 +117,17 @@ func TestThePreviewsToolShowsTheStoredPreviewOfAPullRequest(t *testing.T) { } key, _ := delivery.PreviewKey(repoA, head1, head2) rd := Reader{PreviewState: fakeState{key: data, "junk": []byte("{}")}.read()} - got, bad, err := rd.Previews(repoA, head1[:8]) + got, bad, _, err := rd.Previews(repoA, head1[:8]) if err != nil || len(got) != 1 || got[0].Key != key || got[0].Head != head1 || got[0].Trunk != head2 || string(got[0].Preview) != string(data) { t.Fatalf("previews = %+v, %v", got, err) } if len(bad) != 1 || bad[0] != "junk" { t.Errorf("unreadable = %v", bad) } - if got, _, _ := rd.Previews("novox/other", ""); len(got) != 0 { + if got, _, _, _ := rd.Previews("novox/other", ""); len(got) != 0 { t.Errorf("another repository's = %+v", got) } - if _, _, err := (Reader{}).Previews("", ""); err == nil { + if _, _, _, err := (Reader{}).Previews("", ""); err == nil { t.Error("answered with no state to read") } } @@ -233,3 +235,121 @@ func TestTheStatusCountsAnHQOnlyGroupAsUnjudged(t *testing.T) { t.Errorf("unjudged = %v, want pb-x", st.Unjudged) } } + +// slowBucket wraps a state with a Get that takes delay and counts the values read, the lists and the most Gets in flight. +type slowBucket struct { + vals fakeState + reads, lists atomic.Int64 + inFlight, maxFlight atomic.Int64 +} + +func (b *slowBucket) get(key string) (json.RawMessage, error) { + b.reads.Add(1) + now := b.inFlight.Add(1) + for m := b.maxFlight.Load(); now > m && !b.maxFlight.CompareAndSwap(m, now); m = b.maxFlight.Load() { + } + time.Sleep(time.Millisecond) + b.inFlight.Add(-1) + return b.vals.read().Get(key) +} + +func (b *slowBucket) read() StateRead { + return StateRead{Keys: func() ([]string, error) { b.lists.Add(1); return b.vals.read().Keys() }, Get: b.get} +} + +func (b *slowBucket) Keys() ([]string, error) { return b.read().Keys() } +func (b *slowBucket) Get(key string) (json.RawMessage, error) { return b.get(key) } +func (b *slowBucket) Create(string, []byte) error { return nil } + +// 1827: a read tool lists the keys of a bucket of thousands and reads only the values its answer needs, sixteen at a time, at +// most MaxReads, saying when it cut (on the live shape of 2026-10-26: kept 7990 keys, steps 10034, tries 1000s). +func TestReadToolsReadOnlyWhatTheyNeedInParallelAndSayWhenCut(t *testing.T) { // 1827 + kept := fakeState{} + for i := 0; i < 1800; i++ { + level := map[bool]string{true: "unit", false: "integration"}[i%9 != 0] + k, _ := delivery.KeptKey(level, fmt.Sprintf("%064x", i)) + kept[k] = []byte(`{"v":1}`) + } + previews := fakeState{} + for i := 0; i < 1500; i++ { + repo := map[bool]string{true: repoA, false: repoSDK}[i < 300] + k, _ := delivery.PreviewKey(repo, fmt.Sprintf("%08x%032x", i, 0), trunkA) + previews[k] = []byte(`{"v":1}`) + } + for _, row := range []struct { + name string + call func(*slowBucket) (rows int, cut string, err error) + rows int + reads, lists int64 + cut bool + }{ + {"kept of one key", func(b *slowBucket) (int, string, error) { + m, c, err := Reader{KeptState: b.read()}.Kept("unit", fmt.Sprintf("%064x", 1)) + return len(m), c, err + }, 1, 1, 0, false}, + {"kept of a level", func(b *slowBucket) (int, string, error) { + m, c, err := Reader{KeptState: b.read()}.Kept("integration", "") + return len(m), c, err + }, 200, 200, 1, false}, + {"kept of everything", func(b *slowBucket) (int, string, error) { + m, c, err := Reader{KeptState: b.read()}.Kept("", "") + return len(m), c, err + }, MaxReads, MaxReads, 1, true}, + {"previews of a repository", func(b *slowBucket) (int, string, error) { + p, _, c, err := Reader{PreviewState: b.read()}.Previews(repoSDK, "") + return len(p), c, err + }, 1200, 1000, 1, true}, + {"previews of a repository and a head", func(b *slowBucket) (int, string, error) { + p, _, c, err := Reader{PreviewState: b.read()}.Previews(repoA, fmt.Sprintf("%08x", 7)) + return len(p), c, err + }, 1, 1, 1, false}, + } { + b := &slowBucket{} + b.vals = kept + if strings.HasPrefix(row.name, "previews") { + b.vals = previews + } + start := time.Now() + rows, cut, err := row.call(b) + if err != nil { + t.Fatalf("%s: %v", row.name, err) + } + if rows != min(row.rows, MaxReads) || b.reads.Load() != row.reads || b.lists.Load() != row.lists || (cut != "") != row.cut { + t.Errorf("%s: %d rows, %d reads, %d lists, cut %q; want %d, %d, %d, cut %v", row.name, rows, b.reads.Load(), b.lists.Load(), cut, min(row.rows, MaxReads), row.reads, row.lists, row.cut) + } + if row.reads > 20 && b.maxFlight.Load() < 2 { + t.Errorf("%s: values were read one at a time", row.name) + } + t.Logf("%s: %d values in %v (one after the other: %v)", row.name, b.reads.Load(), time.Since(start), time.Duration(b.reads.Load())*time.Millisecond) + } + + // The status reads the newest steps of a bucket of 1500 first, at most MaxReads, and finds the one that runs. + now := time.Date(2026, 10, 26, 7, 0, 0, 0, time.UTC) + steps := fakeState{} + for i := 0; i < 1500; i++ { + id := fmt.Sprintf("mvo5%04dabcdef", i) + step := delivery.Step{ID: id, Tool: "t", Node: "g14", State: delivery.StepPassed} + if i == 1499 { + step.State, step.Started = delivery.StepRunning, now.Add(-time.Minute) + } + raw, _ := delivery.EncodeStep(step) + steps["g14."+id] = raw + } + sb := &slowBucket{vals: steps} + st, err := Reader{StepStates: map[string]StateRead{"mesh-build-go": sb.read()}, HaveState: fakeState{"g14.mesh-build-go": haveOf(t, delivery.PhaseRunning)}.read(), + Now: func() time.Time { return now }}.Status() + if err != nil || sb.reads.Load() != MaxReads || len(st.Running) != 1 || sb.maxFlight.Load() < 2 { + t.Errorf("status: %d reads (want %d), running %+v, in flight %d, %v", sb.reads.Load(), MaxReads, st.Running, sb.maxFlight.Load(), err) + } + + // A new try reads the newest TryScan of 1500 ended tries, not all, and still finds the same head asked twice. + tries := fakeState{} + for i := 0; i < 1500; i++ { + raw, _ := encodeTry(Try{ID: fmt.Sprintf("try-%d", 1700000000000000000+i), Repository: repoA, Head: head1, Trunk: trunkA, State: TryPass}) + tries[fmt.Sprintf("try-%d", 1700000000000000000+i)] = raw + } + tb := &slowBucket{vals: tries} + if _, err := AskTry(tb, tryTrunks, TryAsk{Repository: repoA, Head: head2, Trunk: trunkA}, now); err != nil || tb.reads.Load() != TryScan { + t.Errorf("AskTry read %d of 1500 tries (want %d), %v", tb.reads.Load(), TryScan, err) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index 5f0be8b..d96e2f0 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -29,6 +29,8 @@ const ( TriesBucket = delivery.CheckerModule + "_" + TriesState // MaxTries is how many tries may wait or run at once, mesh-wide: each holds a build node for minutes. MaxTries = 4 + // TryScan is how many of the newest tries a new ask reads to find the ones that wait or run (issue 1827). + TryScan = 64 // TryKeep is how long an ended try's key is kept before the checker removes it. TryKeep = 7 * 24 * time.Hour ) @@ -155,15 +157,33 @@ func AskTry(store TryStore, trunks map[string]string, a TryAsk, now time.Time) ( if err != nil { return "", fmt.Errorf("the tries cannot be listed: %w", err) } - sort.Strings(keys) - var open []string - for _, k := range keys { - raw, err := store.Get(k) + // An ended try's key stays TryKeep, so the bucket holds thousands; one that waits or runs was asked lately. Only the + // newest TryScan keys are read, in parallel (a try id is the time in nanoseconds, so the newest sorts last). + sort.Slice(keys, func(i, j int) bool { + if len(keys[i]) != len(keys[j]) { + return len(keys[i]) > len(keys[j]) + } + return keys[i] > keys[j] + }) + if len(keys) > TryScan { + keys = keys[:TryScan] + } + tries := make([]Try, len(keys)) + live := make([]bool, len(keys)) + if err := parallel(len(keys), func(i int) error { + raw, err := store.Get(keys[i]) if err != nil || len(raw) == 0 { - continue + return nil } cur, err := DecodeTry(raw) - if err != nil || cur.Over() { + tries[i], live[i] = cur, err == nil && !cur.Over() + return nil + }); err != nil { + return "", err + } + var open []string + for i, cur := range tries { + if !live[i] { continue } if cur.Repository == t.Repository && cur.Head == t.Head && cur.Trunk == t.Trunk && cur.Tests == t.Tests && cur.Level == t.Level {