vendor: update leveldb

vendor/github.com/syndtr/goleveldb/.travis.yml

@@ -1,12 +0,0 @@
-language: go
-
-go:
-  - 1.4
-  - 1.5
-  - 1.6
-  - 1.7
-  - tip
-
-script:
-  - go test -timeout 1h ./...
-  - go test -timeout 30m -race -run "TestDB_(Concurrent|GoleveldbIssue74)" ./leveldb

vendor/github.com/syndtr/goleveldb/README.md

@@ -10,13 +10,15 @@ Installation
 Requirements
 -----------
 
-* Need at least `go1.4` or newer.
+* Need at least `go1.5` or newer.
 
 Usage
 -----------
 
 Create or open a database:
 ```go
+// The returned DB instance is safe for concurrent use. Which mean that all
+// DB's methods may be called concurrently from multiple goroutine.
 db, err := leveldb.OpenFile("path/to/db", nil)
 ...
 defer db.Close()

vendor/github.com/syndtr/goleveldb/leveldb/cache/cache.go

@@ -331,7 +331,6 @@ func (r *Cache) delete(n *Node) bool {
 			return deleted
 		}
 	}
-	return false
 }
 
 // Nodes returns number of 'cache node' in the map.

vendor/github.com/syndtr/goleveldb/leveldb/comparer/bytes_comparer.go

@@ -29,7 +29,7 @@ func (bytesComparer) Separator(dst, a, b []byte) []byte {
 		// Do not shorten if one string is a prefix of the other
 	} else if c := a[i]; c < 0xff && c+1 < b[i] {
 		dst = append(dst, a[:i+1]...)
-		dst[i]++
+		dst[len(dst)-1]++
 		return dst
 	}
 	return nil
@@ -39,7 +39,7 @@ func (bytesComparer) Successor(dst, b []byte) []byte {
 	for i, c := range b {
 		if c != 0xff {
 			dst = append(dst, b[:i+1]...)
-			dst[i]++
+			dst[len(dst)-1]++
 			return dst
 		}
 	}

vendor/github.com/syndtr/goleveldb/leveldb/comparer/comparer.go

@@ -36,7 +36,7 @@ type Comparer interface {
 	// by any users of this package.
 	Name() string
 
-	// Bellow are advanced functions used used to reduce the space requirements
+	// Bellow are advanced functions used to reduce the space requirements
 	// for internal data structures such as index blocks.
 
 	// Separator appends a sequence of bytes x to dst such that a <= x && x < b,

vendor/github.com/syndtr/goleveldb/leveldb/db.go

@@ -32,6 +32,12 @@ type DB struct {
 	// Need 64-bit alignment.
 	seq uint64
 
+	// Stats. Need 64-bit alignment.
+	cWriteDelay            int64 // The cumulative duration of write delays
+	cWriteDelayN           int32 // The cumulative number of write delays
+	inWritePaused          int32 // The indicator whether write operation is paused by compaction
+	aliveSnaps, aliveIters int32
+
 	// Session.
 	s *session
 
@@ -49,9 +55,6 @@ type DB struct {
 	snapsMu   sync.Mutex
 	snapsList *list.List
 
-	// Stats.
-	aliveSnaps, aliveIters int32
-
 	// Write.
 	batchPool    sync.Pool
 	writeMergeC  chan writeMerge
@@ -179,7 +182,7 @@ func Open(stor storage.Storage, o *opt.Options) (db *DB, err error) {
 
 	err = s.recover()
 	if err != nil {
-		if !os.IsNotExist(err) || s.o.GetErrorIfMissing() {
+		if !os.IsNotExist(err) || s.o.GetErrorIfMissing() || s.o.GetReadOnly() {
 			return
 		}
 		err = s.create()
@@ -321,7 +324,7 @@ func recoverTable(s *session, o *opt.Options) error {
 			}
 		}
 		err = iter.Error()
-		if err != nil {
+		if err != nil && !errors.IsCorrupted(err) {
 			return
 		}
 		err = tw.Close()
@@ -392,7 +395,7 @@ func recoverTable(s *session, o *opt.Options) error {
 			}
 			imax = append(imax[:0], key...)
 		}
-		if err := iter.Error(); err != nil {
+		if err := iter.Error(); err != nil && !errors.IsCorrupted(err) {
 			iter.Release()
 			return err
 		}
@@ -844,7 +847,7 @@ func (db *DB) Get(key []byte, ro *opt.ReadOptions) (value []byte, err error) {
 
 // Has returns true if the DB does contains the given key.
 //
-// It is safe to modify the contents of the argument after Get returns.
+// It is safe to modify the contents of the argument after Has returns.
 func (db *DB) Has(key []byte, ro *opt.ReadOptions) (ret bool, err error) {
 	err = db.ok()
 	if err != nil {
@@ -904,6 +907,10 @@ func (db *DB) GetSnapshot() (*Snapshot, error) {
 //		Returns the number of files at level 'n'.
 //	leveldb.stats
 //		Returns statistics of the underlying DB.
+//	leveldb.iostats
+//		Returns statistics of effective disk read and write.
+//	leveldb.writedelay
+//		Returns cumulative write delay caused by compaction.
 //	leveldb.sstables
 //		Returns sstables list for each level.
 //	leveldb.blockpool
@@ -955,6 +962,14 @@ func (db *DB) GetProperty(name string) (value string, err error) {
 				level, len(tables), float64(tables.size())/1048576.0, duration.Seconds(),
 				float64(read)/1048576.0, float64(write)/1048576.0)
 		}
+	case p == "iostats":
+		value = fmt.Sprintf("Read(MB):%.5f Write(MB):%.5f",
+			float64(db.s.stor.reads())/1048576.0,
+			float64(db.s.stor.writes())/1048576.0)
+	case p == "writedelay":
+		writeDelayN, writeDelay := atomic.LoadInt32(&db.cWriteDelayN), time.Duration(atomic.LoadInt64(&db.cWriteDelay))
+		paused := atomic.LoadInt32(&db.inWritePaused) == 1
+		value = fmt.Sprintf("DelayN:%d Delay:%s Paused:%t", writeDelayN, writeDelay, paused)
 	case p == "sstables":
 		for level, tables := range v.levels {
 			value += fmt.Sprintf("--- level %d ---\n", level)
@@ -983,6 +998,75 @@ func (db *DB) GetProperty(name string) (value string, err error) {
 	return
 }
 
+// DBStats is database statistics.
+type DBStats struct {
+	WriteDelayCount    int32
+	WriteDelayDuration time.Duration
+	WritePaused        bool
+
+	AliveSnapshots int32
+	AliveIterators int32
+
+	IOWrite uint64
+	IORead  uint64
+
+	BlockCacheSize    int
+	OpenedTablesCount int
+
+	LevelSizes        []int64
+	LevelTablesCounts []int
+	LevelRead         []int64
+	LevelWrite        []int64
+	LevelDurations    []time.Duration
+}
+
+// Stats populates s with database statistics.
+func (db *DB) Stats(s *DBStats) error {
+	err := db.ok()
+	if err != nil {
+		return err
+	}
+
+	s.IORead = db.s.stor.reads()
+	s.IOWrite = db.s.stor.writes()
+	s.WriteDelayCount = atomic.LoadInt32(&db.cWriteDelayN)
+	s.WriteDelayDuration = time.Duration(atomic.LoadInt64(&db.cWriteDelay))
+	s.WritePaused = atomic.LoadInt32(&db.inWritePaused) == 1
+
+	s.OpenedTablesCount = db.s.tops.cache.Size()
+	if db.s.tops.bcache != nil {
+		s.BlockCacheSize = db.s.tops.bcache.Size()
+	} else {
+		s.BlockCacheSize = 0
+	}
+
+	s.AliveIterators = atomic.LoadInt32(&db.aliveIters)
+	s.AliveSnapshots = atomic.LoadInt32(&db.aliveSnaps)
+
+	s.LevelDurations = s.LevelDurations[:0]
+	s.LevelRead = s.LevelRead[:0]
+	s.LevelWrite = s.LevelWrite[:0]
+	s.LevelSizes = s.LevelSizes[:0]
+	s.LevelTablesCounts = s.LevelTablesCounts[:0]
+
+	v := db.s.version()
+	defer v.release()
+
+	for level, tables := range v.levels {
+		duration, read, write := db.compStats.getStat(level)
+		if len(tables) == 0 && duration == 0 {
+			continue
+		}
+		s.LevelDurations = append(s.LevelDurations, duration)
+		s.LevelRead = append(s.LevelRead, read)
+		s.LevelWrite = append(s.LevelWrite, write)
+		s.LevelSizes = append(s.LevelSizes, tables.size())
+		s.LevelTablesCounts = append(s.LevelTablesCounts, len(tables))
+	}
+
+	return nil
+}
+
 // SizeOf calculates approximate sizes of the given key ranges.
 // The length of the returned sizes are equal with the length of the given
 // ranges. The returned sizes measure storage space usage, so if the user

vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go

@@ -289,7 +289,7 @@ func (db *DB) memCompaction() {
 		close(resumeC)
 		resumeC = nil
 	case <-db.closeC:
-		return
+		db.compactionExitTransact()
 	}
 
 	var (
@@ -338,7 +338,7 @@ func (db *DB) memCompaction() {
 		case <-resumeC:
 			close(resumeC)
 		case <-db.closeC:
-			return
+			db.compactionExitTransact()
 		}
 	}
 
@@ -640,6 +640,16 @@ func (db *DB) tableNeedCompaction() bool {
 	return v.needCompaction()
 }
 
+// resumeWrite returns an indicator whether we should resume write operation if enough level0 files are compacted.
+func (db *DB) resumeWrite() bool {
+	v := db.s.version()
+	defer v.release()
+	if v.tLen(0) < db.s.o.GetWriteL0PauseTrigger() {
+		return true
+	}
+	return false
+}
+
 func (db *DB) pauseCompaction(ch chan<- struct{}) {
 	select {
 	case ch <- struct{}{}:
@@ -653,6 +663,7 @@ type cCmd interface {
 }
 
 type cAuto struct {
+	// Note for table compaction, an non-empty ackC represents it's a compaction waiting command.
 	ackC chan<- error
 }
 
@@ -765,8 +776,10 @@ func (db *DB) mCompaction() {
 }
 
 func (db *DB) tCompaction() {
-	var x cCmd
-	var ackQ []cCmd
+	var (
+		x     cCmd
+		waitQ []cCmd
+	)
 
 	defer func() {
 		if x := recover(); x != nil {
@@ -774,9 +787,9 @@ func (db *DB) tCompaction() {
 				panic(x)
 			}
 		}
-		for i := range ackQ {
-			ackQ[i].ack(ErrClosed)
-			ackQ[i] = nil
+		for i := range waitQ {
+			waitQ[i].ack(ErrClosed)
+			waitQ[i] = nil
 		}
 		if x != nil {
 			x.ack(ErrClosed)
@@ -795,12 +808,20 @@ func (db *DB) tCompaction() {
 				return
 			default:
 			}
-		} else {
-			for i := range ackQ {
-				ackQ[i].ack(nil)
-				ackQ[i] = nil
+			// Resume write operation as soon as possible.
+			if len(waitQ) > 0 && db.resumeWrite() {
+				for i := range waitQ {
+					waitQ[i].ack(nil)
+					waitQ[i] = nil
+				}
+				waitQ = waitQ[:0]
 			}
-			ackQ = ackQ[:0]
+		} else {
+			for i := range waitQ {
+				waitQ[i].ack(nil)
+				waitQ[i] = nil
+			}
+			waitQ = waitQ[:0]
 			select {
 			case x = <-db.tcompCmdC:
 			case ch := <-db.tcompPauseC:
@@ -813,7 +834,14 @@ func (db *DB) tCompaction() {
 		if x != nil {
 			switch cmd := x.(type) {
 			case cAuto:
-				ackQ = append(ackQ, x)
+				if cmd.ackC != nil {
+					// Check the write pause state before caching it.
+					if db.resumeWrite() {
+						x.ack(nil)
+					} else {
+						waitQ = append(waitQ, x)
+					}
+				}
 			case cRange:
 				x.ack(db.tableRangeCompaction(cmd.level, cmd.min, cmd.max))
 			default:

vendor/github.com/syndtr/goleveldb/leveldb/db_state.go

@@ -7,6 +7,7 @@
 package leveldb
 
 import (
+	"errors"
 	"sync/atomic"
 	"time"
 
@@ -15,6 +16,10 @@ import (
 	"github.com/syndtr/goleveldb/leveldb/storage"
 )
 
+var (
+	errHasFrozenMem = errors.New("has frozen mem")
+)
+
 type memDB struct {
 	db *DB
 	*memdb.DB
@@ -126,7 +131,7 @@ func (db *DB) newMem(n int) (mem *memDB, err error) {
 	defer db.memMu.Unlock()
 
 	if db.frozenMem != nil {
-		panic("still has frozen mem")
+		return nil, errHasFrozenMem
 	}
 
 	if db.journal == nil {

vendor/github.com/syndtr/goleveldb/leveldb/db_util.go

@@ -84,7 +84,7 @@ func (db *DB) checkAndCleanFiles() error {
 		var mfds []storage.FileDesc
 		for num, present := range tmap {
 			if !present {
-				mfds = append(mfds, storage.FileDesc{storage.TypeTable, num})
+				mfds = append(mfds, storage.FileDesc{Type: storage.TypeTable, Num: num})
 				db.logf("db@janitor table missing @%d", num)
 			}
 		}

vendor/github.com/syndtr/goleveldb/leveldb/db_write.go

@@ -7,6 +7,7 @@
 package leveldb
 
 import (
+	"sync/atomic"
 	"time"
 
 	"github.com/syndtr/goleveldb/leveldb/memdb"
@@ -32,15 +33,24 @@ func (db *DB) writeJournal(batches []*Batch, seq uint64, sync bool) error {
 }
 
 func (db *DB) rotateMem(n int, wait bool) (mem *memDB, err error) {
+	retryLimit := 3
+retry:
 	// Wait for pending memdb compaction.
 	err = db.compTriggerWait(db.mcompCmdC)
 	if err != nil {
 		return
 	}
+	retryLimit--
 
 	// Create new memdb and journal.
 	mem, err = db.newMem(n)
 	if err != nil {
+		if err == errHasFrozenMem {
+			if retryLimit <= 0 {
+				panic("BUG: still has frozen memdb")
+			}
+			goto retry
+		}
 		return
 	}
 
@@ -79,7 +89,11 @@ func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) {
 			return false
 		case tLen >= pauseTrigger:
 			delayed = true
+			// Set the write paused flag explicitly.
+			atomic.StoreInt32(&db.inWritePaused, 1)
 			err = db.compTriggerWait(db.tcompCmdC)
+			// Unset the write paused flag.
+			atomic.StoreInt32(&db.inWritePaused, 0)
 			if err != nil {
 				return false
 			}
@@ -108,6 +122,8 @@ func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) {
 		db.writeDelayN++
 	} else if db.writeDelayN > 0 {
 		db.logf("db@write was delayed N·%d T·%v", db.writeDelayN, db.writeDelay)
+		atomic.AddInt32(&db.cWriteDelayN, int32(db.writeDelayN))
+		atomic.AddInt64(&db.cWriteDelay, int64(db.writeDelay))
 		db.writeDelay = 0
 		db.writeDelayN = 0
 	}
@@ -134,7 +150,7 @@ func (db *DB) unlockWrite(overflow bool, merged int, err error) {
 	}
 }
 
-// ourBatch if defined should equal with batch.
+// ourBatch is batch that we can modify.
 func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error {
 	// Try to flush memdb. This method would also trying to throttle writes
 	// if it is too fast and compaction cannot catch-up.
@@ -203,6 +219,11 @@ func (db *DB) writeLocked(batch, ourBatch *Batch, merge, sync bool) error {
 		}
 	}
 
+	// Release ourBatch if any.
+	if ourBatch != nil {
+		defer db.batchPool.Put(ourBatch)
+	}
+
 	// Seq number.
 	seq := db.seq + 1
 

vendor/github.com/syndtr/goleveldb/leveldb/doc.go

@@ -8,6 +8,8 @@
 //
 // Create or open a database:
 //
+//	// The returned DB instance is safe for concurrent use. Which mean that all
+//	// DB's methods may be called concurrently from multiple goroutine.
 //	db, err := leveldb.OpenFile("path/to/db", nil)
 //	...
 //	defer db.Close()

vendor/github.com/syndtr/goleveldb/leveldb/iterator/iter.go

@@ -40,11 +40,11 @@ type IteratorSeeker interface {
 	Seek(key []byte) bool
 
 	// Next moves the iterator to the next key/value pair.
-	// It returns whether the iterator is exhausted.
+	// It returns false if the iterator is exhausted.
 	Next() bool
 
 	// Prev moves the iterator to the previous key/value pair.
-	// It returns whether the iterator is exhausted.
+	// It returns false if the iterator is exhausted.
 	Prev() bool
 }
 
@@ -88,7 +88,7 @@ type Iterator interface {
 	// its contents may change on the next call to any 'seeks method'.
 	Key() []byte
 
-	// Value returns the key of the current key/value pair, or nil if done.
+	// Value returns the value of the current key/value pair, or nil if done.
 	// The caller should not modify the contents of the returned slice, and
 	// its contents may change on the next call to any 'seeks method'.
 	Value() []byte

vendor/github.com/syndtr/goleveldb/leveldb/memdb/memdb.go

@@ -329,7 +329,7 @@ func (p *DB) Delete(key []byte) error {
 
 	h := p.nodeData[node+nHeight]
 	for i, n := range p.prevNode[:h] {
-		m := n + 4 + i
+		m := n + nNext + i
 		p.nodeData[m] = p.nodeData[p.nodeData[m]+nNext+i]
 	}
 

vendor/github.com/syndtr/goleveldb/leveldb/opt/options.go

@@ -158,6 +158,12 @@ type Options struct {
 	// The default value is 8MiB.
 	BlockCacheCapacity int
 
+	// BlockCacheEvictRemoved allows enable forced-eviction on cached block belonging
+	// to removed 'sorted table'.
+	//
+	// The default if false.
+	BlockCacheEvictRemoved bool
+
 	// BlockRestartInterval is the number of keys between restart points for
 	// delta encoding of keys.
 	//
@@ -384,6 +390,13 @@ func (o *Options) GetBlockCacheCapacity() int {
 	return o.BlockCacheCapacity
 }
 
+func (o *Options) GetBlockCacheEvictRemoved() bool {
+	if o == nil {
+		return false
+	}
+	return o.BlockCacheEvictRemoved
+}
+
 func (o *Options) GetBlockRestartInterval() int {
 	if o == nil || o.BlockRestartInterval <= 0 {
 		return DefaultBlockRestartInterval

vendor/github.com/syndtr/goleveldb/leveldb/session.go

@@ -42,7 +42,7 @@ type session struct {
 	stTempFileNum    int64
 	stSeqNum         uint64 // last mem compacted seq; need external synchronization
 
-	stor     storage.Storage
+	stor     *iStorage
 	storLock storage.Locker
 	o        *cachedOptions
 	icmp     *iComparer
@@ -68,7 +68,7 @@ func newSession(stor storage.Storage, o *opt.Options) (s *session, err error) {
 		return
 	}
 	s = &session{
-		stor:     stor,
+		stor:     newIStorage(stor),
 		storLock: storLock,
 		fileRef:  make(map[int64]int),
 	}

vendor/github.com/syndtr/goleveldb/leveldb/session_util.go

@@ -36,7 +36,7 @@ func (s *session) logf(format string, v ...interface{}) { s.stor.Log(fmt.Sprintf
 
 func (s *session) newTemp() storage.FileDesc {
 	num := atomic.AddInt64(&s.stTempFileNum, 1) - 1
-	return storage.FileDesc{storage.TypeTemp, num}
+	return storage.FileDesc{Type: storage.TypeTemp, Num: num}
 }
 
 func (s *session) addFileRef(fd storage.FileDesc, ref int) int {
@@ -190,7 +190,7 @@ func (s *session) recordCommited(rec *sessionRecord) {
 
 // Create a new manifest file; need external synchronization.
 func (s *session) newManifest(rec *sessionRecord, v *version) (err error) {
-	fd := storage.FileDesc{storage.TypeManifest, s.allocFileNum()}
+	fd := storage.FileDesc{Type: storage.TypeManifest, Num: s.allocFileNum()}
 	writer, err := s.stor.Create(fd)
 	if err != nil {
 		return

vendor/github.com/syndtr/goleveldb/leveldb/storage.go

@@ -0,0 +1,63 @@
+package leveldb
+
+import (
+	"github.com/syndtr/goleveldb/leveldb/storage"
+	"sync/atomic"
+)
+
+type iStorage struct {
+	storage.Storage
+	read  uint64
+	write uint64
+}
+
+func (c *iStorage) Open(fd storage.FileDesc) (storage.Reader, error) {
+	r, err := c.Storage.Open(fd)
+	return &iStorageReader{r, c}, err
+}
+
+func (c *iStorage) Create(fd storage.FileDesc) (storage.Writer, error) {
+	w, err := c.Storage.Create(fd)
+	return &iStorageWriter{w, c}, err
+}
+
+func (c *iStorage) reads() uint64 {
+	return atomic.LoadUint64(&c.read)
+}
+
+func (c *iStorage) writes() uint64 {
+	return atomic.LoadUint64(&c.write)
+}
+
+// newIStorage returns the given storage wrapped by iStorage.
+func newIStorage(s storage.Storage) *iStorage {
+	return &iStorage{s, 0, 0}
+}
+
+type iStorageReader struct {
+	storage.Reader
+	c *iStorage
+}
+
+func (r *iStorageReader) Read(p []byte) (n int, err error) {
+	n, err = r.Reader.Read(p)
+	atomic.AddUint64(&r.c.read, uint64(n))
+	return n, err
+}
+
+func (r *iStorageReader) ReadAt(p []byte, off int64) (n int, err error) {
+	n, err = r.Reader.ReadAt(p, off)
+	atomic.AddUint64(&r.c.read, uint64(n))
+	return n, err
+}
+
+type iStorageWriter struct {
+	storage.Writer
+	c *iStorage
+}
+
+func (w *iStorageWriter) Write(p []byte) (n int, err error) {
+	n, err = w.Writer.Write(p)
+	atomic.AddUint64(&w.c.write, uint64(n))
+	return n, err
+}

vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage.go

@@ -9,10 +9,12 @@ package storage
 import (
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"runtime"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -42,6 +44,30 @@ func (lock *fileStorageLock) Unlock() {
 	}
 }
 
+type int64Slice []int64
+
+func (p int64Slice) Len() int           { return len(p) }
+func (p int64Slice) Less(i, j int) bool { return p[i] < p[j] }
+func (p int64Slice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
+
+func writeFileSynced(filename string, data []byte, perm os.FileMode) error {
+	f, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
+	if err != nil {
+		return err
+	}
+	n, err := f.Write(data)
+	if err == nil && n < len(data) {
+		err = io.ErrShortWrite
+	}
+	if err1 := f.Sync(); err == nil {
+		err = err1
+	}
+	if err1 := f.Close(); err == nil {
+		err = err1
+	}
+	return err
+}
+
 const logSizeThreshold = 1024 * 1024 // 1 MiB
 
 // fileStorage is a file-system backed storage.
@@ -60,7 +86,7 @@ type fileStorage struct {
 	day  int
 }
 
-// OpenFile returns a new filesytem-backed storage implementation with the given
+// OpenFile returns a new filesystem-backed storage implementation with the given
 // path. This also acquire a file lock, so any subsequent attempt to open the
 // same path will fail.
 //
@@ -189,7 +215,8 @@ func (fs *fileStorage) doLog(t time.Time, str string) {
 	// write
 	fs.buf = append(fs.buf, []byte(str)...)
 	fs.buf = append(fs.buf, '\n')
-	fs.logw.Write(fs.buf)
+	n, _ := fs.logw.Write(fs.buf)
+	fs.logSize += int64(n)
 }
 
 func (fs *fileStorage) Log(str string) {
@@ -210,7 +237,46 @@ func (fs *fileStorage) log(str string) {
 	}
 }
 
-func (fs *fileStorage) SetMeta(fd FileDesc) (err error) {
+func (fs *fileStorage) setMeta(fd FileDesc) error {
+	content := fsGenName(fd) + "\n"
+	// Check and backup old CURRENT file.
+	currentPath := filepath.Join(fs.path, "CURRENT")
+	if _, err := os.Stat(currentPath); err == nil {
+		b, err := ioutil.ReadFile(currentPath)
+		if err != nil {
+			fs.log(fmt.Sprintf("backup CURRENT: %v", err))
+			return err
+		}
+		if string(b) == content {
+			// Content not changed, do nothing.
+			return nil
+		}
+		if err := writeFileSynced(currentPath+".bak", b, 0644); err != nil {
+			fs.log(fmt.Sprintf("backup CURRENT: %v", err))
+			return err
+		}
+	} else if !os.IsNotExist(err) {
+		return err
+	}
+	path := fmt.Sprintf("%s.%d", filepath.Join(fs.path, "CURRENT"), fd.Num)
+	if err := writeFileSynced(path, []byte(content), 0644); err != nil {
+		fs.log(fmt.Sprintf("create CURRENT.%d: %v", fd.Num, err))
+		return err
+	}
+	// Replace CURRENT file.
+	if err := rename(path, currentPath); err != nil {
+		fs.log(fmt.Sprintf("rename CURRENT.%d: %v", fd.Num, err))
+		return err
+	}
+	// Sync root directory.
+	if err := syncDir(fs.path); err != nil {
+		fs.log(fmt.Sprintf("syncDir: %v", err))
+		return err
+	}
+	return nil
+}
+
+func (fs *fileStorage) SetMeta(fd FileDesc) error {
 	if !FileDescOk(fd) {
 		return ErrInvalidFile
 	}
@@ -223,28 +289,10 @@ func (fs *fileStorage) SetMeta(fd FileDesc) (err error) {
 	if fs.open < 0 {
 		return ErrClosed
 	}
-	defer func() {
-		if err != nil {
-			fs.log(fmt.Sprintf("CURRENT: %v", err))
-		}
-	}()
-	path := fmt.Sprintf("%s.%d", filepath.Join(fs.path, "CURRENT"), fd.Num)
-	w, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
-	if err != nil {
-		return
-	}
-	_, err = fmt.Fprintln(w, fsGenName(fd))
-	// Close the file first.
-	if cerr := w.Close(); cerr != nil {
-		fs.log(fmt.Sprintf("close CURRENT.%d: %v", fd.Num, cerr))
-	}
-	if err != nil {
-		return
-	}
-	return rename(path, filepath.Join(fs.path, "CURRENT"))
+	return fs.setMeta(fd)
 }
 
-func (fs *fileStorage) GetMeta() (fd FileDesc, err error) {
+func (fs *fileStorage) GetMeta() (FileDesc, error) {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 	if fs.open < 0 {
@@ -252,7 +300,7 @@ func (fs *fileStorage) GetMeta() (fd FileDesc, err error) {
 	}
 	dir, err := os.Open(fs.path)
 	if err != nil {
-		return
+		return FileDesc{}, err
 	}
 	names, err := dir.Readdirnames(0)
 	// Close the dir first before checking for Readdirnames error.
@@ -260,94 +308,134 @@ func (fs *fileStorage) GetMeta() (fd FileDesc, err error) {
 		fs.log(fmt.Sprintf("close dir: %v", ce))
 	}
 	if err != nil {
-		return
+		return FileDesc{}, err
 	}
-	// Find latest CURRENT file.
-	var rem []string
-	var pend bool
-	var cerr error
+	// Try this in order:
+	// - CURRENT.[0-9]+ ('pending rename' file, descending order)
+	// - CURRENT
+	// - CURRENT.bak
+	//
+	// Skip corrupted file or file that point to a missing target file.
+	type currentFile struct {
+		name string
+		fd   FileDesc
+	}
+	tryCurrent := func(name string) (*currentFile, error) {
+		b, err := ioutil.ReadFile(filepath.Join(fs.path, name))
+		if err != nil {
+			if os.IsNotExist(err) {
+				err = os.ErrNotExist
+			}
+			return nil, err
+		}
+		var fd FileDesc
+		if len(b) < 1 || b[len(b)-1] != '\n' || !fsParseNamePtr(string(b[:len(b)-1]), &fd) {
+			fs.log(fmt.Sprintf("%s: corrupted content: %q", name, b))
+			err := &ErrCorrupted{
+				Err: errors.New("leveldb/storage: corrupted or incomplete CURRENT file"),
+			}
+			return nil, err
+		}
+		if _, err := os.Stat(filepath.Join(fs.path, fsGenName(fd))); err != nil {
+			if os.IsNotExist(err) {
+				fs.log(fmt.Sprintf("%s: missing target file: %s", name, fd))
+				err = os.ErrNotExist
+			}
+			return nil, err
+		}
+		return &currentFile{name: name, fd: fd}, nil
+	}
+	tryCurrents := func(names []string) (*currentFile, error) {
+		var (
+			cur *currentFile
+			// Last corruption error.
+			lastCerr error
+		)
+		for _, name := range names {
+			var err error
+			cur, err = tryCurrent(name)
+			if err == nil {
+				break
+			} else if err == os.ErrNotExist {
+				// Fallback to the next file.
+			} else if isCorrupted(err) {
+				lastCerr = err
+				// Fallback to the next file.
+			} else {
+				// In case the error is due to permission, etc.
+				return nil, err
+			}
+		}
+		if cur == nil {
+			err := os.ErrNotExist
+			if lastCerr != nil {
+				err = lastCerr
+			}
+			return nil, err
+		}
+		return cur, nil
+	}
+
+	// Try 'pending rename' files.
+	var nums []int64
 	for _, name := range names {
-		if strings.HasPrefix(name, "CURRENT") {
-			pend1 := len(name) > 7
-			var pendNum int64
-			// Make sure it is valid name for a CURRENT file, otherwise skip it.
-			if pend1 {
-				if name[7] != '.' || len(name) < 9 {
-					fs.log(fmt.Sprintf("skipping %s: invalid file name", name))
-					continue
-				}
-				var e1 error
-				if pendNum, e1 = strconv.ParseInt(name[8:], 10, 0); e1 != nil {
-					fs.log(fmt.Sprintf("skipping %s: invalid file num: %v", name, e1))
-					continue
-				}
+		if strings.HasPrefix(name, "CURRENT.") && name != "CURRENT.bak" {
+			i, err := strconv.ParseInt(name[8:], 10, 64)
+			if err == nil {
+				nums = append(nums, i)
 			}
-			path := filepath.Join(fs.path, name)
-			r, e1 := os.OpenFile(path, os.O_RDONLY, 0)
-			if e1 != nil {
-				return FileDesc{}, e1
-			}
-			b, e1 := ioutil.ReadAll(r)
-			if e1 != nil {
-				r.Close()
-				return FileDesc{}, e1
-			}
-			var fd1 FileDesc
-			if len(b) < 1 || b[len(b)-1] != '\n' || !fsParseNamePtr(string(b[:len(b)-1]), &fd1) {
-				fs.log(fmt.Sprintf("skipping %s: corrupted or incomplete", name))
-				if pend1 {
-					rem = append(rem, name)
-				}
-				if !pend1 || cerr == nil {
-					metaFd, _ := fsParseName(name)
-					cerr = &ErrCorrupted{
-						Fd:  metaFd,
-						Err: errors.New("leveldb/storage: corrupted or incomplete meta file"),
+		}
+	}
+	var (
+		pendCur   *currentFile
+		pendErr   = os.ErrNotExist
+		pendNames []string
+	)
+	if len(nums) > 0 {
+		sort.Sort(sort.Reverse(int64Slice(nums)))
+		pendNames = make([]string, len(nums))
+		for i, num := range nums {
+			pendNames[i] = fmt.Sprintf("CURRENT.%d", num)
+		}
+		pendCur, pendErr = tryCurrents(pendNames)
+		if pendErr != nil && pendErr != os.ErrNotExist && !isCorrupted(pendErr) {
+			return FileDesc{}, pendErr
+		}
+	}
+
+	// Try CURRENT and CURRENT.bak.
+	curCur, curErr := tryCurrents([]string{"CURRENT", "CURRENT.bak"})
+	if curErr != nil && curErr != os.ErrNotExist && !isCorrupted(curErr) {
+		return FileDesc{}, curErr
+	}
+
+	// pendCur takes precedence, but guards against obsolete pendCur.
+	if pendCur != nil && (curCur == nil || pendCur.fd.Num > curCur.fd.Num) {
+		curCur = pendCur
+	}
+
+	if curCur != nil {
+		// Restore CURRENT file to proper state.
+		if !fs.readOnly && (curCur.name != "CURRENT" || len(pendNames) != 0) {
+			// Ignore setMeta errors, however don't delete obsolete files if we
+			// catch error.
+			if err := fs.setMeta(curCur.fd); err == nil {
+				// Remove 'pending rename' files.
+				for _, name := range pendNames {
+					if err := os.Remove(filepath.Join(fs.path, name)); err != nil {
+						fs.log(fmt.Sprintf("remove %s: %v", name, err))
 					}
 				}
-			} else if pend1 && pendNum != fd1.Num {
-				fs.log(fmt.Sprintf("skipping %s: inconsistent pending-file num: %d vs %d", name, pendNum, fd1.Num))
-				rem = append(rem, name)
-			} else if fd1.Num < fd.Num {
-				fs.log(fmt.Sprintf("skipping %s: obsolete", name))
-				if pend1 {
-					rem = append(rem, name)
-				}
-			} else {
-				fd = fd1
-				pend = pend1
-			}
-			if err := r.Close(); err != nil {
-				fs.log(fmt.Sprintf("close %s: %v", name, err))
 			}
 		}
+		return curCur.fd, nil
 	}
-	// Don't remove any files if there is no valid CURRENT file.
-	if fd.Zero() {
-		if cerr != nil {
-			err = cerr
-		} else {
-			err = os.ErrNotExist
-		}
-		return
+
+	// Nothing found.
+	if isCorrupted(pendErr) {
+		return FileDesc{}, pendErr
 	}
-	if !fs.readOnly {
-		// Rename pending CURRENT file to an effective CURRENT.
-		if pend {
-			path := fmt.Sprintf("%s.%d", filepath.Join(fs.path, "CURRENT"), fd.Num)
-			if err := rename(path, filepath.Join(fs.path, "CURRENT")); err != nil {
-				fs.log(fmt.Sprintf("CURRENT.%d -> CURRENT: %v", fd.Num, err))
-			}
-		}
-		// Remove obsolete or incomplete pending CURRENT files.
-		for _, name := range rem {
-			path := filepath.Join(fs.path, name)
-			if err := os.Remove(path); err != nil {
-				fs.log(fmt.Sprintf("remove %s: %v", name, err))
-			}
-		}
-	}
-	return
+	return FileDesc{}, curErr
 }
 
 func (fs *fileStorage) List(ft FileType) (fds []FileDesc, err error) {

vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_plan9.go

@@ -8,7 +8,6 @@ package storage
 
 import (
 	"os"
-	"path/filepath"
 )
 
 type plan9FileLock struct {
@@ -48,8 +47,7 @@ func rename(oldpath, newpath string) error {
 		}
 	}
 
-	_, fname := filepath.Split(newpath)
-	return os.Rename(oldpath, fname)
+	return os.Rename(oldpath, newpath)
 }
 
 func syncDir(name string) error {

vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_unix.go

@@ -67,13 +67,25 @@ func isErrInvalid(err error) bool {
 	if err == os.ErrInvalid {
 		return true
 	}
+	// Go < 1.8
 	if syserr, ok := err.(*os.SyscallError); ok && syserr.Err == syscall.EINVAL {
 		return true
 	}
+	// Go >= 1.8 returns *os.PathError instead
+	if patherr, ok := err.(*os.PathError); ok && patherr.Err == syscall.EINVAL {
+		return true
+	}
 	return false
 }
 
 func syncDir(name string) error {
+	// As per fsync manpage, Linux seems to expect fsync on directory, however
+	// some system don't support this, so we will ignore syscall.EINVAL.
+	//
+	// From fsync(2):
+	//   Calling fsync() does not necessarily ensure that the entry in the
+	//   directory containing the file has also reached disk. For that an
+	//   explicit fsync() on a file descriptor for the directory is also needed.
 	f, err := os.Open(name)
 	if err != nil {
 		return err

vendor/github.com/syndtr/goleveldb/leveldb/storage/mem_storage.go

@@ -12,7 +12,11 @@ import (
 	"sync"
 )
 
-const typeShift = 3
+const typeShift = 4
+
+// Verify at compile-time that typeShift is large enough to cover all FileType
+// values by confirming that 0 == 0.
+var _ [0]struct{} = [TypeAll >> typeShift]struct{}{}
 
 type memStorageLock struct {
 	ms *memStorage
@@ -143,7 +147,7 @@ func (ms *memStorage) Remove(fd FileDesc) error {
 }
 
 func (ms *memStorage) Rename(oldfd, newfd FileDesc) error {
-	if FileDescOk(oldfd) || FileDescOk(newfd) {
+	if !FileDescOk(oldfd) || !FileDescOk(newfd) {
 		return ErrInvalidFile
 	}
 	if oldfd == newfd {

vendor/github.com/syndtr/goleveldb/leveldb/storage/storage.go

@@ -55,6 +55,14 @@ type ErrCorrupted struct {
 	Err error
 }
 
+func isCorrupted(err error) bool {
+	switch err.(type) {
+	case *ErrCorrupted:
+		return true
+	}
+	return false
+}
+
 func (e *ErrCorrupted) Error() string {
 	if !e.Fd.Zero() {
 		return fmt.Sprintf("%v [file=%v]", e.Err, e.Fd)

vendor/github.com/syndtr/goleveldb/leveldb/table.go

@@ -78,7 +78,7 @@ func newTableFile(fd storage.FileDesc, size int64, imin, imax internalKey) *tFil
 }
 
 func tableFileFromRecord(r atRecord) *tFile {
-	return newTableFile(storage.FileDesc{storage.TypeTable, r.num}, r.size, r.imin, r.imax)
+	return newTableFile(storage.FileDesc{Type: storage.TypeTable, Num: r.num}, r.size, r.imin, r.imax)
 }
 
 // tFiles hold multiple tFile.
@@ -290,16 +290,17 @@ func (x *tFilesSortByNum) Less(i, j int) bool {
 
 // Table operations.
 type tOps struct {
-	s      *session
-	noSync bool
-	cache  *cache.Cache
-	bcache *cache.Cache
-	bpool  *util.BufferPool
+	s            *session
+	noSync       bool
+	evictRemoved bool
+	cache        *cache.Cache
+	bcache       *cache.Cache
+	bpool        *util.BufferPool
 }
 
 // Creates an empty table and returns table writer.
 func (t *tOps) create() (*tWriter, error) {
-	fd := storage.FileDesc{storage.TypeTable, t.s.allocFileNum()}
+	fd := storage.FileDesc{Type: storage.TypeTable, Num: t.s.allocFileNum()}
 	fw, err := t.s.stor.Create(fd)
 	if err != nil {
 		return nil, err
@@ -422,7 +423,7 @@ func (t *tOps) remove(f *tFile) {
 		} else {
 			t.s.logf("table@remove removed @%d", f.fd.Num)
 		}
-		if t.bcache != nil {
+		if t.evictRemoved && t.bcache != nil {
 			t.bcache.EvictNS(uint64(f.fd.Num))
 		}
 	})
@@ -451,7 +452,7 @@ func newTableOps(s *session) *tOps {
 	if !s.o.GetDisableBlockCache() {
 		var bcacher cache.Cacher
 		if s.o.GetBlockCacheCapacity() > 0 {
-			bcacher = cache.NewLRU(s.o.GetBlockCacheCapacity())
+			bcacher = s.o.GetBlockCacher().New(s.o.GetBlockCacheCapacity())
 		}
 		bcache = cache.NewCache(bcacher)
 	}
@@ -459,11 +460,12 @@ func newTableOps(s *session) *tOps {
 		bpool = util.NewBufferPool(s.o.GetBlockSize() + 5)
 	}
 	return &tOps{
-		s:      s,
-		noSync: s.o.GetNoSync(),
-		cache:  cache.NewCache(cacher),
-		bcache: bcache,
-		bpool:  bpool,
+		s:            s,
+		noSync:       s.o.GetNoSync(),
+		evictRemoved: s.o.GetBlockCacheEvictRemoved(),
+		cache:        cache.NewCache(cacher),
+		bcache:       bcache,
+		bpool:        bpool,
 	}
 }
 

vendor/github.com/syndtr/goleveldb/leveldb/table/reader.go

@@ -581,6 +581,7 @@ func (r *Reader) readRawBlock(bh blockHandle, verifyChecksum bool) ([]byte, erro
 	case blockTypeSnappyCompression:
 		decLen, err := snappy.DecodedLen(data[:bh.length])
 		if err != nil {
+			r.bpool.Put(data)
 			return nil, r.newErrCorruptedBH(bh, err.Error())
 		}
 		decData := r.bpool.Get(decLen)

vendor/github.com/syndtr/goleveldb/leveldb/util.go

@@ -20,7 +20,7 @@ func shorten(str string) string {
 	return str[:3] + ".." + str[len(str)-3:]
 }
 
-var bunits = [...]string{"", "Ki", "Mi", "Gi"}
+var bunits = [...]string{"", "Ki", "Mi", "Gi", "Ti"}
 
 func shortenb(bytes int) string {
 	i := 0

vendor/github.com/syndtr/goleveldb/leveldb/util/util.go

@@ -19,7 +19,7 @@ var (
 // Releaser is the interface that wraps the basic Release method.
 type Releaser interface {
 	// Release releases associated resources. Release should always success
-	// and can be called multipe times without causing error.
+	// and can be called multiple times without causing error.
 	Release()
 }