perf(precompute): index oldest-series eviction

asap-precompute-go/precompute.go

@@ -320,6 +320,8 @@ func New(initialCfg *PrecomputeConfig, sketchFactory SketchFactory, observer Ske
 		snapshotCache: NewSnapshotCache(),
 		stats:         NewPrecomputeStats(),
 	}
+	p.window.snapshotCache = p.snapshotCache
+	p.window.sketchSink = &p.sketchSink
 	if initialCfg != nil {
 		cfgCopy := clonePrecomputeConfig(initialCfg)
 		p.cfg.Store(cfgCopy)

asap-precompute-go/window.go

@@ -1,16 +1,20 @@
 package precompute
 
 import (
+	"container/heap"
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
 // seriesEntry is the per-series state held inside a window. It
 // owns one Sketch instance and the labels needed to reconstruct
 // the SketchEnvelope at flush time.
 type seriesEntry struct {
+	seriesKey string
+	heapIndex int
 	// Sketch is the running sketch for this series. Owned here;
 	// the window calls Reset on rotation when the entry is
 	// recycled in place, OR drops the reference when MaxSeries
@@ -64,6 +68,35 @@ type seriesEntry struct {
 	ackVal         float64
 }
 
+type seriesEvictionHeap []*seriesEntry
+
+func (h seriesEvictionHeap) Len() int { return len(h) }
+func (h seriesEvictionHeap) Less(i, j int) bool {
+	if h[i].LastSeenMs == h[j].LastSeenMs {
+		return h[i].seriesKey < h[j].seriesKey
+	}
+	return h[i].LastSeenMs < h[j].LastSeenMs
+}
+func (h seriesEvictionHeap) Swap(i, j int) {
+	h[i], h[j] = h[j], h[i]
+	h[i].heapIndex = i
+	h[j].heapIndex = j
+}
+func (h *seriesEvictionHeap) Push(value any) {
+	e := value.(*seriesEntry)
+	e.heapIndex = len(*h)
+	*h = append(*h, e)
+}
+func (h *seriesEvictionHeap) Pop() any {
+	old := *h
+	n := len(old)
+	e := old[n-1]
+	old[n-1] = nil
+	e.heapIndex = -1
+	*h = old[:n-1]
+	return e
+}
+
 // windowState is the per-Precompute window manager. It supports
 // Tumbling, Batch, and Sliding modes.
 //
@@ -86,6 +119,9 @@ type seriesEntry struct {
 type windowState struct {
 	mu            sync.RWMutex
 	series        map[string]*seriesEntry
+	eviction      seriesEvictionHeap
+	snapshotCache *SnapshotCache
+	sketchSink    *atomic.Pointer[SketchSink]
 	activeStartMs uint64
 	activeEndMs   uint64
 	// initialized tracks whether activeStart/End have been
@@ -411,38 +447,49 @@ func (w *windowState) admitSeriesLocked(
 	}
 	entry := &seriesEntry{
 		Sketch:         sketch,
+		seriesKey:      key,
+		heapIndex:      -1,
 		ResourceLabels: resourceCopy,
 		Labels:         labelsCopy,
 		LastSeenMs:     obs.TimestampMs,
 	}
 	w.series[key] = entry
+	heap.Push(&w.eviction, entry)
 	if stats != nil {
 		stats.ActiveSeries.Add(1)
 	}
 	return entry, nil
 }
 
 func (w *windowState) evictOldestLocked() (string, *seriesEntry) {
-	var oldestKey string
-	oldestMs := ^uint64(0)
-	for key, entry := range w.series {
-		if entry.LastSeenMs < oldestMs {
-			oldestMs = entry.LastSeenMs
-			oldestKey = key
+	if len(w.eviction) == 0 {
+		return "", nil
+	}
+	entry := heap.Pop(&w.eviction).(*seriesEntry)
+	delete(w.series, entry.seriesKey)
+	if w.snapshotCache != nil {
+		w.snapshotCache.Delete(entry.seriesKey)
+	}
+	if w.sketchSink != nil {
+		if sink := w.sketchSink.Load(); sink != nil && *sink != nil {
+			(*sink)(entry.Sketch)
+			entry.Sketch = nil
 		}
 	}
-	entry := w.series[oldestKey]
-	if oldestKey != "" {
-		delete(w.series, oldestKey)
+	if entry.Sketch != nil {
+		entry.Sketch.Reset()
 	}
-	return oldestKey, entry
+	return entry.seriesKey, entry
 }
 
 // recordLocked feeds one observation into a series' sketch and advances its
 // bookkeeping. Caller holds w.mu. Shared by observe and observeKeyed.
 func (w *windowState) recordLocked(entry *seriesEntry, obs *Observation, observer SketchObserver) error {
 	if obs.TimestampMs > entry.LastSeenMs {
 		entry.LastSeenMs = obs.TimestampMs
+		if entry.heapIndex >= 0 {
+			heap.Fix(&w.eviction, entry.heapIndex)
+		}
 	}
 	if err := observer.Observe(entry.Sketch, obs.Value); err != nil {
 		return fmt.Errorf("sketch observe: %w", err)
@@ -526,10 +573,13 @@ func (w *windowState) observeEnvelope(
 		copy(labelsCopy, env.Labels)
 		entry = &seriesEntry{
 			Sketch:     sketch,
+			seriesKey:  key,
+			heapIndex:  -1,
 			Labels:     labelsCopy,
 			LastSeenMs: refMs,
 		}
 		w.series[key] = entry
+		heap.Push(&w.eviction, entry)
 		if stats != nil {
 			stats.ActiveSeries.Add(1)
 		}
@@ -593,6 +643,12 @@ func (w *windowState) observeEnvelope(
 	// Envelopes with Count==0 (older senders that don't populate
 	// the field) contribute zero, which is a no-op.
 	entry.Count += env.Count
+	if refMs > entry.LastSeenMs {
+		entry.LastSeenMs = refMs
+		if entry.heapIndex >= 0 {
+			heap.Fix(&w.eviction, entry.heapIndex)
+		}
+	}
 	return nil
 }
 
@@ -694,6 +750,7 @@ func (w *windowState) rotateLocked(nowMs uint64, cfg *PrecomputeConfig) ([]*seri
 
 	// Reset the series map for the next window.
 	w.series = make(map[string]*seriesEntry)
+	w.eviction = nil
 	w.advanceWindow(nowMs, cfg)
 	return closedSeries, rng
 }
@@ -722,6 +779,7 @@ func (w *windowState) rotateSlidingLocked(nowMs uint64, cfg *PrecomputeConfig) (
 	})
 	// Start a fresh current pane and advance the bounds by one slide.
 	w.series = make(map[string]*seriesEntry)
+	w.eviction = nil
 	w.advanceWindow(nowMs, cfg)
 
 	// Trim the ring to the most recent N panes (drop the oldest,
@@ -866,6 +924,7 @@ func (w *windowState) resetForScopeChange() {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	w.series = make(map[string]*seriesEntry)
+	w.eviction = nil
 	w.panes = nil
 }
 

asap-precompute-go/window_eviction_bench_test.go

@@ -0,0 +1,33 @@
+package precompute
+
+import (
+	"strconv"
+	"testing"
+	"time"
+)
+
+func BenchmarkEvictionAtCap(b *testing.B) {
+	for _, capacity := range []int{1_000, 10_000, 100_000} {
+		b.Run(strconv.Itoa(capacity), func(b *testing.B) {
+			cfg := &PrecomputeConfig{AggID: 1, SketchType: SketchTypeDDSketch,
+				Mode: Tumbling, Window: WindowSpec{Size: 24 * time.Hour},
+				MaxSeries: uint64(capacity), OnOverflow: OnOverflowEvictOldest}
+			w := newWindowState()
+			factory, observer, stats := newFakeFactory(), &fakeObserver{}, NewStats()
+			for i := 0; i < capacity; i++ {
+				obs := &Observation{TimestampMs: 1, Labels: []KeyValue{{Key: "k", Value: strconv.Itoa(i)}}, Value: FloatValue(1)}
+				if err := w.observe(obs, cfg, factory, observer, stats); err != nil {
+					b.Fatal(err)
+				}
+			}
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				obs := &Observation{TimestampMs: uint64(i + 2), Labels: []KeyValue{{Key: "k", Value: "new-" + strconv.Itoa(i)}}, Value: FloatValue(1)}
+				if err := w.observe(obs, cfg, factory, observer, stats); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}