From 3dfc82170121a05473d4521189c8823b377232a2 Mon Sep 17 00:00:00 2001 From: Julian Yap Date: Wed, 12 Dec 2018 23:42:11 -0800 Subject: [PATCH] vendor: update leveldb --- .../github.com/syndtr/goleveldb/.travis.yml | 12 - vendor/github.com/syndtr/goleveldb/README.md | 4 +- .../syndtr/goleveldb/leveldb/cache/cache.go | 1 - .../leveldb/comparer/bytes_comparer.go | 4 +- .../goleveldb/leveldb/comparer/comparer.go | 2 +- .../github.com/syndtr/goleveldb/leveldb/db.go | 98 +++++- .../syndtr/goleveldb/leveldb/db_compaction.go | 54 +++- .../syndtr/goleveldb/leveldb/db_state.go | 7 +- .../syndtr/goleveldb/leveldb/db_util.go | 2 +- .../syndtr/goleveldb/leveldb/db_write.go | 23 +- .../syndtr/goleveldb/leveldb/doc.go | 2 + .../syndtr/goleveldb/leveldb/iterator/iter.go | 6 +- .../syndtr/goleveldb/leveldb/memdb/memdb.go | 2 +- .../syndtr/goleveldb/leveldb/opt/options.go | 13 + .../syndtr/goleveldb/leveldb/session.go | 4 +- .../syndtr/goleveldb/leveldb/session_util.go | 4 +- .../syndtr/goleveldb/leveldb/storage.go | 63 ++++ .../goleveldb/leveldb/storage/file_storage.go | 294 ++++++++++++------ .../leveldb/storage/file_storage_plan9.go | 4 +- .../leveldb/storage/file_storage_unix.go | 12 + .../goleveldb/leveldb/storage/mem_storage.go | 8 +- .../goleveldb/leveldb/storage/storage.go | 8 + .../syndtr/goleveldb/leveldb/table.go | 30 +- .../syndtr/goleveldb/leveldb/table/reader.go | 1 + .../syndtr/goleveldb/leveldb/util.go | 2 +- .../syndtr/goleveldb/leveldb/util/util.go | 2 +- 26 files changed, 490 insertions(+), 172 deletions(-) delete mode 100644 vendor/github.com/syndtr/goleveldb/.travis.yml create mode 100644 vendor/github.com/syndtr/goleveldb/leveldb/storage.go diff --git a/vendor/github.com/syndtr/goleveldb/.travis.yml b/vendor/github.com/syndtr/goleveldb/.travis.yml deleted file mode 100644 index 82de37735d..0000000000 --- a/vendor/github.com/syndtr/goleveldb/.travis.yml +++ /dev/null @@ -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 diff --git a/vendor/github.com/syndtr/goleveldb/README.md b/vendor/github.com/syndtr/goleveldb/README.md index 259286f550..41a47614ca 100644 --- a/vendor/github.com/syndtr/goleveldb/README.md +++ b/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() diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/cache/cache.go b/vendor/github.com/syndtr/goleveldb/leveldb/cache/cache.go index c5940b232c..c36ad32359 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/cache/cache.go +++ b/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. diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/comparer/bytes_comparer.go b/vendor/github.com/syndtr/goleveldb/leveldb/comparer/bytes_comparer.go index 14dddf88dd..abf9fb65c7 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/comparer/bytes_comparer.go +++ b/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 } } diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/comparer/comparer.go b/vendor/github.com/syndtr/goleveldb/leveldb/comparer/comparer.go index 14a28f16fc..2c522db23b 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/comparer/comparer.go +++ b/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, diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/db.go b/vendor/github.com/syndtr/goleveldb/leveldb/db.go index a02cb2c500..b27c38d37e 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/db.go +++ 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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go b/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go index 2d0ad0753c..0c1b9a53b8 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go +++ b/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: diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/db_state.go b/vendor/github.com/syndtr/goleveldb/leveldb/db_state.go index 85b02d24bf..65e1c54bb4 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/db_state.go +++ b/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 { diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/db_util.go b/vendor/github.com/syndtr/goleveldb/leveldb/db_util.go index 7ecd960d2c..3f0654894b 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/db_util.go +++ b/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) } } diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/db_write.go b/vendor/github.com/syndtr/goleveldb/leveldb/db_write.go index cc428b695c..db0c1bece1 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/db_write.go +++ b/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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/doc.go b/vendor/github.com/syndtr/goleveldb/leveldb/doc.go index 53f13bb24c..be768e5739 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/doc.go +++ b/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() diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/iterator/iter.go b/vendor/github.com/syndtr/goleveldb/leveldb/iterator/iter.go index 3b55532746..96fb0f6859 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/iterator/iter.go +++ b/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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/memdb/memdb.go b/vendor/github.com/syndtr/goleveldb/leveldb/memdb/memdb.go index 18a19ed424..b661c08a93 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/memdb/memdb.go +++ b/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] } diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/opt/options.go b/vendor/github.com/syndtr/goleveldb/leveldb/opt/options.go index 44e7d9adce..528b164233 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/opt/options.go +++ b/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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/session.go b/vendor/github.com/syndtr/goleveldb/leveldb/session.go index ad68a87039..3f391f9346 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/session.go +++ b/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), } diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/session_util.go b/vendor/github.com/syndtr/goleveldb/leveldb/session_util.go index 92328933cc..40cb2cf957 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/session_util.go +++ b/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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/storage.go b/vendor/github.com/syndtr/goleveldb/leveldb/storage.go new file mode 100644 index 0000000000..d45fb5dfeb --- /dev/null +++ b/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 +} diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage.go b/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage.go index e53434cab7..9ba71fd6d1 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage.go +++ b/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 ¤tFile{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) { diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_plan9.go b/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_plan9.go index bab62bfcee..b829798012 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_plan9.go +++ b/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 { diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_unix.go b/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_unix.go index 7e29915379..d75f66a9ef 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage_unix.go +++ b/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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/storage/mem_storage.go b/vendor/github.com/syndtr/goleveldb/leveldb/storage/mem_storage.go index 9b0421f035..838f1bee1b 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/storage/mem_storage.go +++ b/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 { diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/storage/storage.go b/vendor/github.com/syndtr/goleveldb/leveldb/storage/storage.go index c16bce6b66..4e4a724258 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/storage/storage.go +++ b/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) diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/table.go b/vendor/github.com/syndtr/goleveldb/leveldb/table.go index 81d18a531e..1fac60d050 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/table.go +++ b/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, } } diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/table/reader.go b/vendor/github.com/syndtr/goleveldb/leveldb/table/reader.go index c5be420b3f..16cfbaa006 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/table/reader.go +++ b/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) diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/util.go b/vendor/github.com/syndtr/goleveldb/leveldb/util.go index e572a329e9..0e2b519e5c 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/util.go +++ b/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 diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/util/util.go b/vendor/github.com/syndtr/goleveldb/leveldb/util/util.go index f35976865b..80614afc58 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/util/util.go +++ b/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() }