db: change MaxConcurrentCompactions() to return a range

`MaxConcurrentCompactions()` returns an upper limit on the compaction
concurrency. Within this upper limit, Pebble decides what the
concurrency limit is (depending on L0 amplification and compaction debt).

There are cases where we want more concurrency for other reasons (like
space amp). Having a knob can be useful to tweak things in a
production environment.

This change changes `MaxConcurrentCompactions()` to
`CompactionConcurrencyRange()` which returns both a lower and upper
value. The lower limit is the "baseline" value for the concurrency,
which Pebble can increase dynamically up to the upper limit. Prior to
this change, the (implicit) lower limit was always 1.

Author: RaduBerinde

cmd/pebble/db.go

@@ -72,8 +72,8 @@ func newPebbleDB(dir string) DB {
 		Merger: &pebble.Merger{
 			Name: "cockroach_merge_operator",
 		},
-		MaxConcurrentCompactions: func() int {
-			return 3
+		CompactionConcurrencyRange: func() (int, int) {
+			return 1, 3
 		},
 	}
 	// In FormatColumnarBlocks (the value of FormatNewest at the time of

cmd/pebble/replay_test.go

@@ -21,7 +21,15 @@ func TestParseOptionsStr(t *testing.T) {
 	testCases := []testCase{
 		{
 			c:       replayConfig{optionsString: `[Options] max_concurrent_compactions=9`},
-			options: &pebble.Options{MaxConcurrentCompactions: func() int { return 9 }},
+			options: &pebble.Options{CompactionConcurrencyRange: func() (int, int) { return 1, 9 }},
+		},
+		{
+			c:       replayConfig{optionsString: `[Options] concurrent_compactions=4`},
+			options: &pebble.Options{CompactionConcurrencyRange: func() (int, int) { return 4, 4 }},
+		},
+		{
+			c:       replayConfig{optionsString: `[Options] concurrent_compactions=4 max_concurrent_compactions=9`},
+			options: &pebble.Options{CompactionConcurrencyRange: func() (int, int) { return 4, 9 }},
 		},
 		{
 			c:       replayConfig{optionsString: `[Options] bytes_per_sync=90000`},

compaction.go

@@ -57,8 +57,9 @@ func expandedCompactionByteSizeLimit(opts *Options, level int, availBytes uint64
 	// than this threshold before expansion.
 	//
 	// NB: this heuristic is an approximation since we may run more compactions
-	// than MaxConcurrentCompactions.
-	diskMax := (availBytes / 2) / uint64(opts.MaxConcurrentCompactions())
+	// than the upper concurrency limit.
+	_, maxConcurrency := opts.CompactionConcurrencyRange()
+	diskMax := (availBytes / 2) / uint64(maxConcurrency)
 	if v > diskMax {
 		v = diskMax
 	}
@@ -1918,7 +1919,7 @@ func (d *DB) GetAllowedWithoutPermission() int {
 	allowedBasedOnManual := 0
 	manualBacklog := int(d.mu.compact.manualLen.Load())
 	if manualBacklog > 0 {
-		maxAllowed := d.opts.MaxConcurrentCompactions()
+		_, maxAllowed := d.opts.CompactionConcurrencyRange()
 		allowedBasedOnManual = min(maxAllowed, manualBacklog+allowedBasedOnBacklog)
 	}
 	return max(allowedBasedOnBacklog, allowedBasedOnManual)
@@ -1982,10 +1983,12 @@ func (d *DB) pickManualCompaction(env compactionEnv) (pc *pickedCompaction) {
 // Returns true iff a compaction was started.
 func (d *DB) tryScheduleDeleteOnlyCompaction() bool {
 	if d.opts.private.disableDeleteOnlyCompactions || d.opts.DisableAutomaticCompactions ||
-		d.mu.compact.compactingCount >= d.opts.MaxConcurrentCompactions() ||
 		len(d.mu.compact.deletionHints) == 0 {
 		return false
 	}
+	if _, maxConcurrency := d.opts.CompactionConcurrencyRange(); d.mu.compact.compactingCount >= maxConcurrency {
+		return false
+	}
 	v := d.mu.versions.currentVersion()
 	snapshots := d.mu.snapshots.toSlice()
 	// We need to save the value of exciseEnabled in the compaction itself, as

compaction_picker.go

@@ -1221,7 +1221,10 @@ func responsibleForGarbageBytes(
 }
 
 func (p *compactionPickerByScore) getCompactionConcurrency() int {
-	maxConcurrentCompactions := p.opts.MaxConcurrentCompactions()
+	lower, upper := p.opts.CompactionConcurrencyRange()
+	if lower >= upper {
+		return upper
+	}
 	// Compaction concurrency is controlled by L0 read-amp. We allow one
 	// additional compaction per L0CompactionConcurrency sublevels, as well as
 	// one additional compaction per CompactionDebtConcurrency bytes of
@@ -1238,23 +1241,25 @@ func (p *compactionPickerByScore) getCompactionConcurrency() int {
 	// Rearranging,
 	// n <= l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency.
 	// So we can run up to
-	// l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency + 1 compactions
-	l0ReadAmpCompactions := 1
+	// l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency extra compactions.
+	l0ReadAmpCompactions := 0
 	if p.opts.Experimental.L0CompactionConcurrency > 0 {
 		l0ReadAmp := p.l0Organizer.MaxDepthAfterOngoingCompactions()
-		l0ReadAmpCompactions = (l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency) + 1
+		l0ReadAmpCompactions = (l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency)
 	}
 	// compactionDebt >= ccSignal2 then can run another compaction, where
 	// ccSignal2 = uint64(n) * p.opts.Experimental.CompactionDebtConcurrency
 	// Rearranging,
 	// n <= compactionDebt / p.opts.Experimental.CompactionDebtConcurrency
 	// So we can run up to
-	// compactionDebt / p.opts.Experimental.CompactionDebtConcurrency + 1 compactions.
-	compactionDebtCompactions := 1
+	// compactionDebt / p.opts.Experimental.CompactionDebtConcurrency extra
+	// compactions.
+	compactionDebtCompactions := 0
 	if p.opts.Experimental.CompactionDebtConcurrency > 0 {
 		compactionDebt := p.estimatedCompactionDebt(0)
-		compactionDebtCompactions = int(compactionDebt/p.opts.Experimental.CompactionDebtConcurrency) + 1
+		compactionDebtCompactions = int(compactionDebt / p.opts.Experimental.CompactionDebtConcurrency)
 	}
+
 	compactableGarbageCompactions := 0
 	garbageFractionLimit := p.opts.Experimental.CompactionGarbageFractionForMaxConcurrency()
 	if garbageFractionLimit > 0 && p.dbSizeBytes > 0 {
@@ -1263,11 +1268,12 @@ func (p *compactionPickerByScore) getCompactionConcurrency() int {
 				*rangeDeletionsBytesEstimateAnnotator.MultiLevelAnnotation(p.vers.Levels[:])
 		garbageFraction := float64(compactableGarbageBytes) / float64(p.dbSizeBytes)
 		compactableGarbageCompactions =
-			int((garbageFraction / garbageFractionLimit) * float64(maxConcurrentCompactions))
+			int((garbageFraction / garbageFractionLimit) * float64(upper-lower))
 	}
-	concurrencyBasedOnSignals :=
-		max(l0ReadAmpCompactions, compactionDebtCompactions, compactableGarbageCompactions)
-	return min(maxConcurrentCompactions, max(concurrencyBasedOnSignals, 1))
+
+	extraCompactions := max(l0ReadAmpCompactions, compactionDebtCompactions, compactableGarbageCompactions, 0)
+
+	return min(lower+extraCompactions, upper)
 }
 
 // TODO(sumeer): remove unless someone actually finds this useful.

compaction_picker_test.go

@@ -218,8 +218,8 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 				if d.HasArg("max-concurrent-compactions") {
 					d.ScanArgs(t, "max-concurrent-compactions", &maxConcurrentCompactions)
 				}
-				opts.MaxConcurrentCompactions = func() int {
-					return maxConcurrentCompactions
+				opts.CompactionConcurrencyRange = func() (int, int) {
+					return 1, maxConcurrentCompactions
 				}
 				if d.HasArg("compaction-debt-concurrency") {
 					d.ScanArgs(t, "compaction-debt-concurrency", &opts.Experimental.CompactionDebtConcurrency)
@@ -647,7 +647,8 @@ func TestCompactionPickerL0(t *testing.T) {
 func TestCompactionPickerConcurrency(t *testing.T) {
 	opts := DefaultOptions()
 	opts.Experimental.L0CompactionConcurrency = 1
-	opts.MaxConcurrentCompactions = func() int { return 4 }
+	lowerConcurrencyLimit, upperConcurrencyLimit := 1, 4
+	opts.CompactionConcurrencyRange = func() (int, int) { return lowerConcurrencyLimit, upperConcurrencyLimit }
 
 	parseMeta := func(s string) (*tableMetadata, error) {
 		parts := strings.Split(s, ":")
@@ -815,6 +816,7 @@ func TestCompactionPickerConcurrency(t *testing.T) {
 			td.MaybeScanArgs(t, "l0_compaction_threshold", &opts.L0CompactionThreshold)
 			td.MaybeScanArgs(t, "l0_compaction_concurrency", &opts.Experimental.L0CompactionConcurrency)
 			td.MaybeScanArgs(t, "compaction_debt_concurrency", &opts.Experimental.CompactionDebtConcurrency)
+			td.MaybeScanArgs(t, "concurrency", &lowerConcurrencyLimit, &upperConcurrencyLimit)
 
 			env := compactionEnv{
 				earliestUnflushedSeqNum: math.MaxUint64,
@@ -842,8 +844,10 @@ func TestCompactionPickerConcurrency(t *testing.T) {
 				fmt.Fprintf(&result, "nil")
 			}
 			return result.String()
+
+		default:
+			return fmt.Sprintf("unrecognized command: %s", td.Cmd)
 		}
-		return fmt.Sprintf("unrecognized command: %s", td.Cmd)
 	})
 }
 

compaction_test.go

@@ -1157,10 +1157,12 @@ func TestCompaction(t *testing.T) {
 				return ""
 
 			case "set-concurrent-compactions":
-				var concurrentCompactions int
-				td.ScanArgs(t, "num", &concurrentCompactions)
-				d.opts.MaxConcurrentCompactions = func() int {
-					return concurrentCompactions
+				lower := 1
+				upper := 1
+				td.MaybeScanArgs(t, "max", &upper)
+				td.MaybeScanArgs(t, "range", &lower, upper)
+				d.opts.CompactionConcurrencyRange = func() (int, int) {
+					return lower, upper
 				}
 				return ""
 

db_test.go

@@ -1187,8 +1187,8 @@ func TestDBConcurrentCompactClose(t *testing.T) {
 	for i := 0; i < 100; i++ {
 		opts := &Options{
 			FS: mem,
-			MaxConcurrentCompactions: func() int {
-				return 2
+			CompactionConcurrencyRange: func() (int, int) {
+				return 1, 2
 			},
 		}
 		d, err := Open("", testingRandomized(t, opts))
@@ -1522,8 +1522,8 @@ func TestMemtableIngestInversion(t *testing.T) {
 		MemTableStopWritesThreshold: 1000,
 		L0StopWritesThreshold:       1000,
 		L0CompactionThreshold:       2,
-		MaxConcurrentCompactions: func() int {
-			return 1000
+		CompactionConcurrencyRange: func() (int, int) {
+			return 1, 1000
 		},
 	}
 

error_test.go

@@ -460,7 +460,7 @@ func TestDBCompactionCrash(t *testing.T) {
 			FS:                          fs,
 			Logger:                      testLogger{t: t},
 			MemTableSize:                128 << 10,
-			MaxConcurrentCompactions:    func() int { return maxConcurrentCompactions },
+			CompactionConcurrencyRange:  func() (int, int) { return 1, maxConcurrentCompactions },
 			LBaseMaxBytes:               64 << 10,
 			L0CompactionThreshold:       2,
 			L0CompactionFileThreshold:   2,

iterator_test.go

@@ -2130,10 +2130,10 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 	maxProcs := runtime.GOMAXPROCS(0)
 
 	opts1 := &Options{
-		FS:                       vfs.NewCrashableMem(),
-		Comparer:                 testkeys.Comparer,
-		FormatMajorVersion:       FormatNewest,
-		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
+		FS:                         vfs.NewCrashableMem(),
+		Comparer:                   testkeys.Comparer,
+		FormatMajorVersion:         FormatNewest,
+		CompactionConcurrencyRange: func() (int, int) { return 1, maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
 			sstable.NewTestKeysBlockPropertyCollector,
 		},
@@ -2143,10 +2143,10 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 	require.NoError(t, err)
 
 	opts2 := &Options{
-		FS:                       vfs.NewCrashableMem(),
-		Comparer:                 testkeys.Comparer,
-		FormatMajorVersion:       FormatNewest,
-		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
+		FS:                         vfs.NewCrashableMem(),
+		Comparer:                   testkeys.Comparer,
+		FormatMajorVersion:         FormatNewest,
+		CompactionConcurrencyRange: func() (int, int) { return 1, maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
 			sstable.NewTestKeysBlockPropertyCollector,
 		},
@@ -2305,10 +2305,10 @@ func BenchmarkIterator_RangeKeyMasking(b *testing.B) {
 	mem := vfs.NewMem()
 	maxProcs := runtime.GOMAXPROCS(0)
 	opts := &Options{
-		FS:                       mem,
-		Comparer:                 testkeys.Comparer,
-		FormatMajorVersion:       FormatNewest,
-		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
+		FS:                         mem,
+		Comparer:                   testkeys.Comparer,
+		FormatMajorVersion:         FormatNewest,
+		CompactionConcurrencyRange: func() (int, int) { return 1, maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
 			sstable.NewTestKeysBlockPropertyCollector,
 		},

metamorphic/options.go

@@ -687,8 +687,12 @@ func RandomOptions(
 	}
 	opts.LBaseMaxBytes = 1 << uint(rng.IntN(30)) // 1B - 1GB
 	maxConcurrentCompactions := rng.IntN(3) + 1  // 1-3
-	opts.MaxConcurrentCompactions = func() int {
-		return maxConcurrentCompactions
+	minConcurrentCompactions := 1
+	if rng.IntN(4) == 0 {
+		minConcurrentCompactions = rng.IntN(maxConcurrentCompactions) + 1
+	}
+	opts.CompactionConcurrencyRange = func() (int, int) {
+		return minConcurrentCompactions, maxConcurrentCompactions
 	}
 	// [-0.2, 0.4], in steps of 0.2.
 	garbageFrac := float64(rng.IntN(5))/5.0 - 0.2