delete pacer: count in-queue bytes as obsolete bytes

Informs #5424

Author: RaduBerinde

obsolete_files.go

@@ -150,7 +150,7 @@ func (cm *cleanupManager) EnqueueJob(
 		}
 	}
 	if pacingBytes > 0 {
-		cm.deletePacer.ReportDeletion(crtime.NowMono(), pacingBytes)
+		cm.deletePacer.DeletionEnqueued(crtime.NowMono(), pacingBytes)
 	}
 
 	cm.mu.Lock()
@@ -216,6 +216,9 @@ func (cm *cleanupManager) deleteObsoleteFilesInJob(
 				cm.maybePace(tb, &of, paceTimer)
 			}
 			cm.deleteObsoleteObject(of.fileType, job.jobID, of.fileNum)
+			if of.needsPacing() {
+				cm.deletePacer.DeletionPerformed(of.fileSize)
+			}
 		default:
 			cm.deleteObsoleteFile(of.fs, of.fileType, job.jobID, of.path, of.fileNum)
 		}

pacer.go

@@ -9,13 +9,16 @@ import (
 	"time"
 
 	"github.com/cockroachdb/crlib/crtime"
+	"github.com/cockroachdb/pebble/internal/invariants"
 )
 
 // deletionPacerInfo contains any info from the db necessary to make deletion
 // pacing decisions (to limit background IO usage so that it does not contend
 // with foreground traffic).
 type deletionPacerInfo struct {
-	freeBytes     uint64
+	freeBytes uint64
+	// obsoleteBytes is the total size of obsolete files in the latest version;
+	// these are files that have not yet been enqueued for deletion.
 	obsoleteBytes uint64
 	liveBytes     uint64
 }
@@ -44,6 +47,10 @@ type deletionPacer struct {
 		// history keeps rack of recent deletion history; it used to increase the
 		// deletion rate to match the pace of deletions.
 		history history
+
+		// bytesToDelete is the sum of pacing bytes for all deletions that were
+		// reported but not yet performed.
+		bytesToDelete uint64
 	}
 
 	targetByteDeletionRate func() int
@@ -83,17 +90,33 @@ func newDeletionPacer(
 	return d
 }
 
-// ReportDeletion is used to report a deletion to the pacer. The pacer uses it
-// to keep track of the recent rate of deletions and potentially increase the
-// deletion rate accordingly.
+// DeletionEnqueued is used to report a new deletion request to the pacer. The
+// pacer uses it to keep track of the recent rate of deletions and potentially
+// increase the deletion rate accordingly.
 //
-// ReportDeletion is thread-safe.
-func (p *deletionPacer) ReportDeletion(now crtime.Mono, bytesToDelete uint64) {
+// DeletionEnqueued is thread-safe.
+func (p *deletionPacer) DeletionEnqueued(now crtime.Mono, bytesToDelete uint64) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	p.mu.bytesToDelete += bytesToDelete
 	p.mu.history.Add(now, int64(bytesToDelete))
 }
 
+// DeletionPerformed is used to report that a deletion was performed. The pacer
+// uses it to keep track of how many bytes are in the queue.
+func (p *deletionPacer) DeletionPerformed(bytesToDelete uint64) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.mu.bytesToDelete < bytesToDelete {
+		if invariants.Enabled {
+			panic("underflow")
+		}
+		p.mu.bytesToDelete = 0
+	} else {
+		p.mu.bytesToDelete -= bytesToDelete
+	}
+}
+
 // PacingDelay returns the recommended pacing wait time (in seconds) for
 // deleting the given number of bytes.
 //
