perf(cluster): rotate the grid cache instead of emptying it, cap 100k
The cache dropped EVERY entry once it passed 20 000. That was survivable while the only feed was this station's own WSJT-X decodes, which never reached the ceiling — it is a cliff for anything larger, and every locator in the cluster list would disappear at once, periodically, for no reason the operator could see. Two generations: when the current map fills it becomes the previous one and a fresh map takes over; lookups consult both. A rotation therefore costs the older half and nothing more. It needs no insertion order, no per-entry timestamp and no bookkeeping on the write path — all of which "evict the oldest thousand" would require, on a path that runs once per decode. The cap is 100 000, measured at 82 bytes an entry: 8 MB a generation, 16 MB for both. 20 000 was chosen when the ceiling was unreachable anyway. Both maps now go through rememberDecodeGrid / lookupDecodeGrid. Rotation swaps the map headers, so the read had to be guarded; routing every access through one accessor is what makes that checkable rather than remembered.
This commit is contained in:
@@ -611,7 +611,16 @@ type App struct {
|
||||
//
|
||||
// In memory only, and bounded: it is a session-local view of who is on the
|
||||
// air now, not a database.
|
||||
decodeGrids map[string]string
|
||||
//
|
||||
// Bounded in TWO generations. The cap used to drop the whole map, which was
|
||||
// survivable while the only source was this station's own decodes — it never
|
||||
// reached the ceiling. It is a cliff for any larger feed: every locator in the
|
||||
// list would vanish at once, periodically. Keeping the previous generation
|
||||
// means a rotation costs the older half and nothing more, and it needs no
|
||||
// insertion order, no per-entry timestamp and no bookkeeping on the write
|
||||
// path — which "evict the oldest thousand" would all require.
|
||||
decodeGrids map[string]string // current generation, written to
|
||||
decodeGridsOld map[string]string // previous generation, still readable
|
||||
decodeGridsMu sync.RWMutex
|
||||
// pskr is the PSK Reporter MQTT feed, up only while the opening watch is on.
|
||||
// It is the source that makes VHF detection work at all: the cluster and RBN
|
||||
@@ -11937,15 +11946,7 @@ func (a *App) consumeUDPEvents() {
|
||||
// Remember the grid before anything else: a CQ is the one message that
|
||||
// carries it, and the station may never send another.
|
||||
if ev.DecodeGrid != "" {
|
||||
a.decodeGridsMu.Lock()
|
||||
if a.decodeGrids == nil {
|
||||
a.decodeGrids = make(map[string]string, 512)
|
||||
}
|
||||
if len(a.decodeGrids) > 20000 {
|
||||
a.decodeGrids = make(map[string]string, 512) // bound a long session
|
||||
}
|
||||
a.decodeGrids[strings.ToUpper(ev.DecodeCall)] = ev.DecodeGrid
|
||||
a.decodeGridsMu.Unlock()
|
||||
a.rememberDecodeGrid(ev.DecodeCall, ev.DecodeGrid)
|
||||
}
|
||||
// A WSJT-X decode (heard station). Render it on the FlexRadio
|
||||
// panadapter when the option is on; green + SNR comment, auto-expiring
|
||||
@@ -17014,6 +17015,50 @@ func (a *App) clusterStatusMaps() *clusterStatusCache {
|
||||
// was ambiguous and the frontend couldn't infer) we degrade gracefully
|
||||
// to band-only — saying "worked" rather than wrongly flagging "new-slot"
|
||||
// just because we don't know the mode.
|
||||
// decodeGridsCap is how many callsigns one generation holds before it rotates.
|
||||
//
|
||||
// 100 000 measured at 82 bytes an entry — 8 MB a generation, 16 MB for both,
|
||||
// which is nothing next to what it buys: a locator on a spot instead of an empty
|
||||
// column. The old 20 000 was chosen when the only feed was this station's own
|
||||
// decodes and the ceiling was never reached anyway.
|
||||
const decodeGridsCap = 100000
|
||||
|
||||
// rememberDecodeGrid records the grid a station announced.
|
||||
//
|
||||
// Rotation, not eviction: when the current generation fills it becomes the
|
||||
// previous one and a fresh map takes over. Nothing is scanned, nothing is
|
||||
// timestamped, and the write stays a single map assignment — which matters
|
||||
// because this runs once per decode.
|
||||
func (a *App) rememberDecodeGrid(call, grid string) {
|
||||
call = strings.ToUpper(strings.TrimSpace(call))
|
||||
if call == "" || grid == "" {
|
||||
return
|
||||
}
|
||||
a.decodeGridsMu.Lock()
|
||||
if a.decodeGrids == nil {
|
||||
a.decodeGrids = make(map[string]string, 512)
|
||||
}
|
||||
if len(a.decodeGrids) >= decodeGridsCap {
|
||||
a.decodeGridsOld = a.decodeGrids
|
||||
a.decodeGrids = make(map[string]string, 512)
|
||||
}
|
||||
a.decodeGrids[call] = grid
|
||||
a.decodeGridsMu.Unlock()
|
||||
}
|
||||
|
||||
// lookupDecodeGrid returns the last grid heard for a callsign, "" if unknown.
|
||||
// The previous generation is consulted second, so a station that has gone quiet
|
||||
// keeps its locator across one rotation instead of losing it at the cliff.
|
||||
func (a *App) lookupDecodeGrid(call string) string {
|
||||
call = strings.ToUpper(call)
|
||||
a.decodeGridsMu.RLock()
|
||||
defer a.decodeGridsMu.RUnlock()
|
||||
if g := a.decodeGrids[call]; g != "" {
|
||||
return g
|
||||
}
|
||||
return a.decodeGridsOld[call]
|
||||
}
|
||||
|
||||
func (a *App) ClusterSpotStatuses(spots []SpotQuery) []SpotStatus {
|
||||
out := make([]SpotStatus, len(spots))
|
||||
if a.qso == nil {
|
||||
@@ -17102,12 +17147,10 @@ func (a *App) ClusterSpotStatuses(spots []SpotQuery) []SpotStatus {
|
||||
// is part of the key, so the "group digital modes" option decides whether a
|
||||
// grid worked on FT8 still counts as new on FT4 — one rule, no branch here.
|
||||
{
|
||||
// The length check has to be INSIDE the lock: the decode goroutine
|
||||
// replaces this map wholesale when it grows too large, so reading len()
|
||||
// unguarded is a race on the map header, not a cheap fast path.
|
||||
a.decodeGridsMu.RLock()
|
||||
g := a.decodeGrids[strings.ToUpper(q.Call)]
|
||||
a.decodeGridsMu.RUnlock()
|
||||
// Read through the accessor: the decode goroutine swaps these maps on
|
||||
// rotation, so touching them unguarded is a race on the map header,
|
||||
// not a cheap fast path.
|
||||
g := a.lookupDecodeGrid(q.Call)
|
||||
if g != "" {
|
||||
out[i].Grid = g
|
||||
cm := out[i].Mode
|
||||
|
||||
+6
-2
@@ -2,8 +2,12 @@
|
||||
{
|
||||
"version": "0.24.8",
|
||||
"date": "",
|
||||
"en": [],
|
||||
"fr": []
|
||||
"en": [
|
||||
"Cluster: the grid cache now holds 100,000 callsigns and rotates instead of emptying itself, so locators stop vanishing from the list."
|
||||
],
|
||||
"fr": [
|
||||
"Cluster : le cache de locators garde 100 000 indicatifs et tourne au lieu de se vider, les locators ne disparaissent donc plus de la liste."
|
||||
]
|
||||
},
|
||||
{
|
||||
"version": "0.24.7",
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The grid cache used to drop EVERYTHING at its ceiling. That was survivable
|
||||
// while the only feed was this station's own decodes, which never reached it.
|
||||
// Fed by anything larger it is a cliff: every locator in the cluster list
|
||||
// disappears at once, periodically, and the operator sees the column empty
|
||||
// itself for no reason.
|
||||
//
|
||||
// Rotation keeps the previous generation, so a full cache costs the older half
|
||||
// and nothing more.
|
||||
func TestDecodeGridRotationKeepsThePreviousGeneration(t *testing.T) {
|
||||
a := &App{}
|
||||
|
||||
// Fill exactly one generation.
|
||||
for i := 0; i < decodeGridsCap; i++ {
|
||||
a.rememberDecodeGrid(fmt.Sprintf("CALL%06d", i), "JN36")
|
||||
}
|
||||
if got := a.lookupDecodeGrid("CALL000000"); got != "JN36" {
|
||||
t.Fatalf("first entry lost before any rotation: %q", got)
|
||||
}
|
||||
if a.decodeGridsOld != nil {
|
||||
t.Fatal("rotated early — the cap is the ceiling of ONE generation")
|
||||
}
|
||||
|
||||
// One more entry rotates.
|
||||
a.rememberDecodeGrid("NEWCALL", "IO91")
|
||||
if a.decodeGridsOld == nil {
|
||||
t.Fatal("did not rotate at the cap")
|
||||
}
|
||||
if got := a.lookupDecodeGrid("NEWCALL"); got != "IO91" {
|
||||
t.Errorf("the entry that caused the rotation was lost: %q", got)
|
||||
}
|
||||
// The whole previous generation is still readable — this is the point.
|
||||
if got := a.lookupDecodeGrid("CALL000000"); got != "JN36" {
|
||||
t.Errorf("a locator from the previous generation was dropped: %q — that is the cliff again", got)
|
||||
}
|
||||
if got := a.lookupDecodeGrid("CALL099999"); got != "JN36" {
|
||||
t.Errorf("previous generation incomplete: %q", got)
|
||||
}
|
||||
|
||||
// A second rotation is what finally retires the oldest half.
|
||||
for i := 0; i < decodeGridsCap; i++ {
|
||||
a.rememberDecodeGrid(fmt.Sprintf("SECOND%06d", i), "KP20")
|
||||
}
|
||||
if got := a.lookupDecodeGrid("CALL000000"); got != "" {
|
||||
t.Errorf("the cache is unbounded: %q survived two rotations", got)
|
||||
}
|
||||
if got := a.lookupDecodeGrid("NEWCALL"); got != "IO91" {
|
||||
t.Errorf("an entry one generation old was retired too early: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Callsigns are normalised on the way in AND on the way out, or a spot for
|
||||
// "f4bpo" would miss a grid learnt as "F4BPO".
|
||||
func TestDecodeGridCaseAndBlanks(t *testing.T) {
|
||||
a := &App{}
|
||||
a.rememberDecodeGrid(" f4bpo ", "JN36")
|
||||
if got := a.lookupDecodeGrid("F4BPO"); got != "JN36" {
|
||||
t.Errorf("lookup of the upper-case form failed: %q", got)
|
||||
}
|
||||
a.rememberDecodeGrid("", "JN36")
|
||||
a.rememberDecodeGrid("K1ABC", "")
|
||||
if got := a.lookupDecodeGrid("K1ABC"); got != "" {
|
||||
t.Errorf("stored an empty grid: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// The write path runs on the decode goroutine while the cluster status builder
|
||||
// reads. Rotation swaps the map headers, so an unguarded read is a data race
|
||||
// rather than a stale value.
|
||||
//
|
||||
// This build is CGO-free, so -race is not available here; the test exercises the
|
||||
// interleaving and would fault on a concurrent map access even without it.
|
||||
func TestDecodeGridConcurrentAccess(t *testing.T) {
|
||||
a := &App{}
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 20000; i++ {
|
||||
a.rememberDecodeGrid(fmt.Sprintf("W%05d", i), "FN31")
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 20000; i++ {
|
||||
_ = a.lookupDecodeGrid(fmt.Sprintf("W%05d", i))
|
||||
}
|
||||
}()
|
||||
wg.Wait()
|
||||
}
|
||||
Reference in New Issue
Block a user