From 41f25d3129fda2a6b36ae90f8afc6c57d7da49f0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 22:33:02 +0100 Subject: [PATCH] Read only the keys a view tool needs, in parallel, so node_haves and assigner_wants answer within the call's 30 s On 2026-10-25 node_haves and assigner_wants timed out at 30 s four times, even narrowed to one node and one module, while assigner_want answered at once. Each listed every key of its bucket and then read every value one after the other, filtering only afterwards; every value is a round trip from the tool through the runtime to the bus, which opens the bucket and gets the key, so a few hundred keys outlasted the call. Now a node and a module read that one key directly, a node or a module alone filters the keys before any value is read, and the values are read sixteen at a time, at most 1000 a call, the answer saying when it was cut. indexer_modules, indexer_candidates and builder_builds read their values the same parallel way. The kind: a read tool that reads a whole bucket key by key; every list tool of the view is covered. --- .../cmd/mesh-delivery-view-tools/main.go | 9 +- .../cmd/mesh-delivery-view-tools/main_test.go | 125 ++++++++- .../mesh-delivery-view/internal/view/read.go | 258 ++++++++++++++---- 3 files changed, 325 insertions(+), 67 deletions(-) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 3ffe7d1..2cbe478 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -85,7 +85,8 @@ func Tools(r view.Reader) []stdio.Tool { }}, {Name: view.WantsReadTool, Description: "The assigner's wants and holds as stored in mesh-delivery-assigner_wants: wants by key (want..) and " + - "held fingerprints by key (held..). Narrow by node and module (a hold has no node and is shown when no node is given). Read-only.", + "held fingerprints by key (held..). Narrow by node and module (a hold has no node and is shown when no node is given); a node and a module read that one key. " + + "At most 1000 values a call; the answer says when it was cut (cut, note). Read-only.", Input: map[string]any{"type": "object", "properties": nodeModule}, Run: func(args map[string]any) (any, error) { node, module := nodeModuleOf(args) @@ -100,7 +101,8 @@ func Tools(r view.Reader) []stdio.Tool { }}, {Name: view.DifferencesReadTool, Description: "The assigner's differences between a want and what the controller sends, as stored in mesh-delivery-assigner_differences, " + - "by key (.), with the rule that explains each or unexplained. Narrow by node and module. Read-only.", + "by key (.), with the rule that explains each or unexplained. Narrow by node and module (both: that one key). " + + "At most 1000 values a call; the answer says when it was cut (cut, note). Read-only.", Input: map[string]any{"type": "object", "properties": nodeModule}, Run: func(args map[string]any) (any, error) { node, module := nodeModuleOf(args) @@ -108,7 +110,8 @@ func Tools(r view.Reader) []stdio.Tool { }}, {Name: view.HavesReadTool, Description: "What each node has, as the node-engine stored it in node-engine_have: by key . (the fingerprint it runs, its phase and health), " + - "and .node (the node's own facts). Narrow by node and module. Read-only.", + "and .node (the node's own facts). Narrow by node and module (both: that one key). " + + "At most 1000 values a call; the answer says when it was cut (cut, note). Read-only.", Input: map[string]any{"type": "object", "properties": nodeModule}, Run: func(args map[string]any) (any, error) { node, module := nodeModuleOf(args) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index 008e7d1..84cb6b1 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -4,7 +4,9 @@ import ( "encoding/json" "fmt" "strings" + "sync/atomic" "testing" + "time" "mesh-delivery-view/internal/view" ) @@ -164,7 +166,7 @@ func TestBuilderBuildJoinsTheIndexersFingerprint(t *testing.T) { // bigBuilds is a bucket of n keys over 100 modules that counts the values read. type bigBuilds struct { fake fake - reads *int + reads *atomic.Int64 } func newBig(n int) bigBuilds { @@ -172,15 +174,15 @@ func newBig(n int) bigBuilds { for i := 0; i < n; i++ { f[fmt.Sprintf("mod%02d.%064x", i%100, i)] = doneBuild } - return bigBuilds{fake: f, reads: new(int)} + return bigBuilds{fake: f, reads: new(atomic.Int64)} } func (b bigBuilds) reader() view.Reader { r := b.fake.reader() get := r.State.Get - r.Builds = view.StateRead{Keys: r.State.Keys, Get: func(k string) (json.RawMessage, error) { *b.reads++; return get(k) }, + r.Builds = view.StateRead{Keys: r.State.Keys, Get: func(k string) (json.RawMessage, error) { b.reads.Add(1); return get(k) }, GetEntry: func(k string) (json.RawMessage, uint64, error) { - *b.reads++ + b.reads.Add(1) raw, err := get(k) var rev uint64 fmt.Sscanf(k[strings.Index(k, ".")+1:], "%x", &rev) // the fake's revision is the key's number @@ -198,8 +200,8 @@ func TestBuilderBuildsFiltersByPrefixBeforeAnyGet(t *testing.T) { t.Fatal(err) } a := out.(view.BuildsAnswer) - if *b.reads != 20 { - t.Errorf("read %d values for a module of 20 keys, want 20", *b.reads) + if b.reads.Load() != 20 { + t.Errorf("read %d values for a module of 20 keys, want 20", b.reads.Load()) } if len(a.Builds) != 5 || !a.Cut || !strings.Contains(a.Note, "5") { t.Errorf("answer = %d rows, cut %v, note %q", len(a.Builds), a.Cut, a.Note) @@ -222,13 +224,13 @@ func TestBuilderBuildsWithoutAModuleIsBounded(t *testing.T) { t.Fatal(err) } a := out.(view.BuildsAnswer) - if len(a.Builds) != view.DefaultBuildsLimit || !a.Cut || *b.reads > view.DefaultBuildsLimit { - t.Errorf("%d rows, cut %v, %d reads", len(a.Builds), a.Cut, *b.reads) + if len(a.Builds) != view.DefaultBuildsLimit || !a.Cut || b.reads.Load() > view.DefaultBuildsLimit { + t.Errorf("%d rows, cut %v, %d reads", len(a.Builds), a.Cut, b.reads.Load()) } b = newBig(2000) out, _ = run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{"state": "failed"}) - if a := out.(view.BuildsAnswer); *b.reads != view.MaxBuildReads || !a.Cut || len(a.Builds) != 0 { - t.Errorf("a state matching nothing: %d reads (want the bound %d), cut %v", *b.reads, view.MaxBuildReads, a.Cut) + if a := out.(view.BuildsAnswer); b.reads.Load() != view.MaxBuildReads || !a.Cut || len(a.Builds) != 0 { + t.Errorf("a state matching nothing: %d reads (want the bound %d), cut %v", b.reads.Load(), view.MaxBuildReads, a.Cut) } } @@ -280,6 +282,41 @@ func keptReader() view.Reader { } } +// slowBucket is a bucket of the live shape (2026-10-25: node_haves and assigner_wants timed out at 30 s): keys over 4 nodes +// and perNode modules each, written by key(node, module), whose Get takes delay and counts the values read, the lists of +// the bucket and the most Gets in flight at once. +type slowBucket struct { + fake fake + delay time.Duration + reads, lists atomic.Int64 + inFlight, maxFlight atomic.Int64 +} + +func newSlow(perNode int, delay time.Duration, key func(node string, i int) string) *slowBucket { + b := &slowBucket{fake: fake{}, delay: delay} + for _, n := range []string{"ace", "g14", "novox", "shanks"} { + for i := 0; i < perNode; i++ { + b.fake[key(n, i)] = `{"fp":"x"}` + } + } + return b +} + +func (b *slowBucket) state() view.StateRead { + f := b.fake.reader().State + return view.StateRead{ + Keys: func() ([]string, error) { b.lists.Add(1); return f.Keys() }, + Get: func(k 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(b.delay) + b.inFlight.Add(-1) + return f.Get(k) + }} +} + func TestAssignerWantsToolAnswersAsStoredAndNarrows(t *testing.T) { r := keptReader() out, err := run(t, r, view.WantsReadTool, map[string]any{}) @@ -298,6 +335,39 @@ func TestAssignerWantsToolAnswersAsStoredAndNarrows(t *testing.T) { if got := out.(view.WantsAnswer).Wants; len(got) != 2 { t.Errorf("module zsh = %v", got) } + + // On the live shape (4 nodes, 150 wants each, 1 ms a Get): a node and a module read that one key and list nothing; + // a node reads only its own keys; the whole answer reads its values in parallel. + slow := newSlow(150, time.Millisecond, func(n string, i int) string { return fmt.Sprintf("want.%s.m%03d", n, i) }) + r = view.Reader{WantsState: slow.state()} + for _, row := range []struct { + args map[string]any + wants int + reads, lists int64 + }{ + {map[string]any{"node": "ace", "module": "m007"}, 1, 1, 0}, + {map[string]any{"node": "ace", "module": "gone"}, 0, 1, 0}, + {map[string]any{"node": "g14"}, 150, 150, 1}, + {map[string]any{"module": "m007"}, 4, 4, 1}, + {map[string]any{}, 600, 600, 1}, + } { + slow.reads.Store(0) + slow.lists.Store(0) + start := time.Now() + out, err := run(t, r, view.WantsReadTool, row.args) + took := time.Since(start) + if err != nil { + t.Fatalf("%v: %v", row.args, err) + } + a := out.(view.WantsAnswer) + if len(a.Wants) != row.wants || slow.reads.Load() != row.reads || slow.lists.Load() != row.lists || a.Cut { + t.Errorf("%v: %d wants, %d reads, %d lists, cut %v; want %d, %d, %d", row.args, len(a.Wants), slow.reads.Load(), slow.lists.Load(), a.Cut, row.wants, row.reads, row.lists) + } + t.Logf("assigner_wants %v: %d values in %v (one after the other: %v)", row.args, slow.reads.Load(), took, time.Duration(slow.reads.Load())*slow.delay) + } + if slow.maxFlight.Load() < 2 { + t.Errorf("the values were read one at a time (at most %d in flight)", slow.maxFlight.Load()) + } } func TestAssignerWantToolAnswersOneAndSaysWhenThereIsNone(t *testing.T) { @@ -328,4 +398,39 @@ func TestAssignerDifferencesAndNodeHavesToolsAnswerAsStored(t *testing.T) { if got := out.(view.KeptAnswer).Keys; len(got) != 2 || got["ace.node"] == nil { t.Errorf("haves of ace (the node's own key too) = %v", got) } + + // More haves than one call reads (4 nodes, 300 each, 1 ms a Get): a node and a module read that one key and list + // nothing; the whole answer reads MaxStateReads values in parallel and says it was cut. + slow := newSlow(300, time.Millisecond, func(n string, i int) string { return fmt.Sprintf("%s.m%03d", n, i) }) + r = view.Reader{HavesState: slow.state(), DifferencesState: slow.state()} + for _, row := range []struct { + tool string + args map[string]any + keys int + reads, lists int64 + cut bool + }{ + {view.HavesReadTool, map[string]any{"node": "shanks", "module": "m042"}, 1, 1, 0, false}, + {view.DifferencesReadTool, map[string]any{"node": "shanks", "module": "m042"}, 1, 1, 0, false}, + {view.HavesReadTool, map[string]any{"node": "ace"}, 300, 300, 1, false}, + {view.HavesReadTool, map[string]any{"module": "m042"}, 4, 4, 1, false}, + {view.HavesReadTool, map[string]any{}, view.MaxStateReads, view.MaxStateReads, 1, true}, + } { + slow.reads.Store(0) + slow.lists.Store(0) + start := time.Now() + out, err := run(t, r, row.tool, row.args) + took := time.Since(start) + if err != nil { + t.Fatalf("%s %v: %v", row.tool, row.args, err) + } + a := out.(view.KeptAnswer) + if len(a.Keys) != row.keys || slow.reads.Load() != row.reads || slow.lists.Load() != row.lists || a.Cut != row.cut || (a.Cut && !strings.Contains(a.Note, "narrow")) { + t.Errorf("%s %v: %d keys, %d reads, %d lists, cut %v %q; want %d, %d, %d, %v", row.tool, row.args, len(a.Keys), slow.reads.Load(), slow.lists.Load(), a.Cut, a.Note, row.keys, row.reads, row.lists, row.cut) + } + t.Logf("%s %v: %d values in %v (one after the other: %v)", row.tool, row.args, slow.reads.Load(), took, time.Duration(slow.reads.Load())*slow.delay) + } + if slow.maxFlight.Load() < 2 { + t.Errorf("the values were read one at a time (at most %d in flight)", slow.maxFlight.Load()) + } } diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 8dd8342..31be53d 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -10,6 +10,8 @@ import ( "slices" "sort" "strings" + "sync" + "sync/atomic" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -70,12 +72,13 @@ func (r Reader) Modules(none2 bool) (map[string]json.RawMessage, error) { return nil, err } sort.Strings(keys) + values, err := r.State.values(keys) + if err != nil { + return nil, err + } out := map[string]json.RawMessage{} - for _, k := range keys { - raw, err := r.State.Get(k) - if err != nil { - return nil, err - } + for i, k := range keys { + raw := values[i] if len(raw) == 0 { continue } @@ -129,13 +132,14 @@ func (r Reader) Candidates(module string) (map[string]json.RawMessage, []string, return nil, nil, err } sort.Strings(keys) + values, err := r.CandidatesState.values(keys) + if err != nil { + return nil, nil, err + } out := map[string]json.RawMessage{} var unreadable []string - for _, k := range keys { - raw, err := r.CandidatesState.Get(k) - if err != nil { - return nil, nil, err - } + for i, k := range keys { + raw := values[i] if len(raw) == 0 { continue } @@ -242,35 +246,51 @@ func (r Reader) BuilderBuilds(module, state string, limit int) (BuildsAnswer, er } candidates = append(candidates, k) } + // The values are read in batches, each batch in parallel and taken in key order: without a module or a state a batch + // is only the rows still missing, so no more is read than the answer needs; otherwise up to readWorkers*4. reads := 0 - for _, k := range candidates { +walk: + for reads < len(candidates) { if reads >= MaxBuildReads { out.Cut, out.Note = true, fmt.Sprintf("read %d values and stopped; give a module (and a state) to narrow the answer", MaxBuildReads) break } - reads++ - row, ok, err := r.builderRow(k) - if err != nil { + size := readWorkers * 4 + if prefix == "" && state == "" { + size = limit - len(out.Builds) + } + size = min(size, MaxBuildReads-reads, len(candidates)-reads) + batch := candidates[reads : reads+size] + rows := make([]BuilderRow, len(batch)) + found := make([]bool, len(batch)) + if err := parallel(len(batch), func(i int) error { + var err error + rows[i], found[i], err = r.builderRow(batch[i]) + return err + }); err != nil { return BuildsAnswer{}, err } - if !ok { - continue - } - if state != "" { - var b struct { - State string `json:"state"` - } - if json.Unmarshal(row.Build, &b) != nil || b.State != state { + reads += len(batch) + for i, row := range rows { + if !found[i] { continue } - } - out.Builds = append(out.Builds, row) - // Without a module the walk is in key order and stops at the limit; with one, all its (few) keys are read to order them. - if prefix == "" && len(out.Builds) >= limit { - if reads < len(candidates) { - out.Cut, out.Note = true, fmt.Sprintf("stopped at the limit of %d builds in key order, not newest first; give a module for its newest", limit) + if state != "" { + var b struct { + State string `json:"state"` + } + if json.Unmarshal(row.Build, &b) != nil || b.State != state { + continue + } + } + out.Builds = append(out.Builds, row) + // Without a module the walk is in key order and stops at the limit; with one, all its (few) keys are read to order them. + if prefix == "" && len(out.Builds) >= limit { + if reads-len(batch)+i+1 < len(candidates) { + out.Cut, out.Note = true, fmt.Sprintf("stopped at the limit of %d builds in key order, not newest first; give a module for its newest", limit) + } + break walk } - break } } sort.SliceStable(out.Builds, func(i, j int) bool { @@ -352,6 +372,9 @@ type WantsAnswer struct { Wants map[string]json.RawMessage `json:"wants"` Held map[string]json.RawMessage `json:"held"` Unreadable []string `json:"unreadable,omitempty"` + // Cut says the answer stopped at MaxStateReads values, and Note how to narrow it. + Cut bool `json:"cut,omitempty"` + Note string `json:"note,omitempty"` } // KeptAnswer is the answer of assigner_differences and node_haves: the keys of the bucket as stored, and the keys that @@ -359,60 +382,158 @@ type WantsAnswer struct { type KeptAnswer struct { Keys map[string]json.RawMessage `json:"keys"` Unreadable []string `json:"unreadable,omitempty"` + Cut bool `json:"cut,omitempty"` + Note string `json:"note,omitempty"` } -func (s StateRead) all(what string) (map[string]json.RawMessage, error) { - if s.Keys == nil || s.Get == nil { - return nil, fmt.Errorf("%s cannot be read here", what) - } - keys, err := s.Keys() - if err != nil { - return nil, err +// 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 (the runtime opens the bucket and gets the key: two bus requests), so reading a bucket of a few +// hundred keys one after the other outlasted the 30 s a tool call is given on 2026-10-25 (node_haves, assigner_wants). +const readWorkers = 16 + +// MaxStateReads bounds the values one call of assigner_wants, assigner_differences or node_haves reads; past it the +// answer says it was cut and how to narrow it. +const MaxStateReads = 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 + ) + workers := min(readWorkers, n) + for w := 0; w < workers; 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 +} + +// values reads the stored value of each key, in parallel, in the order of keys (nil for a key with none). +func (s StateRead) values(keys []string) ([]json.RawMessage, error) { + out := make([]json.RawMessage, len(keys)) + err := parallel(len(keys), func(i int) error { + raw, err := s.Get(keys[i]) + out[i] = raw + return err + }) + return out, err +} + +// read reads the values of keys (sorted first) into a map, at most MaxStateReads of them; cut says it stopped there. +func (s StateRead) read(keys []string) (map[string]json.RawMessage, bool, error) { sort.Strings(keys) + cut := len(keys) > MaxStateReads + if cut { + keys = keys[:MaxStateReads] + } + values, err := s.values(keys) + if err != nil { + return nil, false, err + } out := map[string]json.RawMessage{} - for _, k := range keys { - raw, err := s.Get(k) - if err != nil { - return nil, err - } - if len(raw) > 0 { - out[k] = raw + for i, k := range keys { + if len(values[i]) > 0 { + out[k] = values[i] } } - return out, nil + return out, cut, nil +} + +func cutNote(cut bool, n int) string { + if !cut { + return "" + } + return fmt.Sprintf("read the first %d of %d matching keys in key order and stopped; give a node and a module to narrow the answer", MaxStateReads, n) } // Wants is every want and hold of the assigner, narrowed by node and module (a hold has no node and is kept when the module // matches). func (r Reader) Wants(node, module string) (WantsAnswer, error) { - all, err := r.WantsState.all("the assigner's wants") + s := r.WantsState + if s.Get == nil { + return WantsAnswer{}, fmt.Errorf("the assigner's wants cannot be read here") + } + out := WantsAnswer{Wants: map[string]json.RawMessage{}, Held: map[string]json.RawMessage{}} + if node != "" && module != "" { + // One key, read directly: no list of the bucket (a hold has no node, so none is shown). + key, err := delivery.WantKey(node, module) + if module == "node" { + key, err = delivery.WantNodeKey(node) + } + if err != nil { + return WantsAnswer{}, err + } + raw, err := s.Get(key) + if err != nil { + return WantsAnswer{}, err + } + if len(raw) > 0 { + out.Wants[key] = raw + } + return out, nil + } + if s.Keys == nil { + return WantsAnswer{}, fmt.Errorf("the assigner's wants cannot be read here") + } + keys, err := s.Keys() if err != nil { return WantsAnswer{}, err } - out := WantsAnswer{Wants: map[string]json.RawMessage{}, Held: map[string]json.RawMessage{}} - for k, raw := range all { + // Narrowed by the key before any value is read. + var matching []string + for _, k := range keys { if n, m, err := delivery.ParseWantKey(k); err == nil { if (node == "" || n == node) && (module == "" || m == module) { - out.Wants[k] = raw + matching = append(matching, k) } continue } // A node's own want, want..node (ADR 0539 clause 6): listed with the wants, under the module name "node". if n, err := delivery.ParseWantNodeKey(k); err == nil { if (node == "" || n == node) && (module == "" || module == "node") { - out.Wants[k] = raw + matching = append(matching, k) } continue } if m, _, err := delivery.ParseHeldKey(k); err == nil { if node == "" && (module == "" || m == module) { - out.Held[k] = raw + matching = append(matching, k) } continue } out.Unreadable = append(out.Unreadable, k) } + sort.Strings(out.Unreadable) + values, cut, err := s.read(matching) + if err != nil { + return WantsAnswer{}, err + } + out.Cut, out.Note = cut, cutNote(cut, len(matching)) + for k, raw := range values { + if _, _, err := delivery.ParseHeldKey(k); err == nil { + out.Held[k] = raw + } else { + out.Wants[k] = raw + } + } return out, nil } @@ -436,21 +557,50 @@ func (r Reader) Want(node, module string) (json.RawMessage, error) { } func (r Reader) kept(s StateRead, what, node, module string) (KeptAnswer, error) { - all, err := s.all(what) + if s.Get == nil { + return KeptAnswer{}, fmt.Errorf("%s cannot be read here", what) + } + out := KeptAnswer{Keys: map[string]json.RawMessage{}} + if node != "" && module != "" { + // One key, read directly: no list of the bucket. + key, err := delivery.HaveKey(node, module) + if err != nil { + return KeptAnswer{}, err + } + raw, err := s.Get(key) + if err != nil { + return KeptAnswer{}, err + } + if len(raw) > 0 { + out.Keys[key] = raw + } + return out, nil + } + if s.Keys == nil { + return KeptAnswer{}, fmt.Errorf("%s cannot be read here", what) + } + keys, err := s.Keys() if err != nil { return KeptAnswer{}, err } - out := KeptAnswer{Keys: map[string]json.RawMessage{}} - for k, raw := range all { + // Narrowed by the key before any value is read. + var matching []string + for _, k := range keys { n, m, isNode, err := delivery.ParseHaveKey(k) if err != nil { out.Unreadable = append(out.Unreadable, k) continue } if (node == "" || n == node) && (module == "" || (!isNode && m == module)) { - out.Keys[k] = raw + matching = append(matching, k) } } + sort.Strings(out.Unreadable) + values, cut, err := s.read(matching) + if err != nil { + return KeptAnswer{}, err + } + out.Keys, out.Cut, out.Note = values, cut, cutNote(cut, len(matching)) return out, nil }