fix cleanupManager.Close() hanging when deletion pacing is active

When `TargetByteDeletionRate` is set to a low value, `DB.Close()` can hang for a
long time because deletion pacing uses `time.Sleep()` that can't be interrupted.

This change modifies `maybePace()` to check for early termination and use
interruptible `select` instead of `time.Sleep()`.

Resolves: #5347

Author: ijsong

obsolete_files.go

@@ -41,6 +41,8 @@ type cleanupManager struct {
 	jobsCh chan *cleanupJob
 	// waitGroup is used to wait for the background goroutine to exit.
 	waitGroup sync.WaitGroup
+	// stopCh is closed when the pacer is disabled after closing the cleanup manager.
+	stopCh chan struct{}
 
 	mu struct {
 		sync.Mutex
@@ -103,6 +105,7 @@ func openCleanupManager(
 			getDeletePacerInfo,
 		),
 		jobsCh: make(chan *cleanupJob, jobsQueueDepth),
+		stopCh: make(chan struct{}),
 	}
 	cm.mu.completedJobsCond.L = &cm.mu.Mutex
 	cm.waitGroup.Add(1)
@@ -120,6 +123,7 @@ func openCleanupManager(
 // Delete pacing is disabled for the remaining jobs.
 func (cm *cleanupManager) Close() {
 	close(cm.jobsCh)
+	close(cm.stopCh)
 	cm.waitGroup.Wait()
 }
 
@@ -174,21 +178,18 @@ func (cm *cleanupManager) Wait() {
 func (cm *cleanupManager) mainLoop() {
 	defer cm.waitGroup.Done()
 
+	paceTimer := time.NewTimer(time.Duration(0))
+	defer paceTimer.Stop()
+
 	var tb tokenbucket.TokenBucket
 	// Use a token bucket with 1 token / second refill rate and 1 token burst.
 	tb.Init(1.0, 1.0)
 	for job := range cm.jobsCh {
-		for _, of := range job.obsoleteFiles {
-			switch of.fileType {
-			case base.FileTypeTable:
-				cm.maybePace(&tb, &of)
-				cm.deleteObsoleteObject(of.fileType, job.jobID, of.fileNum)
-			case base.FileTypeBlob:
-				cm.maybePace(&tb, &of)
-				cm.deleteObsoleteObject(of.fileType, job.jobID, of.fileNum)
-			default:
-				cm.deleteObsoleteFile(of.fs, of.fileType, job.jobID, of.path, of.fileNum)
-			}
+		select {
+		case <-cm.stopCh:
+			cm.deleteObsoleteFilesInJob(job, nil, nil)
+		default:
+			cm.deleteObsoleteFilesInJob(job, &tb, paceTimer)
 		}
 		cm.mu.Lock()
 		cm.mu.completedJobs++
@@ -199,9 +200,31 @@ func (cm *cleanupManager) mainLoop() {
 	}
 }
 
+// deleteObsoleteFilesInJob deletes all obsolete files in the given job. If tb
+// and paceTimer are provided, files that need pacing will be throttled
+// according to the deletion rate. If tb is nil, files are deleted immediately
+// without pacing (used when the cleanup manager is being closed).
+func (cm *cleanupManager) deleteObsoleteFilesInJob(
+	job *cleanupJob, tb *tokenbucket.TokenBucket, paceTimer *time.Timer,
+) {
+	for _, of := range job.obsoleteFiles {
+		switch of.fileType {
+		case base.FileTypeTable, base.FileTypeBlob:
+			if tb != nil {
+				cm.maybePace(tb, &of, paceTimer)
+			}
+			cm.deleteObsoleteObject(of.fileType, job.jobID, of.fileNum)
+		default:
+			cm.deleteObsoleteFile(of.fs, of.fileType, job.jobID, of.path, of.fileNum)
+		}
+	}
+}
+
 // maybePace sleeps before deleting an object if appropriate. It is always
 // called from the background goroutine.
-func (cm *cleanupManager) maybePace(tb *tokenbucket.TokenBucket, of *obsoleteFile) {
+func (cm *cleanupManager) maybePace(
+	tb *tokenbucket.TokenBucket, of *obsoleteFile, paceTimer *time.Timer,
+) {
 	if !of.needsPacing() {
 		return
 	}
@@ -220,7 +243,12 @@ func (cm *cleanupManager) maybePace(tb *tokenbucket.TokenBucket, of *obsoleteFil
 		if ok {
 			break
 		}
-		time.Sleep(d)
+		paceTimer.Reset(d)
+		select {
+		case <-paceTimer.C:
+		case <-cm.stopCh:
+			return
+		}
 	}
 }
 

obsolete_files_test.go

@@ -10,10 +10,12 @@ import (
 	"sort"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/testutils"
+	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -139,3 +141,55 @@ func TestCleaner(t *testing.T) {
 		}
 	})
 }
+
+func TestCleanupManagerCloseWithPacing(t *testing.T) {
+	mem := vfs.NewMem()
+	opts := &Options{
+		FS:                     mem,
+		TargetByteDeletionRate: 1024, // 1 KB/s - slow pacing
+	}
+	opts.EnsureDefaults()
+
+	objProvider, err := objstorageprovider.Open(objstorageprovider.Settings{
+		FS:        mem,
+		FSDirName: "/",
+	})
+	require.NoError(t, err)
+	defer objProvider.Close()
+
+	getDeletePacerInfo := func() deletionPacerInfo {
+		return deletionPacerInfo{
+			freeBytes: 10 << 30,
+		}
+	}
+
+	cm := openCleanupManager(opts, objProvider, getDeletePacerInfo)
+
+	// Create obsolete files that would normally take a long time to delete.
+	// At 1 KB/s, 100 files of 10 KB each would take 1000 seconds.
+	largeFiles := make([]obsoleteFile, 100)
+	for i := range largeFiles {
+		largeFiles[i] = obsoleteFile{
+			fileType: base.FileTypeTable,
+			fs:       mem,
+			path:     fmt.Sprintf("test%02d.sst", i+1),
+			fileNum:  base.DiskFileNum(i + 1),
+			fileSize: 10 << 10,
+			isLocal:  true,
+		}
+	}
+
+	cm.EnqueueJob(1, largeFiles, obsoleteObjectStats{})
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		cm.Close()
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatalf("timed out waiting for cleanupManager.Close() to return")
+	}
+}