From 4c14442b4c2f59e6b6d86977df4360304af04d23 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:49:04 -0600 Subject: [PATCH] perf(precompute): parallelize independent series updates --- asap-precompute-go/window.go | 65 +++++++++++-------- .../window_parallel_bench_test.go | 37 +++++++++++ 2 files changed, 74 insertions(+), 28 deletions(-) create mode 100644 asap-precompute-go/window_parallel_bench_test.go diff --git a/asap-precompute-go/window.go b/asap-precompute-go/window.go index 5caf62fd..137fbbbe 100644 --- a/asap-precompute-go/window.go +++ b/asap-precompute-go/window.go @@ -13,6 +13,7 @@ import ( // owns one Sketch instance and the labels needed to reconstruct // the SketchEnvelope at flush time. type seriesEntry struct { + mu sync.Mutex seriesKey string heapIndex int // Sketch is the running sketch for this series. Owned here; @@ -120,6 +121,7 @@ type windowState struct { mu sync.RWMutex series map[string]*seriesEntry eviction seriesEvictionHeap + evictionMu sync.Mutex snapshotCache *SnapshotCache sketchSink *atomic.Pointer[SketchSink] activeStartMs uint64 @@ -303,18 +305,6 @@ func (w *windowState) observe( observer SketchObserver, stats *PrecomputeStats, ) error { - w.mu.Lock() - defer w.mu.Unlock() - - if w.sketchFactory == nil { - w.sketchFactory = sketchFactory - } - w.initWindow(obs.TimestampMs, cfg) - - if err := w.validateTimestampLocked(obs.TimestampMs, cfg); err != nil { - return err - } - // Build the lookup key into a pooled byte buffer so the common // case (an already-admitted series) costs no allocation: the // `w.series[string(sc.buf)]` index is the compiler's zero-alloc @@ -323,14 +313,7 @@ func (w *windowState) observe( sc := getSeriesKeyScratch() defer putSeriesKeyScratch(sc) cfg.buildSeriesKey(sc, obs) - entry, ok := w.series[string(sc.buf)] - if !ok { - var err error - if entry, err = w.admitSeriesLocked(string(sc.buf), obs, cfg, sketchFactory, stats); err != nil { - return err - } - } - return w.recordLocked(entry, obs, observer) + return w.observeWithKey(string(sc.buf), obs, cfg, sketchFactory, observer, stats) } // observeKeyed is the shared-key entry point for the fused asap_edge @@ -346,26 +329,48 @@ func (w *windowState) observeKeyed( observer SketchObserver, stats *PrecomputeStats, ) error { + return w.observeWithKey(key, obs, cfg, sketchFactory, observer, stats) +} + +func (w *windowState) observeWithKey(key string, obs *Observation, cfg *PrecomputeConfig, sketchFactory SketchFactory, observer SketchObserver, stats *PrecomputeStats) error { + // The generation read lock prevents rotation while independent entries use + // their own locks. Established series therefore update concurrently. + w.mu.RLock() + if w.initialized { + if err := w.validateTimestampLocked(obs.TimestampMs, cfg); err != nil { + w.mu.RUnlock() + return err + } + if entry := w.series[key]; entry != nil { + entry.mu.Lock() + err := w.recordLocked(entry, obs, observer, cfg.MaxSeries > 0 && cfg.OnOverflow == OnOverflowEvictOldest) + entry.mu.Unlock() + w.mu.RUnlock() + return err + } + } + w.mu.RUnlock() + + // New-series admission mutates the map/cardinality index and is rare after + // warm-up. Recheck after upgrading because another goroutine may admit it. w.mu.Lock() defer w.mu.Unlock() - if w.sketchFactory == nil { w.sketchFactory = sketchFactory } w.initWindow(obs.TimestampMs, cfg) - if err := w.validateTimestampLocked(obs.TimestampMs, cfg); err != nil { return err } - - entry, ok := w.series[key] - if !ok { + entry := w.series[key] + if entry == nil { var err error - if entry, err = w.admitSeriesLocked(key, obs, cfg, sketchFactory, stats); err != nil { + entry, err = w.admitSeriesLocked(key, obs, cfg, sketchFactory, stats) + if err != nil { return err } } - return w.recordLocked(entry, obs, observer) + return w.recordLocked(entry, obs, observer, cfg.MaxSeries > 0 && cfg.OnOverflow == OnOverflowEvictOldest) } // validateTimestampLocked prevents host scheduling jitter from changing @@ -484,8 +489,12 @@ func (w *windowState) evictOldestLocked() (string, *seriesEntry) { // 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 { +func (w *windowState) recordLocked(entry *seriesEntry, obs *Observation, observer SketchObserver, trackEviction bool) error { if obs.TimestampMs > entry.LastSeenMs { + if trackEviction { + w.evictionMu.Lock() + defer w.evictionMu.Unlock() + } entry.LastSeenMs = obs.TimestampMs if entry.heapIndex >= 0 { heap.Fix(&w.eviction, entry.heapIndex) diff --git a/asap-precompute-go/window_parallel_bench_test.go b/asap-precompute-go/window_parallel_bench_test.go new file mode 100644 index 00000000..45789580 --- /dev/null +++ b/asap-precompute-go/window_parallel_bench_test.go @@ -0,0 +1,37 @@ +package precompute + +import ( + "strconv" + "sync/atomic" + "testing" + "time" +) + +func BenchmarkParallelExistingSeries(b *testing.B) { + const series = 1024 + cfg := &PrecomputeConfig{AggID: 1, SketchType: SketchTypeDDSketch, + Mode: Tumbling, Window: WindowSpec{Size: time.Hour}} + w := newWindowState() + factory, observer, stats := newFakeFactory(), &fakeObserver{}, NewStats() + observations := make([]*Observation, series) + keys := make([]string, series) + for i := range observations { + observations[i] = &Observation{TimestampMs: 1, Labels: []KeyValue{{Key: "k", Value: strconv.Itoa(i)}}, Value: FloatValue(1)} + keys[i] = cfg.SeriesKeyFor(observations[i]) + if err := w.observeKeyed(keys[i], observations[i], cfg, factory, observer, stats); err != nil { + b.Fatal(err) + } + } + var next atomic.Uint64 + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + i := int(next.Add(1)) & (series - 1) + if err := w.observeKeyed(keys[i], observations[i], cfg, factory, observer, stats); err != nil { + b.Error(err) + return + } + } + }) +}