db: remove MaxWriterConcurrency and CPUWorkPermissionGranter

Remove the MaxWriterConcurrency option and the CPUWorkPermissionGranter
interface. The new columnar-block sstable writer does not support parallel
compression. We expect we do not want to parallelize work beneath the
granularity of compactions, in part because it is difficult to integrate with
admission control.

Author: jbowens

compaction.go

@@ -3118,12 +3118,11 @@ func (d *DB) compactAndWrite(
 		}
 		// Create a new table.
 		writerOpts := d.opts.MakeWriterOptions(c.outputLevel.level, tableFormat)
-		objMeta, tw, cpuWorkHandle, err := d.newCompactionOutput(jobID, c, writerOpts)
+		objMeta, tw, err := d.newCompactionOutput(jobID, c, writerOpts)
 		if err != nil {
 			return runner.Finish().WithError(err)
 		}
 		runner.WriteTable(objMeta, tw)
-		d.opts.Experimental.CPUWorkPermissionGranter.CPUWorkDone(cpuWorkHandle)
 	}
 	result = runner.Finish()
 	if result.Err == nil {
@@ -3246,10 +3245,10 @@ func (c *compaction) makeVersionEdit(result compact.Result) (*versionEdit, error
 // compaction or flush.
 func (d *DB) newCompactionOutput(
 	jobID JobID, c *compaction, writerOpts sstable.WriterOptions,
-) (objstorage.ObjectMetadata, sstable.RawWriter, CPUWorkHandle, error) {
+) (objstorage.ObjectMetadata, sstable.RawWriter, error) {
 	writable, objMeta, err := d.newCompactionOutputObj(jobID, c, base.FileTypeTable)
 	if err != nil {
-		return objstorage.ObjectMetadata{}, nil, nil, err
+		return objstorage.ObjectMetadata{}, nil, err
 	}
 
 	var reason string
@@ -3272,16 +3271,10 @@ func (d *DB) newCompactionOutput(
 		},
 	})
 
-	const MaxFileWriteAdditionalCPUTime = time.Millisecond * 100
-	cpuWorkHandle := d.opts.Experimental.CPUWorkPermissionGranter.GetPermission(
-		MaxFileWriteAdditionalCPUTime,
-	)
-	writerOpts.Parallelism =
-		d.opts.Experimental.MaxWriterConcurrency > 0 &&
-			(cpuWorkHandle.Permitted() || d.opts.Experimental.ForceWriterParallelism)
+	writerOpts.Parallelism = d.opts.Experimental.ForceWriterParallelism
 
 	tw := sstable.NewRawWriter(writable, writerOpts)
-	return objMeta, tw, cpuWorkHandle, nil
+	return objMeta, tw, nil
 }
 
 // newCompactionOutputObj creates an object produced by a compaction or flush.

compaction_test.go

@@ -541,73 +541,6 @@ func TestPickCompaction(t *testing.T) {
 	}
 }
 
