+++ /dev/null
-// Copyright (c) 2015-2016 The btcsuite developers
-// Use of this source code is governed by an ISC
-// license that can be found in the LICENSE file.
-
-package ffldb
-
-import (
- "bytes"
- "encoding/binary"
- "fmt"
- "os"
- "path/filepath"
- "runtime"
- "sort"
- "sync"
-
- "github.com/btcsuite/btcd/chaincfg/chainhash"
- "github.com/btcsuite/btcd/database"
- "github.com/btcsuite/btcd/database/internal/treap"
- "github.com/btcsuite/btcd/wire"
- "github.com/btcsuite/btcutil"
- "github.com/btcsuite/goleveldb/leveldb"
- "github.com/btcsuite/goleveldb/leveldb/comparer"
- ldberrors "github.com/btcsuite/goleveldb/leveldb/errors"
- "github.com/btcsuite/goleveldb/leveldb/filter"
- "github.com/btcsuite/goleveldb/leveldb/iterator"
- "github.com/btcsuite/goleveldb/leveldb/opt"
- "github.com/btcsuite/goleveldb/leveldb/util"
-)
-
-const (
- // metadataDbName is the name used for the metadata database.
- metadataDbName = "metadata"
-
- // blockHdrSize is the size of a block header. This is simply the
- // constant from wire and is only provided here for convenience since
- // wire.MaxBlockHeaderPayload is quite long.
- blockHdrSize = wire.MaxBlockHeaderPayload
-
- // blockHdrOffset defines the offsets into a block index row for the
- // block header.
- //
- // The serialized block index row format is:
- // <blocklocation><blockheader>
- blockHdrOffset = blockLocSize
-)
-
-var (
- // byteOrder is the preferred byte order used through the database and
- // block files. Sometimes big endian will be used to allow ordered byte
- // sortable integer values.
- byteOrder = binary.LittleEndian
-
- // bucketIndexPrefix is the prefix used for all entries in the bucket
- // index.
- bucketIndexPrefix = []byte("bidx")
-
- // curBucketIDKeyName is the name of the key used to keep track of the
- // current bucket ID counter.
- curBucketIDKeyName = []byte("bidx-cbid")
-
- // metadataBucketID is the ID of the top-level metadata bucket.
- // It is the value 0 encoded as an unsigned big-endian uint32.
- metadataBucketID = [4]byte{}
-
- // blockIdxBucketID is the ID of the internal block metadata bucket.
- // It is the value 1 encoded as an unsigned big-endian uint32.
- blockIdxBucketID = [4]byte{0x00, 0x00, 0x00, 0x01}
-
- // blockIdxBucketName is the bucket used internally to track block
- // metadata.
- blockIdxBucketName = []byte("ffldb-blockidx")
-
- // writeLocKeyName is the key used to store the current write file
- // location.
- writeLocKeyName = []byte("ffldb-writeloc")
-)
-
-// Common error strings.
-const (
- // errDbNotOpenStr is the text to use for the database.ErrDbNotOpen
- // error code.
- errDbNotOpenStr = "database is not open"
-
- // errTxClosedStr is the text to use for the database.ErrTxClosed error
- // code.
- errTxClosedStr = "database tx is closed"
-)
-
-// bulkFetchData is allows a block location to be specified along with the
-// index it was requested from. This in turn allows the bulk data loading
-// functions to sort the data accesses based on the location to improve
-// performance while keeping track of which result the data is for.
-type bulkFetchData struct {
- *blockLocation
- replyIndex int
-}
-
-// bulkFetchDataSorter implements sort.Interface to allow a slice of
-// bulkFetchData to be sorted. In particular it sorts by file and then
-// offset so that reads from files are grouped and linear.
-type bulkFetchDataSorter []bulkFetchData
-
-// Len returns the number of items in the slice. It is part of the
-// sort.Interface implementation.
-func (s bulkFetchDataSorter) Len() int {
- return len(s)
-}
-
-// Swap swaps the items at the passed indices. It is part of the
-// sort.Interface implementation.
-func (s bulkFetchDataSorter) Swap(i, j int) {
- s[i], s[j] = s[j], s[i]
-}
-
-// Less returns whether the item with index i should sort before the item with
-// index j. It is part of the sort.Interface implementation.
-func (s bulkFetchDataSorter) Less(i, j int) bool {
- if s[i].blockFileNum < s[j].blockFileNum {
- return true
- }
- if s[i].blockFileNum > s[j].blockFileNum {
- return false
- }
-
- return s[i].fileOffset < s[j].fileOffset
-}
-
-// makeDbErr creates a database.Error given a set of arguments.
-func makeDbErr(c database.ErrorCode, desc string, err error) database.Error {
- return database.Error{ErrorCode: c, Description: desc, Err: err}
-}
-
-// convertErr converts the passed leveldb error into a database error with an
-// equivalent error code and the passed description. It also sets the passed
-// error as the underlying error.
-func convertErr(desc string, ldbErr error) database.Error {
- // Use the driver-specific error code by default. The code below will
- // update this with the converted error if it's recognized.
- var code = database.ErrDriverSpecific
-
- switch {
- // Database corruption errors.
- case ldberrors.IsCorrupted(ldbErr):
- code = database.ErrCorruption
-
- // Database open/create errors.
- case ldbErr == leveldb.ErrClosed:
- code = database.ErrDbNotOpen
-
- // Transaction errors.
- case ldbErr == leveldb.ErrSnapshotReleased:
- code = database.ErrTxClosed
- case ldbErr == leveldb.ErrIterReleased:
- code = database.ErrTxClosed
- }
-
- return database.Error{ErrorCode: code, Description: desc, Err: ldbErr}
-}
-
-// copySlice returns a copy of the passed slice. This is mostly used to copy
-// leveldb iterator keys and values since they are only valid until the iterator
-// is moved instead of during the entirety of the transaction.
-func copySlice(slice []byte) []byte {
- ret := make([]byte, len(slice))
- copy(ret, slice)
- return ret
-}
-
-// cursor is an internal type used to represent a cursor over key/value pairs
-// and nested buckets of a bucket and implements the database.Cursor interface.
-type cursor struct {
- bucket *bucket
- dbIter iterator.Iterator
- pendingIter iterator.Iterator
- currentIter iterator.Iterator
-}
-
-// Enforce cursor implements the database.Cursor interface.
-var _ database.Cursor = (*cursor)(nil)
-
-// Bucket returns the bucket the cursor was created for.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Bucket() database.Bucket {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return nil
- }
-
- return c.bucket
-}
-
-// Delete removes the current key/value pair the cursor is at without
-// invalidating the cursor.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrIncompatibleValue if attempted when the cursor points to a nested
-// bucket
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Delete() error {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return err
- }
-
- // Error if the cursor is exhausted.
- if c.currentIter == nil {
- str := "cursor is exhausted"
- return makeDbErr(database.ErrIncompatibleValue, str, nil)
- }
-
- // Do not allow buckets to be deleted via the cursor.
- key := c.currentIter.Key()
- if bytes.HasPrefix(key, bucketIndexPrefix) {
- str := "buckets may not be deleted from a cursor"
- return makeDbErr(database.ErrIncompatibleValue, str, nil)
- }
-
- c.bucket.tx.deleteKey(copySlice(key), true)
- return nil
-}
-
-// skipPendingUpdates skips any keys at the current database iterator position
-// that are being updated by the transaction. The forwards flag indicates the
-// direction the cursor is moving.
-func (c *cursor) skipPendingUpdates(forwards bool) {
- for c.dbIter.Valid() {
- var skip bool
- key := c.dbIter.Key()
- if c.bucket.tx.pendingRemove.Has(key) {
- skip = true
- } else if c.bucket.tx.pendingKeys.Has(key) {
- skip = true
- }
- if !skip {
- break
- }
-
- if forwards {
- c.dbIter.Next()
- } else {
- c.dbIter.Prev()
- }
- }
-}
-
-// chooseIterator first skips any entries in the database iterator that are
-// being updated by the transaction and sets the current iterator to the
-// appropriate iterator depending on their validity and the order they compare
-// in while taking into account the direction flag. When the cursor is being
-// moved forwards and both iterators are valid, the iterator with the smaller
-// key is chosen and vice versa when the cursor is being moved backwards.
-func (c *cursor) chooseIterator(forwards bool) bool {
- // Skip any keys at the current database iterator position that are
- // being updated by the transaction.
- c.skipPendingUpdates(forwards)
-
- // When both iterators are exhausted, the cursor is exhausted too.
- if !c.dbIter.Valid() && !c.pendingIter.Valid() {
- c.currentIter = nil
- return false
- }
-
- // Choose the database iterator when the pending keys iterator is
- // exhausted.
- if !c.pendingIter.Valid() {
- c.currentIter = c.dbIter
- return true
- }
-
- // Choose the pending keys iterator when the database iterator is
- // exhausted.
- if !c.dbIter.Valid() {
- c.currentIter = c.pendingIter
- return true
- }
-
- // Both iterators are valid, so choose the iterator with either the
- // smaller or larger key depending on the forwards flag.
- compare := bytes.Compare(c.dbIter.Key(), c.pendingIter.Key())
- if (forwards && compare > 0) || (!forwards && compare < 0) {
- c.currentIter = c.pendingIter
- } else {
- c.currentIter = c.dbIter
- }
- return true
-}
-
-// First positions the cursor at the first key/value pair and returns whether or
-// not the pair exists.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) First() bool {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return false
- }
-
- // Seek to the first key in both the database and pending iterators and
- // choose the iterator that is both valid and has the smaller key.
- c.dbIter.First()
- c.pendingIter.First()
- return c.chooseIterator(true)
-}
-
-// Last positions the cursor at the last key/value pair and returns whether or
-// not the pair exists.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Last() bool {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return false
- }
-
- // Seek to the last key in both the database and pending iterators and
- // choose the iterator that is both valid and has the larger key.
- c.dbIter.Last()
- c.pendingIter.Last()
- return c.chooseIterator(false)
-}
-
-// Next moves the cursor one key/value pair forward and returns whether or not
-// the pair exists.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Next() bool {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return false
- }
-
- // Nothing to return if cursor is exhausted.
- if c.currentIter == nil {
- return false
- }
-
- // Move the current iterator to the next entry and choose the iterator
- // that is both valid and has the smaller key.
- c.currentIter.Next()
- return c.chooseIterator(true)
-}
-
-// Prev moves the cursor one key/value pair backward and returns whether or not
-// the pair exists.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Prev() bool {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return false
- }
-
- // Nothing to return if cursor is exhausted.
- if c.currentIter == nil {
- return false
- }
-
- // Move the current iterator to the previous entry and choose the
- // iterator that is both valid and has the larger key.
- c.currentIter.Prev()
- return c.chooseIterator(false)
-}
-
-// Seek positions the cursor at the first key/value pair that is greater than or
-// equal to the passed seek key. Returns false if no suitable key was found.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Seek(seek []byte) bool {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return false
- }
-
- // Seek to the provided key in both the database and pending iterators
- // then choose the iterator that is both valid and has the larger key.
- seekKey := bucketizedKey(c.bucket.id, seek)
- c.dbIter.Seek(seekKey)
- c.pendingIter.Seek(seekKey)
- return c.chooseIterator(true)
-}
-
-// rawKey returns the current key the cursor is pointing to without stripping
-// the current bucket prefix or bucket index prefix.
-func (c *cursor) rawKey() []byte {
- // Nothing to return if cursor is exhausted.
- if c.currentIter == nil {
- return nil
- }
-
- return copySlice(c.currentIter.Key())
-}
-
-// Key returns the current key the cursor is pointing to.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Key() []byte {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return nil
- }
-
- // Nothing to return if cursor is exhausted.
- if c.currentIter == nil {
- return nil
- }
-
- // Slice out the actual key name and make a copy since it is no longer
- // valid after iterating to the next item.
- //
- // The key is after the bucket index prefix and parent ID when the
- // cursor is pointing to a nested bucket.
- key := c.currentIter.Key()
- if bytes.HasPrefix(key, bucketIndexPrefix) {
- key = key[len(bucketIndexPrefix)+4:]
- return copySlice(key)
- }
-
- // The key is after the bucket ID when the cursor is pointing to a
- // normal entry.
- key = key[len(c.bucket.id):]
- return copySlice(key)
-}
-
-// rawValue returns the current value the cursor is pointing to without
-// stripping without filtering bucket index values.
-func (c *cursor) rawValue() []byte {
- // Nothing to return if cursor is exhausted.
- if c.currentIter == nil {
- return nil
- }
-
- return copySlice(c.currentIter.Value())
-}
-
-// Value returns the current value the cursor is pointing to. This will be nil
-// for nested buckets.
-//
-// This function is part of the database.Cursor interface implementation.
-func (c *cursor) Value() []byte {
- // Ensure transaction state is valid.
- if err := c.bucket.tx.checkClosed(); err != nil {
- return nil
- }
-
- // Nothing to return if cursor is exhausted.
- if c.currentIter == nil {
- return nil
- }
-
- // Return nil for the value when the cursor is pointing to a nested
- // bucket.
- if bytes.HasPrefix(c.currentIter.Key(), bucketIndexPrefix) {
- return nil
- }
-
- return copySlice(c.currentIter.Value())
-}
-
-// cursorType defines the type of cursor to create.
-type cursorType int
-
-// The following constants define the allowed cursor types.
-const (
- // ctKeys iterates through all of the keys in a given bucket.
- ctKeys cursorType = iota
-
- // ctBuckets iterates through all directly nested buckets in a given
- // bucket.
- ctBuckets
-
- // ctFull iterates through both the keys and the directly nested buckets
- // in a given bucket.
- ctFull
-)
-
-// cursorFinalizer is either invoked when a cursor is being garbage collected or
-// called manually to ensure the underlying cursor iterators are released.
-func cursorFinalizer(c *cursor) {
- c.dbIter.Release()
- c.pendingIter.Release()
-}
-
-// newCursor returns a new cursor for the given bucket, bucket ID, and cursor
-// type.
-//
-// NOTE: The caller is responsible for calling the cursorFinalizer function on
-// the returned cursor.
-func newCursor(b *bucket, bucketID []byte, cursorTyp cursorType) *cursor {
- var dbIter, pendingIter iterator.Iterator
- switch cursorTyp {
- case ctKeys:
- keyRange := util.BytesPrefix(bucketID)
- dbIter = b.tx.snapshot.NewIterator(keyRange)
- pendingKeyIter := newLdbTreapIter(b.tx, keyRange)
- pendingIter = pendingKeyIter
-
- case ctBuckets:
- // The serialized bucket index key format is:
- // <bucketindexprefix><parentbucketid><bucketname>
-
- // Create an iterator for the both the database and the pending
- // keys which are prefixed by the bucket index identifier and
- // the provided bucket ID.
- prefix := make([]byte, len(bucketIndexPrefix)+4)
- copy(prefix, bucketIndexPrefix)
- copy(prefix[len(bucketIndexPrefix):], bucketID)
- bucketRange := util.BytesPrefix(prefix)
-
- dbIter = b.tx.snapshot.NewIterator(bucketRange)
- pendingBucketIter := newLdbTreapIter(b.tx, bucketRange)
- pendingIter = pendingBucketIter
-
- case ctFull:
- fallthrough
- default:
- // The serialized bucket index key format is:
- // <bucketindexprefix><parentbucketid><bucketname>
- prefix := make([]byte, len(bucketIndexPrefix)+4)
- copy(prefix, bucketIndexPrefix)
- copy(prefix[len(bucketIndexPrefix):], bucketID)
- bucketRange := util.BytesPrefix(prefix)
- keyRange := util.BytesPrefix(bucketID)
-
- // Since both keys and buckets are needed from the database,
- // create an individual iterator for each prefix and then create
- // a merged iterator from them.
- dbKeyIter := b.tx.snapshot.NewIterator(keyRange)
- dbBucketIter := b.tx.snapshot.NewIterator(bucketRange)
- iters := []iterator.Iterator{dbKeyIter, dbBucketIter}
- dbIter = iterator.NewMergedIterator(iters,
- comparer.DefaultComparer, true)
-
- // Since both keys and buckets are needed from the pending keys,
- // create an individual iterator for each prefix and then create
- // a merged iterator from them.
- pendingKeyIter := newLdbTreapIter(b.tx, keyRange)
- pendingBucketIter := newLdbTreapIter(b.tx, bucketRange)
- iters = []iterator.Iterator{pendingKeyIter, pendingBucketIter}
- pendingIter = iterator.NewMergedIterator(iters,
- comparer.DefaultComparer, true)
- }
-
- // Create the cursor using the iterators.
- return &cursor{bucket: b, dbIter: dbIter, pendingIter: pendingIter}
-}
-
-// bucket is an internal type used to represent a collection of key/value pairs
-// and implements the database.Bucket interface.
-type bucket struct {
- tx *transaction
- id [4]byte
-}
-
-// Enforce bucket implements the database.Bucket interface.
-var _ database.Bucket = (*bucket)(nil)
-
-// bucketIndexKey returns the actual key to use for storing and retrieving a
-// child bucket in the bucket index. This is required because additional
-// information is needed to distinguish nested buckets with the same name.
-func bucketIndexKey(parentID [4]byte, key []byte) []byte {
- // The serialized bucket index key format is:
- // <bucketindexprefix><parentbucketid><bucketname>
- indexKey := make([]byte, len(bucketIndexPrefix)+4+len(key))
- copy(indexKey, bucketIndexPrefix)
- copy(indexKey[len(bucketIndexPrefix):], parentID[:])
- copy(indexKey[len(bucketIndexPrefix)+4:], key)
- return indexKey
-}
-
-// bucketizedKey returns the actual key to use for storing and retrieving a key
-// for the provided bucket ID. This is required because bucketizing is handled
-// through the use of a unique prefix per bucket.
-func bucketizedKey(bucketID [4]byte, key []byte) []byte {
- // The serialized block index key format is:
- // <bucketid><key>
- bKey := make([]byte, 4+len(key))
- copy(bKey, bucketID[:])
- copy(bKey[4:], key)
- return bKey
-}
-
-// Bucket retrieves a nested bucket with the given key. Returns nil if
-// the bucket does not exist.
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) Bucket(key []byte) database.Bucket {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return nil
- }
-
- // Attempt to fetch the ID for the child bucket. The bucket does not
- // exist if the bucket index entry does not exist.
- childID := b.tx.fetchKey(bucketIndexKey(b.id, key))
- if childID == nil {
- return nil
- }
-
- childBucket := &bucket{tx: b.tx}
- copy(childBucket.id[:], childID)
- return childBucket
-}
-
-// CreateBucket creates and returns a new nested bucket with the given key.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBucketExists if the bucket already exists
-// - ErrBucketNameRequired if the key is empty
-// - ErrIncompatibleValue if the key is otherwise invalid for the particular
-// implementation
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) CreateBucket(key []byte) (database.Bucket, error) {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // Ensure the transaction is writable.
- if !b.tx.writable {
- str := "create bucket requires a writable database transaction"
- return nil, makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Ensure a key was provided.
- if len(key) == 0 {
- str := "create bucket requires a key"
- return nil, makeDbErr(database.ErrBucketNameRequired, str, nil)
- }
-
- // Ensure bucket does not already exist.
- bidxKey := bucketIndexKey(b.id, key)
- if b.tx.hasKey(bidxKey) {
- str := "bucket already exists"
- return nil, makeDbErr(database.ErrBucketExists, str, nil)
- }
-
- // Find the appropriate next bucket ID to use for the new bucket. In
- // the case of the special internal block index, keep the fixed ID.
- var childID [4]byte
- if b.id == metadataBucketID && bytes.Equal(key, blockIdxBucketName) {
- childID = blockIdxBucketID
- } else {
- var err error
- childID, err = b.tx.nextBucketID()
- if err != nil {
- return nil, err
- }
- }
-
- // Add the new bucket to the bucket index.
- if err := b.tx.putKey(bidxKey, childID[:]); err != nil {
- str := fmt.Sprintf("failed to create bucket with key %q", key)
- return nil, convertErr(str, err)
- }
- return &bucket{tx: b.tx, id: childID}, nil
-}
-
-// CreateBucketIfNotExists creates and returns a new nested bucket with the
-// given key if it does not already exist.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBucketNameRequired if the key is empty
-// - ErrIncompatibleValue if the key is otherwise invalid for the particular
-// implementation
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) CreateBucketIfNotExists(key []byte) (database.Bucket, error) {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // Ensure the transaction is writable.
- if !b.tx.writable {
- str := "create bucket requires a writable database transaction"
- return nil, makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Return existing bucket if it already exists, otherwise create it.
- if bucket := b.Bucket(key); bucket != nil {
- return bucket, nil
- }
- return b.CreateBucket(key)
-}
-
-// DeleteBucket removes a nested bucket with the given key.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBucketNotFound if the specified bucket does not exist
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) DeleteBucket(key []byte) error {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return err
- }
-
- // Ensure the transaction is writable.
- if !b.tx.writable {
- str := "delete bucket requires a writable database transaction"
- return makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Attempt to fetch the ID for the child bucket. The bucket does not
- // exist if the bucket index entry does not exist. In the case of the
- // special internal block index, keep the fixed ID.
- bidxKey := bucketIndexKey(b.id, key)
- childID := b.tx.fetchKey(bidxKey)
- if childID == nil {
- str := fmt.Sprintf("bucket %q does not exist", key)
- return makeDbErr(database.ErrBucketNotFound, str, nil)
- }
-
- // Remove all nested buckets and their keys.
- childIDs := [][]byte{childID}
- for len(childIDs) > 0 {
- childID = childIDs[len(childIDs)-1]
- childIDs = childIDs[:len(childIDs)-1]
-
- // Delete all keys in the nested bucket.
- keyCursor := newCursor(b, childID, ctKeys)
- for ok := keyCursor.First(); ok; ok = keyCursor.Next() {
- b.tx.deleteKey(keyCursor.rawKey(), false)
- }
- cursorFinalizer(keyCursor)
-
- // Iterate through all nested buckets.
- bucketCursor := newCursor(b, childID, ctBuckets)
- for ok := bucketCursor.First(); ok; ok = bucketCursor.Next() {
- // Push the id of the nested bucket onto the stack for
- // the next iteration.
- childID := bucketCursor.rawValue()
- childIDs = append(childIDs, childID)
-
- // Remove the nested bucket from the bucket index.
- b.tx.deleteKey(bucketCursor.rawKey(), false)
- }
- cursorFinalizer(bucketCursor)
- }
-
- // Remove the nested bucket from the bucket index. Any buckets nested
- // under it were already removed above.
- b.tx.deleteKey(bidxKey, true)
- return nil
-}
-
-// Cursor returns a new cursor, allowing for iteration over the bucket's
-// key/value pairs and nested buckets in forward or backward order.
-//
-// You must seek to a position using the First, Last, or Seek functions before
-// calling the Next, Prev, Key, or Value functions. Failure to do so will
-// result in the same return values as an exhausted cursor, which is false for
-// the Prev and Next functions and nil for Key and Value functions.
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) Cursor() database.Cursor {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return &cursor{bucket: b}
- }
-
- // Create the cursor and setup a runtime finalizer to ensure the
- // iterators are released when the cursor is garbage collected.
- c := newCursor(b, b.id[:], ctFull)
- runtime.SetFinalizer(c, cursorFinalizer)
- return c
-}
-
-// ForEach invokes the passed function with every key/value pair in the bucket.
-// This does not include nested buckets or the key/value pairs within those
-// nested buckets.
-//
-// WARNING: It is not safe to mutate data while iterating with this method.
-// Doing so may cause the underlying cursor to be invalidated and return
-// unexpected keys and/or values.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrTxClosed if the transaction has already been closed
-//
-// NOTE: The values returned by this function are only valid during a
-// transaction. Attempting to access them after a transaction has ended will
-// likely result in an access violation.
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) ForEach(fn func(k, v []byte) error) error {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return err
- }
-
- // Invoke the callback for each cursor item. Return the error returned
- // from the callback when it is non-nil.
- c := newCursor(b, b.id[:], ctKeys)
- defer cursorFinalizer(c)
- for ok := c.First(); ok; ok = c.Next() {
- err := fn(c.Key(), c.Value())
- if err != nil {
- return err
- }
- }
-
- return nil
-}
-
-// ForEachBucket invokes the passed function with the key of every nested bucket
-// in the current bucket. This does not include any nested buckets within those
-// nested buckets.
-//
-// WARNING: It is not safe to mutate data while iterating with this method.
-// Doing so may cause the underlying cursor to be invalidated and return
-// unexpected keys.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrTxClosed if the transaction has already been closed
-//
-// NOTE: The values returned by this function are only valid during a
-// transaction. Attempting to access them after a transaction has ended will
-// likely result in an access violation.
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) ForEachBucket(fn func(k []byte) error) error {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return err
- }
-
- // Invoke the callback for each cursor item. Return the error returned
- // from the callback when it is non-nil.
- c := newCursor(b, b.id[:], ctBuckets)
- defer cursorFinalizer(c)
- for ok := c.First(); ok; ok = c.Next() {
- err := fn(c.Key())
- if err != nil {
- return err
- }
- }
-
- return nil
-}
-
-// Writable returns whether or not the bucket is writable.
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) Writable() bool {
- return b.tx.writable
-}
-
-// Put saves the specified key/value pair to the bucket. Keys that do not
-// already exist are added and keys that already exist are overwritten.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrKeyRequired if the key is empty
-// - ErrIncompatibleValue if the key is the same as an existing bucket
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) Put(key, value []byte) error {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return err
- }
-
- // Ensure the transaction is writable.
- if !b.tx.writable {
- str := "setting a key requires a writable database transaction"
- return makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Ensure a key was provided.
- if len(key) == 0 {
- str := "put requires a key"
- return makeDbErr(database.ErrKeyRequired, str, nil)
- }
-
- return b.tx.putKey(bucketizedKey(b.id, key), value)
-}
-
-// Get returns the value for the given key. Returns nil if the key does not
-// exist in this bucket. An empty slice is returned for keys that exist but
-// have no value assigned.
-//
-// NOTE: The value returned by this function is only valid during a transaction.
-// Attempting to access it after a transaction has ended results in undefined
-// behavior. Additionally, the value must NOT be modified by the caller.
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) Get(key []byte) []byte {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return nil
- }
-
- // Nothing to return if there is no key.
- if len(key) == 0 {
- return nil
- }
-
- return b.tx.fetchKey(bucketizedKey(b.id, key))
-}
-
-// Delete removes the specified key from the bucket. Deleting a key that does
-// not exist does not return an error.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrKeyRequired if the key is empty
-// - ErrIncompatibleValue if the key is the same as an existing bucket
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Bucket interface implementation.
-func (b *bucket) Delete(key []byte) error {
- // Ensure transaction state is valid.
- if err := b.tx.checkClosed(); err != nil {
- return err
- }
-
- // Ensure the transaction is writable.
- if !b.tx.writable {
- str := "deleting a value requires a writable database transaction"
- return makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Nothing to do if there is no key.
- if len(key) == 0 {
- return nil
- }
-
- b.tx.deleteKey(bucketizedKey(b.id, key), true)
- return nil
-}
-
-// pendingBlock houses a block that will be written to disk when the database
-// transaction is committed.
-type pendingBlock struct {
- hash *chainhash.Hash
- bytes []byte
-}
-
-// transaction represents a database transaction. It can either be read-only or
-// read-write and implements the database.Bucket interface. The transaction
-// provides a root bucket against which all read and writes occur.
-type transaction struct {
- managed bool // Is the transaction managed?
- closed bool // Is the transaction closed?
- writable bool // Is the transaction writable?
- db *db // DB instance the tx was created from.
- snapshot *dbCacheSnapshot // Underlying snapshot for txns.
- metaBucket *bucket // The root metadata bucket.
- blockIdxBucket *bucket // The block index bucket.
-
- // Blocks that need to be stored on commit. The pendingBlocks map is
- // kept to allow quick lookups of pending data by block hash.
- pendingBlocks map[chainhash.Hash]int
- pendingBlockData []pendingBlock
-
- // Keys that need to be stored or deleted on commit.
- pendingKeys *treap.Mutable
- pendingRemove *treap.Mutable
-
- // Active iterators that need to be notified when the pending keys have
- // been updated so the cursors can properly handle updates to the
- // transaction state.
- activeIterLock sync.RWMutex
- activeIters []*treap.Iterator
-}
-
-// Enforce transaction implements the database.Tx interface.
-var _ database.Tx = (*transaction)(nil)
-
-// removeActiveIter removes the passed iterator from the list of active
-// iterators against the pending keys treap.
-func (tx *transaction) removeActiveIter(iter *treap.Iterator) {
- // An indexing for loop is intentionally used over a range here as range
- // does not reevaluate the slice on each iteration nor does it adjust
- // the index for the modified slice.
- tx.activeIterLock.Lock()
- for i := 0; i < len(tx.activeIters); i++ {
- if tx.activeIters[i] == iter {
- copy(tx.activeIters[i:], tx.activeIters[i+1:])
- tx.activeIters[len(tx.activeIters)-1] = nil
- tx.activeIters = tx.activeIters[:len(tx.activeIters)-1]
- }
- }
- tx.activeIterLock.Unlock()
-}
-
-// addActiveIter adds the passed iterator to the list of active iterators for
-// the pending keys treap.
-func (tx *transaction) addActiveIter(iter *treap.Iterator) {
- tx.activeIterLock.Lock()
- tx.activeIters = append(tx.activeIters, iter)
- tx.activeIterLock.Unlock()
-}
-
-// notifyActiveIters notifies all of the active iterators for the pending keys
-// treap that it has been updated.
-func (tx *transaction) notifyActiveIters() {
- tx.activeIterLock.RLock()
- for _, iter := range tx.activeIters {
- iter.ForceReseek()
- }
- tx.activeIterLock.RUnlock()
-}
-
-// checkClosed returns an error if the the database or transaction is closed.
-func (tx *transaction) checkClosed() error {
- // The transaction is no longer valid if it has been closed.
- if tx.closed {
- return makeDbErr(database.ErrTxClosed, errTxClosedStr, nil)
- }
-
- return nil
-}
-
-// hasKey returns whether or not the provided key exists in the database while
-// taking into account the current transaction state.
-func (tx *transaction) hasKey(key []byte) bool {
- // When the transaction is writable, check the pending transaction
- // state first.
- if tx.writable {
- if tx.pendingRemove.Has(key) {
- return false
- }
- if tx.pendingKeys.Has(key) {
- return true
- }
- }
-
- // Consult the database cache and underlying database.
- return tx.snapshot.Has(key)
-}
-
-// putKey adds the provided key to the list of keys to be updated in the
-// database when the transaction is committed.
-//
-// NOTE: This function must only be called on a writable transaction. Since it
-// is an internal helper function, it does not check.
-func (tx *transaction) putKey(key, value []byte) error {
- // Prevent the key from being deleted if it was previously scheduled
- // to be deleted on transaction commit.
- tx.pendingRemove.Delete(key)
-
- // Add the key/value pair to the list to be written on transaction
- // commit.
- tx.pendingKeys.Put(key, value)
- tx.notifyActiveIters()
- return nil
-}
-
-// fetchKey attempts to fetch the provided key from the database cache (and
-// hence underlying database) while taking into account the current transaction
-// state. Returns nil if the key does not exist.
-func (tx *transaction) fetchKey(key []byte) []byte {
- // When the transaction is writable, check the pending transaction
- // state first.
- if tx.writable {
- if tx.pendingRemove.Has(key) {
- return nil
- }
- if value := tx.pendingKeys.Get(key); value != nil {
- return value
- }
- }
-
- // Consult the database cache and underlying database.
- return tx.snapshot.Get(key)
-}
-
-// deleteKey adds the provided key to the list of keys to be deleted from the
-// database when the transaction is committed. The notify iterators flag is
-// useful to delay notifying iterators about the changes during bulk deletes.
-//
-// NOTE: This function must only be called on a writable transaction. Since it
-// is an internal helper function, it does not check.
-func (tx *transaction) deleteKey(key []byte, notifyIterators bool) {
- // Remove the key from the list of pendings keys to be written on
- // transaction commit if needed.
- tx.pendingKeys.Delete(key)
-
- // Add the key to the list to be deleted on transaction commit.
- tx.pendingRemove.Put(key, nil)
-
- // Notify the active iterators about the change if the flag is set.
- if notifyIterators {
- tx.notifyActiveIters()
- }
-}
-
-// nextBucketID returns the next bucket ID to use for creating a new bucket.
-//
-// NOTE: This function must only be called on a writable transaction. Since it
-// is an internal helper function, it does not check.
-func (tx *transaction) nextBucketID() ([4]byte, error) {
- // Load the currently highest used bucket ID.
- curIDBytes := tx.fetchKey(curBucketIDKeyName)
- curBucketNum := binary.BigEndian.Uint32(curIDBytes)
-
- // Increment and update the current bucket ID and return it.
- var nextBucketID [4]byte
- binary.BigEndian.PutUint32(nextBucketID[:], curBucketNum+1)
- if err := tx.putKey(curBucketIDKeyName, nextBucketID[:]); err != nil {
- return [4]byte{}, err
- }
- return nextBucketID, nil
-}
-
-// Metadata returns the top-most bucket for all metadata storage.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) Metadata() database.Bucket {
- return tx.metaBucket
-}
-
-// hasBlock returns whether or not a block with the given hash exists.
-func (tx *transaction) hasBlock(hash *chainhash.Hash) bool {
- // Return true if the block is pending to be written on commit since
- // it exists from the viewpoint of this transaction.
- if _, exists := tx.pendingBlocks[*hash]; exists {
- return true
- }
-
- return tx.hasKey(bucketizedKey(blockIdxBucketID, hash[:]))
-}
-
-// StoreBlock stores the provided block into the database. There are no checks
-// to ensure the block connects to a previous block, contains double spends, or
-// any additional functionality such as transaction indexing. It simply stores
-// the block in the database.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockExists when the block hash already exists
-// - ErrTxNotWritable if attempted against a read-only transaction
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) StoreBlock(block *btcutil.Block) error {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return err
- }
-
- // Ensure the transaction is writable.
- if !tx.writable {
- str := "store block requires a writable database transaction"
- return makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Reject the block if it already exists.
- blockHash := block.Hash()
- if tx.hasBlock(blockHash) {
- str := fmt.Sprintf("block %s already exists", blockHash)
- return makeDbErr(database.ErrBlockExists, str, nil)
- }
-
- blockBytes, err := block.Bytes()
- if err != nil {
- str := fmt.Sprintf("failed to get serialized bytes for block %s",
- blockHash)
- return makeDbErr(database.ErrDriverSpecific, str, err)
- }
-
- // Add the block to be stored to the list of pending blocks to store
- // when the transaction is committed. Also, add it to pending blocks
- // map so it is easy to determine the block is pending based on the
- // block hash.
- if tx.pendingBlocks == nil {
- tx.pendingBlocks = make(map[chainhash.Hash]int)
- }
- tx.pendingBlocks[*blockHash] = len(tx.pendingBlockData)
- tx.pendingBlockData = append(tx.pendingBlockData, pendingBlock{
- hash: blockHash,
- bytes: blockBytes,
- })
- log.Tracef("Added block %s to pending blocks", blockHash)
-
- return nil
-}
-
-// HasBlock returns whether or not a block with the given hash exists in the
-// database.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) HasBlock(hash *chainhash.Hash) (bool, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return false, err
- }
-
- return tx.hasBlock(hash), nil
-}
-
-// HasBlocks returns whether or not the blocks with the provided hashes
-// exist in the database.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrTxClosed if the transaction has already been closed
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) HasBlocks(hashes []chainhash.Hash) ([]bool, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- results := make([]bool, len(hashes))
- for i := range hashes {
- results[i] = tx.hasBlock(&hashes[i])
- }
-
- return results, nil
-}
-
-// fetchBlockRow fetches the metadata stored in the block index for the provided
-// hash. It will return ErrBlockNotFound if there is no entry.
-func (tx *transaction) fetchBlockRow(hash *chainhash.Hash) ([]byte, error) {
- blockRow := tx.blockIdxBucket.Get(hash[:])
- if blockRow == nil {
- str := fmt.Sprintf("block %s does not exist", hash)
- return nil, makeDbErr(database.ErrBlockNotFound, str, nil)
- }
-
- return blockRow, nil
-}
-
-// FetchBlockHeader returns the raw serialized bytes for the block header
-// identified by the given hash. The raw bytes are in the format returned by
-// Serialize on a wire.BlockHeader.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockNotFound if the requested block hash does not exist
-// - ErrTxClosed if the transaction has already been closed
-// - ErrCorruption if the database has somehow become corrupted
-//
-// NOTE: The data returned by this function is only valid during a
-// database transaction. Attempting to access it after a transaction
-// has ended results in undefined behavior. This constraint prevents
-// additional data copies and allows support for memory-mapped database
-// implementations.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) FetchBlockHeader(hash *chainhash.Hash) ([]byte, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // When the block is pending to be written on commit return the bytes
- // from there.
- if idx, exists := tx.pendingBlocks[*hash]; exists {
- blockBytes := tx.pendingBlockData[idx].bytes
- return blockBytes[0:blockHdrSize:blockHdrSize], nil
- }
-
- // Fetch the block index row and slice off the header. Notice the use
- // of the cap on the subslice to prevent the caller from accidentally
- // appending into the db data.
- blockRow, err := tx.fetchBlockRow(hash)
- if err != nil {
- return nil, err
- }
- endOffset := blockLocSize + blockHdrSize
- return blockRow[blockLocSize:endOffset:endOffset], nil
-}
-
-// FetchBlockHeaders returns the raw serialized bytes for the block headers
-// identified by the given hashes. The raw bytes are in the format returned by
-// Serialize on a wire.BlockHeader.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockNotFound if the any of the requested block hashes do not exist
-// - ErrTxClosed if the transaction has already been closed
-// - ErrCorruption if the database has somehow become corrupted
-//
-// NOTE: The data returned by this function is only valid during a database
-// transaction. Attempting to access it after a transaction has ended results
-// in undefined behavior. This constraint prevents additional data copies and
-// allows support for memory-mapped database implementations.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) FetchBlockHeaders(hashes []chainhash.Hash) ([][]byte, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // NOTE: This could check for the existence of all blocks before loading
- // any of the headers which would be faster in the failure case, however
- // callers will not typically be calling this function with invalid
- // values, so optimize for the common case.
-
- // Load the headers.
- headers := make([][]byte, len(hashes))
- for i := range hashes {
- hash := &hashes[i]
-
- // When the block is pending to be written on commit return the
- // bytes from there.
- if idx, exists := tx.pendingBlocks[*hash]; exists {
- blkBytes := tx.pendingBlockData[idx].bytes
- headers[i] = blkBytes[0:blockHdrSize:blockHdrSize]
- continue
- }
-
- // Fetch the block index row and slice off the header. Notice
- // the use of the cap on the subslice to prevent the caller
- // from accidentally appending into the db data.
- blockRow, err := tx.fetchBlockRow(hash)
- if err != nil {
- return nil, err
- }
- endOffset := blockLocSize + blockHdrSize
- headers[i] = blockRow[blockLocSize:endOffset:endOffset]
- }
-
- return headers, nil
-}
-
-// FetchBlock returns the raw serialized bytes for the block identified by the
-// given hash. The raw bytes are in the format returned by Serialize on a
-// wire.MsgBlock.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockNotFound if the requested block hash does not exist
-// - ErrTxClosed if the transaction has already been closed
-// - ErrCorruption if the database has somehow become corrupted
-//
-// In addition, returns ErrDriverSpecific if any failures occur when reading the
-// block files.
-//
-// NOTE: The data returned by this function is only valid during a database
-// transaction. Attempting to access it after a transaction has ended results
-// in undefined behavior. This constraint prevents additional data copies and
-// allows support for memory-mapped database implementations.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) FetchBlock(hash *chainhash.Hash) ([]byte, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // When the block is pending to be written on commit return the bytes
- // from there.
- if idx, exists := tx.pendingBlocks[*hash]; exists {
- return tx.pendingBlockData[idx].bytes, nil
- }
-
- // Lookup the location of the block in the files from the block index.
- blockRow, err := tx.fetchBlockRow(hash)
- if err != nil {
- return nil, err
- }
- location := deserializeBlockLoc(blockRow)
-
- // Read the block from the appropriate location. The function also
- // performs a checksum over the data to detect data corruption.
- blockBytes, err := tx.db.store.readBlock(hash, location)
- if err != nil {
- return nil, err
- }
-
- return blockBytes, nil
-}
-
-// FetchBlocks returns the raw serialized bytes for the blocks identified by the
-// given hashes. The raw bytes are in the format returned by Serialize on a
-// wire.MsgBlock.
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockNotFound if any of the requested block hashed do not exist
-// - ErrTxClosed if the transaction has already been closed
-// - ErrCorruption if the database has somehow become corrupted
-//
-// In addition, returns ErrDriverSpecific if any failures occur when reading the
-// block files.
-//
-// NOTE: The data returned by this function is only valid during a database
-// transaction. Attempting to access it after a transaction has ended results
-// in undefined behavior. This constraint prevents additional data copies and
-// allows support for memory-mapped database implementations.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) FetchBlocks(hashes []chainhash.Hash) ([][]byte, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // NOTE: This could check for the existence of all blocks before loading
- // any of them which would be faster in the failure case, however
- // callers will not typically be calling this function with invalid
- // values, so optimize for the common case.
-
- // Load the blocks.
- blocks := make([][]byte, len(hashes))
- for i := range hashes {
- var err error
- blocks[i], err = tx.FetchBlock(&hashes[i])
- if err != nil {
- return nil, err
- }
- }
-
- return blocks, nil
-}
-
-// fetchPendingRegion attempts to fetch the provided region from any block which
-// are pending to be written on commit. It will return nil for the byte slice
-// when the region references a block which is not pending. When the region
-// does reference a pending block, it is bounds checked and returns
-// ErrBlockRegionInvalid if invalid.
-func (tx *transaction) fetchPendingRegion(region *database.BlockRegion) ([]byte, error) {
- // Nothing to do if the block is not pending to be written on commit.
- idx, exists := tx.pendingBlocks[*region.Hash]
- if !exists {
- return nil, nil
- }
-
- // Ensure the region is within the bounds of the block.
- blockBytes := tx.pendingBlockData[idx].bytes
- blockLen := uint32(len(blockBytes))
- endOffset := region.Offset + region.Len
- if endOffset < region.Offset || endOffset > blockLen {
- str := fmt.Sprintf("block %s region offset %d, length %d "+
- "exceeds block length of %d", region.Hash,
- region.Offset, region.Len, blockLen)
- return nil, makeDbErr(database.ErrBlockRegionInvalid, str, nil)
- }
-
- // Return the bytes from the pending block.
- return blockBytes[region.Offset:endOffset:endOffset], nil
-}
-
-// FetchBlockRegion returns the raw serialized bytes for the given block region.
-//
-// For example, it is possible to directly extract Bitcoin transactions and/or
-// scripts from a block with this function. Depending on the backend
-// implementation, this can provide significant savings by avoiding the need to
-// load entire blocks.
-//
-// The raw bytes are in the format returned by Serialize on a wire.MsgBlock and
-// the Offset field in the provided BlockRegion is zero-based and relative to
-// the start of the block (byte 0).
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockNotFound if the requested block hash does not exist
-// - ErrBlockRegionInvalid if the region exceeds the bounds of the associated
-// block
-// - ErrTxClosed if the transaction has already been closed
-// - ErrCorruption if the database has somehow become corrupted
-//
-// In addition, returns ErrDriverSpecific if any failures occur when reading the
-// block files.
-//
-// NOTE: The data returned by this function is only valid during a database
-// transaction. Attempting to access it after a transaction has ended results
-// in undefined behavior. This constraint prevents additional data copies and
-// allows support for memory-mapped database implementations.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) FetchBlockRegion(region *database.BlockRegion) ([]byte, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // When the block is pending to be written on commit return the bytes
- // from there.
- if tx.pendingBlocks != nil {
- regionBytes, err := tx.fetchPendingRegion(region)
- if err != nil {
- return nil, err
- }
- if regionBytes != nil {
- return regionBytes, nil
- }
- }
-
- // Lookup the location of the block in the files from the block index.
- blockRow, err := tx.fetchBlockRow(region.Hash)
- if err != nil {
- return nil, err
- }
- location := deserializeBlockLoc(blockRow)
-
- // Ensure the region is within the bounds of the block.
- endOffset := region.Offset + region.Len
- if endOffset < region.Offset || endOffset > location.blockLen {
- str := fmt.Sprintf("block %s region offset %d, length %d "+
- "exceeds block length of %d", region.Hash,
- region.Offset, region.Len, location.blockLen)
- return nil, makeDbErr(database.ErrBlockRegionInvalid, str, nil)
-
- }
-
- // Read the region from the appropriate disk block file.
- regionBytes, err := tx.db.store.readBlockRegion(location, region.Offset,
- region.Len)
- if err != nil {
- return nil, err
- }
-
- return regionBytes, nil
-}
-
-// FetchBlockRegions returns the raw serialized bytes for the given block
-// regions.
-//
-// For example, it is possible to directly extract Bitcoin transactions and/or
-// scripts from various blocks with this function. Depending on the backend
-// implementation, this can provide significant savings by avoiding the need to
-// load entire blocks.
-//
-// The raw bytes are in the format returned by Serialize on a wire.MsgBlock and
-// the Offset fields in the provided BlockRegions are zero-based and relative to
-// the start of the block (byte 0).
-//
-// Returns the following errors as required by the interface contract:
-// - ErrBlockNotFound if any of the request block hashes do not exist
-// - ErrBlockRegionInvalid if one or more region exceed the bounds of the
-// associated block
-// - ErrTxClosed if the transaction has already been closed
-// - ErrCorruption if the database has somehow become corrupted
-//
-// In addition, returns ErrDriverSpecific if any failures occur when reading the
-// block files.
-//
-// NOTE: The data returned by this function is only valid during a database
-// transaction. Attempting to access it after a transaction has ended results
-// in undefined behavior. This constraint prevents additional data copies and
-// allows support for memory-mapped database implementations.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) FetchBlockRegions(regions []database.BlockRegion) ([][]byte, error) {
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return nil, err
- }
-
- // NOTE: This could check for the existence of all blocks before
- // deserializing the locations and building up the fetch list which
- // would be faster in the failure case, however callers will not
- // typically be calling this function with invalid values, so optimize
- // for the common case.
-
- // NOTE: A potential optimization here would be to combine adjacent
- // regions to reduce the number of reads.
-
- // In order to improve efficiency of loading the bulk data, first grab
- // the block location for all of the requested block hashes and sort
- // the reads by filenum:offset so that all reads are grouped by file
- // and linear within each file. This can result in quite a significant
- // performance increase depending on how spread out the requested hashes
- // are by reducing the number of file open/closes and random accesses
- // needed. The fetchList is intentionally allocated with a cap because
- // some of the regions might be fetched from the pending blocks and
- // hence there is no need to fetch those from disk.
- blockRegions := make([][]byte, len(regions))
- fetchList := make([]bulkFetchData, 0, len(regions))
- for i := range regions {
- region := ®ions[i]
-
- // When the block is pending to be written on commit grab the
- // bytes from there.
- if tx.pendingBlocks != nil {
- regionBytes, err := tx.fetchPendingRegion(region)
- if err != nil {
- return nil, err
- }
- if regionBytes != nil {
- blockRegions[i] = regionBytes
- continue
- }
- }
-
- // Lookup the location of the block in the files from the block
- // index.
- blockRow, err := tx.fetchBlockRow(region.Hash)
- if err != nil {
- return nil, err
- }
- location := deserializeBlockLoc(blockRow)
-
- // Ensure the region is within the bounds of the block.
- endOffset := region.Offset + region.Len
- if endOffset < region.Offset || endOffset > location.blockLen {
- str := fmt.Sprintf("block %s region offset %d, length "+
- "%d exceeds block length of %d", region.Hash,
- region.Offset, region.Len, location.blockLen)
- return nil, makeDbErr(database.ErrBlockRegionInvalid, str, nil)
- }
-
- fetchList = append(fetchList, bulkFetchData{&location, i})
- }
- sort.Sort(bulkFetchDataSorter(fetchList))
-
- // Read all of the regions in the fetch list and set the results.
- for i := range fetchList {
- fetchData := &fetchList[i]
- ri := fetchData.replyIndex
- region := ®ions[ri]
- location := fetchData.blockLocation
- regionBytes, err := tx.db.store.readBlockRegion(*location,
- region.Offset, region.Len)
- if err != nil {
- return nil, err
- }
- blockRegions[ri] = regionBytes
- }
-
- return blockRegions, nil
-}
-
-// close marks the transaction closed then releases any pending data, the
-// underlying snapshot, the transaction read lock, and the write lock when the
-// transaction is writable.
-func (tx *transaction) close() {
- tx.closed = true
-
- // Clear pending blocks that would have been written on commit.
- tx.pendingBlocks = nil
- tx.pendingBlockData = nil
-
- // Clear pending keys that would have been written or deleted on commit.
- tx.pendingKeys = nil
- tx.pendingRemove = nil
-
- // Release the snapshot.
- if tx.snapshot != nil {
- tx.snapshot.Release()
- tx.snapshot = nil
- }
-
- tx.db.closeLock.RUnlock()
-
- // Release the writer lock for writable transactions to unblock any
- // other write transaction which are possibly waiting.
- if tx.writable {
- tx.db.writeLock.Unlock()
- }
-}
-
-// serializeBlockRow serializes a block row into a format suitable for storage
-// into the block index.
-func serializeBlockRow(blockLoc blockLocation, blockHdr []byte) []byte {
- // The serialized block index row format is:
- //
- // [0:blockLocSize] Block location
- // [blockLocSize:blockLocSize+blockHdrSize] Block header
- serializedRow := make([]byte, blockLocSize+blockHdrSize)
- copy(serializedRow, serializeBlockLoc(blockLoc))
- copy(serializedRow[blockHdrOffset:], blockHdr)
- return serializedRow
-}
-
-// writePendingAndCommit writes pending block data to the flat block files,
-// updates the metadata with their locations as well as the new current write
-// location, and commits the metadata to the memory database cache. It also
-// properly handles rollback in the case of failures.
-//
-// This function MUST only be called when there is pending data to be written.
-func (tx *transaction) writePendingAndCommit() error {
- // Save the current block store write position for potential rollback.
- // These variables are only updated here in this function and there can
- // only be one write transaction active at a time, so it's safe to store
- // them for potential rollback.
- wc := tx.db.store.writeCursor
- wc.RLock()
- oldBlkFileNum := wc.curFileNum
- oldBlkOffset := wc.curOffset
- wc.RUnlock()
-
- // rollback is a closure that is used to rollback all writes to the
- // block files.
- rollback := func() {
- // Rollback any modifications made to the block files if needed.
- tx.db.store.handleRollback(oldBlkFileNum, oldBlkOffset)
- }
-
- // Loop through all of the pending blocks to store and write them.
- for _, blockData := range tx.pendingBlockData {
- log.Tracef("Storing block %s", blockData.hash)
- location, err := tx.db.store.writeBlock(blockData.bytes)
- if err != nil {
- rollback()
- return err
- }
-
- // Add a record in the block index for the block. The record
- // includes the location information needed to locate the block
- // on the filesystem as well as the block header since they are
- // so commonly needed.
- blockHdr := blockData.bytes[0:blockHdrSize]
- blockRow := serializeBlockRow(location, blockHdr)
- err = tx.blockIdxBucket.Put(blockData.hash[:], blockRow)
- if err != nil {
- rollback()
- return err
- }
- }
-
- // Update the metadata for the current write file and offset.
- writeRow := serializeWriteRow(wc.curFileNum, wc.curOffset)
- if err := tx.metaBucket.Put(writeLocKeyName, writeRow); err != nil {
- rollback()
- return convertErr("failed to store write cursor", err)
- }
-
- // Atomically update the database cache. The cache automatically
- // handles flushing to the underlying persistent storage database.
- return tx.db.cache.commitTx(tx)
-}
-
-// Commit commits all changes that have been made to the root metadata bucket
-// and all of its sub-buckets to the database cache which is periodically synced
-// to persistent storage. In addition, it commits all new blocks directly to
-// persistent storage bypassing the db cache. Blocks can be rather large, so
-// this help increase the amount of cache available for the metadata updates and
-// is safe since blocks are immutable.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) Commit() error {
- // Prevent commits on managed transactions.
- if tx.managed {
- tx.close()
- panic("managed transaction commit not allowed")
- }
-
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return err
- }
-
- // Regardless of whether the commit succeeds, the transaction is closed
- // on return.
- defer tx.close()
-
- // Ensure the transaction is writable.
- if !tx.writable {
- str := "Commit requires a writable database transaction"
- return makeDbErr(database.ErrTxNotWritable, str, nil)
- }
-
- // Write pending data. The function will rollback if any errors occur.
- return tx.writePendingAndCommit()
-}
-
-// Rollback undoes all changes that have been made to the root bucket and all of
-// its sub-buckets.
-//
-// This function is part of the database.Tx interface implementation.
-func (tx *transaction) Rollback() error {
- // Prevent rollbacks on managed transactions.
- if tx.managed {
- tx.close()
- panic("managed transaction rollback not allowed")
- }
-
- // Ensure transaction state is valid.
- if err := tx.checkClosed(); err != nil {
- return err
- }
-
- tx.close()
- return nil
-}
-
-// db represents a collection of namespaces which are persisted and implements
-// the database.DB interface. All database access is performed through
-// transactions which are obtained through the specific Namespace.
-type db struct {
- writeLock sync.Mutex // Limit to one write transaction at a time.
- closeLock sync.RWMutex // Make database close block while txns active.
- closed bool // Is the database closed?
- store *blockStore // Handles read/writing blocks to flat files.
- cache *dbCache // Cache layer which wraps underlying leveldb DB.
-}
-
-// Enforce db implements the database.DB interface.
-var _ database.DB = (*db)(nil)
-
-// Type returns the database driver type the current database instance was
-// created with.
-//
-// This function is part of the database.DB interface implementation.
-func (db *db) Type() string {
- return dbType
-}
-
-// begin is the implementation function for the Begin database method. See its
-// documentation for more details.
-//
-// This function is only separate because it returns the internal transaction
-// which is used by the managed transaction code while the database method
-// returns the interface.
-func (db *db) begin(writable bool) (*transaction, error) {
- // Whenever a new writable transaction is started, grab the write lock
- // to ensure only a single write transaction can be active at the same
- // time. This lock will not be released until the transaction is
- // closed (via Rollback or Commit).
- if writable {
- db.writeLock.Lock()
- }
-
- // Whenever a new transaction is started, grab a read lock against the
- // database to ensure Close will wait for the transaction to finish.
- // This lock will not be released until the transaction is closed (via
- // Rollback or Commit).
- db.closeLock.RLock()
- if db.closed {
- db.closeLock.RUnlock()
- if writable {
- db.writeLock.Unlock()
- }
- return nil, makeDbErr(database.ErrDbNotOpen, errDbNotOpenStr,
- nil)
- }
-
- // Grab a snapshot of the database cache (which in turn also handles the
- // underlying database).
- snapshot, err := db.cache.Snapshot()
- if err != nil {
- db.closeLock.RUnlock()
- if writable {
- db.writeLock.Unlock()
- }
-
- return nil, err
- }
-
- // The metadata and block index buckets are internal-only buckets, so
- // they have defined IDs.
- tx := &transaction{
- writable: writable,
- db: db,
- snapshot: snapshot,
- pendingKeys: treap.NewMutable(),
- pendingRemove: treap.NewMutable(),
- }
- tx.metaBucket = &bucket{tx: tx, id: metadataBucketID}
- tx.blockIdxBucket = &bucket{tx: tx, id: blockIdxBucketID}
- return tx, nil
-}
-
-// Begin starts a transaction which is either read-only or read-write depending
-// on the specified flag. Multiple read-only transactions can be started
-// simultaneously while only a single read-write transaction can be started at a
-// time. The call will block when starting a read-write transaction when one is
-// already open.
-//
-// NOTE: The transaction must be closed by calling Rollback or Commit on it when
-// it is no longer needed. Failure to do so will result in unclaimed memory.
-//
-// This function is part of the database.DB interface implementation.
-func (db *db) Begin(writable bool) (database.Tx, error) {
- return db.begin(writable)
-}
-
-// rollbackOnPanic rolls the passed transaction back if the code in the calling
-// function panics. This is needed since the mutex on a transaction must be
-// released and a panic in called code would prevent that from happening.
-//
-// NOTE: This can only be handled manually for managed transactions since they
-// control the life-cycle of the transaction. As the documentation on Begin
-// calls out, callers opting to use manual transactions will have to ensure the
-// transaction is rolled back on panic if it desires that functionality as well
-// or the database will fail to close since the read-lock will never be
-// released.
-func rollbackOnPanic(tx *transaction) {
- if err := recover(); err != nil {
- tx.managed = false
- _ = tx.Rollback()
- panic(err)
- }
-}
-
-// View invokes the passed function in the context of a managed read-only
-// transaction with the root bucket for the namespace. Any errors returned from
-// the user-supplied function are returned from this function.
-//
-// This function is part of the database.DB interface implementation.
-func (db *db) View(fn func(database.Tx) error) error {
- // Start a read-only transaction.
- tx, err := db.begin(false)
- if err != nil {
- return err
- }
-
- // Since the user-provided function might panic, ensure the transaction
- // releases all mutexes and resources. There is no guarantee the caller
- // won't use recover and keep going. Thus, the database must still be
- // in a usable state on panics due to caller issues.
- defer rollbackOnPanic(tx)
-
- tx.managed = true
- err = fn(tx)
- tx.managed = false
- if err != nil {
- // The error is ignored here because nothing was written yet
- // and regardless of a rollback failure, the tx is closed now
- // anyways.
- _ = tx.Rollback()
- return err
- }
-
- return tx.Rollback()
-}
-
-// Update invokes the passed function in the context of a managed read-write
-// transaction with the root bucket for the namespace. Any errors returned from
-// the user-supplied function will cause the transaction to be rolled back and
-// are returned from this function. Otherwise, the transaction is committed
-// when the user-supplied function returns a nil error.
-//
-// This function is part of the database.DB interface implementation.
-func (db *db) Update(fn func(database.Tx) error) error {
- // Start a read-write transaction.
- tx, err := db.begin(true)
- if err != nil {
- return err
- }
-
- // Since the user-provided function might panic, ensure the transaction
- // releases all mutexes and resources. There is no guarantee the caller
- // won't use recover and keep going. Thus, the database must still be
- // in a usable state on panics due to caller issues.
- defer rollbackOnPanic(tx)
-
- tx.managed = true
- err = fn(tx)
- tx.managed = false
- if err != nil {
- // The error is ignored here because nothing was written yet
- // and regardless of a rollback failure, the tx is closed now
- // anyways.
- _ = tx.Rollback()
- return err
- }
-
- return tx.Commit()
-}
-
-// Close cleanly shuts down the database and syncs all data. It will block
-// until all database transactions have been finalized (rolled back or
-// committed).
-//
-// This function is part of the database.DB interface implementation.
-func (db *db) Close() error {
- // Since all transactions have a read lock on this mutex, this will
- // cause Close to wait for all readers to complete.
- db.closeLock.Lock()
- defer db.closeLock.Unlock()
-
- if db.closed {
- return makeDbErr(database.ErrDbNotOpen, errDbNotOpenStr, nil)
- }
- db.closed = true
-
- // NOTE: Since the above lock waits for all transactions to finish and
- // prevents any new ones from being started, it is safe to flush the
- // cache and clear all state without the individual locks.
-
- // Close the database cache which will flush any existing entries to
- // disk and close the underlying leveldb database. Any error is saved
- // and returned at the end after the remaining cleanup since the
- // database will be marked closed even if this fails given there is no
- // good way for the caller to recover from a failure here anyways.
- closeErr := db.cache.Close()
-
- // Close any open flat files that house the blocks.
- wc := db.store.writeCursor
- if wc.curFile.file != nil {
- _ = wc.curFile.file.Close()
- wc.curFile.file = nil
- }
- for _, blockFile := range db.store.openBlockFiles {
- _ = blockFile.file.Close()
- }
- db.store.openBlockFiles = nil
- db.store.openBlocksLRU.Init()
- db.store.fileNumToLRUElem = nil
-
- return closeErr
-}
-
-// filesExists reports whether the named file or directory exists.
-func fileExists(name string) bool {
- if _, err := os.Stat(name); err != nil {
- if os.IsNotExist(err) {
- return false
- }
- }
- return true
-}
-
-// initDB creates the initial buckets and values used by the package. This is
-// mainly in a separate function for testing purposes.
-func initDB(ldb *leveldb.DB) error {
- // The starting block file write cursor location is file num 0, offset
- // 0.
- batch := new(leveldb.Batch)
- batch.Put(bucketizedKey(metadataBucketID, writeLocKeyName),
- serializeWriteRow(0, 0))
-
- // Create block index bucket and set the current bucket id.
- //
- // NOTE: Since buckets are virtualized through the use of prefixes,
- // there is no need to store the bucket index data for the metadata
- // bucket in the database. However, the first bucket ID to use does
- // need to account for it to ensure there are no key collisions.
- batch.Put(bucketIndexKey(metadataBucketID, blockIdxBucketName),
- blockIdxBucketID[:])
- batch.Put(curBucketIDKeyName, blockIdxBucketID[:])
-
- // Write everything as a single batch.
- if err := ldb.Write(batch, nil); err != nil {
- str := fmt.Sprintf("failed to initialize metadata database: %v",
- err)
- return convertErr(str, err)
- }
-
- return nil
-}
-
-// openDB opens the database at the provided path. database.ErrDbDoesNotExist
-// is returned if the database doesn't exist and the create flag is not set.
-func openDB(dbPath string, network wire.BitcoinNet, create bool) (database.DB, error) {
- // Error if the database doesn't exist and the create flag is not set.
- metadataDbPath := filepath.Join(dbPath, metadataDbName)
- dbExists := fileExists(metadataDbPath)
- if !create && !dbExists {
- str := fmt.Sprintf("database %q does not exist", metadataDbPath)
- return nil, makeDbErr(database.ErrDbDoesNotExist, str, nil)
- }
-
- // Ensure the full path to the database exists.
- if !dbExists {
- // The error can be ignored here since the call to
- // leveldb.OpenFile will fail if the directory couldn't be
- // created.
- _ = os.MkdirAll(dbPath, 0700)
- }
-
- // Open the metadata database (will create it if needed).
- opts := opt.Options{
- ErrorIfExist: create,
- Strict: opt.DefaultStrict,
- Compression: opt.NoCompression,
- Filter: filter.NewBloomFilter(10),
- }
- ldb, err := leveldb.OpenFile(metadataDbPath, &opts)
- if err != nil {
- return nil, convertErr(err.Error(), err)
- }
-
- // Create the block store which includes scanning the existing flat
- // block files to find what the current write cursor position is
- // according to the data that is actually on disk. Also create the
- // database cache which wraps the underlying leveldb database to provide
- // write caching.
- store := newBlockStore(dbPath, network)
- cache := newDbCache(ldb, store, defaultCacheSize, defaultFlushSecs)
- pdb := &db{store: store, cache: cache}
-
- // Perform any reconciliation needed between the block and metadata as
- // well as database initialization, if needed.
- return reconcileDB(pdb, create)
-}
OSDN Git Service
