diff --git a/cmd/server/analytics_pause_when_idle_test.go b/cmd/server/analytics_pause_when_idle_test.go new file mode 100644 index 00000000..a501997b --- /dev/null +++ b/cmd/server/analytics_pause_when_idle_test.go @@ -0,0 +1,142 @@ +package main + +import ( + "encoding/json" + "sync/atomic" + "testing" + "time" +) + +// The analytics recomputers (#1240) recompute every endpoint on a fixed +// interval whether or not anyone reads it. On a production instance the +// topology, RF and hash-size endpoints were each read 23 times in 45 hours +// while being recomputed every 5 minutes, and the recompute loop accounted for +// 30% of all bytes allocated (121 GB in 45 h), which drives the collector. +// +// With pauseWhenIdle, a tick with no read since the previous compute is +// skipped, and the next Load serves the existing snapshot and kicks a refresh. + +func startCountingRecomputer(t *testing.T, interval time.Duration, pause bool) (*analyticsRecomputer, *int64) { + t.Helper() + var runs int64 + r := newAnalyticsRecomputer("test", interval, func() interface{} { + return atomic.AddInt64(&runs, 1) + }) + r.pauseWhenIdle = pause + r.Start() + t.Cleanup(r.Stop) + return r, &runs +} + +func TestAnalyticsRecomputer_PauseWhenIdleSkipsUnreadTicks(t *testing.T) { + _, runs := startCountingRecomputer(t, 20*time.Millisecond, true) + time.Sleep(300 * time.Millisecond) // ~15 ticks, no reads + if got := atomic.LoadInt64(runs); got > 2 { + t.Fatalf("expected the idle recomputer to stop after its first pass, got %d computes", got) + } +} + +func TestAnalyticsRecomputer_ReadAfterPauseServesSnapshotAndRefreshes(t *testing.T) { + r, runs := startCountingRecomputer(t, 20*time.Millisecond, true) + time.Sleep(150 * time.Millisecond) // pause + before := atomic.LoadInt64(runs) + + if v := r.Load(); v == nil { + t.Fatal("expected the existing snapshot while paused, got nil") + } + deadline := time.Now().Add(2 * time.Second) + for atomic.LoadInt64(runs) == before && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if atomic.LoadInt64(runs) == before { + t.Fatal("expected a read on a paused recomputer to start a refresh") + } + if got, ok := r.Load().(int64); !ok || got <= before { + t.Fatalf("expected a newer snapshot after the refresh, got %v", r.Load()) + } +} + +func TestAnalyticsRecomputer_KeepsTickingWhileRead(t *testing.T) { + r, runs := startCountingRecomputer(t, 20*time.Millisecond, true) + stop := time.Now().Add(300 * time.Millisecond) + for time.Now().Before(stop) { + r.Load() + time.Sleep(5 * time.Millisecond) + } + if got := atomic.LoadInt64(runs); got < 6 { + t.Fatalf("expected a recomputer that is being read to keep refreshing, got %d computes", got) + } +} + +// Default behaviour is unchanged: without pauseWhenIdle it recomputes on every +// tick, read or not. +func TestAnalyticsRecomputer_DefaultKeepsTickingUnread(t *testing.T) { + _, runs := startCountingRecomputer(t, 20*time.Millisecond, false) + time.Sleep(300 * time.Millisecond) + if got := atomic.LoadInt64(runs); got < 6 { + t.Fatalf("expected the default recomputer to keep ticking, got %d computes", got) + } +} + +func TestConfig_AnalyticsPauseWhenIdle(t *testing.T) { + var on, off Config + if err := json.Unmarshal([]byte(`{"analytics":{"pauseWhenIdle":true}}`), &on); err != nil { + t.Fatal(err) + } + if err := json.Unmarshal([]byte(`{"analytics":{"defaultIntervalSeconds":300}}`), &off); err != nil { + t.Fatal(err) + } + if !on.AnalyticsPauseWhenIdle() { + t.Error("expected pauseWhenIdle:true to enable it") + } + if off.AnalyticsPauseWhenIdle() { + t.Error("expected it to default to off") + } + if (*Config)(nil).AnalyticsPauseWhenIdle() { + t.Error("expected a nil config to default to off") + } +} + +func TestStartAnalyticsRecomputers_AppliesPauseWhenIdle(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + stop := store.StartAnalyticsRecomputers(time.Hour, AnalyticsRecomputeIntervals{PauseWhenIdle: true}) + defer stop() + store.analyticsRecomputerMu.Lock() + all := store.analyticsRecomputersLocked() + store.analyticsRecomputerMu.Unlock() + for _, rc := range all { + if !rc.pauseWhenIdle { + t.Errorf("recomputer %q did not get pauseWhenIdle", rc.name) + } + } +} + +// The node-health "Heard By" card reads the direct-heard snapshot through +// loadDirectHeard, not through the recomputer's Load. With pauseWhenIdle on, +// that read must still count, or direct-heard pauses after startup and never +// recomputes again: nodes first heard later show "nobody hears this node". +func TestDirectHeard_PausedRecomputerResumesWhenTheCardIsRead(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + stop := store.StartAnalyticsRecomputers(20*time.Millisecond, AnalyticsRecomputeIntervals{PauseWhenIdle: true}) + defer stop() + store.analyticsRecomputerMu.Lock() + rc := store.recompDirectHeard + store.analyticsRecomputerMu.Unlock() + if rc == nil { + t.Fatal("no direct-heard recomputer") + } + time.Sleep(150 * time.Millisecond) // unread: pauses + before := rc.computeRuns.Load() + deadline := time.Now().Add(300 * time.Millisecond) + for time.Now().Before(deadline) { + store.loadDirectHeard() // what the Heard By card calls + time.Sleep(10 * time.Millisecond) + } + if got := rc.computeRuns.Load() - before; got < 2 { + t.Errorf("direct-heard recomputed %d times in 300 ms of card reads at a 20 ms interval, want it to keep refreshing", got) + } +} diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go index db6b5dbf..13ae6f58 100644 --- a/cmd/server/analytics_recomputer.go +++ b/cmd/server/analytics_recomputer.go @@ -45,6 +45,15 @@ type analyticsRecomputer struct { startOnce sync.Once stopOnce sync.Once + // pauseWhenIdle: see AnalyticsConfig.PauseWhenIdle. A tick with no + // Load since the previous compute is skipped and the recomputer is + // marked paused; the next Load serves the existing snapshot and sends + // on kick, and the loop recomputes then. + pauseWhenIdle bool + readSinceCompute atomic.Bool + paused atomic.Bool + kick chan struct{} // buffered(1): a paused recomputer was read + // Stats (atomic). computeRuns atomic.Int64 lastComputeNs atomic.Int64 // duration of last compute in nanoseconds @@ -71,6 +80,7 @@ func newAnalyticsRecomputer(name string, interval time.Duration, compute func() stop: make(chan struct{}), done: make(chan struct{}), recomputeReq: make(chan chan struct{}), + kick: make(chan struct{}, 1), } } @@ -100,7 +110,16 @@ func (r *analyticsRecomputer) loop() { for { select { case <-t.C: + if r.pauseWhenIdle && !r.readSinceCompute.Load() { + r.paused.Store(true) + continue + } r.runOnce() + case <-r.kick: + if r.paused.Load() { + r.runOnce() + t.Reset(r.interval) + } case ack := <-r.recomputeReq: r.runOnce() t.Reset(r.interval) @@ -127,6 +146,9 @@ func (r *analyticsRecomputer) runOnce() { // started on a partially loaded store must not end the warm-up, // even if the load finishes while it runs. ready := r.warmupReadyGateOpen_1659() + // Reads from here on count towards the next tick (pauseWhenIdle). + r.readSinceCompute.Store(false) + r.paused.Store(false) t0 := time.Now() result := r.compute() r.lastComputeNs.Store(int64(time.Since(t0))) @@ -190,10 +212,30 @@ func recomputeWhenLoaded(loaded, stop <-chan struct{}, rcs []*analyticsRecompute len(rcs), time.Since(t0).Round(time.Millisecond), strings.Join(parts, " ")) } +// markRead counts a read for pauseWhenIdle: the next tick recomputes, and a +// paused recomputer is kicked to refresh now. Readers that take the +// snapshot some other way (direct-heard publishes its own) call it +// directly. Never blocks; a no-op when pauseWhenIdle is off. +func (r *analyticsRecomputer) markRead() { + if !r.pauseWhenIdle { + return + } + r.readSinceCompute.Store(true) + if r.paused.Load() { + select { + case r.kick <- struct{}{}: + default: // a refresh is already queued + } + } +} + // Load returns the most recently computed snapshot, or nil if Start // has not been called (or the very first compute returned nil). -// Never blocks beyond a single atomic load. +// Never blocks: beyond the atomic load, a paused recomputer (pauseWhenIdle) +// gets a non-blocking send on kick, and the snapshot it already holds is +// returned while the refresh runs. func (r *analyticsRecomputer) Load() interface{} { + r.markRead() v := r.cache.Load() if v == nil { return nil @@ -238,6 +280,9 @@ type AnalyticsRecomputeIntervals struct { Roles time.Duration ObserversClockSkew time.Duration NodesClockSkew time.Duration + + // PauseWhenIdle is applied to every recomputer (AnalyticsConfig.PauseWhenIdle). + PauseWhenIdle bool } func pickInterval(override, def time.Duration) time.Duration { @@ -352,6 +397,9 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o }, ) all := s.analyticsRecomputersLocked() + for _, rc := range all { + rc.pauseWhenIdle = ov.PauseWhenIdle + } s.analyticsRecomputerMu.Unlock() // Issue #1659 (PR #1688 r1, munger #5): wire the loader readiness diff --git a/cmd/server/config.go b/cmd/server/config.go index af4477a4..5c4de116 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -1138,6 +1138,11 @@ func (c *Config) IsObserverBlacklisted(id string) bool { type AnalyticsConfig struct { DefaultIntervalSeconds int `json:"defaultIntervalSeconds,omitempty"` RecomputeIntervalSeconds map[string]int `json:"recomputeIntervalSeconds,omitempty"` + // PauseWhenIdle skips a scheduled recompute when nothing has read that + // endpoint's snapshot since the previous recompute. The first read + // after a pause is served the existing snapshot at once and starts a + // refresh. Off by default (always recompute on the interval). + PauseWhenIdle bool `json:"pauseWhenIdle,omitempty"` // LoRaPreset is the assumed PHY preset used by the relay-airtime-share // metric to compute true Time-on-Air (issue #1768). Defaults to the // EU MeshCore deployment: 869.6 MHz / BW 62.5 kHz / SF 8 / CR 4/5. @@ -1164,11 +1169,16 @@ func (c *Config) AnalyticsDefaultRecomputeInterval() time.Duration { return 5 * time.Minute } +// AnalyticsPauseWhenIdle reports whether idle analytics recomputes are skipped. +func (c *Config) AnalyticsPauseWhenIdle() bool { + return c != nil && c.Analytics != nil && c.Analytics.PauseWhenIdle +} + // AnalyticsRecomputeIntervals returns the per-endpoint override map. // Returns the zero value (all defaults) if the analytics block is // absent or empty. func (c *Config) AnalyticsRecomputeIntervals() AnalyticsRecomputeIntervals { - out := AnalyticsRecomputeIntervals{} + out := AnalyticsRecomputeIntervals{PauseWhenIdle: c.AnalyticsPauseWhenIdle()} if c == nil || c.Analytics == nil || c.Analytics.RecomputeIntervalSeconds == nil { return out } diff --git a/cmd/server/direct_heard.go b/cmd/server/direct_heard.go index 9118a2d6..0fa3e01c 100644 --- a/cmd/server/direct_heard.go +++ b/cmd/server/direct_heard.go @@ -216,7 +216,17 @@ func (s *PacketStore) publishDirectHeard(idx directHeardIndex) { // compute has published one. A nil index simply yields empty direct rows: // the card degrades to "nobody hears this node" until the first pass lands, // never to a wrong attribution. +// +// It counts as a read of the direct-heard recomputer, so with pauseWhenIdle +// on the card keeps it refreshing (the snapshot is read here, not through +// the recomputer's Load). func (s *PacketStore) loadDirectHeard() directHeardIndex { + s.analyticsRecomputerMu.RLock() + rc := s.recompDirectHeard + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + rc.markRead() + } idx, _ := s.directHeardSnap.Load().(directHeardIndex) return idx } diff --git a/config.example.json b/config.example.json index cf951d6c..2332ccc1 100644 --- a/config.example.json +++ b/config.example.json @@ -414,6 +414,8 @@ "analytics": { "defaultIntervalSeconds": 300, + "pauseWhenIdle": false, + "_comment_pauseWhenIdle": "Opt-in. When true, a scheduled recompute is skipped if nothing has read that endpoint's snapshot since the previous recompute, so an instance whose analytics pages are rarely open stops recomputing (and allocating) for them every interval. The first read after a pause is served the existing snapshot at once, which can be older than the interval, and starts a refresh; reads after that see fresh data. Default false: every endpoint is recomputed on its interval whether read or not (#1240).", "recomputeIntervalSeconds": { "topology": 300, "rf": 300,