Repository: petermattis/pebble (go)

Issue:
db: rework DB.deleteObsoleteFiles
`DB.deleteObsoleteFiles` is called after each successful flush and compaction. The current implementations performs a directory listing and then looks for files which are not in use by any `version`. The directory listing is fairly expensive, especially as the number of files in the database increases into the thousands and tens of thousands. RocksDB has a different implementation where the normal case avoids a directory listing. Each `fileMetadata` is given a reference count and that reference count is implemented when a file is added to a `version` and decremented when the `version` is unreferenced. This is still an O(n) operation, but it is all in the process and should be much faster than a directory listing.

I think we can do even better. A `version` is constructed by taking an existing `version` and applying a `versionEdit`. The `versionEdit` contains a `deletedFiles` set. Currently, the `versionEdit` is discarded once the new `version` is created, but we could keep it around. A `version` would thus contain a `deletedFiles` set from the previous version. When a `version` is unreferenced, we can merge its `deletedFiles` with the next `version`. When the head of the version list is unreferenced, we add any `deletedFiles` to a `versionSet.obsoleteFiles` set. 

Log file deletion also needs attention. When a flush completes, we can add the log files being referenced by the memtables to a `versionSet.obsoleteLogs` set.

Pull request description:
Rework DB.deleteObsoleteFiles
Previously, `deleteObsoleteFiles` would perform a directory listing to
find obsolete files. The directory listing is fairly expensive,
especially as the number of tables gets large. Switch to an approach
that is similar to RocksDB, where in steady state we keep track of
obsolete logs and tables as memtables are flushed and versions are
unreferenced.

Fix a bug in handling of `versionSet.logNumber`. Previously that field
was being bumped to the most recent log file number any time a flush was
performed, but we can only bump it to the most recent log file that
hasn't been flushed.

Fixes #96

Patch:
diff --git a/batch.go b/batch.go
index 338b752..29b7bce 100644
--- a/batch.go
+++ b/batch.go
@@ -846,6 +846,8 @@ type flushableBatch struct {
 	tombstones []rangedel.Tombstone
 
 	flushedCh chan struct{}
+
+	logNum uint64
 }
 
 var _ flushable = (*flushableBatch)(nil)
@@ -972,6 +974,10 @@ func (b *flushableBatch) readyForFlush() bool {
 	return true
 }
 