@@ -108,10 +131,10 @@ func (p *deletionPacer) PacingDelay(now crtime.Mono, bytesToDelete uint64) (wait
 	baseRate := float64(targetByteDeletionRate)
 	// If recent deletion rate is more than our target, use that so that we don't
 	// fall behind.
-	historicRate := func() float64 {
+	historicRate, totalBytesToDelete := func() (float64, uint64) {
 		p.mu.Lock()
 		defer p.mu.Unlock()
-		return float64(p.mu.history.Sum(now)) / deletePacerHistory.Seconds()
+		return float64(p.mu.history.Sum(now)) / deletePacerHistory.Seconds(), p.mu.bytesToDelete
 	}()
 	if historicRate > baseRate {
 		baseRate = historicRate
@@ -128,7 +151,7 @@ func (p *deletionPacer) PacingDelay(now crtime.Mono, bytesToDelete uint64) (wait
 		// We don't know the obsolete bytes ratio. Disable pacing altogether.
 		return 0.0
 	}
-	obsoleteBytesRatio := float64(info.obsoleteBytes) / float64(info.liveBytes)
+	obsoleteBytesRatio := float64(info.obsoleteBytes+totalBytesToDelete) / float64(info.liveBytes)
 	if obsoleteBytesRatio >= p.obsoleteBytesMaxRatio {
 		// Increase the rate so that we can free up enough bytes within the timeframe.
 		r := (obsoleteBytesRatio - p.obsoleteBytesMaxRatio) * float64(info.liveBytes) / p.obsoleteBytesTimeframe.Seconds()

pacer_test.go

@@ -29,7 +29,8 @@ func TestDeletionPacer(t *testing.T) {
 		// history of deletion reporting; first value in the pair is the time,
 		// second value is the deleted bytes. The time of pacing is the same as the
 		// last time in the history.
-		history [][2]int64
+		history      [][2]int64
+		bytesInQueue uint64
 		// expected pacing rate in MB/s.
 		expected float64
 	}{
@@ -73,7 +74,7 @@ func TestDeletionPacer(t *testing.T) {
 		{
 			freeBytes:     160 * GB,
 			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
+			liveBytes:     1000 * GB,
 			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
 			expected:      200.0,
 		},
@@ -82,7 +83,7 @@ func TestDeletionPacer(t *testing.T) {
 		{
 			freeBytes:     6 * GB,
 			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
+			liveBytes:     100 * GB,
 			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
 			expected:      1224.0,
 		},
@@ -95,6 +96,27 @@ func TestDeletionPacer(t *testing.T) {
 			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
 			expected:      302.4,
 		},
+		// History shows 200MB/sec deletions on average over last 5 minutes and
+		// obsoleteBytesRatio is 50%, from bytes still in queue.
+		{
+			freeBytes:     500 * GB,
+			obsoleteBytes: 0 * GB,
+			liveBytes:     100 * GB,
+			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
+			bytesInQueue:  50 * GB,
+			expected:      302.4,
+		},
+		// History shows 200MB/sec deletions on average over last 5 minutes and
+		// obsoleteBytesRatio is 50%, from a combination of obsoleteBytes and bytes
+		// still in queue.
+		{
+			freeBytes:     500 * GB,
+			obsoleteBytes: 25 * GB,
+			liveBytes:     100 * GB,
+			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
+			bytesInQueue:  25 * GB,
+			expected:      302.4,
+		},
 		// History shows 1000MB/sec deletions on average over last 5 minutes.
 		{
 			freeBytes:     160 * GB,
@@ -134,10 +156,13 @@ func TestDeletionPacer(t *testing.T) {
 				opts.ObsoleteBytesTimeframe,
 				getInfo,
 			)
+			var inQueue uint64
 			for _, h := range tc.history {
 				last = start + crtime.Mono(time.Second*time.Duration(h[0]))
-				pacer.ReportDeletion(last, uint64(h[1]))
+				pacer.DeletionEnqueued(last, uint64(h[1]))
+				inQueue += uint64(h[1])
 			}
+			pacer.DeletionPerformed(inQueue - tc.bytesInQueue)
 			result := 1.0 / pacer.PacingDelay(last, 1*MB)
 			require.InDelta(t, tc.expected, result, 1e-7)
 		})