db: delete pacer rewrite

This change reimplements the delete pacer with an improved design. The
previous implementation was prone to falling behind on deletions; the
"obsolete ratio" feedback mechanism was broken because we didn't count
the queued deletions as obsolete bytes. The implementation was also
unnecessarily complicated and intertwined with the rest of the DB
code.

The new delete pacer is in a separate package with clear API
boundaries. It internally maintains all relevant metrics.  It is
designed to increase the deletion rate whenever deletions stay in the
queue for more than 5 minutes (it no longer depends on the obsolete
bytes ratio). In addition, instead of proportional control (where the
rate is proportional to the backlog, thus going down as we clear it)
we switch to constant-horizon control where we keep the same rate
until the backlog is cleared.

The implementation uses an unbounded `fifo.Queue` instead of a channel
of limited size (which can cause unwanted blocking). We still have a
safety valve to disable pacing if the queue gets too big.

We add delete pacer metrics so that we can monitor the deletion rate
and queue depth.

Fixes #5424

Author: RaduBerinde

blob_rewrite.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/problemspans"
 	"github.com/cockroachdb/pebble/objstorage"
@@ -249,17 +250,17 @@ func (c *blobFileRewriteCompaction) Execute(jobID JobID, d *DB) error {
 	// Ensure we clean up the blob file we created on failure.
 	if err != nil {
 		if objMeta.DiskFileNum != 0 {
-			d.mu.versions.obsoleteBlobs = mergeObsoleteFiles(d.mu.versions.obsoleteBlobs, []obsoleteFile{
+			d.mu.versions.obsoleteBlobs = mergeObsoleteFiles(d.mu.versions.obsoleteBlobs, []deletepacer.ObsoleteFile{
 				{
-					fileType: base.FileTypeBlob,
-					fs:       d.opts.FS,
-					path:     d.objProvider.Path(objMeta),
-					fileNum:  objMeta.DiskFileNum,
+					FileType: base.FileTypeBlob,
+					FS:       d.opts.FS,
+					Path:     d.objProvider.Path(objMeta),
+					FileNum:  objMeta.DiskFileNum,
 					// We don't know the size of the output blob file--it may have
 					// been half-written. We use the input blob file size as an
 					// approximation for deletion pacing.
-					fileSize: c.input.Physical.Size,
-					isLocal:  true,
+					FileSize: c.input.Physical.Size,
+					IsLocal:  true,
 				},
 			})
 		}

compaction.go

@@ -45,7 +45,6 @@ var errEmptyTable = errors.New("pebble: empty table")
 var ErrCancelledCompaction = errors.New("pebble: compaction cancelled by a concurrent operation, will retry compaction")
 
 var flushLabels = pprof.Labels("pebble", "flush", "output-level", "L0")
-var gcLabels = pprof.Labels("pebble", "gc")
 
 // expandedCompactionByteSizeLimit is the maximum number of bytes in all
 // compacted files. We avoid expanding the lower level file set of a compaction

compaction_test.go

@@ -3099,7 +3099,7 @@ func TestCompaction_UpdateVersionFails(t *testing.T) {
 }
 
 // TestSharedObjectDeletePacing tests that we don't throttle shared object
-// deletes (see the TargetBytesDeletionRate option).
+// deletes (see the DeletePacer.BaselineRate option).
 func TestSharedObjectDeletePacing(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -3109,7 +3109,7 @@ func TestSharedObjectDeletePacing(t *testing.T) {
 		"": remote.NewInMem(),
 	})
 	opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
-	opts.TargetByteDeletionRate = func() int { return 1 }
+	opts.DeletionPacing.BaselineRate = func() uint64 { return 1 }
 	opts.Logger = testutils.Logger{T: t}
 
 	d, err := Open("", &opts)

db.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/arenaskl"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invalidating"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -310,7 +311,7 @@ type DB struct {
 	closed   *atomic.Value
 	closedCh chan struct{}
 
-	cleanupManager *cleanupManager
+	deletePacer *deletepacer.DeletePacer
 
 	compactionScheduler CompactionScheduler
 
@@ -485,10 +486,6 @@ type DB struct {
 			// count acts as a reference count to prohibit file cleaning. See
 			// DB.{disable,enable}FileDeletions().
 			disableCount int
-			// queuedStats holds cumulative stats for files that have been
-			// queued for deletion by the cleanup manager. These stats are
-			// monotonically increasing for the *DB's lifetime.
-			queuedStats obsoleteObjectStats
 		}
 
 		snapshots struct {
@@ -561,7 +558,7 @@ var _ Writer = (*DB)(nil)
 
 // TestOnlyWaitForCleaning MUST only be used in tests.
 func (d *DB) TestOnlyWaitForCleaning() {
-	d.cleanupManager.Wait()
+	d.deletePacer.WaitForTesting()
 }
 
 // Set sets the value for the given key. It overwrites any previous value
@@ -1594,7 +1591,7 @@ func (d *DB) Close() error {
 	d.mu.Unlock()
 
 	// Wait for all cleaning jobs to finish.
-	d.cleanupManager.Close()
+	d.deletePacer.Close()
 
 	d.mu.Lock()
 	// Sanity check compaction metrics.
@@ -1883,7 +1880,7 @@ func (d *DB) AsyncFlush() (<-chan struct{}, error) {
 func (d *DB) Metrics() *Metrics {
 	metrics := &Metrics{}
 	walStats := d.mu.log.manager.Stats()
-	completedObsoleteFileStats := d.cleanupManager.CompletedStats()
+	deletePacerMetrics := d.deletePacer.Metrics()
 
 	d.mu.Lock()
 	vers := d.mu.versions.currentVersion()
@@ -1947,29 +1944,19 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Table.ZombieSize = d.mu.versions.zombieTables.TotalSize()
 	metrics.Table.Local.ZombieCount, metrics.Table.Local.ZombieSize = d.mu.versions.zombieTables.LocalStats()
 
-	// The obsolete blob/table metrics have a subtle calculation:
-	//
-	// (A) The vs.metrics.{Table,BlobFiles}.[Local.]{ObsoleteCount,ObsoleteSize}
-	// fields reflect the set of files currently sitting in
-	// vs.obsolete{Tables,Blobs} but not yet enqueued to the cleanup manager.
-	//
-	// (B) The d.mu.fileDeletions.queuedStats field holds the set of files that have
-	// been queued for deletion by the cleanup manager.
-	//
-	// (C) The cleanup manager also maintains cumulative stats for the set of
-	// files that have been deleted.
-	//
-	// The value of currently pending obsolete files is (A) + (B) - (C).
-	pendingObsoleteFileStats := d.mu.fileDeletions.queuedStats
-	pendingObsoleteFileStats.Sub(completedObsoleteFileStats)
-	metrics.Table.Local.ObsoleteCount += pendingObsoleteFileStats.tablesLocal.count
-	metrics.Table.Local.ObsoleteSize += pendingObsoleteFileStats.tablesLocal.size
-	metrics.Table.ObsoleteCount += int64(pendingObsoleteFileStats.tablesAll.count)
-	metrics.Table.ObsoleteSize += pendingObsoleteFileStats.tablesAll.size
-	metrics.BlobFiles.Local.ObsoleteCount += pendingObsoleteFileStats.blobFilesLocal.count
-	metrics.BlobFiles.Local.ObsoleteSize += pendingObsoleteFileStats.blobFilesLocal.size
-	metrics.BlobFiles.ObsoleteCount += pendingObsoleteFileStats.blobFilesAll.count
-	metrics.BlobFiles.ObsoleteSize += pendingObsoleteFileStats.blobFilesAll.size
+	// The obsolete blob/table metrics have a subtle calculation: the initial
+	// metrics we copied from vs.metrics reflect (in
+	// {Table,BlobFiles}.[Local.]{ObsoleteCount,ObsoleteSize}) the set of files
+	// currently sitting in vs.obsolete{Tables,Blobs} but not yet enqueued to the
+	// delete pacer. To complete the metrics, we add the currently enqueued files.
+	metrics.Table.Local.ObsoleteCount += deletePacerMetrics.InQueue.Tables.Local.Count
+	metrics.Table.Local.ObsoleteSize += deletePacerMetrics.InQueue.Tables.Local.Bytes
+	metrics.Table.ObsoleteCount += int64(deletePacerMetrics.InQueue.Tables.All.Count)
+	metrics.Table.ObsoleteSize += deletePacerMetrics.InQueue.Tables.All.Bytes
+	metrics.BlobFiles.Local.ObsoleteCount += deletePacerMetrics.InQueue.BlobFiles.Local.Count
+	metrics.BlobFiles.Local.ObsoleteSize += deletePacerMetrics.InQueue.BlobFiles.Local.Bytes
+	metrics.BlobFiles.ObsoleteCount += deletePacerMetrics.InQueue.BlobFiles.All.Count
+	metrics.BlobFiles.ObsoleteSize += deletePacerMetrics.InQueue.BlobFiles.All.Bytes
 	metrics.private.optionsFileSize = d.optionsFileSize
 
 	d.mu.versions.logLock()
@@ -2022,6 +2009,8 @@ func (d *DB) Metrics() *Metrics {
 	metrics.TableIters = d.fileCache.IterCount()
 	metrics.CategoryStats = d.fileCache.SSTStatsCollector().GetStats()
 
+	metrics.DeletePacer = deletePacerMetrics
+
 	metrics.SecondaryCacheMetrics = d.objProvider.Metrics()
 
 	metrics.Uptime = d.opts.private.timeNow().Sub(d.openedAt)