Build modules' step and toolchain lists read only the newest keys they answer, in parallel (issue 1827) #485

Merged
mesh-tools merged 2 commits from fix/1827-bounded-reads into main 2026-10-25 23:56:13 +00:00
6 changed files with 464 additions and 60 deletions
+97 -22
View File
@@ -7,9 +7,12 @@ package main
import (
"encoding/json"
"fmt"
"log"
"sort"
"strconv"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -39,6 +42,53 @@ func readOf(s stdio.KeptState) StateRead {
}}
}
// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them
// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read
// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB).
const (
MaxStateReads = 1000
stateReadParallel = 16
)
// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It
// stops starting reads after the first error and returns it.
func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) {
out := make([]json.RawMessage, len(keys))
var (
wg sync.WaitGroup
mu sync.Mutex
fail error
next = make(chan int)
)
for w := 0; w < stateReadParallel && w < len(keys); w++ {
wg.Add(1)
go func() {
defer wg.Done()
for i := range next {
raw, err := get(keys[i])
mu.Lock()
if err != nil && fail == nil {
fail = err
}
mu.Unlock()
out[i] = raw
}
}()
}
for i := range keys {
mu.Lock()
stop := fail != nil
mu.Unlock()
if stop {
break
}
next <- i
}
close(next)
wg.Wait()
return out, fail
}
// stepKey is the key of a step of this node: the id alone, or the whole key <node>.<id> as a step tool answered it.
func (m *Module) stepKey(id string) (string, error) {
id = strings.TrimSpace(id)
@@ -80,27 +130,41 @@ type StepLine struct {
Reason string `json:"reason,omitempty"`
}
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero).
func (m *Module) RecentSteps(limit int) ([]StepLine, error) {
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not
// read. The values still order the lines, since an ask step's id carries the ask time.
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
if limit <= 0 {
limit = 10
}
if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil {
return nil, fmt.Errorf("the steps state cannot be read here")
return nil, false, fmt.Errorf("the steps state cannot be read here")
}
start := time.Now()
keys, err := m.ReadSteps.Keys()
if err != nil {
return nil, err
return nil, false, err
}
own := keys[:0:0]
for _, k := range keys {
if strings.HasPrefix(k, m.Node+".") {
if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node {
own = append(own, k)
}
}
}
sort.Sort(sort.Reverse(sort.StringSlice(own)))
if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
own, cut = own[:n], true
}
raws, err := readValues(own, m.ReadSteps.Get)
log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut)
if err != nil {
return nil, cut, err
}
var steps []delivery.Step
for _, k := range keys {
if node, _, err := delivery.ParseStepKey(k); err != nil || node != m.Node {
continue
}
raw, err := m.ReadSteps.Get(k)
if err != nil {
return nil, err
}
for _, raw := range raws {
if len(raw) == 0 {
continue
}
@@ -122,7 +186,7 @@ func (m *Module) RecentSteps(limit int) ([]StepLine, error) {
}
out = append(out, l)
}
return out, nil
return out, cut, nil
}
// ReadToolchains is every key of the toolchains state (<node>.<language>) with its stored value.
@@ -130,24 +194,30 @@ func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) {
if m.ReadToolchainsState.Keys == nil || m.ReadToolchainsState.Get == nil {
return nil, fmt.Errorf("the toolchains state cannot be read here")
}
start := time.Now()
keys, err := m.ReadToolchainsState.Keys()
if err != nil {
return nil, err
}
cut := len(keys) > MaxStateReads
if cut {
keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
}
raws, err := readValues(keys, m.ReadToolchainsState.Get)
log.Printf("toolchains read: %d keys read in %v (cut %v: keys past %d were not read)", len(keys), time.Since(start).Round(time.Millisecond), cut, MaxStateReads)
if err != nil {
return nil, err
}
out := map[string]delivery.Toolchain{}
for _, k := range keys {
raw, err := m.ReadToolchainsState.Get(k)
if err != nil {
return nil, err
}
for i, raw := range raws {
if len(raw) == 0 {
continue
}
t, err := delivery.DecodeToolchain(raw)
if err != nil {
return nil, fmt.Errorf("%s: %w", k, err)
return nil, fmt.Errorf("%s: %w", keys[i], err)
}
out[k] = t
out[keys[i]] = t
}
return out, nil
}
@@ -171,8 +241,13 @@ func (m *Module) ReadTools() []stdio.Tool {
Input: map[string]any{"type": "object", "properties": map[string]any{
"limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}},
Run: func(args map[string]any) (any, error) {
steps, err := m.RecentSteps(intArg(args["limit"]))
return map[string]any{"node": m.Node, "steps": steps}, err
steps, cut, err := m.RecentSteps(intArg(args["limit"]))
ans := map[string]any{"node": m.Node, "steps": steps}
if cut {
ans["cut"] = true
ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads)
}
return ans, err
}},
{Name: ToolchainsReadTool,
Description: "What the module's toolchains state holds: each key <node>.<language> with the version installed, the " +
@@ -3,8 +3,10 @@ package main
import (
"context"
"encoding/json"
"fmt"
"slices"
"strings"
"sync/atomic"
"testing"
"time"
@@ -145,3 +147,62 @@ func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) {
t.Errorf("the read refused: %v", err)
}
}
// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool
// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut.
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
for _, row := range []struct {
perNode, limit int
reads int64
steps int
cut bool
}{
{20, 0, 20, 10, false},
{600, 0, 100, 10, true},
{1500, 300, MaxStateReads, 300, true},
} {
var reads, inFlight, maxFlight atomic.Int64
stored := map[string][]byte{}
for _, n := range []string{"ace", "g14", "novox", "shanks"} {
for i := 0; i < row.perNode; i++ {
b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)})
stored[n+".s"+fmt.Sprintf("%04d", i)] = b
}
}
m := &Module{Node: "shanks"}
m.ReadSteps = StateRead{
Keys: func() ([]string, error) {
var k []string
for key := range stored {
k = append(k, key)
}
return k, nil
},
Get: func(key string) (json.RawMessage, error) {
reads.Add(1)
now := inFlight.Add(1)
for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() {
}
time.Sleep(time.Millisecond)
inFlight.Add(-1)
if !strings.HasPrefix(key, "shanks.") {
t.Errorf("read %s, a key of another node", key)
}
return stored[key], nil
},
}
start := time.Now()
lines, cut, err := m.RecentSteps(row.limit)
if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads)
}
if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) {
t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID)
}
if maxFlight.Load() < 2 {
t.Errorf("%d a node: values read one at a time", row.perNode)
}
t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond)
}
}
@@ -7,9 +7,12 @@ package main
import (
"encoding/json"
"fmt"
"log"
"sort"
"strconv"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -38,6 +41,53 @@ func readOf(s stdio.KeptState) StateRead {
}}
}
// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them
// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read
// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB).
const (
MaxStateReads = 1000
stateReadParallel = 16
)
// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It
// stops starting reads after the first error and returns it.
func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) {
out := make([]json.RawMessage, len(keys))
var (
wg sync.WaitGroup
mu sync.Mutex
fail error
next = make(chan int)
)
for w := 0; w < stateReadParallel && w < len(keys); w++ {
wg.Add(1)
go func() {
defer wg.Done()
for i := range next {
raw, err := get(keys[i])
mu.Lock()
if err != nil && fail == nil {
fail = err
}
mu.Unlock()
out[i] = raw
}
}()
}
for i := range keys {
mu.Lock()
stop := fail != nil
mu.Unlock()
if stop {
break
}
next <- i
}
close(next)
wg.Wait()
return out, fail
}
// stepKey is the key of a step of this node: the id alone, or the whole key <node>.<id> as a step tool answered it.
func (m *Module) stepKey(id string) (string, error) {
id = strings.TrimSpace(id)
@@ -79,27 +129,41 @@ type StepLine struct {
Reason string `json:"reason,omitempty"`
}
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero).
func (m *Module) RecentSteps(limit int) ([]StepLine, error) {
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not
// read. The values still order the lines, since an ask step's id carries the ask time.
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
if limit <= 0 {
limit = 10
}
if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil {
return nil, fmt.Errorf("the steps state cannot be read here")
return nil, false, fmt.Errorf("the steps state cannot be read here")
}
start := time.Now()
keys, err := m.ReadSteps.Keys()
if err != nil {
return nil, err
return nil, false, err
}
own := keys[:0:0]
for _, k := range keys {
if strings.HasPrefix(k, m.Node+".") {
if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node {
own = append(own, k)
}
}
}
sort.Sort(sort.Reverse(sort.StringSlice(own)))
if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
own, cut = own[:n], true
}
raws, err := readValues(own, m.ReadSteps.Get)
log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut)
if err != nil {
return nil, cut, err
}
var steps []delivery.Step
for _, k := range keys {
if node, _, err := delivery.ParseStepKey(k); err != nil || node != m.Node {
continue
}
raw, err := m.ReadSteps.Get(k)
if err != nil {
return nil, err
}
for _, raw := range raws {
if len(raw) == 0 {
continue
}
@@ -121,7 +185,7 @@ func (m *Module) RecentSteps(limit int) ([]StepLine, error) {
}
out = append(out, l)
}
return out, nil
return out, cut, nil
}
// ReadTools are the module's read tools. All are read-only.
@@ -143,8 +207,13 @@ func (m *Module) ReadTools() []stdio.Tool {
Input: map[string]any{"type": "object", "properties": map[string]any{
"limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}},
Run: func(args map[string]any) (any, error) {
steps, err := m.RecentSteps(intArg(args["limit"]))
return map[string]any{"node": m.Node, "steps": steps}, err
steps, cut, err := m.RecentSteps(intArg(args["limit"]))
ans := map[string]any{"node": m.Node, "steps": steps}
if cut {
ans["cut"] = true
ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads)
}
return ans, err
}},
}
}
@@ -2,7 +2,11 @@ package main
import (
"encoding/json"
"fmt"
"strings"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -39,8 +43,67 @@ func TestTheReadToolsAnswerAStepsResult(t *testing.T) {
if err := json.Unmarshal(got.Result, &back); err != nil || back.Digest != res.Digest {
t.Errorf("result = %s, %v", got.Result, err)
}
lines, err := m.RecentSteps(0)
lines, _, err := m.RecentSteps(0)
if err != nil || len(lines) != 1 || lines[0].Tool != BuildTool {
t.Errorf("steps = %+v, %v", lines, err)
}
}
// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool
// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut.
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
for _, row := range []struct {
perNode, limit int
reads int64
steps int
cut bool
}{
{20, 0, 20, 10, false},
{600, 0, 100, 10, true},
{1500, 300, MaxStateReads, 300, true},
} {
var reads, inFlight, maxFlight atomic.Int64
stored := map[string][]byte{}
for _, n := range []string{"ace", "g14", "novox", "shanks"} {
for i := 0; i < row.perNode; i++ {
b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)})
stored[n+".s"+fmt.Sprintf("%04d", i)] = b
}
}
m := &Module{Node: "shanks"}
m.ReadSteps = StateRead{
Keys: func() ([]string, error) {
var k []string
for key := range stored {
k = append(k, key)
}
return k, nil
},
Get: func(key string) (json.RawMessage, error) {
reads.Add(1)
now := inFlight.Add(1)
for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() {
}
time.Sleep(time.Millisecond)
inFlight.Add(-1)
if !strings.HasPrefix(key, "shanks.") {
t.Errorf("read %s, a key of another node", key)
}
return stored[key], nil
},
}
start := time.Now()
lines, cut, err := m.RecentSteps(row.limit)
if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads)
}
if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) {
t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID)
}
if maxFlight.Load() < 2 {
t.Errorf("%d a node: values read one at a time", row.perNode)
}
t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond)
}
}
@@ -7,9 +7,12 @@ package main
import (
"encoding/json"
"fmt"
"log"
"sort"
"strconv"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -39,6 +42,53 @@ func readOf(s stdio.KeptState) StateRead {
}}
}
// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them
// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read
// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB).
const (
MaxStateReads = 1000
stateReadParallel = 16
)
// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It
// stops starting reads after the first error and returns it.
func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) {
out := make([]json.RawMessage, len(keys))
var (
wg sync.WaitGroup
mu sync.Mutex
fail error
next = make(chan int)
)
for w := 0; w < stateReadParallel && w < len(keys); w++ {
wg.Add(1)
go func() {
defer wg.Done()
for i := range next {
raw, err := get(keys[i])
mu.Lock()
if err != nil && fail == nil {
fail = err
}
mu.Unlock()
out[i] = raw
}
}()
}
for i := range keys {
mu.Lock()
stop := fail != nil
mu.Unlock()
if stop {
break
}
next <- i
}
close(next)
wg.Wait()
return out, fail
}
// stepKey is the key of a step of this node: the id alone, or the whole key <node>.<id> as a step tool answered it.
func (m *Module) stepKey(id string) (string, error) {
id = strings.TrimSpace(id)
@@ -80,27 +130,41 @@ type StepLine struct {
Reason string `json:"reason,omitempty"`
}
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero).
func (m *Module) RecentSteps(limit int) ([]StepLine, error) {
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not
// read. The values still order the lines, since an ask step's id carries the ask time.
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
if limit <= 0 {
limit = 10
}
if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil {
return nil, fmt.Errorf("the steps state cannot be read here")
return nil, false, fmt.Errorf("the steps state cannot be read here")
}
start := time.Now()
keys, err := m.ReadSteps.Keys()
if err != nil {
return nil, err
return nil, false, err
}
own := keys[:0:0]
for _, k := range keys {
if strings.HasPrefix(k, m.Node+".") {
if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node {
own = append(own, k)
}
}
}
sort.Sort(sort.Reverse(sort.StringSlice(own)))
if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
own, cut = own[:n], true
}
raws, err := readValues(own, m.ReadSteps.Get)
log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut)
if err != nil {
return nil, cut, err
}
var steps []delivery.Step
for _, k := range keys {
if node, _, err := delivery.ParseStepKey(k); err != nil || node != m.Node {
continue
}
raw, err := m.ReadSteps.Get(k)
if err != nil {
return nil, err
}
for _, raw := range raws {
if len(raw) == 0 {
continue
}
@@ -122,7 +186,7 @@ func (m *Module) RecentSteps(limit int) ([]StepLine, error) {
}
out = append(out, l)
}
return out, nil
return out, cut, nil
}
// ReadToolchains is every key of the toolchains state (<node>.<language>) with its stored value.
@@ -130,24 +194,30 @@ func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) {
if m.ReadToolchainsState.Keys == nil || m.ReadToolchainsState.Get == nil {
return nil, fmt.Errorf("the toolchains state cannot be read here")
}
start := time.Now()
keys, err := m.ReadToolchainsState.Keys()
if err != nil {
return nil, err
}
cut := len(keys) > MaxStateReads
if cut {
keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
}
raws, err := readValues(keys, m.ReadToolchainsState.Get)
log.Printf("toolchains read: %d keys read in %v (cut %v: keys past %d were not read)", len(keys), time.Since(start).Round(time.Millisecond), cut, MaxStateReads)
if err != nil {
return nil, err
}
out := map[string]delivery.Toolchain{}
for _, k := range keys {
raw, err := m.ReadToolchainsState.Get(k)
if err != nil {
return nil, err
}
for i, raw := range raws {
if len(raw) == 0 {
continue
}
t, err := delivery.DecodeToolchain(raw)
if err != nil {
return nil, fmt.Errorf("%s: %w", k, err)
return nil, fmt.Errorf("%s: %w", keys[i], err)
}
out[k] = t
out[keys[i]] = t
}
return out, nil
}
@@ -171,8 +241,13 @@ func (m *Module) ReadTools() []stdio.Tool {
Input: map[string]any{"type": "object", "properties": map[string]any{
"limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}},
Run: func(args map[string]any) (any, error) {
steps, err := m.RecentSteps(intArg(args["limit"]))
return map[string]any{"node": m.Node, "steps": steps}, err
steps, cut, err := m.RecentSteps(intArg(args["limit"]))
ans := map[string]any{"node": m.Node, "steps": steps}
if cut {
ans["cut"] = true
ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads)
}
return ans, err
}},
{Name: ToolchainsReadTool,
Description: "What the module's toolchains state holds: each key <node>.<language> with the version installed, the " +
@@ -3,8 +3,10 @@ package main
import (
"context"
"encoding/json"
"fmt"
"slices"
"strings"
"sync/atomic"
"testing"
"time"
@@ -145,3 +147,62 @@ func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) {
t.Errorf("the read refused: %v", err)
}
}
// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool
// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut.
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
for _, row := range []struct {
perNode, limit int
reads int64
steps int
cut bool
}{
{20, 0, 20, 10, false},
{600, 0, 100, 10, true},
{1500, 300, MaxStateReads, 300, true},
} {
var reads, inFlight, maxFlight atomic.Int64
stored := map[string][]byte{}
for _, n := range []string{"ace", "g14", "novox", "shanks"} {
for i := 0; i < row.perNode; i++ {
b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)})
stored[n+".s"+fmt.Sprintf("%04d", i)] = b
}
}
m := &Module{Node: "shanks"}
m.ReadSteps = StateRead{
Keys: func() ([]string, error) {
var k []string
for key := range stored {
k = append(k, key)
}
return k, nil
},
Get: func(key string) (json.RawMessage, error) {
reads.Add(1)
now := inFlight.Add(1)
for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() {
}
time.Sleep(time.Millisecond)
inFlight.Add(-1)
if !strings.HasPrefix(key, "shanks.") {
t.Errorf("read %s, a key of another node", key)
}
return stored[key], nil
},
}
start := time.Now()
lines, cut, err := m.RecentSteps(row.limit)
if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads)
}
if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) {
t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID)
}
if maxFlight.Load() < 2 {
t.Errorf("%d a node: values read one at a time", row.perNode)
}
t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond)
}
}