Compose the declared-packages mirror only when what it composes from changed (issue 1945)
The mirror composed every node in full on the record mirrors' 10 s tick; a pass takes about 45 s on the control node, so the controller never rested (130-175 % CPU, issue 1945 note 22). It now passes on a watch of the compositions' digests and of the assignments, one pass at a time with the wakes of a pass coalesced into one after it (at least 30 s apart), skips the composition when the stores' Input digest is unchanged, and keeps a safety pass every 10 minutes that composes nothing when nothing changed.
This commit is contained in:
@@ -0,0 +1,191 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
)
|
||||
|
||||
// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
|
||||
// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
|
||||
|
||||
// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
|
||||
func scaled() packagesTiming {
|
||||
return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
|
||||
}
|
||||
|
||||
const scaledMinute = 60 * time.Millisecond
|
||||
|
||||
// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
|
||||
type fakeMesh struct {
|
||||
mu sync.Mutex
|
||||
assigned map[string][]string
|
||||
composed atomic.Int64 // node compositions
|
||||
synced atomic.Int64
|
||||
lastSync map[string][]byte
|
||||
// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
|
||||
block chan struct{}
|
||||
entered chan struct{}
|
||||
}
|
||||
|
||||
func newFakeMesh(nodes int) *fakeMesh {
|
||||
f := &fakeMesh{assigned: map[string][]string{}}
|
||||
for i := 0; i < nodes; i++ {
|
||||
f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *fakeMesh) assign(node, module string) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
|
||||
}
|
||||
|
||||
func (f *fakeMesh) keeper() *packagesKeeper {
|
||||
read := func(context.Context) (*meshInput, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
|
||||
for _, n := range nodesOfFake(f.assigned) {
|
||||
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
|
||||
}
|
||||
return &meshInput{Input: in}, nil
|
||||
}
|
||||
composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
|
||||
f.mu.Lock()
|
||||
block, entered := f.block, f.entered
|
||||
f.block = nil
|
||||
f.mu.Unlock()
|
||||
if block != nil {
|
||||
close(entered)
|
||||
<-block
|
||||
}
|
||||
out := map[string][]byte{}
|
||||
for _, n := range in.Nodes {
|
||||
f.composed.Add(1)
|
||||
out[n.Name] = []byte(fmt.Sprint(n.Assigned))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
syncAll := func(_ context.Context, want map[string][]byte) error {
|
||||
f.mu.Lock()
|
||||
f.lastSync = want
|
||||
f.mu.Unlock()
|
||||
f.synced.Add(1)
|
||||
return nil
|
||||
}
|
||||
return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
|
||||
}
|
||||
|
||||
func nodesOfFake(m map[string][]string) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
for i := range out {
|
||||
for j := i + 1; j < len(out); j++ {
|
||||
if out[j] < out[i] {
|
||||
out[i], out[j] = out[j], out[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
k.run(ctx, scaled(), func(string) {})
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
<-done
|
||||
})
|
||||
return cancel
|
||||
}
|
||||
|
||||
func eventually(t *testing.T, what string, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for !cond() {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("not within a second: %s", what)
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
|
||||
// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
|
||||
func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
|
||||
const nodes = 5
|
||||
f := newFakeMesh(nodes)
|
||||
runKeeper(t, f.keeper())
|
||||
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
|
||||
start := f.composed.Load()
|
||||
time.Sleep(scaledMinute)
|
||||
perMinute := f.composed.Load() - start
|
||||
t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
|
||||
if perMinute != 0 {
|
||||
t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
|
||||
// never two at once and never one per change.
|
||||
func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
|
||||
f := newFakeMesh(1)
|
||||
release := make(chan struct{})
|
||||
f.block, f.entered = release, make(chan struct{})
|
||||
k := f.keeper()
|
||||
runKeeper(t, k)
|
||||
<-f.entered
|
||||
for i := 0; i < 50; i++ {
|
||||
f.assign("node0", fmt.Sprintf("m%d", i))
|
||||
k.poke()
|
||||
}
|
||||
close(release)
|
||||
eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
|
||||
time.Sleep(2*scaled().gap + 20*time.Millisecond)
|
||||
if got := f.synced.Load(); got != 2 {
|
||||
t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
|
||||
func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
|
||||
f := newFakeMesh(2)
|
||||
k := f.keeper()
|
||||
runKeeper(t, k)
|
||||
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
|
||||
f.assign("node1", "viewer")
|
||||
k.poke()
|
||||
eventually(t, "the change mirrored", func() bool {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return string(f.lastSync["node1"]) == "[base viewer]"
|
||||
})
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
|
||||
func TestTheInputDigestFollowsTheInput(t *testing.T) {
|
||||
of := func(assigned ...string) *meshInput {
|
||||
return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
|
||||
}
|
||||
a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
|
||||
if a == "" || a != b {
|
||||
t.Errorf("the same input gave the digests %q and %q", a, b)
|
||||
}
|
||||
if a == c {
|
||||
t.Errorf("an added assignment left the digest at %q", a)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user