-type cpuPermissionGranter struct {
-	// requestCount is used to confirm that every GetPermission function call
-	// has a corresponding CPUWorkDone function call.
-	requestCount int
-	used         bool
-	permit       bool
-}
-
-type cpuWorkHandle struct {
-	permit bool
-}
-
-func (c cpuWorkHandle) Permitted() bool {
-	return c.permit
-}
-
-func (t *cpuPermissionGranter) GetPermission(dur time.Duration) CPUWorkHandle {
-	t.requestCount++
-	t.used = true
-	return cpuWorkHandle{t.permit}
-}
-
-func (t *cpuPermissionGranter) CPUWorkDone(_ CPUWorkHandle) {
-	t.requestCount--
-}
-
-// Simple test to check if compactions are using the granter, and if exactly
-// the acquired handles are returned.
-func TestCompactionCPUGranter(t *testing.T) {
-	mem := vfs.NewMem()
-	opts := &Options{FS: mem}
-	opts.WithFSDefaults()
-	g := &cpuPermissionGranter{permit: true}
-	opts.Experimental.CPUWorkPermissionGranter = g
-	d, err := Open("", opts)
-	if err != nil {
-		t.Fatalf("Open: %v", err)
-	}
-	defer d.Close()
-
-	d.Set([]byte{'a'}, []byte{'a'}, nil)
-	err = d.Compact([]byte{'a'}, []byte{'b'}, true)
-	if err != nil {
-		t.Fatalf("Compact: %v", err)
-	}
-	require.True(t, g.used)
-	require.Equal(t, g.requestCount, 0)
-}
-
-// Tests that there's no errors or panics when the default CPU granter is used.
-func TestCompactionCPUGranterDefault(t *testing.T) {
-	mem := vfs.NewMem()
-	opts := &Options{FS: mem}
-	opts.WithFSDefaults()
-	d, err := Open("", opts)
-	if err != nil {
-		t.Fatalf("Open: %v", err)
-	}
-	defer d.Close()
-
-	d.Set([]byte{'a'}, []byte{'a'}, nil)
-	err = d.Compact([]byte{'a'}, []byte{'b'}, true)
-	if err != nil {
-		t.Fatalf("Compact: %v", err)
-	}
-}
-
 func TestCompaction(t *testing.T) {
 	const memTableSize = 10000
 	// Tuned so that 2 values can reside in the memtable before a flush, but a

db.go

@@ -220,40 +220,6 @@ type Writer interface {
 	RangeKeyDelete(start, end []byte, opts *WriteOptions) error
 }
 
-// CPUWorkHandle represents a handle used by the CPUWorkPermissionGranter API.
-type CPUWorkHandle interface {
-	// Permitted indicates whether Pebble can use additional CPU resources.
-	Permitted() bool
-}
-
-// CPUWorkPermissionGranter is used to request permission to opportunistically
-// use additional CPUs to speed up internal background work.
-type CPUWorkPermissionGranter interface {
-	// GetPermission returns a handle regardless of whether permission is granted
-	// or not. In the latter case, the handle is only useful for recording
-	// the CPU time actually spent on this calling goroutine.
-	GetPermission(time.Duration) CPUWorkHandle
-	// CPUWorkDone must be called regardless of whether CPUWorkHandle.Permitted
-	// returns true or false.
-	CPUWorkDone(CPUWorkHandle)
-}
-
-// Use a default implementation for the CPU work granter to avoid excessive nil
-// checks in the code.
-type defaultCPUWorkHandle struct{}
-
-func (d defaultCPUWorkHandle) Permitted() bool {
-	return false
-}
-
-type defaultCPUWorkGranter struct{}
-
-func (d defaultCPUWorkGranter) GetPermission(_ time.Duration) CPUWorkHandle {
-	return defaultCPUWorkHandle{}
-}
-
-func (d defaultCPUWorkGranter) CPUWorkDone(_ CPUWorkHandle) {}
-
 // DB provides a concurrent, persistent ordered key/value store.
 //
 // A DB's basic operations (Get, Set, Delete) should be self-explanatory. Get

metamorphic/options.go

@@ -738,10 +738,7 @@ func RandomOptions(
 		}
 	}
 	if rng.IntN(4) == 0 {
-		// Enable Writer parallelism for 25% of the random options. Setting
-		// MaxWriterConcurrency to any value greater than or equal to 1 has the
-		// same effect currently.
-		opts.Experimental.MaxWriterConcurrency = 2
+		// Enable Writer parallelism for 25% of the random options.
 		opts.Experimental.ForceWriterParallelism = true
 	}
 	if rng.IntN(2) == 0 {

options.go

@@ -660,24 +660,12 @@ type Options struct {
 		// compaction will never get triggered.
 		MultiLevelCompactionHeuristic MultiLevelHeuristic
 
-		// MaxWriterConcurrency is used to indicate the maximum number of
-		// compression workers the compression queue is allowed to use. If
-		// MaxWriterConcurrency > 0, then the Writer will use parallelism, to
-		// compress and write blocks to disk. Otherwise, the writer will
-		// compress and write blocks to disk synchronously.
-		MaxWriterConcurrency int
-
 		// ForceWriterParallelism is used to force parallelism in the sstable
 		// Writer for the metamorphic tests. Even with the MaxWriterConcurrency
 		// option set, we only enable parallelism in the sstable Writer if there
 		// is enough CPU available, and this option bypasses that.
 		ForceWriterParallelism bool
 
-		// CPUWorkPermissionGranter should be set if Pebble should be given the
-		// ability to optionally schedule additional CPU. See the documentation
-		// for CPUWorkPermissionGranter for more details.
-		CPUWorkPermissionGranter CPUWorkPermissionGranter
-
 		// EnableColumnarBlocks is used to decide whether to enable writing
 		// TableFormatPebblev5 sstables. This setting is only respected by
 		// FormatColumnarBlocks. In lower format major versions, the
@@ -1287,9 +1275,6 @@ func (o *Options) EnsureDefaults() {
 	if o.Experimental.FileCacheShards <= 0 {
 		o.Experimental.FileCacheShards = runtime.GOMAXPROCS(0)
 	}
-	if o.Experimental.CPUWorkPermissionGranter == nil {
-		o.Experimental.CPUWorkPermissionGranter = defaultCPUWorkGranter{}
-	}
 	if o.Experimental.MultiLevelCompactionHeuristic == nil {
 		o.Experimental.MultiLevelCompactionHeuristic = WriteAmpHeuristic{}
 	}
@@ -1433,7 +1418,6 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  validate_on_ingest=%t\n", o.Experimental.ValidateOnIngest)
 	fmt.Fprintf(&buf, "  wal_dir=%s\n", o.WALDir)
 	fmt.Fprintf(&buf, "  wal_bytes_per_sync=%d\n", o.WALBytesPerSync)
-	fmt.Fprintf(&buf, "  max_writer_concurrency=%d\n", o.Experimental.MaxWriterConcurrency)
 	fmt.Fprintf(&buf, "  force_writer_parallelism=%t\n", o.Experimental.ForceWriterParallelism)
 	fmt.Fprintf(&buf, "  secondary_cache_size_bytes=%d\n", o.Experimental.SecondaryCacheSizeBytes)
 	fmt.Fprintf(&buf, "  create_on_shared=%d\n", o.Experimental.CreateOnShared)
@@ -1833,7 +1817,7 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 			case "wal_bytes_per_sync":
 				o.WALBytesPerSync, err = strconv.Atoi(value)
 			case "max_writer_concurrency":
-				o.Experimental.MaxWriterConcurrency, err = strconv.Atoi(value)
+				// No longer implemented; ignore.
 			case "force_writer_parallelism":
 				o.Experimental.ForceWriterParallelism, err = strconv.ParseBool(value)
 			case "secondary_cache_size_bytes":

options_test.go

@@ -115,7 +115,6 @@ func TestDefaultOptionsString(t *testing.T) {
   validate_on_ingest=false
   wal_dir=
   wal_bytes_per_sync=0
-  max_writer_concurrency=0
   force_writer_parallelism=false
   secondary_cache_size_bytes=0
   create_on_shared=0
@@ -296,7 +295,6 @@ func TestOptionsParse(t *testing.T) {
 			opts.Experimental.DeletionSizeRatioThreshold = 0.7
 			opts.Experimental.TombstoneDenseCompactionThreshold = 0.2
 			opts.Experimental.FileCacheShards = 500
-			opts.Experimental.MaxWriterConcurrency = 1
 			opts.Experimental.ForceWriterParallelism = true
 			opts.Experimental.SecondaryCacheSizeBytes = 1024
 			opts.EnsureDefaults()

replay/testdata/replay

@@ -11,7 +11,7 @@ tree
      614      000007.sst
        0      LOCK
      133      MANIFEST-000001
-    1418      OPTIONS-000003
+    1391      OPTIONS-000003
        0      marker.format-version.000001.013
        0      marker.manifest.000001.MANIFEST-000001
             simple/
@@ -21,7 +21,7 @@ tree
       25        000004.log
      586        000005.sst
       85        MANIFEST-000001
-    1418        OPTIONS-000003
+    1391        OPTIONS-000003
        0        marker.format-version.000001.013
        0        marker.manifest.000001.MANIFEST-000001
 
@@ -67,7 +67,6 @@ cat build/OPTIONS-000003
   validate_on_ingest=false
   wal_dir=
   wal_bytes_per_sync=0
-  max_writer_concurrency=0
   force_writer_parallelism=false
   secondary_cache_size_bytes=0
   create_on_shared=0

replay/testdata/replay_paced

@@ -14,7 +14,7 @@ tree
        0      LOCK
      133      MANIFEST-000001
      205      MANIFEST-000010
-    1418      OPTIONS-000003
+    1391      OPTIONS-000003
        0      marker.format-version.000001.013
        0      marker.manifest.000002.MANIFEST-000010
             high_read_amp/
@@ -26,7 +26,7 @@ tree
       39        000008.log
      560        000009.sst
      157        MANIFEST-000010
-    1418        OPTIONS-000003
+    1391        OPTIONS-000003
        0        marker.format-version.000001.013
        0        marker.manifest.000001.MANIFEST-000010
 

testdata/metrics

@@ -238,7 +238,7 @@ Iter category stats:
 
 disk-usage
 ----
-2.8KB
+2.7KB
 
 # Closing iter b will release the last zombie sstable and the last zombie memtable.