mirror of https://github.com/poanetwork/quorum.git
756 lines
17 KiB
Go
756 lines
17 KiB
Go
|
// Copyright (c) 2012, Suryandaru Triandana <syndtr@gmail.com>
|
||
|
// 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 (
|
||
|
"errors"
|
||
|
"fmt"
|
||
|
"io"
|
||
|
"os"
|
||
|
"runtime"
|
||
|
"strings"
|
||
|
"sync"
|
||
|
"time"
|
||
|
|
||
|
"github.com/syndtr/goleveldb/leveldb/iterator"
|
||
|
"github.com/syndtr/goleveldb/leveldb/journal"
|
||
|
"github.com/syndtr/goleveldb/leveldb/memdb"
|
||
|
"github.com/syndtr/goleveldb/leveldb/opt"
|
||
|
"github.com/syndtr/goleveldb/leveldb/storage"
|
||
|
"github.com/syndtr/goleveldb/leveldb/table"
|
||
|
"github.com/syndtr/goleveldb/leveldb/util"
|
||
|
)
|
||
|
|
||
|
// DB is a LevelDB database.
|
||
|
type DB struct {
|
||
|
// Need 64-bit alignment.
|
||
|
seq uint64
|
||
|
|
||
|
s *session
|
||
|
|
||
|
// MemDB
|
||
|
memMu sync.RWMutex
|
||
|
mem *memdb.DB
|
||
|
frozenMem *memdb.DB
|
||
|
journal *journal.Writer
|
||
|
journalWriter storage.Writer
|
||
|
journalFile storage.File
|
||
|
frozenJournalFile storage.File
|
||
|
frozenSeq uint64
|
||
|
|
||
|
// Snapshot
|
||
|
snapsMu sync.Mutex
|
||
|
snapsRoot snapshotElement
|
||
|
|
||
|
// Write
|
||
|
writeC chan *Batch
|
||
|
writeMergedC chan bool
|
||
|
writeLockC chan struct{}
|
||
|
writeAckC chan error
|
||
|
journalC chan *Batch
|
||
|
journalAckC chan error
|
||
|
|
||
|
// Compaction
|
||
|
tcompCmdC chan cCmd
|
||
|
tcompPauseC chan chan<- struct{}
|
||
|
tcompTriggerC chan struct{}
|
||
|
mcompCmdC chan cCmd
|
||
|
mcompTriggerC chan struct{}
|
||
|
compErrC chan error
|
||
|
compErrSetC chan error
|
||
|
compStats [kNumLevels]cStats
|
||
|
|
||
|
// Close
|
||
|
closeW sync.WaitGroup
|
||
|
closeC chan struct{}
|
||
|
closed uint32
|
||
|
closer io.Closer
|
||
|
}
|
||
|
|
||
|
func openDB(s *session) (*DB, error) {
|
||
|
s.log("db@open opening")
|
||
|
start := time.Now()
|
||
|
db := &DB{
|
||
|
s: s,
|
||
|
// Initial sequence
|
||
|
seq: s.stSeq,
|
||
|
// Write
|
||
|
writeC: make(chan *Batch),
|
||
|
writeMergedC: make(chan bool),
|
||
|
writeLockC: make(chan struct{}, 1),
|
||
|
writeAckC: make(chan error),
|
||
|
journalC: make(chan *Batch),
|
||
|
journalAckC: make(chan error),
|
||
|
// Compaction
|
||
|
tcompCmdC: make(chan cCmd),
|
||
|
tcompPauseC: make(chan chan<- struct{}),
|
||
|
tcompTriggerC: make(chan struct{}, 1),
|
||
|
mcompCmdC: make(chan cCmd),
|
||
|
mcompTriggerC: make(chan struct{}, 1),
|
||
|
compErrC: make(chan error),
|
||
|
compErrSetC: make(chan error),
|
||
|
// Close
|
||
|
closeC: make(chan struct{}),
|
||
|
}
|
||
|
db.initSnapshot()
|
||
|
|
||
|
if err := db.recoverJournal(); err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
// Remove any obsolete files.
|
||
|
if err := db.checkAndCleanFiles(); err != nil {
|
||
|
// Close journal.
|
||
|
if db.journal != nil {
|
||
|
db.journal.Close()
|
||
|
db.journalWriter.Close()
|
||
|
}
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
// Don't include compaction error goroutine into wait group.
|
||
|
go db.compactionError()
|
||
|
|
||
|
db.closeW.Add(3)
|
||
|
go db.tCompaction()
|
||
|
go db.mCompaction()
|
||
|
go db.jWriter()
|
||
|
|
||
|
s.logf("db@open done T·%v", time.Since(start))
|
||
|
|
||
|
runtime.SetFinalizer(db, (*DB).Close)
|
||
|
return db, nil
|
||
|
}
|
||
|
|
||
|
// Open opens or creates a DB for the given storage.
|
||
|
// The DB will be created if not exist, unless ErrorIfMissing is true.
|
||
|
// Also, if ErrorIfExist is true and the DB exist Open will returns
|
||
|
// os.ErrExist error.
|
||
|
//
|
||
|
// Open will return an error with type of ErrCorrupted if corruption
|
||
|
// detected in the DB. Corrupted DB can be recovered with Recover
|
||
|
// function.
|
||
|
//
|
||
|
// The DB must be closed after use, by calling Close method.
|
||
|
func Open(p storage.Storage, o *opt.Options) (db *DB, err error) {
|
||
|
s, err := newSession(p, o)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
defer func() {
|
||
|
if err != nil {
|
||
|
s.close()
|
||
|
s.release()
|
||
|
}
|
||
|
}()
|
||
|
|
||
|
err = s.recover()
|
||
|
if err != nil {
|
||
|
if !os.IsNotExist(err) || s.o.GetErrorIfMissing() {
|
||
|
return
|
||
|
}
|
||
|
err = s.create()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
} else if s.o.GetErrorIfExist() {
|
||
|
err = os.ErrExist
|
||
|
return
|
||
|
}
|
||
|
|
||
|
return openDB(s)
|
||
|
}
|
||
|
|
||
|
// OpenFile opens or creates a DB for the given path.
|
||
|
// The DB will be created if not exist, unless ErrorIfMissing is true.
|
||
|
// Also, if ErrorIfExist is true and the DB exist OpenFile will returns
|
||
|
// os.ErrExist error.
|
||
|
//
|
||
|
// OpenFile uses standard file-system backed storage implementation as
|
||
|
// desribed in the leveldb/storage package.
|
||
|
//
|
||
|
// OpenFile will return an error with type of ErrCorrupted if corruption
|
||
|
// detected in the DB. Corrupted DB can be recovered with Recover
|
||
|
// function.
|
||
|
//
|
||
|
// The DB must be closed after use, by calling Close method.
|
||
|
func OpenFile(path string, o *opt.Options) (db *DB, err error) {
|
||
|
stor, err := storage.OpenFile(path)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
db, err = Open(stor, o)
|
||
|
if err != nil {
|
||
|
stor.Close()
|
||
|
} else {
|
||
|
db.closer = stor
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Recover recovers and opens a DB with missing or corrupted manifest files
|
||
|
// for the given storage. It will ignore any manifest files, valid or not.
|
||
|
// The DB must already exist or it will returns an error.
|
||
|
// Also, Recover will ignore ErrorIfMissing and ErrorIfExist options.
|
||
|
//
|
||
|
// The DB must be closed after use, by calling Close method.
|
||
|
func Recover(p storage.Storage, o *opt.Options) (db *DB, err error) {
|
||
|
s, err := newSession(p, o)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
defer func() {
|
||
|
if err != nil {
|
||
|
s.close()
|
||
|
s.release()
|
||
|
}
|
||
|
}()
|
||
|
|
||
|
err = recoverTable(s, o)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
return openDB(s)
|
||
|
}
|
||
|
|
||
|
// RecoverFile recovers and opens a DB with missing or corrupted manifest files
|
||
|
// for the given path. It will ignore any manifest files, valid or not.
|
||
|
// The DB must already exist or it will returns an error.
|
||
|
// Also, Recover will ignore ErrorIfMissing and ErrorIfExist options.
|
||
|
//
|
||
|
// RecoverFile uses standard file-system backed storage implementation as desribed
|
||
|
// in the leveldb/storage package.
|
||
|
//
|
||
|
// The DB must be closed after use, by calling Close method.
|
||
|
func RecoverFile(path string, o *opt.Options) (db *DB, err error) {
|
||
|
stor, err := storage.OpenFile(path)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
db, err = Recover(stor, o)
|
||
|
if err != nil {
|
||
|
stor.Close()
|
||
|
} else {
|
||
|
db.closer = stor
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
func recoverTable(s *session, o *opt.Options) error {
|
||
|
ff0, err := s.getFiles(storage.TypeTable)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
ff1 := files(ff0)
|
||
|
ff1.sort()
|
||
|
|
||
|
var mSeq uint64
|
||
|
var good, corrupted int
|
||
|
rec := new(sessionRecord)
|
||
|
buildTable := func(iter iterator.Iterator) (tmp storage.File, size int64, err error) {
|
||
|
tmp = s.newTemp()
|
||
|
writer, err := tmp.Create()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
defer func() {
|
||
|
writer.Close()
|
||
|
if err != nil {
|
||
|
tmp.Remove()
|
||
|
tmp = nil
|
||
|
}
|
||
|
}()
|
||
|
tw := table.NewWriter(writer, o)
|
||
|
// Copy records.
|
||
|
for iter.Next() {
|
||
|
key := iter.Key()
|
||
|
if validIkey(key) {
|
||
|
err = tw.Append(key, iter.Value())
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
err = iter.Error()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
err = tw.Close()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
err = writer.Sync()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
size = int64(tw.BytesLen())
|
||
|
return
|
||
|
}
|
||
|
recoverTable := func(file storage.File) error {
|
||
|
s.logf("table@recovery recovering @%d", file.Num())
|
||
|
reader, err := file.Open()
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
defer reader.Close()
|
||
|
// Get file size.
|
||
|
size, err := reader.Seek(0, 2)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
var tSeq uint64
|
||
|
var tgood, tcorrupted, blockerr int
|
||
|
var min, max []byte
|
||
|
tr := table.NewReader(reader, size, nil, o)
|
||
|
iter := tr.NewIterator(nil, nil)
|
||
|
iter.(iterator.ErrorCallbackSetter).SetErrorCallback(func(err error) {
|
||
|
s.logf("table@recovery found error @%d %q", file.Num(), err)
|
||
|
blockerr++
|
||
|
})
|
||
|
// Scan the table.
|
||
|
for iter.Next() {
|
||
|
key := iter.Key()
|
||
|
_, seq, _, ok := parseIkey(key)
|
||
|
if !ok {
|
||
|
tcorrupted++
|
||
|
continue
|
||
|
}
|
||
|
tgood++
|
||
|
if seq > tSeq {
|
||
|
tSeq = seq
|
||
|
}
|
||
|
if min == nil {
|
||
|
min = append([]byte{}, key...)
|
||
|
}
|
||
|
max = append(max[:0], key...)
|
||
|
}
|
||
|
if err := iter.Error(); err != nil {
|
||
|
iter.Release()
|
||
|
return err
|
||
|
}
|
||
|
iter.Release()
|
||
|
if tgood > 0 {
|
||
|
if tcorrupted > 0 || blockerr > 0 {
|
||
|
// Rebuild the table.
|
||
|
s.logf("table@recovery rebuilding @%d", file.Num())
|
||
|
iter := tr.NewIterator(nil, nil)
|
||
|
tmp, newSize, err := buildTable(iter)
|
||
|
iter.Release()
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
reader.Close()
|
||
|
if err := file.Replace(tmp); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
size = newSize
|
||
|
}
|
||
|
if tSeq > mSeq {
|
||
|
mSeq = tSeq
|
||
|
}
|
||
|
// Add table to level 0.
|
||
|
rec.addTable(0, file.Num(), uint64(size), min, max)
|
||
|
s.logf("table@recovery recovered @%d N·%d C·%d B·%d S·%d Q·%d", file.Num(), tgood, tcorrupted, blockerr, size, tSeq)
|
||
|
} else {
|
||
|
s.logf("table@recovery unrecoverable @%d C·%d B·%d S·%d", file.Num(), tcorrupted, blockerr, size)
|
||
|
}
|
||
|
|
||
|
good += tgood
|
||
|
corrupted += tcorrupted
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
// Recover all tables.
|
||
|
if len(ff1) > 0 {
|
||
|
s.logf("table@recovery F·%d", len(ff1))
|
||
|
s.markFileNum(ff1[len(ff1)-1].Num())
|
||
|
for _, file := range ff1 {
|
||
|
if err := recoverTable(file); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
s.logf("table@recovery recovered F·%d N·%d C·%d Q·%d", len(ff1), good, corrupted, mSeq)
|
||
|
}
|
||
|
// Set sequence number.
|
||
|
rec.setSeq(mSeq + 1)
|
||
|
// Create new manifest.
|
||
|
if err := s.create(); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
// Commit.
|
||
|
return s.commit(rec)
|
||
|
}
|
||
|
|
||
|
func (d *DB) recoverJournal() error {
|
||
|
s := d.s
|
||
|
|
||
|
ff0, err := s.getFiles(storage.TypeJournal)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
ff1 := files(ff0)
|
||
|
ff1.sort()
|
||
|
ff2 := make([]storage.File, 0, len(ff1))
|
||
|
for _, file := range ff1 {
|
||
|
if file.Num() >= s.stJournalNum || file.Num() == s.stPrevJournalNum {
|
||
|
s.markFileNum(file.Num())
|
||
|
ff2 = append(ff2, file)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
var jr *journal.Reader
|
||
|
var of storage.File
|
||
|
var mem *memdb.DB
|
||
|
batch := new(Batch)
|
||
|
cm := newCMem(s)
|
||
|
buf := new(util.Buffer)
|
||
|
// Options.
|
||
|
strict := s.o.GetStrict(opt.StrictJournal)
|
||
|
checksum := s.o.GetStrict(opt.StrictJournalChecksum)
|
||
|
writeBuffer := s.o.GetWriteBuffer()
|
||
|
recoverJournal := func(file storage.File) error {
|
||
|
s.logf("journal@recovery recovering @%d", file.Num())
|
||
|
reader, err := file.Open()
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
defer reader.Close()
|
||
|
if jr == nil {
|
||
|
jr = journal.NewReader(reader, dropper{s, file}, strict, checksum)
|
||
|
} else {
|
||
|
jr.Reset(reader, dropper{s, file}, strict, checksum)
|
||
|
}
|
||
|
if of != nil {
|
||
|
if mem.Len() > 0 {
|
||
|
if err := cm.flush(mem, 0); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
if err := cm.commit(file.Num(), d.seq); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
cm.reset()
|
||
|
of.Remove()
|
||
|
of = nil
|
||
|
}
|
||
|
// Reset memdb.
|
||
|
mem.Reset()
|
||
|
for {
|
||
|
r, err := jr.Next()
|
||
|
if err != nil {
|
||
|
if err == io.EOF {
|
||
|
break
|
||
|
}
|
||
|
return err
|
||
|
}
|
||
|
buf.Reset()
|
||
|
if _, err := buf.ReadFrom(r); err != nil {
|
||
|
if strict {
|
||
|
return err
|
||
|
}
|
||
|
continue
|
||
|
}
|
||
|
if err := batch.decode(buf.Bytes()); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
if err := batch.memReplay(mem); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
d.seq = batch.seq + uint64(batch.len())
|
||
|
if mem.Size() >= writeBuffer {
|
||
|
// Large enough, flush it.
|
||
|
if err := cm.flush(mem, 0); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
// Reset memdb.
|
||
|
mem.Reset()
|
||
|
}
|
||
|
}
|
||
|
of = file
|
||
|
return nil
|
||
|
}
|
||
|
// Recover all journals.
|
||
|
if len(ff2) > 0 {
|
||
|
s.logf("journal@recovery F·%d", len(ff2))
|
||
|
mem = memdb.New(s.icmp, writeBuffer)
|
||
|
for _, file := range ff2 {
|
||
|
if err := recoverJournal(file); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
// Flush the last journal.
|
||
|
if mem.Len() > 0 {
|
||
|
if err := cm.flush(mem, 0); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
// Create a new journal.
|
||
|
if _, err := d.newMem(0); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
// Commit.
|
||
|
if err := cm.commit(d.journalFile.Num(), d.seq); err != nil {
|
||
|
// Close journal.
|
||
|
if d.journal != nil {
|
||
|
d.journal.Close()
|
||
|
d.journalWriter.Close()
|
||
|
}
|
||
|
return err
|
||
|
}
|
||
|
// Remove the last journal.
|
||
|
if of != nil {
|
||
|
of.Remove()
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (d *DB) get(key []byte, seq uint64, ro *opt.ReadOptions) (value []byte, err error) {
|
||
|
s := d.s
|
||
|
|
||
|
ikey := newIKey(key, seq, tSeek)
|
||
|
|
||
|
em, fm := d.getMems()
|
||
|
for _, m := range [...]*memdb.DB{em, fm} {
|
||
|
if m == nil {
|
||
|
continue
|
||
|
}
|
||
|
mk, mv, me := m.Find(ikey)
|
||
|
if me == nil {
|
||
|
ukey, _, t, ok := parseIkey(mk)
|
||
|
if ok && s.icmp.uCompare(ukey, key) == 0 {
|
||
|
if t == tDel {
|
||
|
return nil, ErrNotFound
|
||
|
}
|
||
|
return mv, nil
|
||
|
}
|
||
|
} else if me != ErrNotFound {
|
||
|
return nil, me
|
||
|
}
|
||
|
}
|
||
|
|
||
|
v := s.version()
|
||
|
value, cSched, err := v.get(ikey, ro)
|
||
|
v.release()
|
||
|
if cSched {
|
||
|
// Trigger table compaction.
|
||
|
d.compTrigger(d.tcompTriggerC)
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Get gets the value for the given key. It returns ErrNotFound if the
|
||
|
// DB does not contain the key.
|
||
|
//
|
||
|
// The caller should not modify the contents of the returned slice, but
|
||
|
// it is safe to modify the contents of the argument after Get returns.
|
||
|
func (d *DB) Get(key []byte, ro *opt.ReadOptions) (value []byte, err error) {
|
||
|
err = d.ok()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
return d.get(key, d.getSeq(), ro)
|
||
|
}
|
||
|
|
||
|
// NewIterator returns an iterator for the latest snapshot of the
|
||
|
// uderlying DB.
|
||
|
// The returned iterator is not goroutine-safe, but it is safe to use
|
||
|
// multiple iterators concurrently, with each in a dedicated goroutine.
|
||
|
// It is also safe to use an iterator concurrently with modifying its
|
||
|
// underlying DB. The resultant key/value pairs are guaranteed to be
|
||
|
// consistent.
|
||
|
//
|
||
|
// Slice allows slicing the iterator to only contains keys in the given
|
||
|
// range. A nil Range.Start is treated as a key before all keys in the
|
||
|
// DB. And a nil Range.Limit is treated as a key after all keys in
|
||
|
// the DB.
|
||
|
//
|
||
|
// The iterator must be released after use, by calling Release method.
|
||
|
//
|
||
|
// Also read Iterator documentation of the leveldb/iterator package.
|
||
|
func (d *DB) NewIterator(slice *util.Range, ro *opt.ReadOptions) iterator.Iterator {
|
||
|
if err := d.ok(); err != nil {
|
||
|
return iterator.NewEmptyIterator(err)
|
||
|
}
|
||
|
|
||
|
p := d.newSnapshot()
|
||
|
defer p.Release()
|
||
|
return p.NewIterator(slice, ro)
|
||
|
}
|
||
|
|
||
|
// GetSnapshot returns a latest snapshot of the underlying DB. A snapshot
|
||
|
// is a frozen snapshot of a DB state at a particular point in time. The
|
||
|
// content of snapshot are guaranteed to be consistent.
|
||
|
//
|
||
|
// The snapshot must be released after use, by calling Release method.
|
||
|
func (d *DB) GetSnapshot() (*Snapshot, error) {
|
||
|
if err := d.ok(); err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
return d.newSnapshot(), nil
|
||
|
}
|
||
|
|
||
|
// GetProperty returns value of the given property name.
|
||
|
//
|
||
|
// Property names:
|
||
|
// leveldb.num-files-at-level{n}
|
||
|
// Returns the number of filer at level 'n'.
|
||
|
// leveldb.stats
|
||
|
// Returns statistics of the underlying DB.
|
||
|
// leveldb.sstables
|
||
|
// Returns sstables list for each level.
|
||
|
func (d *DB) GetProperty(name string) (value string, err error) {
|
||
|
err = d.ok()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
const prefix = "leveldb."
|
||
|
if !strings.HasPrefix(name, prefix) {
|
||
|
return "", errors.New("leveldb: GetProperty: unknown property: " + name)
|
||
|
}
|
||
|
|
||
|
p := name[len(prefix):]
|
||
|
|
||
|
s := d.s
|
||
|
v := s.version()
|
||
|
defer v.release()
|
||
|
|
||
|
switch {
|
||
|
case strings.HasPrefix(p, "num-files-at-level"):
|
||
|
var level uint
|
||
|
var rest string
|
||
|
n, _ := fmt.Scanf("%d%s", &level, &rest)
|
||
|
if n != 1 || level >= kNumLevels {
|
||
|
err = errors.New("leveldb: GetProperty: invalid property: " + name)
|
||
|
} else {
|
||
|
value = fmt.Sprint(v.tLen(int(level)))
|
||
|
}
|
||
|
case p == "stats":
|
||
|
value = "Compactions\n" +
|
||
|
" Level | Tables | Size(MB) | Time(sec) | Read(MB) | Write(MB)\n" +
|
||
|
"-------+------------+---------------+---------------+---------------+---------------\n"
|
||
|
for level, tt := range v.tables {
|
||
|
duration, read, write := d.compStats[level].get()
|
||
|
if len(tt) == 0 && duration == 0 {
|
||
|
continue
|
||
|
}
|
||
|
value += fmt.Sprintf(" %3d | %10d | %13.5f | %13.5f | %13.5f | %13.5f\n",
|
||
|
level, len(tt), float64(tt.size())/1048576.0, duration.Seconds(),
|
||
|
float64(read)/1048576.0, float64(write)/1048576.0)
|
||
|
}
|
||
|
case p == "sstables":
|
||
|
for level, tt := range v.tables {
|
||
|
value += fmt.Sprintf("--- level %d ---\n", level)
|
||
|
for _, t := range tt {
|
||
|
value += fmt.Sprintf("%d:%d[%q .. %q]\n", t.file.Num(), t.size, t.min, t.max)
|
||
|
}
|
||
|
}
|
||
|
default:
|
||
|
err = errors.New("leveldb: GetProperty: unknown property: " + name)
|
||
|
}
|
||
|
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// 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
|
||
|
// data compresses by a factor of ten, the returned sizes will be one-tenth
|
||
|
// the size of the corresponding user data size.
|
||
|
// The results may not include the sizes of recently written data.
|
||
|
func (d *DB) SizeOf(ranges []util.Range) (Sizes, error) {
|
||
|
if err := d.ok(); err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
v := d.s.version()
|
||
|
defer v.release()
|
||
|
|
||
|
sizes := make(Sizes, 0, len(ranges))
|
||
|
for _, r := range ranges {
|
||
|
min := newIKey(r.Start, kMaxSeq, tSeek)
|
||
|
max := newIKey(r.Limit, kMaxSeq, tSeek)
|
||
|
start, err := v.offsetOf(min)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
limit, err := v.offsetOf(max)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
var size uint64
|
||
|
if limit >= start {
|
||
|
size = limit - start
|
||
|
}
|
||
|
sizes = append(sizes, size)
|
||
|
}
|
||
|
|
||
|
return sizes, nil
|
||
|
}
|
||
|
|
||
|
// Close closes the DB. This will also releases any outstanding snapshot.
|
||
|
//
|
||
|
// It is not safe to close a DB until all outstanding iterators are released.
|
||
|
// It is valid to call Close multiple times. Other methods should not be
|
||
|
// called after the DB has been closed.
|
||
|
func (d *DB) Close() error {
|
||
|
if !d.setClosed() {
|
||
|
return ErrClosed
|
||
|
}
|
||
|
|
||
|
s := d.s
|
||
|
start := time.Now()
|
||
|
s.log("db@close closing")
|
||
|
|
||
|
// Clear the finalizer.
|
||
|
runtime.SetFinalizer(d, nil)
|
||
|
|
||
|
// Get compaction error.
|
||
|
var err error
|
||
|
select {
|
||
|
case err = <-d.compErrC:
|
||
|
default:
|
||
|
}
|
||
|
|
||
|
close(d.closeC)
|
||
|
|
||
|
// Wait for the close WaitGroup.
|
||
|
d.closeW.Wait()
|
||
|
|
||
|
// Close journal.
|
||
|
if d.journal != nil {
|
||
|
d.journal.Close()
|
||
|
d.journalWriter.Close()
|
||
|
}
|
||
|
|
||
|
// Close session.
|
||
|
s.close()
|
||
|
s.logf("db@close done T·%v", time.Since(start))
|
||
|
s.release()
|
||
|
|
||
|
if d.closer != nil {
|
||
|
if err1 := d.closer.Close(); err == nil {
|
||
|
err = err1
|
||
|
}
|
||
|
}
|
||
|
|
||
|
d.s = nil
|
||
|
d.mem = nil
|
||
|
d.frozenMem = nil
|
||
|
d.journal = nil
|
||
|
d.journalWriter = nil
|
||
|
d.journalFile = nil
|
||
|
d.frozenJournalFile = nil
|
||
|
d.snapsRoot = snapshotElement{}
|
||
|
d.closer = nil
|
||
|
|
||
|
return err
|
||
|
}
|