diff --git a/docs/design/2026_06_12_proposed_scaling_roadmap.md b/docs/design/2026_06_12_proposed_scaling_roadmap.md index 6da71b736..ba670f160 100644 --- a/docs/design/2026_06_12_proposed_scaling_roadmap.md +++ b/docs/design/2026_06_12_proposed_scaling_roadmap.md @@ -102,10 +102,10 @@ Multi-region blockers: ### 2.3 Storage tier (Pebble) -`store/lsm_store.go`: per-shard `pebble.Open`, default block cache -`defaultPebbleCacheBytes = 256 MiB` **per store** (not shared). -WAL sync via `ELASTICKV_FSM_SYNC_MODE` (default `pebble.Sync` on -FSM apply; `nosync` opt-in). +`store/lsm_store.go`: per-shard `pebble.Open`, process-wide shared block cache +`defaultPebbleCacheBytes = 256 MiB` **per node** (shared by all stores in the +process). WAL sync via `ELASTICKV_FSM_SYNC_MODE` (default `pebble.Sync` on FSM +apply; `nosync` opt-in). `store/mvcc_store.go`: encoded as `UserKey ++ 0x00 ++ inverted_TS`, `maxSnapshotVersionCount = 1 M`, `maxSnapshotValueSize = 256 MiB`. @@ -129,8 +129,9 @@ Storage breakage at 1–10 TB/shard: is tens of minutes. - Pebble L0CompactionThreshold / LBaseMaxBytes / compaction concurrency are defaults; write-heavy shards hit stall thresholds. - 256 MiB block cache per shard × N shards/node = N × 256 MiB - resident memory (shared-cache TODO not landed). + Shared block cache M1 has landed, so resident block-cache memory is capped per + process rather than N × 256 MiB. Shared memtable / compaction concurrency + budgeting and per-group cache fairness remain open. - MVCC retention 30 min + per-key 1 M version cap means a key written ≥ 555/s for 30 min trips the cap; compactor runs every 5 min so read tail latency spikes during accumulation. @@ -366,11 +367,11 @@ control-plane (`*_proposed_*` doc TBD).** work. - Streamed in parallel across the leader's outgoing transport. -**M2 — Shared block cache + per-shard tuning (`*_proposed_*` doc +**M2 — Shared block-cache follow-ups + per-shard tuning (`*_proposed_*` doc TBD).** -- Land the existing shared-cache TODO: one `pebble.Cache` per node - shared across all shards' stores, sized as a per-node config - fraction of available RAM (default 25%). +- M1 shared block cache has landed: one `pebble.Cache` per process is shared + across all shards' stores and sized by `ELASTICKV_PEBBLE_CACHE_MB`. +- Add shared memtable / compaction concurrency budgets across stores. - Surface `L0CompactionThreshold`, `LBaseMaxBytes`, `MaxConcurrentCompactions` as per-shard config so a write-heavy shard can be tuned without touching the cluster default. diff --git a/docs/design/2026_06_23_proposed_scaling_roadmap.md b/docs/design/2026_06_23_proposed_scaling_roadmap.md index b51827adb..cd41d7968 100644 --- a/docs/design/2026_06_23_proposed_scaling_roadmap.md +++ b/docs/design/2026_06_23_proposed_scaling_roadmap.md @@ -39,13 +39,11 @@ What bounds a single elastickv deployment today: is not wired (see §2(b), §2(e)). 2. **Per-group memory.** Each Raft group owns a private Pebble store, and - `NewPebbleStore` → `defaultPebbleOptionsWithCache` allocates a *fresh* - block cache per store (`store/lsm_store.go:273-279`, `pebble.NewCache(pebbleCacheBytes)` - at `:274`), default 256 MiB (`defaultPebbleCacheBytes` at `:66`). N groups - on a node means N × 256 MiB of block cache alone, plus N memtables, N WALs, - N compaction budgets. The TODO at `store/lsm_store.go:117-120` records the - intended fix (a process-wide shared cache plumbed through `NewPebbleStore`) - but it is a comment, not a design. + `NewPebbleStore` → `defaultPebbleOptionsWithCache` now borrows one + process-wide shared block cache through `processPebbleCacheRef`, default + 256 MiB (`defaultPebbleCacheBytes`). N groups on a node therefore share one + block-cache LRU instead of reserving N × 256 MiB for block cache alone. + Per-group memtables, WALs, and compaction budgets are still independent. 3. **Leader concentration.** Leadership of each group is elected independently by etcd/raft; nothing spreads leaderships across nodes. One @@ -136,12 +134,11 @@ memory each group's private cache/memtable pins. bounds WAL/snapshot growth to O(manifest), which is the data-volume lever for the S3 surface specifically. Still proposed; the legacy `BlobKey`-on-Raft path is what runs today. -- **Shared Pebble cache / resource pools — TODO only, promoted to a design - item here.** The `store/lsm_store.go:117-120` TODO is the single biggest - per-node memory tax as group count grows: N independent 256 MiB caches with - no shared eviction pool. This deserves a real design (see §3, Gap 2), not a - comment. It blocks high group counts on a single node, which in turn blocks - the "many small ranges" model that split (a) produces. +- **Shared Pebble cache / resource pools — M1 shipped, follow-ups remain.** + The block-cache tax is closed by the process-wide `pebble.Cache` in + `store/lsm_store.go`, so group count no longer multiplies block-cache memory. + Shared memtable / compaction budgets and per-group fairness still need the + follow-up design slices in §3, Gap 2. ### (b) Write throughput @@ -303,14 +300,18 @@ true multi-node multi-group workloads, tracked as a follow-on rather than a bootstrap design blocker. ### Gap 2 — Shared Pebble cache / resource pools -**Problem.** Each group's store allocates a private 256 MiB block cache -(`store/lsm_store.go:273-279`); N groups = N × 256 MiB plus N memtables/WALs, -with no shared eviction. This caps how many groups (hence how many ranges) one -node can hold, throttling the "many small ranges" model that split produces. -**Rough milestones:** (M1) process-wide shared `pebble.Cache` plumbed through -`NewPebbleStore` (the existing TODO), with per-node sizing. (M2) shared -memtable / compaction concurrency budget across stores. (M3) per-group -fairness so one hot group cannot evict everyone else's working set. +**Status.** M1 is closed: `defaultPebbleOptionsWithCache` now borrows a +process-wide shared `pebble.Cache` via `processPebbleCacheRef`, and +`NewPebbleStore` / restore reopen paths hold one explicit store/open reference +that is released on close or reopen. `ELASTICKV_PEBBLE_CACHE_MB` now sizes the +node-level shared cache rather than a per-store cache. + +**Remaining problem.** Each group's store still owns independent memtables, +WALs, flush scheduling, and compaction concurrency. That still caps how many +groups one node can hold, but block-cache memory no longer grows as +N × 256 MiB. **Remaining milestones:** (M2) shared memtable / compaction +concurrency budget across stores. (M3) per-group fairness so one hot group +cannot evict everyone else's working set. **Depends-on:** none functionally; pairs naturally with Gap 1 (high group counts only matter once multi-node groups exist). diff --git a/monitoring/pebble.go b/monitoring/pebble.go index 59133099d..91d857523 100644 --- a/monitoring/pebble.go +++ b/monitoring/pebble.go @@ -16,12 +16,13 @@ import ( // dashboard. // // The point-in-time fields (Sublevels, NumFiles, EstimatedDebt, -// MemTable.*, NumInProgress, BlockCache.Size) are exposed as -// Prometheus GAUGES — each poll overwrites the previous value. -// Monotonic fields (Compact.Count, BlockCache.Hits/Misses) are exposed -// as COUNTERS; the collector emits only the positive delta against the -// last snapshot so a store reset (Restore/swap) does not produce -// negative values. +// MemTable.*, NumInProgress, BlockCache.Size) are exposed as Prometheus +// GAUGES — each poll overwrites the previous value. LSM / memtable / +// compaction signals remain group-scoped; the block cache is process-wide and +// is emitted once per node because all stores share one pebble.Cache. +// Monotonic fields (Compact.Count, BlockCache.Hits/Misses) are exposed as +// COUNTERS; the collector emits only the positive delta against the last +// snapshot so a store reset (Restore/swap) does not produce negative values. // // Name convention: elastickv_pebble_* to keep a consistent node_id / // node_address label prefix with the rest of the registry. @@ -121,30 +122,30 @@ func newPebbleMetrics(registerer prometheus.Registerer) *PebbleMetrics { blockCacheSizeBytes: prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: "elastickv_pebble_block_cache_size_bytes", - Help: "Current bytes in use by Pebble's block cache.", + Help: "Current bytes in use by the process-wide Pebble block cache.", }, - []string{"group"}, + nil, ), blockCacheCapacityBytes: prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: "elastickv_pebble_block_cache_capacity_bytes", - Help: "Configured maximum size of Pebble's block cache in bytes. Paired with elastickv_pebble_block_cache_size_bytes so operators can see usage relative to capacity and with the hit/miss counters so they can reason about whether a low hit rate reflects a cold cache or an undersized one.", + Help: "Configured maximum size of the process-wide Pebble block cache in bytes. Paired with elastickv_pebble_block_cache_size_bytes so operators can see usage relative to capacity and with the hit/miss counters so they can reason about whether a low hit rate reflects a cold cache or an undersized one.", }, - []string{"group"}, + nil, ), blockCacheHitsTotal: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "elastickv_pebble_block_cache_hits_total", - Help: "Cumulative block cache hits reported by Pebble.", + Help: "Cumulative process-wide block cache hits reported by Pebble.", }, - []string{"group"}, + nil, ), blockCacheMissesTotal: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "elastickv_pebble_block_cache_misses_total", - Help: "Cumulative block cache misses reported by Pebble.", + Help: "Cumulative process-wide block cache misses reported by Pebble.", }, - []string{"group"}, + nil, ), fsmApplySyncMode: prometheus.NewGaugeVec( prometheus.GaugeOpts{ @@ -238,14 +239,18 @@ type PebbleSource struct { type PebbleCollector struct { metrics *PebbleMetrics - mu sync.Mutex - previous map[uint64]pebbleSnapshot + mu sync.Mutex + previous map[uint64]pebbleSnapshot + previousBlockCache pebbleBlockCacheSnapshot } type pebbleSnapshot struct { - compactCount int64 - blockCacheHits int64 - blockCacheMisses int64 + compactCount int64 +} + +type pebbleBlockCacheSnapshot struct { + hits int64 + misses int64 } func newPebbleCollector(metrics *PebbleMetrics) *PebbleCollector { @@ -293,6 +298,8 @@ func (c *PebbleCollector) observeOnce(sources []PebbleSource) { } c.mu.Lock() defer c.mu.Unlock() + blockCacheObserved := false + blockCacheCapacityObserved := false for _, src := range sources { if src.Source == nil { continue @@ -302,6 +309,13 @@ func (c *PebbleCollector) observeOnce(sources []PebbleSource) { continue } c.observeSource(src, snap) + if !blockCacheObserved { + c.observeBlockCache(snap) + blockCacheObserved = true + } + if !blockCacheCapacityObserved { + blockCacheCapacityObserved = c.observeBlockCacheCapacity(src) + } } } @@ -324,35 +338,45 @@ func (c *PebbleCollector) observeSource(src PebbleSource, snap *pebble.Metrics) c.metrics.memtableSizeBytes.WithLabelValues(group).Set(float64(snap.MemTable.Size)) c.metrics.memtableZombieCount.WithLabelValues(group).Set(float64(snap.MemTable.ZombieCount)) - // Block cache gauges: current usage (always) + configured capacity - // (when the source exposes it). Capacity is static for the lifetime - // of a DB in practice, but we re-read each tick so operators observe - // the new value immediately after a restart with a different - // ELASTICKV_PEBBLE_CACHE_MB. - c.metrics.blockCacheSizeBytes.WithLabelValues(group).Set(float64(snap.BlockCache.Size)) - if capSrc, ok := src.Source.(PebbleCacheCapacitySource); ok { - if capBytes := capSrc.BlockCacheCapacityBytes(); capBytes > 0 { - c.metrics.blockCacheCapacityBytes.WithLabelValues(group).Set(float64(capBytes)) - } - } - // Monotonic counters: emit only the positive delta. A smaller value // means the source was reset (store reopened); rebase silently // without emitting negative. prev := c.previous[src.GroupID] curr := pebbleSnapshot{ - compactCount: snap.Compact.Count, - blockCacheHits: snap.BlockCache.Hits, - blockCacheMisses: snap.BlockCache.Misses, + compactCount: snap.Compact.Count, } if curr.compactCount > prev.compactCount { c.metrics.compactCountTotal.WithLabelValues(group).Add(float64(curr.compactCount - prev.compactCount)) } - if curr.blockCacheHits > prev.blockCacheHits { - c.metrics.blockCacheHitsTotal.WithLabelValues(group).Add(float64(curr.blockCacheHits - prev.blockCacheHits)) + c.previous[src.GroupID] = curr +} + +func (c *PebbleCollector) observeBlockCache(snap *pebble.Metrics) { + c.metrics.blockCacheSizeBytes.WithLabelValues().Set(float64(snap.BlockCache.Size)) + + prev := c.previousBlockCache + curr := pebbleBlockCacheSnapshot{ + hits: snap.BlockCache.Hits, + misses: snap.BlockCache.Misses, } - if curr.blockCacheMisses > prev.blockCacheMisses { - c.metrics.blockCacheMissesTotal.WithLabelValues(group).Add(float64(curr.blockCacheMisses - prev.blockCacheMisses)) + if curr.hits > prev.hits { + c.metrics.blockCacheHitsTotal.WithLabelValues().Add(float64(curr.hits - prev.hits)) } - c.previous[src.GroupID] = curr + if curr.misses > prev.misses { + c.metrics.blockCacheMissesTotal.WithLabelValues().Add(float64(curr.misses - prev.misses)) + } + c.previousBlockCache = curr +} + +func (c *PebbleCollector) observeBlockCacheCapacity(src PebbleSource) bool { + capSrc, ok := src.Source.(PebbleCacheCapacitySource) + if !ok { + return false + } + capBytes := capSrc.BlockCacheCapacityBytes() + if capBytes <= 0 { + return false + } + c.metrics.blockCacheCapacityBytes.WithLabelValues().Set(float64(capBytes)) + return true } diff --git a/monitoring/pebble_test.go b/monitoring/pebble_test.go index dc9cb270f..d7e13097a 100644 --- a/monitoring/pebble_test.go +++ b/monitoring/pebble_test.go @@ -97,16 +97,16 @@ func TestPebbleCollectorMirrorsGaugesAndCounters(t *testing.T) { err := testutil.GatherAndCompare( registry.Gatherer(), strings.NewReader(` -# HELP elastickv_pebble_block_cache_hits_total Cumulative block cache hits reported by Pebble. +# HELP elastickv_pebble_block_cache_hits_total Cumulative process-wide block cache hits reported by Pebble. # TYPE elastickv_pebble_block_cache_hits_total counter -elastickv_pebble_block_cache_hits_total{group="1",node_address="10.0.0.1:50051",node_id="n1"} 150 -# HELP elastickv_pebble_block_cache_misses_total Cumulative block cache misses reported by Pebble. +elastickv_pebble_block_cache_hits_total{node_address="10.0.0.1:50051",node_id="n1"} 150 +# HELP elastickv_pebble_block_cache_misses_total Cumulative process-wide block cache misses reported by Pebble. # TYPE elastickv_pebble_block_cache_misses_total counter -elastickv_pebble_block_cache_misses_total{group="1",node_address="10.0.0.1:50051",node_id="n1"} 25 +elastickv_pebble_block_cache_misses_total{node_address="10.0.0.1:50051",node_id="n1"} 25 -# HELP elastickv_pebble_block_cache_size_bytes Current bytes in use by Pebble's block cache. +# HELP elastickv_pebble_block_cache_size_bytes Current bytes in use by the process-wide Pebble block cache. # TYPE elastickv_pebble_block_cache_size_bytes gauge -elastickv_pebble_block_cache_size_bytes{group="1",node_address="10.0.0.1:50051",node_id="n1"} 16384 +elastickv_pebble_block_cache_size_bytes{node_address="10.0.0.1:50051",node_id="n1"} 16384 # HELP elastickv_pebble_compact_count_total Cumulative number of compactions completed by Pebble since the process started. # TYPE elastickv_pebble_compact_count_total counter elastickv_pebble_compact_count_total{group="1",node_address="10.0.0.1:50051",node_id="n1"} 15 @@ -171,9 +171,9 @@ func TestPebbleCollectorHandlesSourceReset(t *testing.T) { err := testutil.GatherAndCompare( registry.Gatherer(), strings.NewReader(` -# HELP elastickv_pebble_block_cache_hits_total Cumulative block cache hits reported by Pebble. +# HELP elastickv_pebble_block_cache_hits_total Cumulative process-wide block cache hits reported by Pebble. # TYPE elastickv_pebble_block_cache_hits_total counter -elastickv_pebble_block_cache_hits_total{group="7",node_address="10.0.0.1:50051",node_id="n1"} 110 +elastickv_pebble_block_cache_hits_total{node_address="10.0.0.1:50051",node_id="n1"} 110 # HELP elastickv_pebble_compact_count_total Cumulative number of compactions completed by Pebble since the process started. # TYPE elastickv_pebble_compact_count_total counter elastickv_pebble_compact_count_total{group="7",node_address="10.0.0.1:50051",node_id="n1"} 12 @@ -225,12 +225,12 @@ func TestPebbleCollectorEmitsBlockCacheCapacity(t *testing.T) { err := testutil.GatherAndCompare( registry.Gatherer(), strings.NewReader(` -# HELP elastickv_pebble_block_cache_capacity_bytes Configured maximum size of Pebble's block cache in bytes. Paired with elastickv_pebble_block_cache_size_bytes so operators can see usage relative to capacity and with the hit/miss counters so they can reason about whether a low hit rate reflects a cold cache or an undersized one. +# HELP elastickv_pebble_block_cache_capacity_bytes Configured maximum size of the process-wide Pebble block cache in bytes. Paired with elastickv_pebble_block_cache_size_bytes so operators can see usage relative to capacity and with the hit/miss counters so they can reason about whether a low hit rate reflects a cold cache or an undersized one. # TYPE elastickv_pebble_block_cache_capacity_bytes gauge -elastickv_pebble_block_cache_capacity_bytes{group="3",node_address="10.0.0.1:50051",node_id="n1"} 2.68435456e+08 -# HELP elastickv_pebble_block_cache_size_bytes Current bytes in use by Pebble's block cache. +elastickv_pebble_block_cache_capacity_bytes{node_address="10.0.0.1:50051",node_id="n1"} 2.68435456e+08 +# HELP elastickv_pebble_block_cache_size_bytes Current bytes in use by the process-wide Pebble block cache. # TYPE elastickv_pebble_block_cache_size_bytes gauge -elastickv_pebble_block_cache_size_bytes{group="3",node_address="10.0.0.1:50051",node_id="n1"} 4096 +elastickv_pebble_block_cache_size_bytes{node_address="10.0.0.1:50051",node_id="n1"} 4096 `), "elastickv_pebble_block_cache_capacity_bytes", "elastickv_pebble_block_cache_size_bytes", @@ -238,11 +238,60 @@ elastickv_pebble_block_cache_size_bytes{group="3",node_address="10.0.0.1:50051", require.NoError(t, err) } +func TestPebbleCollectorEmitsSharedBlockCacheOnce(t *testing.T) { + registry := NewRegistry("n1", "10.0.0.1:50051") + collector := registry.PebbleCollector() + require.NotNil(t, collector) + + src1 := &fakePebbleCapacitySource{capacity: 256 << 20} + src2 := &fakePebbleCapacitySource{capacity: 256 << 20} + src1.set(newFakeMetrics( + 1, 2, 0, 0, 7, + 1, 1024, 0, + 8192, 100, 20, + )) + src2.set(newFakeMetrics( + 3, 4, 0, 0, 11, + 2, 2048, 0, + 8192, 100, 20, + )) + + collector.ObserveOnce([]PebbleSource{ + {GroupID: 1, GroupIDStr: "1", Source: src1}, + {GroupID: 2, GroupIDStr: "2", Source: src2}, + }) + + err := testutil.GatherAndCompare( + registry.Gatherer(), + strings.NewReader(` +# HELP elastickv_pebble_block_cache_capacity_bytes Configured maximum size of the process-wide Pebble block cache in bytes. Paired with elastickv_pebble_block_cache_size_bytes so operators can see usage relative to capacity and with the hit/miss counters so they can reason about whether a low hit rate reflects a cold cache or an undersized one. +# TYPE elastickv_pebble_block_cache_capacity_bytes gauge +elastickv_pebble_block_cache_capacity_bytes{node_address="10.0.0.1:50051",node_id="n1"} 2.68435456e+08 +# HELP elastickv_pebble_block_cache_hits_total Cumulative process-wide block cache hits reported by Pebble. +# TYPE elastickv_pebble_block_cache_hits_total counter +elastickv_pebble_block_cache_hits_total{node_address="10.0.0.1:50051",node_id="n1"} 100 +# HELP elastickv_pebble_block_cache_misses_total Cumulative process-wide block cache misses reported by Pebble. +# TYPE elastickv_pebble_block_cache_misses_total counter +elastickv_pebble_block_cache_misses_total{node_address="10.0.0.1:50051",node_id="n1"} 20 +# HELP elastickv_pebble_block_cache_size_bytes Current bytes in use by the process-wide Pebble block cache. +# TYPE elastickv_pebble_block_cache_size_bytes gauge +elastickv_pebble_block_cache_size_bytes{node_address="10.0.0.1:50051",node_id="n1"} 8192 +`), + "elastickv_pebble_block_cache_capacity_bytes", + "elastickv_pebble_block_cache_size_bytes", + "elastickv_pebble_block_cache_hits_total", + "elastickv_pebble_block_cache_misses_total", + ) + require.NoError(t, err) + require.Equal(t, 1, testutil.CollectAndCount(registry.pebble.blockCacheSizeBytes)) + require.Equal(t, 1, testutil.CollectAndCount(registry.pebble.blockCacheHitsTotal)) +} + func TestPebbleCollectorSkipsCapacityWhenUnsupported(t *testing.T) { // A PebbleMetricsSource that does NOT additionally implement // PebbleCacheCapacitySource must not cause the capacity gauge to be - // populated for its group. This preserves backward compatibility - // with sources that pre-date the capacity interface. + // populated. This preserves backward compatibility with sources that + // pre-date the capacity interface. registry := NewRegistry("n1", "10.0.0.1:50051") collector := registry.PebbleCollector() require.NotNil(t, collector) diff --git a/store/lsm_store.go b/store/lsm_store.go index d818b7bde..5d0ba3843 100644 --- a/store/lsm_store.go +++ b/store/lsm_store.go @@ -59,17 +59,17 @@ const ( // exceed maxSnapshotKeySize once the timestamp suffix is appended. maxPebbleEncodedKeySize = maxSnapshotKeySize + timestampSize - // defaultPebbleCacheBytes is the default Pebble block-cache capacity per - // store. Pebble's built-in EnsureDefaults() supplies only 8 MiB, which + // defaultPebbleCacheBytes is the default process-wide Pebble block-cache + // capacity. Pebble's built-in EnsureDefaults() supplies only 8 MiB, which // is far too small for our workloads: production observed a block-cache // hit rate of 0.003% (1.8B misses vs 58k hits) because the working set // evicted faster than it filled. 256 MiB is a conservative baseline per - // shard; operators can override via ELASTICKV_PEBBLE_CACHE_MB. + // node; operators can override via ELASTICKV_PEBBLE_CACHE_MB. defaultPebbleCacheBytes int64 = 256 << 20 - // pebbleCacheMBEnv is the env var operators use to override the per-store - // Pebble block-cache capacity. Units are MiB, integer only. Malformed or - // out-of-range values fall back to the default. + // pebbleCacheMBEnv is the env var operators use to override the + // process-wide Pebble block-cache capacity. Units are MiB, integer only. + // Malformed or out-of-range values fall back to the default. pebbleCacheMBEnv = "ELASTICKV_PEBBLE_CACHE_MB" // pebbleCacheMBMin / pebbleCacheMBMax define the accepted range for the @@ -111,17 +111,18 @@ const ( fsmSyncModeNoSync = "nosync" ) -// pebbleCacheBytes is the effective per-store Pebble block-cache capacity, +// pebbleCacheBytes is the effective process-wide Pebble block-cache capacity, // resolved once at process start. Exposed as a package variable so tests can -// swap it via setPebbleCacheBytesForTest; production code treats it as +// swap it via setSmallPebbleCacheForTest; production code treats it as // read-only after init(). -// -// TODO(perf/pebble): introduce a process-wide shared cache plumbed through -// NewPebbleStore so all shards on a node share one LRU eviction pool rather -// than each carrying an independent 256 MiB budget. That requires changing -// NewPebbleStore's signature and is deferred to a follow-up PR. var pebbleCacheBytes = defaultPebbleCacheBytes +var ( + processPebbleCacheMu sync.Mutex + processPebbleCache *pebble.Cache + processPebbleCacheBytes int64 +) + func init() { pebbleCacheBytes = resolvePebbleCacheBytes(os.Getenv(pebbleCacheMBEnv)) } @@ -260,20 +261,18 @@ func WithPebbleLogger(l *slog.Logger) PebbleStoreOption { // defaultPebbleOptionsWithCache returns the standard Pebble options used // throughout the store (including restores) to ensure consistent behaviour // between a freshly opened and a restored/swapped-in database, along with -// the owned *pebble.Cache handle. +// the store-owned *pebble.Cache reference. // // FormatMajorVersion is pinned to ratchet v1-era DBs above pebble v2's // FormatMinSupported (FormatFlushableIngest) before the v2 upgrade lands. // -// The returned options carry a freshly-allocated block cache sized from -// pebbleCacheBytes. pebble.NewCache hands back a refcounted Cache with -// ref=1; pebble.Open adds one reference, so the caller MUST Unref the -// returned cache after the DB is closed (or after pebble.Open fails) to -// fully release the memory. Callers that only need *pebble.Options should -// still take the cache handle and defer its Unref to avoid leaking a -// 256 MiB (default) allocation per call. +// The returned options carry a process-wide shared block cache sized from +// pebbleCacheBytes. The process holds one base reference; each call adds one +// store/open reference and returns it to the caller. The caller MUST Unref the +// returned cache after the DB is closed (or after pebble.Open fails), matching +// the lifetime rules used by NewPebbleStore, Restore, and temp restore DBs. func defaultPebbleOptionsWithCache() (*pebble.Options, *pebble.Cache) { - cache := pebble.NewCache(pebbleCacheBytes) + cache := processPebbleCacheRef() opts := &pebble.Options{ FS: vfs.Default, FormatMajorVersion: pebble.FormatVirtualSSTables, @@ -285,6 +284,22 @@ func defaultPebbleOptionsWithCache() (*pebble.Options, *pebble.Cache) { return opts, cache } +func processPebbleCacheRef() *pebble.Cache { + processPebbleCacheMu.Lock() + defer processPebbleCacheMu.Unlock() + + if processPebbleCache == nil || processPebbleCacheBytes != pebbleCacheBytes { + old := processPebbleCache + processPebbleCache = pebble.NewCache(pebbleCacheBytes) + processPebbleCacheBytes = pebbleCacheBytes + if old != nil { + old.Unref() + } + } + processPebbleCache.Ref() + return processPebbleCache +} + // NewPebbleStore creates a new Pebble-backed MVCC store. func NewPebbleStore(dir string, opts ...PebbleStoreOption) (MVCCStore, error) { fsmOpts, fsmLabel := resolveFSMApplyWriteOpts(os.Getenv(fsmSyncModeEnv)) diff --git a/store/lsm_store_env_test.go b/store/lsm_store_env_test.go index addf7d8de..fbe2a6732 100644 --- a/store/lsm_store_env_test.go +++ b/store/lsm_store_env_test.go @@ -7,15 +7,15 @@ import ( "github.com/stretchr/testify/require" ) -// setPebbleCacheBytesForTest swaps the package-level pebbleCacheBytes value -// for the duration of a single test and restores it during t.Cleanup. The -// real override happens in init() from ELASTICKV_PEBBLE_CACHE_MB; tests use -// this helper to exercise specific cache sizes without relying on process +// setSmallPebbleCacheForTest swaps the package-level pebbleCacheBytes value +// to 16 MiB for the duration of a single test and restores it during +// t.Cleanup. The real override happens in init() from +// ELASTICKV_PEBBLE_CACHE_MB; tests use this helper without relying on process // env state. -func setPebbleCacheBytesForTest(t *testing.T, n int64) { +func setSmallPebbleCacheForTest(t *testing.T) { t.Helper() prev := pebbleCacheBytes - pebbleCacheBytes = n + pebbleCacheBytes = 16 << 20 t.Cleanup(func() { pebbleCacheBytes = prev }) } @@ -66,23 +66,23 @@ func TestPebbleCacheEnvOverride(t *testing.T) { }) } -// TestSetPebbleCacheBytesForTestRestores verifies the helper reinstates the +// TestSetSmallPebbleCacheForTestRestores verifies the helper reinstates the // previous value via t.Cleanup so tests that tweak pebbleCacheBytes do not // leak state to later tests in the package. -func TestSetPebbleCacheBytesForTestRestores(t *testing.T) { +func TestSetSmallPebbleCacheForTestRestores(t *testing.T) { before := pebbleCacheBytes t.Run("inner", func(t *testing.T) { - setPebbleCacheBytesForTest(t, 16<<20) + setSmallPebbleCacheForTest(t) require.Equal(t, int64(16)<<20, pebbleCacheBytes) }) require.Equal(t, before, pebbleCacheBytes) } // TestDefaultPebbleOptionsCarriesCache sanity-checks that the options -// constructor wires a cache through at the configured size and that Unref -// is safe to call after closing the DB (the primary lifecycle path). +// constructor wires the process-shared cache through at the configured size +// and that Unref is safe for each borrowed store/open reference. func TestDefaultPebbleOptionsCarriesCache(t *testing.T) { - setPebbleCacheBytesForTest(t, 16<<20) + setSmallPebbleCacheForTest(t) opts, cache := defaultPebbleOptionsWithCache() require.NotNil(t, cache) require.Same(t, cache, opts.Cache) @@ -90,6 +90,39 @@ func TestDefaultPebbleOptionsCarriesCache(t *testing.T) { cache.Unref() } +func TestDefaultPebbleOptionsSharesProcessCache(t *testing.T) { + setSmallPebbleCacheForTest(t) + opts1, cache1 := defaultPebbleOptionsWithCache() + defer cache1.Unref() + opts2, cache2 := defaultPebbleOptionsWithCache() + defer cache2.Unref() + + require.Same(t, cache1, cache2) + require.Same(t, cache1, opts1.Cache) + require.Same(t, cache1, opts2.Cache) +} + +func TestNewPebbleStoreSharesProcessCache(t *testing.T) { + setSmallPebbleCacheForTest(t) + + s1, err := NewPebbleStore(t.TempDir()) + require.NoError(t, err) + defer s1.Close() + ps1, ok := s1.(*pebbleStore) + require.True(t, ok) + + s2, err := NewPebbleStore(t.TempDir()) + require.NoError(t, err) + defer s2.Close() + ps2, ok := s2.(*pebbleStore) + require.True(t, ok) + + require.NotNil(t, ps1.cache) + require.Same(t, ps1.cache, ps2.cache) + require.Equal(t, int64(16)<<20, ps1.BlockCacheCapacityBytes()) + require.Equal(t, ps1.BlockCacheCapacityBytes(), ps2.BlockCacheCapacityBytes()) +} + // newPebbleStoreWithFSMApplyWriteOptsForTest constructs a pebbleStore // (not the MVCCStore interface) with an explicit *pebble.WriteOptions // and sync-mode label for the FSM commit path, bypassing the