+func (b *flushableBatch) logNumber() uint64 {
+	return b.logNum
+}
+
 // Note: flushableBatchIter mirrors the implementation of batchIter. Keep the
 // two in sync.
 type flushableBatchIter struct {
diff --git a/compaction.go b/compaction.go
index 4a9ad18..de8a57c 100644
--- a/compaction.go
+++ b/compaction.go
@@ -9,7 +9,6 @@ import (
 	"errors"
 	"fmt"
 	"os"
-	"path/filepath"
 	"sort"
 	"unsafe"
 
@@ -500,7 +499,7 @@ func (d *DB) flush1() error {
 	// The flush succeeded or it produced an empty sstable. In either case we
 	// want to bump the log number.
 	ve := &versionEdit{
-		logNumber: d.mu.log.number,
+		logNumber: d.mu.mem.queue[n].logNumber(),
 	}
 	if err != errEmptyTable {
 		ve.newFiles = []newFileEntry{
@@ -964,25 +963,18 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 	return ve, pendingOutputs, nil
 }
 
-// deleteObsoleteFiles deletes those files that are no longer needed.
+// scanObsoleteFiles scans the filesystem for files that are no longer needed
+// and adds those to the internal lists of obsolete files. Note that he files
+// are not actually deleted by this method. A subsequent call to
+// deleteObsoleteFiles must be performed.
 //
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
-func (d *DB) deleteObsoleteFiles(jobID int) {
-	// Only allow a single delete obsolete files job to run at a time.
-	for d.mu.cleaner.cleaning {
-		d.mu.cleaner.cond.Wait()
-	}
-	d.mu.cleaner.cleaning = true
-
+func (d *DB) scanObsoleteFiles() {
 	// Release d.mu while doing I/O
 	// Note the unusual order: Unlock and then Lock.
 	d.mu.Unlock()
-	defer func() {
-		d.mu.Lock()
-		d.mu.cleaner.cleaning = false
-		d.mu.cleaner.cond.Signal()
-	}()
+	defer d.mu.Lock()
 
 	fs := d.opts.VFS
 	list, err := fs.List(d.dirname)
@@ -990,9 +982,6 @@ func (d *DB) deleteObsoleteFiles(jobID int) {
 		// Ignore any filesystem errors.
 		return
 	}
-	// We sort to make the order of deletions deterministic, which is nice for
-	// tests.
-	sort.Strings(list)
 
 	// Grab d.mu again in order to get a snapshot of the live state. Note that we
 	// need to this after the directory list because after releasing the lock
@@ -1007,60 +996,161 @@ func (d *DB) deleteObsoleteFiles(jobID int) {
 	manifestFileNumber := d.mu.versions.manifestFileNumber
 	d.mu.Unlock()
 
+	var obsoleteLogs []uint64
+	var obsoleteTables []uint64
+	var obsoleteManifests []uint64
+	var obsoleteOptions []uint64
+
 	for _, filename := range list {
 		fileType, fileNum, ok := parseDBFilename(filename)
 		if !ok {
 			continue
 		}
-		keep := true
 		switch fileType {
 		case fileTypeLog:
 			// TODO(peter): also look at prevLogNumber?
-			keep = fileNum >= logNumber
-			if !keep {
-				keep = d.logRecycler.add(fileNum)
+			if fileNum >= logNumber {
+				continue
 			}
+			obsoleteLogs = append(obsoleteLogs, fileNum)
 		case fileTypeManifest:
-			keep = fileNum >= manifestFileNumber
+			if fileNum >= manifestFileNumber {
+				continue
+			}
+			obsoleteManifests = append(obsoleteManifests, fileNum)
 		case fileTypeOptions:
-			keep = fileNum >= d.optionsFileNum
+			if fileNum >= d.optionsFileNum {
+				continue
+			}
+			obsoleteOptions = append(obsoleteOptions, fileNum)
 		case fileTypeTable:
-			_, keep = liveFileNums[fileNum]
+			if _, ok := liveFileNums[fileNum]; ok {
+				continue
+			}
+			obsoleteTables = append(obsoleteTables, fileNum)
 		default:
 			// Don't delete files we don't know about.
 			continue
 		}
-		if keep {
-			continue
-		}
-		if fileType == fileTypeTable {
-			d.tableCache.evict(fileNum)
-		}
-		path := filepath.Join(d.dirname, filename)
-		err := fs.Remove(path)
-		if err == os.ErrNotExist {
-			continue
+	}
+
+	d.mu.Lock()
+	d.mu.log.queue = merge(d.mu.log.queue, obsoleteLogs)
+	d.mu.versions.obsoleteTables = merge(d.mu.versions.obsoleteTables, obsoleteTables)
+	d.mu.versions.obsoleteManifests = merge(d.mu.versions.obsoleteManifests, obsoleteManifests)
+	d.mu.versions.obsoleteOptions = merge(d.mu.versions.obsoleteOptions, obsoleteOptions)
+	d.mu.Unlock()
+}
+
+// deleteObsoleteFiles deletes those files that are no longer needed.
+//
+// d.mu must be held when calling this, but the mutex may be dropped and
+// re-acquired during the course of this method.
+func (d *DB) deleteObsoleteFiles(jobID int) {
+	// Only allow a single delete obsolete files job to run at a time.
+	for d.mu.cleaner.cleaning {
+		d.mu.cleaner.cond.Wait()
+	}
+	d.mu.cleaner.cleaning = true
+	defer func() {
+		d.mu.cleaner.cleaning = false
+		d.mu.cleaner.cond.Signal()
+	}()
+
+	var obsoleteLogs []uint64
+	for i := range d.mu.log.queue {
+		// NB: d.mu.versions.logNumber is the file number of the latest log that
+		// has had its contents persisted to the LSM.
+		if d.mu.log.queue[i] >= d.mu.versions.logNumber {
+			obsoleteLogs = d.mu.log.queue[:i]
+			d.mu.log.queue = d.mu.log.queue[i:]
+			break
 		}
+	}
+	obsoleteTables := d.mu.versions.obsoleteTables
+	d.mu.versions.obsoleteTables = nil
 
-		switch fileType {
-		case fileTypeLog:
-			if d.opts.EventListener != nil && d.opts.EventListener.WALDeleted != nil {
-				d.opts.EventListener.WALDeleted(db.WALDeleteInfo{
-					JobID:   jobID,
-					Path:    path,
-					FileNum: fileNum,
-					Err:     err,
-				})
+	obsoleteManifests := d.mu.versions.obsoleteManifests
+	d.mu.versions.obsoleteManifests = nil
+
+	obsoleteOptions := d.mu.versions.obsoleteOptions
+	d.mu.versions.obsoleteOptions = nil
+
+	// Release d.mu while doing I/O
+	// Note the unusual order: Unlock and then Lock.
+	d.mu.Unlock()
+	defer d.mu.Lock()
+
+	files := [4]struct {
+		fileType fileType
+		obsolete []uint64
+	}{
+		{fileTypeLog, obsoleteLogs},
+		{fileTypeTable, obsoleteTables},
+		{fileTypeManifest, obsoleteManifests},
+		{fileTypeOptions, obsoleteOptions},
+	}
+	for _, f := range files {
+		// We sort to make the order of deletions deterministic, which is nice for
+		// tests.
+		sort.Slice(f.obsolete, func(i, j int) bool {
+			return f.obsolete[i] < f.obsolete[j]
+		})
+		for _, fileNum := range f.obsolete {
+			switch f.fileType {
+			case fileTypeLog:
+				if d.logRecycler.add(fileNum) {
+					continue
+				}
+			case fileTypeTable:
+				d.tableCache.evict(fileNum)
 			}
-		case fileTypeTable:
-			if d.opts.EventListener != nil && d.opts.EventListener.TableDeleted != nil {
-				d.opts.EventListener.TableDeleted(db.TableDeleteInfo{
-					JobID:   jobID,
-					Path:    path,
-					FileNum: fileNum,
-					Err:     err,
-				})
+
+			path := dbFilename(d.dirname, f.fileType, fileNum)
+			err := d.opts.VFS.Remove(path)
+
+			if err != os.ErrNotExist && d.opts.EventListener != nil {
+				switch f.fileType {
+				case fileTypeLog:
+					if d.opts.EventListener.WALDeleted != nil {
+						d.opts.EventListener.WALDeleted(db.WALDeleteInfo{
+							JobID:   jobID,
+							Path:    path,
+							FileNum: fileNum,
+							Err:     err,
+						})
+					}
+				case fileTypeTable:
+					if d.opts.EventListener.TableDeleted != nil {
+						d.opts.EventListener.TableDeleted(db.TableDeleteInfo{
+							JobID:   jobID,
+							Path:    path,
+							FileNum: fileNum,
+							Err:     err,
+						})
+					}
+				}
 			}
 		}
 	}
 }
+
+func merge(a, b []uint64) []uint64 {
+	if len(b) == 0 {
+		return a
+	}
+
+	a = append(a, b...)
+	sort.Slice(a, func(i, j int) bool {
+		return a[i] < a[j]
+	})
+
+	n := 0
+	for i := 0; i < len(a); i++ {
+		if n == 0 || a[i] != a[n-1] {
+			a[n] = a[i]
+			n++
+		}
+	}
+	return a[:n]
+}
diff --git a/db.go b/db.go
index e8d0b43..8876d47 100644
--- a/db.go
+++ b/db.go
@@ -33,6 +33,7 @@ type flushable interface {
 	newRangeDelIter(o *db.IterOptions) internalIterator
 	flushed() chan struct{}
 	readyForFlush() bool
+	logNumber() uint64
 }
 
 // Reader is a readable key/value store.
@@ -164,7 +165,7 @@ type DB struct {
 		versions versionSet
 
 		log struct {
-			number uint64
+			queue []uint64
 			*record.LogWriter
 		}
 
@@ -837,18 +838,18 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			panic(err)
 		}
 
-		// NB: When the immutable memtable is flushed to disk it will apply a
-		// versionEdit to the manifest telling it that log files < d.mu.log.number
-		// have been applied.
 		if !d.opts.DisableWAL {
-			d.mu.log.number = newLogNumber
+			d.mu.log.queue = append(d.mu.log.queue, newLogNumber)
 			d.mu.log.LogWriter = record.NewLogWriter(newLogFile, newLogNumber)
 		}
 
+		prevLogNumber := d.mu.mem.mutable.logNum
+
 		var scheduleFlush bool
 		if b != nil && b.flushable != nil {
 			// The batch is too large to fit in the memtable so add it directly to
 			// the immutable queue.
+			b.flushable.logNum = prevLogNumber
 			d.mu.mem.queue = append(d.mu.mem.queue, b.flushable)
 			scheduleFlush = true
 		}
@@ -862,6 +863,10 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		// want to go through the flush path in order to recycle that WAL file.
 		imm := d.mu.mem.mutable
 		d.mu.mem.mutable = newMemTable(d.opts)
+		// NB: When the immutable memtable is flushed to disk it will apply a
+		// versionEdit to the manifest telling it that log files <= newLogNumber
+		// have been applied.
+		d.mu.mem.mutable.logNum = newLogNumber
 		d.mu.mem.queue = append(d.mu.mem.queue, d.mu.mem.mutable)
 		d.updateReadStateLocked()
 		if (imm != nil && imm.unref()) || scheduleFlush {
diff --git a/mem_table.go b/mem_table.go
index f8fc73c..c381370 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -53,6 +53,7 @@ type memTable struct {
 	refs        int32
 	flushedCh   chan struct{}
 	tombstones  rangeTombstoneCache
+	logNum      uint64
 }
 
 // newMemTable returns a new MemTable.
@@ -94,6 +95,10 @@ func (m *memTable) readyForFlush() bool {
 	return atomic.LoadInt32(&m.refs) == 0
 }
 
+func (m *memTable) logNumber() uint64 {
+	return m.logNum
+}
+
 // Get gets the value for the given key. It returns ErrNotFound if the DB does
 // not contain the key.
 func (m *memTable) get(key []byte) (value []byte, err error) {
diff --git a/open.go b/open.go
index 72f12c0..add7583 100644
--- a/open.go
+++ b/open.go
@@ -171,7 +171,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 
 	// Create an empty .log file.
 	ve.logNumber = d.mu.versions.nextFileNum()
-	d.mu.log.number = ve.logNumber
+	d.mu.log.queue = append(d.mu.log.queue, ve.logNumber)
 	logFile, err := opts.VFS.Create(dbFilename(dirname, fileTypeLog, ve.logNumber))
 	if err != nil {
 		return nil, err
@@ -180,7 +180,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		BytesPerSync:    d.opts.BytesPerSync,
 		PreallocateSize: d.walPreallocateSize(),
 	})
-	d.mu.log.LogWriter = record.NewLogWriter(logFile, d.mu.log.number)
+	d.mu.log.LogWriter = record.NewLogWriter(logFile, ve.logNumber)
 
 	// Write a new manifest to disk.
 	if err := d.mu.versions.logAndApply(&ve); err != nil {
@@ -201,6 +201,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 
 	jobID := d.mu.nextJobID
 	d.mu.nextJobID++
+	d.scanObsoleteFiles()
 	d.deleteObsoleteFiles(jobID)
 	d.maybeScheduleFlush()
 	d.maybeScheduleCompaction()
diff --git a/version.go b/version.go
index 6c5e033..c9ea5ca 100644
--- a/version.go
+++ b/version.go
@@ -16,6 +16,12 @@ import (
 
 // fileMetadata holds the metadata for an on-disk table.
 type fileMetadata struct {
+	// reference count for the file: incremented when a file is added to a
+	// version and decremented when the version is unreferenced. The file is
+	// obsolete when the reference count falls to zero. This is a pointer because
+	// fileMetadata is copied by value from version to version, but we want the
+	// reference count to be shared.
+	refs *int32
 	// fileNum is the file number.
 	fileNum uint64
 	// size is the size of the file, in bytes.
@@ -134,6 +140,9 @@ type version struct {
 
 	files [numLevels][]fileMetadata
 
+	// The version set this version is associated with.
+	vs *versionSet
+
 	// The list the version is linked into.
 	list *versionList
 
@@ -179,9 +188,11 @@ func (v *version) ref() {
 
 func (v *version) unref() {
 	if atomic.AddInt32(&v.refs, -1) == 0 {
+		obsolete := v.unrefFiles()
 		l := v.list
 		l.mu.Lock()
 		l.remove(v)
+		v.vs.addObsoleteLocked(obsolete)
 		l.mu.Unlock()
 	}
 }
@@ -189,7 +200,21 @@ func (v *version) unref() {
 func (v *version) unrefLocked() {
 	if atomic.AddInt32(&v.refs, -1) == 0 {
 		v.list.remove(v)
+		v.vs.addObsoleteLocked(v.unrefFiles())
+	}
+}
+
+func (v *version) unrefFiles() []uint64 {
+	var obsolete []uint64
+	for _, files := range v.files {
+		for i := range files {
+			f := &files[i]
+			if atomic.AddInt32(f.refs, -1) == 0 {
+				obsolete = append(obsolete, f.fileNum)
+			}
+		}
 	}
+	return obsolete
 }
 
 // overlaps returns all elements of v.files[level] whose user key range
diff --git a/version_edit.go b/version_edit.go
index a477d70..60a13bb 100644
--- a/version_edit.go
+++ b/version_edit.go
@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"io"
 	"sort"
+	"sync/atomic"
 
 	"github.com/petermattis/pebble/db"
 )
@@ -406,7 +407,12 @@ func (b *bulkVersionEdit) apply(
 			if base == nil {
 				continue
 			}
-			v.files[level] = base.files[level]
+			files := base.files[level]
+			v.files[level] = files
+			// We still have to bump the ref count for all files.
+			for i := range files {
+				atomic.AddInt32(files[i].refs, 1)
+			}
 			continue
 		}
 
@@ -429,6 +435,10 @@ func (b *bulkVersionEdit) apply(
 				if dmap != nil && dmap[f.fileNum] {
 					continue
 				}
+				if f.refs == nil {
+					f.refs = new(int32)
+				}
+				atomic.AddInt32(f.refs, 1)
 				v.files[level] = append(v.files[level], f)
 			}
 		}
diff --git a/version_set.go b/version_set.go
index 17bc48a..ef246cb 100644
--- a/version_set.go
+++ b/version_set.go
@@ -34,6 +34,10 @@ type versionSet struct {
 	versions versionList
 	picker   *compactionPicker
 
+	obsoleteTables    []uint64
+	obsoleteManifests []uint64
+	obsoleteOptions   []uint64
+
 	logNumber          uint64
 	prevLogNumber      uint64
 	nextFileNumber     uint64
@@ -173,20 +177,25 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 	}
 	ve.nextFileNumber = vs.nextFileNumber
 	ve.lastSequence = atomic.LoadUint64(&vs.logSeqNum)
-
-	var bve bulkVersionEdit
-	bve.accumulate(ve)
-	newVersion, err := bve.apply(vs.opts, vs.currentVersion(), vs.cmp)
-	if err != nil {
-		return err
-	}
+	currentVersion := vs.currentVersion()
+	var newVersion *version
 
 	var picker *compactionPicker
 	if err := func() error {
 		vs.mu.Unlock()
 		defer vs.mu.Lock()
 
-		// TODO(peter): if vs.manifest becomes too large, create a new one.
+		var bve bulkVersionEdit
+		bve.accumulate(ve)
+
+		var err error
+		newVersion, err = bve.apply(vs.opts, currentVersion, vs.cmp)
+		if err != nil {
+			return err
+		}
+
+		// TODO(peter): if vs.manifest becomes too large, create a new one. Be sure
+		// to add the old manifest to obsoleteManifests.
 		if vs.manifest == nil {
 			if err := vs.createManifest(vs.dirname); err != nil {
 				return err
@@ -296,6 +305,7 @@ func (vs *versionSet) append(v *version) {
 	if !vs.versions.empty() {
 		vs.versions.back().unrefLocked()
 	}
+	v.vs = vs
 	v.ref()
 	vs.versions.pushBack(v)
 }
@@ -313,3 +323,7 @@ func (vs *versionSet) addLiveFileNums(m map[uint64]struct{}) {
 		}
 	}
 }
+
+func (vs *versionSet) addObsoleteLocked(obsolete []uint64) {
+	vs.obsoleteTables = append(vs.obsoleteTables, obsolete...)
+}


Test patch:
diff --git a/log_recycler_test.go b/log_recycler_test.go
index 01e746b..fd28bb8 100644
--- a/log_recycler_test.go
+++ b/log_recycler_test.go
@@ -85,7 +85,7 @@ func TestRecycleLogs(t *testing.T) {
 	logNum := func() uint64 {
 		d.mu.Lock()
 		defer d.mu.Unlock()
-		return d.mu.log.number
+		return d.mu.log.queue[len(d.mu.log.queue)-1]
 	}
 
 	// Flush the memtable a few times, forcing rotation of the WAL. We should see
diff --git a/version_test.go b/version_test.go
index 87b685d..0f39cdc 100644
--- a/version_test.go
+++ b/version_test.go
@@ -257,7 +257,7 @@ func TestVersionUnref(t *testing.T) {
 		mu: &sync.Mutex{},
 	}
 	list.init()
-	v := &version{}
+	v := &version{vs: &versionSet{}}
 	v.ref()
 	list.pushBack(v)
 	v.unref()