leveldb: implement deleteObsoleteFiles.

Also implement tableCache eviction, parseDBFilename, and delete the
logFileNum function.

R=bradfitz
CC=golang-dev
https://codereview.appspot.com/19710043

Author: nigeltao

leveldb/compaction.go

@@ -214,7 +214,8 @@ func (d *DB) compact1() error {
 	if err := d.versions.logAndApply(d.dirname, ve); err != nil {
 		return err
 	}
-	return d.deleteObsoleteFiles()
+	d.deleteObsoleteFiles()
+	return nil
 }
 
 // compactMemTable runs a compaction that copies d.imm from memory to disk.
@@ -236,7 +237,8 @@ func (d *DB) compactMemTable() error {
 		return err
 	}
 	d.imm = nil
-	return d.deleteObsoleteFiles()
+	d.deleteObsoleteFiles()
+	return nil
 }
 
 // compactDiskTables runs a compaction that produces new on-disk tables from

leveldb/filenames.go

@@ -7,21 +7,24 @@ package leveldb
 import (
 	"fmt"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 
 	"code.google.com/p/leveldb-go/leveldb/db"
 )
 
+type fileType int
+
 const (
-	fileTypeLog = iota
+	fileTypeLog fileType = iota
 	fileTypeLock
 	fileTypeTable
 	fileTypeManifest
 	fileTypeCurrent
 )
 
