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 makes `MaxConcurrentCompactions()` return a "baseline
limit" and an "upper limit". The baseline limit is the "starting"
value for the max concurrency, which Pebble can increase dynamically
up to the upper limit. Prior to this change, the (implicit) baseline
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
+		MaxConcurrentCompactions: 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{MaxConcurrentCompactions: func() (int, int) { return 1, 9 }},
+		},
+		{
+			c:       replayConfig{optionsString: `[Options] baseline_concurrent_compactions=4`},
+			options: &pebble.Options{MaxConcurrentCompactions: func() (int, int) { return 4, 4 }},
+		},
+		{
+			c:       replayConfig{optionsString: `[Options] baseline_concurrent_compactions=4 max_concurrent_compactions=9`},
+			options: &pebble.Options{MaxConcurrentCompactions: func() (int, int) { return 4, 9 }},
 		},
 		{
 			c:       replayConfig{optionsString: `[Options] bytes_per_sync=90000`},

compaction.go

@@ -58,7 +58,8 @@ func expandedCompactionByteSizeLimit(opts *Options, level int, availBytes uint64
 	//
 	// NB: this heuristic is an approximation since we may run more compactions
 	// than MaxConcurrentCompactions.
-	diskMax := (availBytes / 2) / uint64(opts.MaxConcurrentCompactions())
+	_, upperConcurrencyLimit := opts.MaxConcurrentCompactions()
+	diskMax := (availBytes / 2) / uint64(upperConcurrencyLimit)
 	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.MaxConcurrentCompactions()
 		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 _, upperLimit := d.opts.MaxConcurrentCompactions(); d.mu.compact.compactingCount >= upperLimit {
+		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

@@ -1218,7 +1218,7 @@ func responsibleForGarbageBytes(
 }
 
 func (p *compactionPickerByScore) getCompactionConcurrency() int {
-	maxConcurrentCompactions := p.opts.MaxConcurrentCompactions()
+	baselineLimit, upperLimit := p.opts.MaxConcurrentCompactions()
 	// 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
@@ -1235,24 +1235,26 @@ 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)
 	}
-	return max(min(maxConcurrentCompactions, max(l0ReadAmpCompactions, compactionDebtCompactions)), 1)
+	extraCompactions := max(l0ReadAmpCompactions, compactionDebtCompactions)
+	return min(baselineLimit+extraCompactions, upperLimit)
 }
 
 // TODO(sumeer): remove unless someone actually finds this useful.

compaction_picker_test.go

@@ -216,13 +216,13 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 				// <level>: <size> [compensation]
 				var errMsg string
 				vers, l0Organizer, opts, errMsg = loadVersion(t, d)
-				opts.MaxConcurrentCompactions = func() int {
+				opts.MaxConcurrentCompactions = func() (int, int) {
 					// This test only limits the count based on the L0 read amp and
 					// compaction debt, so we would like to return math.MaxInt. But we
 					// don't since it is also used in expandedCompactionByteSizeLimit,
 					// and causes the expanded bytes to reduce. The test cases never
 					// pick more than 4 compactions, so we use 4.
-					return 4
+					return 1, 4
 				}
 				if errMsg != "" {
 					return errMsg
@@ -636,7 +636,8 @@ func TestCompactionPickerL0(t *testing.T) {
 func TestCompactionPickerConcurrency(t *testing.T) {
 	opts := DefaultOptions()
 	opts.Experimental.L0CompactionConcurrency = 1
-	opts.MaxConcurrentCompactions = func() int { return 4 }
+	baselineConcurrencyLimit, upperConcurrencyLimit := 1, 4
+	opts.MaxConcurrentCompactions = func() (int, int) { return baselineConcurrencyLimit, upperConcurrencyLimit }
 
 	parseMeta := func(s string) (*tableMetadata, error) {
 		parts := strings.Split(s, ":")
@@ -804,6 +805,8 @@ 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, "upper_concurrency_limit", &upperConcurrencyLimit)
+			td.MaybeScanArgs(t, "baseline_concurrency_limit", &baselineConcurrencyLimit)
 
 			env := compactionEnv{
 				earliestUnflushedSeqNum: math.MaxUint64,
@@ -831,8 +834,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
+				var upperLimit int
+				baselineLimit := 1
+				td.ScanArgs(t, "max", &upperLimit)
+				td.MaybeScanArgs(t, "baseline", &baselineLimit)
+				d.opts.MaxConcurrentCompactions = func() (int, int) {
+					return baselineLimit, upperLimit
 				}
 				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
+			MaxConcurrentCompactions: 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
+		MaxConcurrentCompactions: 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 },
+			MaxConcurrentCompactions:    func() (int, int) { return 1, maxConcurrentCompactions },
 			LBaseMaxBytes:               64 << 10,
 			L0CompactionThreshold:       2,
 			L0CompactionFileThreshold:   2,

iterator_test.go

@@ -2133,7 +2133,7 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 		FS:                       vfs.NewCrashableMem(),
 		Comparer:                 testkeys.Comparer,
 		FormatMajorVersion:       FormatNewest,
-		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
+		MaxConcurrentCompactions: func() (int, int) { return 1, maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
 			sstable.NewTestKeysBlockPropertyCollector,
 		},
@@ -2146,7 +2146,7 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 		FS:                       vfs.NewCrashableMem(),
 		Comparer:                 testkeys.Comparer,
 		FormatMajorVersion:       FormatNewest,
-		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
+		MaxConcurrentCompactions: func() (int, int) { return 1, maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
 			sstable.NewTestKeysBlockPropertyCollector,
 		},
@@ -2308,7 +2308,7 @@ func BenchmarkIterator_RangeKeyMasking(b *testing.B) {
 		FS:                       mem,
 		Comparer:                 testkeys.Comparer,
 		FormatMajorVersion:       FormatNewest,
-		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
+		MaxConcurrentCompactions: 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
+	baselineConcurrentCompactions := 1
+	if rng.IntN(4) == 0 {
+		baselineConcurrentCompactions = rng.IntN(maxConcurrentCompactions) + 1
+	}
+	opts.MaxConcurrentCompactions = func() (int, int) {
+		return baselineConcurrentCompactions, maxConcurrentCompactions
 	}
 	maxConcurrentDownloads := rng.IntN(3) + 1 // 1-3
 	opts.MaxConcurrentDownloads = func() int {