1176 lines
		
	
	
		
			28 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			1176 lines
		
	
	
		
			28 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 (
 | 
						|
	"container/list"
 | 
						|
	"fmt"
 | 
						|
	"io"
 | 
						|
	"os"
 | 
						|
	"runtime"
 | 
						|
	"strings"
 | 
						|
	"sync"
 | 
						|
	"sync/atomic"
 | 
						|
	"time"
 | 
						|
 | 
						|
	"github.com/syndtr/goleveldb/leveldb/errors"
 | 
						|
	"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
 | 
						|
 | 
						|
	// 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
 | 
						|
 | 
						|
	// MemDB.
 | 
						|
	memMu           sync.RWMutex
 | 
						|
	memPool         chan *memdb.DB
 | 
						|
	mem, frozenMem  *memDB
 | 
						|
	journal         *journal.Writer
 | 
						|
	journalWriter   storage.Writer
 | 
						|
	journalFd       storage.FileDesc
 | 
						|
	frozenJournalFd storage.FileDesc
 | 
						|
	frozenSeq       uint64
 | 
						|
 | 
						|
	// Snapshot.
 | 
						|
	snapsMu   sync.Mutex
 | 
						|
	snapsList *list.List
 | 
						|
 | 
						|
	// Write.
 | 
						|
	batchPool    sync.Pool
 | 
						|
	writeMergeC  chan writeMerge
 | 
						|
	writeMergedC chan bool
 | 
						|
	writeLockC   chan struct{}
 | 
						|
	writeAckC    chan error
 | 
						|
	writeDelay   time.Duration
 | 
						|
	writeDelayN  int
 | 
						|
	tr           *Transaction
 | 
						|
 | 
						|
	// Compaction.
 | 
						|
	compCommitLk     sync.Mutex
 | 
						|
	tcompCmdC        chan cCmd
 | 
						|
	tcompPauseC      chan chan<- struct{}
 | 
						|
	mcompCmdC        chan cCmd
 | 
						|
	compErrC         chan error
 | 
						|
	compPerErrC      chan error
 | 
						|
	compErrSetC      chan error
 | 
						|
	compWriteLocking bool
 | 
						|
	compStats        cStats
 | 
						|
	memdbMaxLevel    int // For testing.
 | 
						|
 | 
						|
	// 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.stSeqNum,
 | 
						|
		// MemDB
 | 
						|
		memPool: make(chan *memdb.DB, 1),
 | 
						|
		// Snapshot
 | 
						|
		snapsList: list.New(),
 | 
						|
		// Write
 | 
						|
		batchPool:    sync.Pool{New: newBatch},
 | 
						|
		writeMergeC:  make(chan writeMerge),
 | 
						|
		writeMergedC: make(chan bool),
 | 
						|
		writeLockC:   make(chan struct{}, 1),
 | 
						|
		writeAckC:    make(chan error),
 | 
						|
		// Compaction
 | 
						|
		tcompCmdC:   make(chan cCmd),
 | 
						|
		tcompPauseC: make(chan chan<- struct{}),
 | 
						|
		mcompCmdC:   make(chan cCmd),
 | 
						|
		compErrC:    make(chan error),
 | 
						|
		compPerErrC: make(chan error),
 | 
						|
		compErrSetC: make(chan error),
 | 
						|
		// Close
 | 
						|
		closeC: make(chan struct{}),
 | 
						|
	}
 | 
						|
 | 
						|
	// Read-only mode.
 | 
						|
	readOnly := s.o.GetReadOnly()
 | 
						|
 | 
						|
	if readOnly {
 | 
						|
		// Recover journals (read-only mode).
 | 
						|
		if err := db.recoverJournalRO(); err != nil {
 | 
						|
			return nil, err
 | 
						|
		}
 | 
						|
	} else {
 | 
						|
		// Recover journals.
 | 
						|
		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
 | 
						|
		}
 | 
						|
 | 
						|
	}
 | 
						|
 | 
						|
	// Doesn't need to be included in the wait group.
 | 
						|
	go db.compactionError()
 | 
						|
	go db.mpoolDrain()
 | 
						|
 | 
						|
	if readOnly {
 | 
						|
		db.SetReadOnly()
 | 
						|
	} else {
 | 
						|
		db.closeW.Add(2)
 | 
						|
		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. Use errors.IsCorrupted to test whether an error is
 | 
						|
// due to corruption. Corrupted DB can be recovered with Recover function.
 | 
						|
//
 | 
						|
// The returned DB instance is safe for concurrent use.
 | 
						|
// The DB must be closed after use, by calling Close method.
 | 
						|
func Open(stor storage.Storage, o *opt.Options) (db *DB, err error) {
 | 
						|
	s, err := newSession(stor, 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() || s.o.GetReadOnly() {
 | 
						|
			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
 | 
						|
// described in the leveldb/storage package.
 | 
						|
//
 | 
						|
// OpenFile will return an error with type of ErrCorrupted if corruption
 | 
						|
// detected in the DB. Use errors.IsCorrupted to test whether an error is
 | 
						|
// due to corruption. Corrupted DB can be recovered with Recover function.
 | 
						|
//
 | 
						|
// The returned DB instance is safe for concurrent use.
 | 
						|
// 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, o.GetReadOnly())
 | 
						|
	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 returned DB instance is safe for concurrent use.
 | 
						|
// The DB must be closed after use, by calling Close method.
 | 
						|
func Recover(stor storage.Storage, o *opt.Options) (db *DB, err error) {
 | 
						|
	s, err := newSession(stor, 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 described
 | 
						|
// in the leveldb/storage package.
 | 
						|
//
 | 
						|
// The returned DB instance is safe for concurrent use.
 | 
						|
// 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, false)
 | 
						|
	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 {
 | 
						|
	o = dupOptions(o)
 | 
						|
	// Mask StrictReader, lets StrictRecovery doing its job.
 | 
						|
	o.Strict &= ^opt.StrictReader
 | 
						|
 | 
						|
	// Get all tables and sort it by file number.
 | 
						|
	fds, err := s.stor.List(storage.TypeTable)
 | 
						|
	if err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
	sortFds(fds)
 | 
						|
 | 
						|
	var (
 | 
						|
		maxSeq                                                            uint64
 | 
						|
		recoveredKey, goodKey, corruptedKey, corruptedBlock, droppedTable int
 | 
						|
 | 
						|
		// We will drop corrupted table.
 | 
						|
		strict = o.GetStrict(opt.StrictRecovery)
 | 
						|
		noSync = o.GetNoSync()
 | 
						|
 | 
						|
		rec   = &sessionRecord{}
 | 
						|
		bpool = util.NewBufferPool(o.GetBlockSize() + 5)
 | 
						|
	)
 | 
						|
	buildTable := func(iter iterator.Iterator) (tmpFd storage.FileDesc, size int64, err error) {
 | 
						|
		tmpFd = s.newTemp()
 | 
						|
		writer, err := s.stor.Create(tmpFd)
 | 
						|
		if err != nil {
 | 
						|
			return
 | 
						|
		}
 | 
						|
		defer func() {
 | 
						|
			writer.Close()
 | 
						|
			if err != nil {
 | 
						|
				s.stor.Remove(tmpFd)
 | 
						|
				tmpFd = storage.FileDesc{}
 | 
						|
			}
 | 
						|
		}()
 | 
						|
 | 
						|
		// Copy entries.
 | 
						|
		tw := table.NewWriter(writer, o)
 | 
						|
		for iter.Next() {
 | 
						|
			key := iter.Key()
 | 
						|
			if validInternalKey(key) {
 | 
						|
				err = tw.Append(key, iter.Value())
 | 
						|
				if err != nil {
 | 
						|
					return
 | 
						|
				}
 | 
						|
			}
 | 
						|
		}
 | 
						|
		err = iter.Error()
 | 
						|
		if err != nil && !errors.IsCorrupted(err) {
 | 
						|
			return
 | 
						|
		}
 | 
						|
		err = tw.Close()
 | 
						|
		if err != nil {
 | 
						|
			return
 | 
						|
		}
 | 
						|
		if !noSync {
 | 
						|
			err = writer.Sync()
 | 
						|
			if err != nil {
 | 
						|
				return
 | 
						|
			}
 | 
						|
		}
 | 
						|
		size = int64(tw.BytesLen())
 | 
						|
		return
 | 
						|
	}
 | 
						|
	recoverTable := func(fd storage.FileDesc) error {
 | 
						|
		s.logf("table@recovery recovering @%d", fd.Num)
 | 
						|
		reader, err := s.stor.Open(fd)
 | 
						|
		if err != nil {
 | 
						|
			return err
 | 
						|
		}
 | 
						|
		var closed bool
 | 
						|
		defer func() {
 | 
						|
			if !closed {
 | 
						|
				reader.Close()
 | 
						|
			}
 | 
						|
		}()
 | 
						|
 | 
						|
		// Get file size.
 | 
						|
		size, err := reader.Seek(0, 2)
 | 
						|
		if err != nil {
 | 
						|
			return err
 | 
						|
		}
 | 
						|
 | 
						|
		var (
 | 
						|
			tSeq                                     uint64
 | 
						|
			tgoodKey, tcorruptedKey, tcorruptedBlock int
 | 
						|
			imin, imax                               []byte
 | 
						|
		)
 | 
						|
		tr, err := table.NewReader(reader, size, fd, nil, bpool, o)
 | 
						|
		if err != nil {
 | 
						|
			return err
 | 
						|
		}
 | 
						|
		iter := tr.NewIterator(nil, nil)
 | 
						|
		if itererr, ok := iter.(iterator.ErrorCallbackSetter); ok {
 | 
						|
			itererr.SetErrorCallback(func(err error) {
 | 
						|
				if errors.IsCorrupted(err) {
 | 
						|
					s.logf("table@recovery block corruption @%d %q", fd.Num, err)
 | 
						|
					tcorruptedBlock++
 | 
						|
				}
 | 
						|
			})
 | 
						|
		}
 | 
						|
 | 
						|
		// Scan the table.
 | 
						|
		for iter.Next() {
 | 
						|
			key := iter.Key()
 | 
						|
			_, seq, _, kerr := parseInternalKey(key)
 | 
						|
			if kerr != nil {
 | 
						|
				tcorruptedKey++
 | 
						|
				continue
 | 
						|
			}
 | 
						|
			tgoodKey++
 | 
						|
			if seq > tSeq {
 | 
						|
				tSeq = seq
 | 
						|
			}
 | 
						|
			if imin == nil {
 | 
						|
				imin = append([]byte{}, key...)
 | 
						|
			}
 | 
						|
			imax = append(imax[:0], key...)
 | 
						|
		}
 | 
						|
		if err := iter.Error(); err != nil && !errors.IsCorrupted(err) {
 | 
						|
			iter.Release()
 | 
						|
			return err
 | 
						|
		}
 | 
						|
		iter.Release()
 | 
						|
 | 
						|
		goodKey += tgoodKey
 | 
						|
		corruptedKey += tcorruptedKey
 | 
						|
		corruptedBlock += tcorruptedBlock
 | 
						|
 | 
						|
		if strict && (tcorruptedKey > 0 || tcorruptedBlock > 0) {
 | 
						|
			droppedTable++
 | 
						|
			s.logf("table@recovery dropped @%d Gk·%d Ck·%d Cb·%d S·%d Q·%d", fd.Num, tgoodKey, tcorruptedKey, tcorruptedBlock, size, tSeq)
 | 
						|
			return nil
 | 
						|
		}
 | 
						|
 | 
						|
		if tgoodKey > 0 {
 | 
						|
			if tcorruptedKey > 0 || tcorruptedBlock > 0 {
 | 
						|
				// Rebuild the table.
 | 
						|
				s.logf("table@recovery rebuilding @%d", fd.Num)
 | 
						|
				iter := tr.NewIterator(nil, nil)
 | 
						|
				tmpFd, newSize, err := buildTable(iter)
 | 
						|
				iter.Release()
 | 
						|
				if err != nil {
 | 
						|
					return err
 | 
						|
				}
 | 
						|
				closed = true
 | 
						|
				reader.Close()
 | 
						|
				if err := s.stor.Rename(tmpFd, fd); err != nil {
 | 
						|
					return err
 | 
						|
				}
 | 
						|
				size = newSize
 | 
						|
			}
 | 
						|
			if tSeq > maxSeq {
 | 
						|
				maxSeq = tSeq
 | 
						|
			}
 | 
						|
			recoveredKey += tgoodKey
 | 
						|
			// Add table to level 0.
 | 
						|
			rec.addTable(0, fd.Num, size, imin, imax)
 | 
						|
			s.logf("table@recovery recovered @%d Gk·%d Ck·%d Cb·%d S·%d Q·%d", fd.Num, tgoodKey, tcorruptedKey, tcorruptedBlock, size, tSeq)
 | 
						|
		} else {
 | 
						|
			droppedTable++
 | 
						|
			s.logf("table@recovery unrecoverable @%d Ck·%d Cb·%d S·%d", fd.Num, tcorruptedKey, tcorruptedBlock, size)
 | 
						|
		}
 | 
						|
 | 
						|
		return nil
 | 
						|
	}
 | 
						|
 | 
						|
	// Recover all tables.
 | 
						|
	if len(fds) > 0 {
 | 
						|
		s.logf("table@recovery F·%d", len(fds))
 | 
						|
 | 
						|
		// Mark file number as used.
 | 
						|
		s.markFileNum(fds[len(fds)-1].Num)
 | 
						|
 | 
						|
		for _, fd := range fds {
 | 
						|
			if err := recoverTable(fd); err != nil {
 | 
						|
				return err
 | 
						|
			}
 | 
						|
		}
 | 
						|
 | 
						|
		s.logf("table@recovery recovered F·%d N·%d Gk·%d Ck·%d Q·%d", len(fds), recoveredKey, goodKey, corruptedKey, maxSeq)
 | 
						|
	}
 | 
						|
 | 
						|
	// Set sequence number.
 | 
						|
	rec.setSeqNum(maxSeq)
 | 
						|
 | 
						|
	// Create new manifest.
 | 
						|
	if err := s.create(); err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	// Commit.
 | 
						|
	return s.commit(rec)
 | 
						|
}
 | 
						|
 | 
						|
func (db *DB) recoverJournal() error {
 | 
						|
	// Get all journals and sort it by file number.
 | 
						|
	rawFds, err := db.s.stor.List(storage.TypeJournal)
 | 
						|
	if err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
	sortFds(rawFds)
 | 
						|
 | 
						|
	// Journals that will be recovered.
 | 
						|
	var fds []storage.FileDesc
 | 
						|
	for _, fd := range rawFds {
 | 
						|
		if fd.Num >= db.s.stJournalNum || fd.Num == db.s.stPrevJournalNum {
 | 
						|
			fds = append(fds, fd)
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	var (
 | 
						|
		ofd storage.FileDesc // Obsolete file.
 | 
						|
		rec = &sessionRecord{}
 | 
						|
	)
 | 
						|
 | 
						|
	// Recover journals.
 | 
						|
	if len(fds) > 0 {
 | 
						|
		db.logf("journal@recovery F·%d", len(fds))
 | 
						|
 | 
						|
		// Mark file number as used.
 | 
						|
		db.s.markFileNum(fds[len(fds)-1].Num)
 | 
						|
 | 
						|
		var (
 | 
						|
			// Options.
 | 
						|
			strict      = db.s.o.GetStrict(opt.StrictJournal)
 | 
						|
			checksum    = db.s.o.GetStrict(opt.StrictJournalChecksum)
 | 
						|
			writeBuffer = db.s.o.GetWriteBuffer()
 | 
						|
 | 
						|
			jr       *journal.Reader
 | 
						|
			mdb      = memdb.New(db.s.icmp, writeBuffer)
 | 
						|
			buf      = &util.Buffer{}
 | 
						|
			batchSeq uint64
 | 
						|
			batchLen int
 | 
						|
		)
 | 
						|
 | 
						|
		for _, fd := range fds {
 | 
						|
			db.logf("journal@recovery recovering @%d", fd.Num)
 | 
						|
 | 
						|
			fr, err := db.s.stor.Open(fd)
 | 
						|
			if err != nil {
 | 
						|
				return err
 | 
						|
			}
 | 
						|
 | 
						|
			// Create or reset journal reader instance.
 | 
						|
			if jr == nil {
 | 
						|
				jr = journal.NewReader(fr, dropper{db.s, fd}, strict, checksum)
 | 
						|
			} else {
 | 
						|
				jr.Reset(fr, dropper{db.s, fd}, strict, checksum)
 | 
						|
			}
 | 
						|
 | 
						|
			// Flush memdb and remove obsolete journal file.
 | 
						|
			if !ofd.Zero() {
 | 
						|
				if mdb.Len() > 0 {
 | 
						|
					if _, err := db.s.flushMemdb(rec, mdb, 0); err != nil {
 | 
						|
						fr.Close()
 | 
						|
						return err
 | 
						|
					}
 | 
						|
				}
 | 
						|
 | 
						|
				rec.setJournalNum(fd.Num)
 | 
						|
				rec.setSeqNum(db.seq)
 | 
						|
				if err := db.s.commit(rec); err != nil {
 | 
						|
					fr.Close()
 | 
						|
					return err
 | 
						|
				}
 | 
						|
				rec.resetAddedTables()
 | 
						|
 | 
						|
				db.s.stor.Remove(ofd)
 | 
						|
				ofd = storage.FileDesc{}
 | 
						|
			}
 | 
						|
 | 
						|
			// Replay journal to memdb.
 | 
						|
			mdb.Reset()
 | 
						|
			for {
 | 
						|
				r, err := jr.Next()
 | 
						|
				if err != nil {
 | 
						|
					if err == io.EOF {
 | 
						|
						break
 | 
						|
					}
 | 
						|
 | 
						|
					fr.Close()
 | 
						|
					return errors.SetFd(err, fd)
 | 
						|
				}
 | 
						|
 | 
						|
				buf.Reset()
 | 
						|
				if _, err := buf.ReadFrom(r); err != nil {
 | 
						|
					if err == io.ErrUnexpectedEOF {
 | 
						|
						// This is error returned due to corruption, with strict == false.
 | 
						|
						continue
 | 
						|
					}
 | 
						|
 | 
						|
					fr.Close()
 | 
						|
					return errors.SetFd(err, fd)
 | 
						|
				}
 | 
						|
				batchSeq, batchLen, err = decodeBatchToMem(buf.Bytes(), db.seq, mdb)
 | 
						|
				if err != nil {
 | 
						|
					if !strict && errors.IsCorrupted(err) {
 | 
						|
						db.s.logf("journal error: %v (skipped)", err)
 | 
						|
						// We won't apply sequence number as it might be corrupted.
 | 
						|
						continue
 | 
						|
					}
 | 
						|
 | 
						|
					fr.Close()
 | 
						|
					return errors.SetFd(err, fd)
 | 
						|
				}
 | 
						|
 | 
						|
				// Save sequence number.
 | 
						|
				db.seq = batchSeq + uint64(batchLen)
 | 
						|
 | 
						|
				// Flush it if large enough.
 | 
						|
				if mdb.Size() >= writeBuffer {
 | 
						|
					if _, err := db.s.flushMemdb(rec, mdb, 0); err != nil {
 | 
						|
						fr.Close()
 | 
						|
						return err
 | 
						|
					}
 | 
						|
 | 
						|
					mdb.Reset()
 | 
						|
				}
 | 
						|
			}
 | 
						|
 | 
						|
			fr.Close()
 | 
						|
			ofd = fd
 | 
						|
		}
 | 
						|
 | 
						|
		// Flush the last memdb.
 | 
						|
		if mdb.Len() > 0 {
 | 
						|
			if _, err := db.s.flushMemdb(rec, mdb, 0); err != nil {
 | 
						|
				return err
 | 
						|
			}
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	// Create a new journal.
 | 
						|
	if _, err := db.newMem(0); err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	// Commit.
 | 
						|
	rec.setJournalNum(db.journalFd.Num)
 | 
						|
	rec.setSeqNum(db.seq)
 | 
						|
	if err := db.s.commit(rec); err != nil {
 | 
						|
		// Close journal on error.
 | 
						|
		if db.journal != nil {
 | 
						|
			db.journal.Close()
 | 
						|
			db.journalWriter.Close()
 | 
						|
		}
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	// Remove the last obsolete journal file.
 | 
						|
	if !ofd.Zero() {
 | 
						|
		db.s.stor.Remove(ofd)
 | 
						|
	}
 | 
						|
 | 
						|
	return nil
 | 
						|
}
 | 
						|
 | 
						|
func (db *DB) recoverJournalRO() error {
 | 
						|
	// Get all journals and sort it by file number.
 | 
						|
	rawFds, err := db.s.stor.List(storage.TypeJournal)
 | 
						|
	if err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
	sortFds(rawFds)
 | 
						|
 | 
						|
	// Journals that will be recovered.
 | 
						|
	var fds []storage.FileDesc
 | 
						|
	for _, fd := range rawFds {
 | 
						|
		if fd.Num >= db.s.stJournalNum || fd.Num == db.s.stPrevJournalNum {
 | 
						|
			fds = append(fds, fd)
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	var (
 | 
						|
		// Options.
 | 
						|
		strict      = db.s.o.GetStrict(opt.StrictJournal)
 | 
						|
		checksum    = db.s.o.GetStrict(opt.StrictJournalChecksum)
 | 
						|
		writeBuffer = db.s.o.GetWriteBuffer()
 | 
						|
 | 
						|
		mdb = memdb.New(db.s.icmp, writeBuffer)
 | 
						|
	)
 | 
						|
 | 
						|
	// Recover journals.
 | 
						|
	if len(fds) > 0 {
 | 
						|
		db.logf("journal@recovery RO·Mode F·%d", len(fds))
 | 
						|
 | 
						|
		var (
 | 
						|
			jr       *journal.Reader
 | 
						|
			buf      = &util.Buffer{}
 | 
						|
			batchSeq uint64
 | 
						|
			batchLen int
 | 
						|
		)
 | 
						|
 | 
						|
		for _, fd := range fds {
 | 
						|
			db.logf("journal@recovery recovering @%d", fd.Num)
 | 
						|
 | 
						|
			fr, err := db.s.stor.Open(fd)
 | 
						|
			if err != nil {
 | 
						|
				return err
 | 
						|
			}
 | 
						|
 | 
						|
			// Create or reset journal reader instance.
 | 
						|
			if jr == nil {
 | 
						|
				jr = journal.NewReader(fr, dropper{db.s, fd}, strict, checksum)
 | 
						|
			} else {
 | 
						|
				jr.Reset(fr, dropper{db.s, fd}, strict, checksum)
 | 
						|
			}
 | 
						|
 | 
						|
			// Replay journal to memdb.
 | 
						|
			for {
 | 
						|
				r, err := jr.Next()
 | 
						|
				if err != nil {
 | 
						|
					if err == io.EOF {
 | 
						|
						break
 | 
						|
					}
 | 
						|
 | 
						|
					fr.Close()
 | 
						|
					return errors.SetFd(err, fd)
 | 
						|
				}
 | 
						|
 | 
						|
				buf.Reset()
 | 
						|
				if _, err := buf.ReadFrom(r); err != nil {
 | 
						|
					if err == io.ErrUnexpectedEOF {
 | 
						|
						// This is error returned due to corruption, with strict == false.
 | 
						|
						continue
 | 
						|
					}
 | 
						|
 | 
						|
					fr.Close()
 | 
						|
					return errors.SetFd(err, fd)
 | 
						|
				}
 | 
						|
				batchSeq, batchLen, err = decodeBatchToMem(buf.Bytes(), db.seq, mdb)
 | 
						|
				if err != nil {
 | 
						|
					if !strict && errors.IsCorrupted(err) {
 | 
						|
						db.s.logf("journal error: %v (skipped)", err)
 | 
						|
						// We won't apply sequence number as it might be corrupted.
 | 
						|
						continue
 | 
						|
					}
 | 
						|
 | 
						|
					fr.Close()
 | 
						|
					return errors.SetFd(err, fd)
 | 
						|
				}
 | 
						|
 | 
						|
				// Save sequence number.
 | 
						|
				db.seq = batchSeq + uint64(batchLen)
 | 
						|
			}
 | 
						|
 | 
						|
			fr.Close()
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	// Set memDB.
 | 
						|
	db.mem = &memDB{db: db, DB: mdb, ref: 1}
 | 
						|
 | 
						|
	return nil
 | 
						|
}
 | 
						|
 | 
						|
func memGet(mdb *memdb.DB, ikey internalKey, icmp *iComparer) (ok bool, mv []byte, err error) {
 | 
						|
	mk, mv, err := mdb.Find(ikey)
 | 
						|
	if err == nil {
 | 
						|
		ukey, _, kt, kerr := parseInternalKey(mk)
 | 
						|
		if kerr != nil {
 | 
						|
			// Shouldn't have had happen.
 | 
						|
			panic(kerr)
 | 
						|
		}
 | 
						|
		if icmp.uCompare(ukey, ikey.ukey()) == 0 {
 | 
						|
			if kt == keyTypeDel {
 | 
						|
				return true, nil, ErrNotFound
 | 
						|
			}
 | 
						|
			return true, mv, nil
 | 
						|
 | 
						|
		}
 | 
						|
	} else if err != ErrNotFound {
 | 
						|
		return true, nil, err
 | 
						|
	}
 | 
						|
	return
 | 
						|
}
 | 
						|
 | 
						|
func (db *DB) get(auxm *memdb.DB, auxt tFiles, key []byte, seq uint64, ro *opt.ReadOptions) (value []byte, err error) {
 | 
						|
	ikey := makeInternalKey(nil, key, seq, keyTypeSeek)
 | 
						|
 | 
						|
	if auxm != nil {
 | 
						|
		if ok, mv, me := memGet(auxm, ikey, db.s.icmp); ok {
 | 
						|
			return append([]byte{}, mv...), me
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	em, fm := db.getMems()
 | 
						|
	for _, m := range [...]*memDB{em, fm} {
 | 
						|
		if m == nil {
 | 
						|
			continue
 | 
						|
		}
 | 
						|
		defer m.decref()
 | 
						|
 | 
						|
		if ok, mv, me := memGet(m.DB, ikey, db.s.icmp); ok {
 | 
						|
			return append([]byte{}, mv...), me
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	v := db.s.version()
 | 
						|
	value, cSched, err := v.get(auxt, ikey, ro, false)
 | 
						|
	v.release()
 | 
						|
	if cSched {
 | 
						|
		// Trigger table compaction.
 | 
						|
		db.compTrigger(db.tcompCmdC)
 | 
						|
	}
 | 
						|
	return
 | 
						|
}
 | 
						|
 | 
						|
func nilIfNotFound(err error) error {
 | 
						|
	if err == ErrNotFound {
 | 
						|
		return nil
 | 
						|
	}
 | 
						|
	return err
 | 
						|
}
 | 
						|
 | 
						|
func (db *DB) has(auxm *memdb.DB, auxt tFiles, key []byte, seq uint64, ro *opt.ReadOptions) (ret bool, err error) {
 | 
						|
	ikey := makeInternalKey(nil, key, seq, keyTypeSeek)
 | 
						|
 | 
						|
	if auxm != nil {
 | 
						|
		if ok, _, me := memGet(auxm, ikey, db.s.icmp); ok {
 | 
						|
			return me == nil, nilIfNotFound(me)
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	em, fm := db.getMems()
 | 
						|
	for _, m := range [...]*memDB{em, fm} {
 | 
						|
		if m == nil {
 | 
						|
			continue
 | 
						|
		}
 | 
						|
		defer m.decref()
 | 
						|
 | 
						|
		if ok, _, me := memGet(m.DB, ikey, db.s.icmp); ok {
 | 
						|
			return me == nil, nilIfNotFound(me)
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	v := db.s.version()
 | 
						|
	_, cSched, err := v.get(auxt, ikey, ro, true)
 | 
						|
	v.release()
 | 
						|
	if cSched {
 | 
						|
		// Trigger table compaction.
 | 
						|
		db.compTrigger(db.tcompCmdC)
 | 
						|
	}
 | 
						|
	if err == nil {
 | 
						|
		ret = true
 | 
						|
	} else if err == ErrNotFound {
 | 
						|
		err = nil
 | 
						|
	}
 | 
						|
	return
 | 
						|
}
 | 
						|
 | 
						|
// Get gets the value for the given key. It returns ErrNotFound if the
 | 
						|
// DB does not contains the key.
 | 
						|
//
 | 
						|
// The returned slice is its own copy, it is safe to modify the contents
 | 
						|
// of the returned slice.
 | 
						|
// It is safe to modify the contents of the argument after Get returns.
 | 
						|
func (db *DB) Get(key []byte, ro *opt.ReadOptions) (value []byte, err error) {
 | 
						|
	err = db.ok()
 | 
						|
	if err != nil {
 | 
						|
		return
 | 
						|
	}
 | 
						|
 | 
						|
	se := db.acquireSnapshot()
 | 
						|
	defer db.releaseSnapshot(se)
 | 
						|
	return db.get(nil, nil, key, se.seq, ro)
 | 
						|
}
 | 
						|
 | 
						|
// Has returns true if the DB does contains the given key.
 | 
						|
//
 | 
						|
// 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 {
 | 
						|
		return
 | 
						|
	}
 | 
						|
 | 
						|
	se := db.acquireSnapshot()
 | 
						|
	defer db.releaseSnapshot(se)
 | 
						|
	return db.has(nil, nil, key, se.seq, ro)
 | 
						|
}
 | 
						|
 | 
						|
// NewIterator returns an iterator for the latest snapshot of the
 | 
						|
// underlying DB.
 | 
						|
// The returned iterator is not safe for concurrent use, 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 (db *DB) NewIterator(slice *util.Range, ro *opt.ReadOptions) iterator.Iterator {
 | 
						|
	if err := db.ok(); err != nil {
 | 
						|
		return iterator.NewEmptyIterator(err)
 | 
						|
	}
 | 
						|
 | 
						|
	se := db.acquireSnapshot()
 | 
						|
	defer db.releaseSnapshot(se)
 | 
						|
	// Iterator holds 'version' lock, 'version' is immutable so snapshot
 | 
						|
	// can be released after iterator created.
 | 
						|
	return db.newIterator(nil, nil, se.seq, 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 (db *DB) GetSnapshot() (*Snapshot, error) {
 | 
						|
	if err := db.ok(); err != nil {
 | 
						|
		return nil, err
 | 
						|
	}
 | 
						|
 | 
						|
	return db.newSnapshot(), nil
 | 
						|
}
 | 
						|
 | 
						|
// GetProperty returns value of the given property name.
 | 
						|
//
 | 
						|
// Property names:
 | 
						|
//	leveldb.num-files-at-level{n}
 | 
						|
//		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
 | 
						|
//		Returns block pool stats.
 | 
						|
//	leveldb.cachedblock
 | 
						|
//		Returns size of cached block.
 | 
						|
//	leveldb.openedtables
 | 
						|
//		Returns number of opened tables.
 | 
						|
//	leveldb.alivesnaps
 | 
						|
//		Returns number of alive snapshots.
 | 
						|
//	leveldb.aliveiters
 | 
						|
//		Returns number of alive iterators.
 | 
						|
func (db *DB) GetProperty(name string) (value string, err error) {
 | 
						|
	err = db.ok()
 | 
						|
	if err != nil {
 | 
						|
		return
 | 
						|
	}
 | 
						|
 | 
						|
	const prefix = "leveldb."
 | 
						|
	if !strings.HasPrefix(name, prefix) {
 | 
						|
		return "", ErrNotFound
 | 
						|
	}
 | 
						|
	p := name[len(prefix):]
 | 
						|
 | 
						|
	v := db.s.version()
 | 
						|
	defer v.release()
 | 
						|
 | 
						|
	numFilesPrefix := "num-files-at-level"
 | 
						|
	switch {
 | 
						|
	case strings.HasPrefix(p, numFilesPrefix):
 | 
						|
		var level uint
 | 
						|
		var rest string
 | 
						|
		n, _ := fmt.Sscanf(p[len(numFilesPrefix):], "%d%s", &level, &rest)
 | 
						|
		if n != 1 {
 | 
						|
			err = ErrNotFound
 | 
						|
		} 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, tables := range v.levels {
 | 
						|
			duration, read, write := db.compStats.getStat(level)
 | 
						|
			if len(tables) == 0 && duration == 0 {
 | 
						|
				continue
 | 
						|
			}
 | 
						|
			value += fmt.Sprintf(" %3d   | %10d | %13.5f | %13.5f | %13.5f | %13.5f\n",
 | 
						|
				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)
 | 
						|
			for _, t := range tables {
 | 
						|
				value += fmt.Sprintf("%d:%d[%q .. %q]\n", t.fd.Num, t.size, t.imin, t.imax)
 | 
						|
			}
 | 
						|
		}
 | 
						|
	case p == "blockpool":
 | 
						|
		value = fmt.Sprintf("%v", db.s.tops.bpool)
 | 
						|
	case p == "cachedblock":
 | 
						|
		if db.s.tops.bcache != nil {
 | 
						|
			value = fmt.Sprintf("%d", db.s.tops.bcache.Size())
 | 
						|
		} else {
 | 
						|
			value = "<nil>"
 | 
						|
		}
 | 
						|
	case p == "openedtables":
 | 
						|
		value = fmt.Sprintf("%d", db.s.tops.cache.Size())
 | 
						|
	case p == "alivesnaps":
 | 
						|
		value = fmt.Sprintf("%d", atomic.LoadInt32(&db.aliveSnaps))
 | 
						|
	case p == "aliveiters":
 | 
						|
		value = fmt.Sprintf("%d", atomic.LoadInt32(&db.aliveIters))
 | 
						|
	default:
 | 
						|
		err = ErrNotFound
 | 
						|
	}
 | 
						|
 | 
						|
	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
 | 
						|
// 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 (db *DB) SizeOf(ranges []util.Range) (Sizes, error) {
 | 
						|
	if err := db.ok(); err != nil {
 | 
						|
		return nil, err
 | 
						|
	}
 | 
						|
 | 
						|
	v := db.s.version()
 | 
						|
	defer v.release()
 | 
						|
 | 
						|
	sizes := make(Sizes, 0, len(ranges))
 | 
						|
	for _, r := range ranges {
 | 
						|
		imin := makeInternalKey(nil, r.Start, keyMaxSeq, keyTypeSeek)
 | 
						|
		imax := makeInternalKey(nil, r.Limit, keyMaxSeq, keyTypeSeek)
 | 
						|
		start, err := v.offsetOf(imin)
 | 
						|
		if err != nil {
 | 
						|
			return nil, err
 | 
						|
		}
 | 
						|
		limit, err := v.offsetOf(imax)
 | 
						|
		if err != nil {
 | 
						|
			return nil, err
 | 
						|
		}
 | 
						|
		var size int64
 | 
						|
		if limit >= start {
 | 
						|
			size = limit - start
 | 
						|
		}
 | 
						|
		sizes = append(sizes, size)
 | 
						|
	}
 | 
						|
 | 
						|
	return sizes, nil
 | 
						|
}
 | 
						|
 | 
						|
// Close closes the DB. This will also releases any outstanding snapshot,
 | 
						|
// abort any in-flight compaction and discard open transaction.
 | 
						|
//
 | 
						|
// 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 (db *DB) Close() error {
 | 
						|
	if !db.setClosed() {
 | 
						|
		return ErrClosed
 | 
						|
	}
 | 
						|
 | 
						|
	start := time.Now()
 | 
						|
	db.log("db@close closing")
 | 
						|
 | 
						|
	// Clear the finalizer.
 | 
						|
	runtime.SetFinalizer(db, nil)
 | 
						|
 | 
						|
	// Get compaction error.
 | 
						|
	var err error
 | 
						|
	select {
 | 
						|
	case err = <-db.compErrC:
 | 
						|
		if err == ErrReadOnly {
 | 
						|
			err = nil
 | 
						|
		}
 | 
						|
	default:
 | 
						|
	}
 | 
						|
 | 
						|
	// Signal all goroutines.
 | 
						|
	close(db.closeC)
 | 
						|
 | 
						|
	// Discard open transaction.
 | 
						|
	if db.tr != nil {
 | 
						|
		db.tr.Discard()
 | 
						|
	}
 | 
						|
 | 
						|
	// Acquire writer lock.
 | 
						|
	db.writeLockC <- struct{}{}
 | 
						|
 | 
						|
	// Wait for all gorotines to exit.
 | 
						|
	db.closeW.Wait()
 | 
						|
 | 
						|
	// Closes journal.
 | 
						|
	if db.journal != nil {
 | 
						|
		db.journal.Close()
 | 
						|
		db.journalWriter.Close()
 | 
						|
		db.journal = nil
 | 
						|
		db.journalWriter = nil
 | 
						|
	}
 | 
						|
 | 
						|
	if db.writeDelayN > 0 {
 | 
						|
		db.logf("db@write was delayed N·%d T·%v", db.writeDelayN, db.writeDelay)
 | 
						|
	}
 | 
						|
 | 
						|
	// Close session.
 | 
						|
	db.s.close()
 | 
						|
	db.logf("db@close done T·%v", time.Since(start))
 | 
						|
	db.s.release()
 | 
						|
 | 
						|
	if db.closer != nil {
 | 
						|
		if err1 := db.closer.Close(); err == nil {
 | 
						|
			err = err1
 | 
						|
		}
 | 
						|
		db.closer = nil
 | 
						|
	}
 | 
						|
 | 
						|
	// Clear memdbs.
 | 
						|
	db.clearMems()
 | 
						|
 | 
						|
	return err
 | 
						|
}
 |