-func dbFilename(dirname string, fileType int, fileNum uint64) string {
+func dbFilename(dirname string, fileType fileType, fileNum uint64) string {
 	for len(dirname) > 0 && dirname[len(dirname)-1] == os.PathSeparator {
 		dirname = dirname[:len(dirname)-1]
 	}
@@ -40,18 +43,36 @@ func dbFilename(dirname string, fileType int, fileNum uint64) string {
 	panic("unreachable")
 }
 
-// logFileNum returns the fileNum of the given log file, or 0 if that file is
-// not a log file.
-func logFileNum(filename string) uint64 {
-	if !strings.HasSuffix(filename, ".log") {
-		return 0
-	}
-	filename = filename[:len(filename)-4]
-	u, err := strconv.ParseUint(filename, 10, 64)
-	if err != nil {
-		return 0
+func parseDBFilename(filename string) (fileType fileType, fileNum uint64, ok bool) {
+	filename = filepath.Base(filename)
+	switch {
+	case filename == "CURRENT":
+		return fileTypeCurrent, 0, true
+	case filename == "LOCK":
+		return fileTypeLock, 0, true
+	case strings.HasPrefix(filename, "MANIFEST-"):
+		u, err := strconv.ParseUint(filename[len("MANIFEST-"):], 10, 64)
+		if err != nil {
+			break
+		}
+		return fileTypeManifest, u, true
+	default:
+		i := strings.IndexByte(filename, '.')
+		if i < 0 {
+			break
+		}
+		u, err := strconv.ParseUint(filename[:i], 10, 64)
+		if err != nil {
+			break
+		}
+		switch filename[i+1:] {
+		case "log":
+			return fileTypeLog, u, true
+		case "sst":
+			return fileTypeTable, u, true
+		}
 	}
-	return u
+	return 0, 0, false
 }
 
 func setCurrentFile(dirname string, fs db.FileSystem, fileNum uint64) error {

leveldb/filenames_test.go

@@ -0,0 +1,67 @@
+// Copyright 2013 The LevelDB-Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package leveldb
+
+import (
+	"path/filepath"
+	"testing"
+)
+
+func TestParseDBFilename(t *testing.T) {
+	testCases := map[string]bool{
+		"000000.log":          true,
+		"000000.log.zip":      false,
+		"000000..log":         false,
+		"a000000.log":         false,
+		"abcdef.log":          false,
+		"000001.sst":          true,
+		"CURRENT":             true,
+		"CURRaNT":             false,
+		"LOCK":                true,
+		"xLOCK":               false,
+		"x.LOCK":              false,
+		"MANIFEST":            false,
+		"MANIFEST123456":      false,
+		"MANIFEST-":           false,
+		"MANIFEST-123456":     true,
+		"MANIFEST-123456.doc": false,
+	}
+	for tc, want := range testCases {
+		_, _, got := parseDBFilename(filepath.Join("foo", tc))
+		if got != want {
+			t.Errorf("%q: got %v, want %v", tc, got, want)
+		}
+	}
+}
+
+func TestFilenameRoundTrip(t *testing.T) {
+	testCases := map[fileType]bool{
+		// CURRENT and LOCK files aren't numbered.
+		fileTypeCurrent: false,
+		fileTypeLock:    false,
+		// The remaining file types are numbered.
+		fileTypeLog:      true,
+		fileTypeManifest: true,
+		fileTypeTable:    true,
+	}
+	for fileType, numbered := range testCases {
+		fileNums := []uint64{0}
+		if numbered {
+			fileNums = []uint64{0, 1, 2, 3, 10, 42, 99, 1001}
+		}
+		for _, fileNum := range fileNums {
+			filename := dbFilename("foo", fileType, fileNum)
+			gotFT, gotFN, gotOK := parseDBFilename(filename)
+			if !gotOK {
+				t.Errorf("could not parse %q", filename)
+				continue
+			}
+			if gotFT != fileType || gotFN != fileNum {
+				t.Errorf("filename=%q: got %v, %v, want %v, %v", filename, gotFT, gotFN, fileType, fileNum)
+				continue
+			}
+		}
+	}
+}

leveldb/leveldb.go

@@ -297,9 +297,9 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	}
 	var logFiles fileNumAndNameSlice
 	for _, filename := range ls {
-		n := logFileNum(filename)
-		if n != 0 && (n >= d.versions.logNumber || n == d.versions.prevLogNumber) {
-			logFiles = append(logFiles, fileNumAndName{n, filename})
+		ft, fn, ok := parseDBFilename(filename)
+		if ok && ft == fileTypeLog && (fn >= d.versions.logNumber || fn == d.versions.prevLogNumber) {
+			logFiles = append(logFiles, fileNumAndName{fn, filename})
 		}
 	}
 	sort.Sort(logFiles)
@@ -333,9 +333,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return nil, err
 	}
 
-	if err := d.deleteObsoleteFiles(); err != nil {
-		return nil, err
-	}
+	d.deleteObsoleteFiles()
 	d.maybeScheduleCompaction()
 
 	d.logFile, logFile = logFile, nil
@@ -608,7 +606,47 @@ func (d *DB) makeRoomForWrite(force bool) error {
 //
 // 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() error {
-	// TODO: implement.
-	return nil
+func (d *DB) deleteObsoleteFiles() {
+	// TODO: (elsewhere) track pending outputs, and refer to them here.
+	liveFileNums := map[uint64]struct{}{}
+
+	d.versions.addLiveFileNums(liveFileNums)
+	logNumber := d.versions.logNumber
+	manifestFileNumber := d.versions.manifestFileNumber
+
+	// Release the d.mu lock while doing I/O.
+	// Note the unusual order: Unlock and then Lock.
+	d.mu.Unlock()
+	defer d.mu.Lock()
+
+	fs := d.opts.GetFileSystem()
+	list, err := fs.List(d.dirname)
+	if err != nil {
+		// Ignore any filesystem errors.
+		return
+	}
+	for _, filename := range list {
+		fileType, fileNum, ok := parseDBFilename(filename)
+		if !ok {
+			return
+		}
+		keep := true
+		switch fileType {
+		case fileTypeLog:
+			// TODO: also look at prevLogNumber?
+			keep = fileNum >= logNumber
+		case fileTypeManifest:
+			keep = fileNum >= manifestFileNumber
+		case fileTypeTable:
+			_, keep = liveFileNums[fileNum]
+		}
+		if keep {
+			continue
+		}
+		if fileType == fileTypeTable {
+			d.tableCache.evict(fileNum)
+		}
+		// Ignore any file system errors.
+		fs.Remove(filepath.Join(d.dirname, filename))
+	}
 }

leveldb/leveldb_test.go

@@ -9,6 +9,9 @@ import (
 	"io"
 	"math/rand"
 	"os"
+	"path/filepath"
+	"reflect"
+	"sort"
 	"strconv"
 	"strings"
 	"testing"
@@ -46,6 +49,34 @@ func TestErrorIfDBExists(t *testing.T) {
 	}
 }
 
+func TestNewDBFilenames(t *testing.T) {
+	fooBar := filepath.Join("foo", "bar")
+	fs := memfs.New()
+	d, err := Open(fooBar, &db.Options{
+		FileSystem: fs,
+	})
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	if err := d.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	got, err := fs.List(fooBar)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	sort.Strings(got)
+	// TODO: should there be a LOCK file here?
+	want := []string{
+		"000003.log",
+		"CURRENT",
+		"MANIFEST-000002",
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("\ngot  %v\nwant %v", got, want)
+	}
+}
+
 // cloneFileSystem returns a new memory-backed file system whose root contains
 // a copy of the directory dirname in the source file system srcFS. The copy
 // is not recursive; directories under dirname are not copied.

leveldb/table_cache.go

@@ -59,6 +59,19 @@ func (c *tableCache) find(fileNum uint64, ikey internalKey) (db.Iterator, error)
 	}, nil
 }
 
+// releaseNode releases a node from the tableCache.
+//
+// c.mu must be held when calling this.
+func (c *tableCache) releaseNode(n *tableCacheNode) {
+	delete(c.nodes, n.fileNum)
+	n.next.prev = n.prev
+	n.prev.next = n.next
+	n.refCount--
+	if n.refCount == 0 {
+		go n.release()
+	}
+}
+
 // findNode returns the node for the table with the given file number, creating
 // that node if it didn't already exist. The caller is responsible for
 // decrementing the returned node's refCount.
@@ -76,14 +89,7 @@ func (c *tableCache) findNode(fileNum uint64) *tableCacheNode {
 		c.nodes[fileNum] = n
 		if len(c.nodes) > c.size {
 			// Release the tail node.
-			tail := c.dummy.prev
-			delete(c.nodes, tail.fileNum)
-			tail.next.prev = tail.prev
-			tail.prev.next = tail.next
-			tail.refCount--
-			if tail.refCount == 0 {
-				go tail.release()
-			}
+			c.releaseNode(c.dummy.prev)
 		}
 		go n.load(c)
 	} else {
@@ -101,6 +107,15 @@ func (c *tableCache) findNode(fileNum uint64) *tableCacheNode {
 	return n
 }
 
+func (c *tableCache) evict(fileNum uint64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if n := c.nodes[fileNum]; n != nil {
+		c.releaseNode(n)
+	}
+}
+
 func (c *tableCache) Close() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()

leveldb/table_cache_test.go

@@ -222,3 +222,50 @@ func TestTableCacheFrequentlyUsed(t *testing.T) {
 		}
 	})
 }
+
+func TestTableCacheEvictions(t *testing.T) {
+	const (
+		N      = 1000
+		lo, hi = 10, 20
+	)
+	c, fs, err := newTableCache()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rng := rand.New(rand.NewSource(2))
+	for i := 0; i < N; i++ {
+		j := rng.Intn(tableCacheTestNumTables)
+		iter, err := c.find(uint64(j), nil)
+		if err != nil {
+			t.Fatalf("i=%d, j=%d: find: %v", i, j, err)
+		}
+		if err := iter.Close(); err != nil {
+			t.Fatalf("i=%d, j=%d: close: %v", i, j, err)
+		}
+
+		c.evict(uint64(lo + rng.Intn(hi-lo)))
+	}
+
+	sumEvicted, nEvicted := 0, 0
+	sumSafe, nSafe := 0, 0
+	fs.validate(t, c, func(i, gotO, gotC int) {
+		if lo <= i && i < hi {
+			sumEvicted += gotO
+			nEvicted++
+		} else {
+			sumSafe += gotO
+			nSafe++
+		}
+	})
+	fEvicted := float64(sumEvicted) / float64(nEvicted)
+	fSafe := float64(sumSafe) / float64(nSafe)
+	// The magic 1.25 number isn't derived from formal modeling. It's just a guess. For
+	// (lo, hi, tableCacheTestCacheSize, tableCacheTestNumTables) = (10, 20, 100, 300),
+	// the ratio seems to converge on roughly 1.5 for large N, compared to 1.0 if we do
+	// not evict any cache entries.
+	if ratio := fEvicted / fSafe; ratio < 1.25 {
+		t.Errorf("evicted tables were opened %.3f times on average, safe tables %.3f, ratio %.3f < 1.250",
+			fEvicted, fSafe, ratio)
+	}
+}

leveldb/version_set.go

@@ -265,3 +265,13 @@ func (vs *versionSet) append(v *version) {
 func (vs *versionSet) currentVersion() *version {
 	return vs.dummyVersion.prev
 }
+
+func (vs *versionSet) addLiveFileNums(m map[uint64]struct{}) {
+	for v := vs.dummyVersion.next; v != &vs.dummyVersion; v = v.next {
+		for _, ff := range v.files {
+			for _, f := range ff {
+				m[f.fileNum] = struct{}{}
+			}
+		}
+	}
+}