storage: fix up compaction concurrency code

Author: RaduBerinde

pkg/storage/metamorphic/options.go

@@ -127,7 +127,7 @@ func randomOptions() *pebble.Options {
 	opts.MemTableSize = 1 << rngIntRange(rng, 11, 28)
 	opts.MemTableStopWritesThreshold = int(rngIntRange(rng, 2, 7))
 	maxConcurrentCompactions := int(rngIntRange(rng, 1, 4))
-	opts.MaxConcurrentCompactions = func() int { return maxConcurrentCompactions }
+	opts.CompactionConcurrencyRange = func() (lower, upper int) { return 1, maxConcurrentCompactions }
 
 	opts.Cache = pebble.NewCache(1 << rngIntRange(rng, 1, 30))
 	defer opts.Cache.Unref()

pkg/storage/mvcc.go

@@ -131,7 +131,7 @@ func getDefaultMaxConcurrentCompactions() int {
 
 // l0SubLevelCompactionConcurrency is the sub-level threshold at which to
 // allow an increase in compaction concurrency. The maximum is still
-// controlled by pebble.Options.MaxConcurrentCompactions. The default of 2
+// controlled by pebble.Options.CompactionConcurrencyRange. The default of 2
 // allows an additional compaction (so total 1 + 1 = 2 compactions) when the
 // sub-level count is 2, and increments concurrency by 1 whenever sub-level
 // count increases by 2 (so 1 + 2 = 3 compactions) when sub-level count is 4,

pkg/storage/open.go

@@ -202,7 +202,7 @@ func RemoteStorageFactory(accessor *cloud.EarlyBootExternalStorageAccessor) Conf
 // compactions an Engine will execute.
 func MaxConcurrentCompactions(n int) ConfigOption {
 	return func(cfg *engineConfig) error {
-		cfg.opts.MaxConcurrentCompactions = func() int { return n }
+		cfg.opts.CompactionConcurrencyRange = func() (lower, upper int) { return 1, n }
 		return nil
 	}
 }

pkg/storage/pebble.go

@@ -360,14 +360,10 @@ func DefaultPebbleOptions() *pebble.Options {
 		KeySchema:  DefaultKeySchema,
 		KeySchemas: sstable.MakeKeySchemas(KeySchemas...),
 		// A value of 2 triggers a compaction when there is 1 sub-level.
-		L0CompactionThreshold: 2,
-		L0StopWritesThreshold: 1000,
-		LBaseMaxBytes:         64 << 20, // 64 MB
-		Levels:                make([]pebble.LevelOptions, 7),
-		// NB: Options.MaxConcurrentCompactions may be "wrapped" in NewPebble to
-		// allow overriding the max at runtime through
-		// Engine.SetCompactionConcurrency.
-		MaxConcurrentCompactions:    func() int { return defaultMaxConcurrentCompactions },
+		L0CompactionThreshold:       2,
+		L0StopWritesThreshold:       1000,
+		LBaseMaxBytes:               64 << 20, // 64 MB
+		Levels:                      make([]pebble.LevelOptions, 7),
 		MemTableSize:                64 << 20, // 64 MB
 		MemTableStopWritesThreshold: 4,
 		Merger:                      MVCCMerger,
@@ -676,17 +672,20 @@ func newPebble(ctx context.Context, cfg engineConfig) (p *Pebble, err error) {
 			return getCompressionAlgorithm(ctx, cfg.settings, CompressionAlgorithmStorage)
 		}
 	}
-
+	// Note: the MaxConcurrentCompactions function will be wrapped below to allow
+	// overriding dynamically via overriding the max at runtime through
+	// Engine.SetCompactionConcurrency.
+	if cfg.opts.CompactionConcurrencyRange == nil {
+		cfg.opts.CompactionConcurrencyRange = func() (lower, upper int) {
+			return 1, defaultMaxConcurrentCompactions
+		}
+	}
 	if cfg.opts.MaxConcurrentDownloads == nil {
 		cfg.opts.MaxConcurrentDownloads = func() int {
 			return int(concurrentDownloadCompactions.Get(&cfg.settings.SV))
 		}
 	}
 
-	if cfg.opts.MaxConcurrentCompactions == nil {
-		cfg.opts.MaxConcurrentCompactions = func() int { return defaultMaxConcurrentCompactions }
-	}
-
 	cfg.opts.EnsureDefaults()
 
 	// The context dance here is done so that we have a clean context without
@@ -791,11 +790,9 @@ func newPebble(ctx context.Context, cfg engineConfig) (p *Pebble, err error) {
 		diskWriteStatsCollector: cfg.DiskWriteStatsCollector,
 	}
 
-	// MaxConcurrentCompactions can be set by multiple sources, but all the
-	// sources will eventually call NewPebble. So, we override
-	// cfg.opts.MaxConcurrentCompactions to a closure which allows ovderriding the
-	// value.
-	cfg.opts.MaxConcurrentCompactions = p.cco.Wrap(cfg.opts.MaxConcurrentCompactions)
+	// Wrap the MaxConcurrentCompactions function to allow overriding dynamically via
+	// overriding the max at runtime through Engine.SetCompactionConcurrency.
+	cfg.opts.CompactionConcurrencyRange = p.cco.Wrap(cfg.opts.CompactionConcurrencyRange)
 
 	// NB: The ordering of the event listeners passed to TeeEventListener is
 	// deliberate. The listener returned by makeMetricEtcEventListener is
@@ -2740,8 +2737,7 @@ func (e *ExceedMaxSizeError) Error() string {
 	return fmt.Sprintf("export size (%d bytes) exceeds max size (%d bytes)", e.reached, e.maxSize)
 }
 
-// compactionConcurrencyOverride allows overriding the max concurrent
-// compactions.
+// compactionConcurrencyOverride allows overriding the compaction concurrency.
 type compactionConcurrencyOverride struct {
 	override atomic.Uint32
 }
@@ -2751,12 +2747,15 @@ func (cco *compactionConcurrencyOverride) Set(value uint32) {
 	cco.override.Store(value)
 }
 
-// Wrap a MaxConcurrentCompactions function to take into account the override.
-func (cco *compactionConcurrencyOverride) Wrap(maxConcurrentCompactions func() int) func() int {
-	return func() int {
+// Wrap a CompactionConcurrencyRange function to take into account the override.
+func (cco *compactionConcurrencyOverride) Wrap(
+	compactionConcurrencyRange func() (lower, upper int),
+) func() (lower, upper int) {
+	return func() (lower, upper int) {
 		if o := cco.override.Load(); o > 0 {
-			return int(o)
+			// We override both the lower and upper limits.
+			return int(o), int(o)
 		}
-		return maxConcurrentCompactions()
+		return compactionConcurrencyRange()
 	}
 }

pkg/storage/pebble_test.go

@@ -1675,9 +1675,11 @@ func TestPebbleSetCompactionConcurrency(t *testing.T) {
 	require.NoError(t, err)
 	defer p.Close()
 
-	require.Equal(t, 4, p.cfg.opts.MaxConcurrentCompactions())
+	require.Equal(t, "1 4", fmt.Sprint(p.cfg.opts.CompactionConcurrencyRange()))
+
 	p.SetCompactionConcurrency(10)
-	require.Equal(t, 10, p.cfg.opts.MaxConcurrentCompactions())
+	require.Equal(t, "10 10", fmt.Sprint(p.cfg.opts.CompactionConcurrencyRange()))
+
 	p.SetCompactionConcurrency(0)
-	require.Equal(t, 4, p.cfg.opts.MaxConcurrentCompactions())
+	require.Equal(t, "1 4", fmt.Sprint(p.cfg.opts.CompactionConcurrencyRange()))
 }