Read only the keys a view tool needs, in parallel, so node_haves and assigner_wants answer within the call's 30 s
mesh/delivery merged: source.main moved to novox/mesh-delivery@ce7d60cb6ed0
mesh/check pass 41f25d3129fd on 73be3df28cd4 | unit 21 run, 579 kept | integration 21 run, 655 kept | composition passed on shanks
mesh/delivery merged: source.main moved to novox/mesh-delivery@ce7d60cb6ed0
mesh/check pass 41f25d3129fd on 73be3df28cd4 | unit 21 run, 579 kept | integration 21 run, 655 kept | composition passed on shanks
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.
This commit is contained in:
@@ -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>.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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user