mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
Delete ethdb directory
This commit is contained in:
Isabel Schöps Thiel
GitHub
parent
1ebc730646
commit
463ba74fc7
12 changed files with 0 additions and 2760 deletions
|
|
@ -1,74 +0,0 @@
|
|||
// Copyright 2018 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package ethdb
|
||||
|
||||
// IdealBatchSize defines the size of the data batches should ideally add in one
|
||||
// write.
|
||||
const IdealBatchSize = 100 * 1024
|
||||
|
||||
// Batch is a write-only database that commits changes to its host database
|
||||
// when Write is called. A batch cannot be used concurrently.
|
||||
type Batch interface {
|
||||
KeyValueWriter
|
||||
|
||||
// ValueSize retrieves the amount of data queued up for writing.
|
||||
ValueSize() int
|
||||
|
||||
// Write flushes any accumulated data to disk.
|
||||
Write() error
|
||||
|
||||
// Reset resets the batch for reuse.
|
||||
Reset()
|
||||
|
||||
// Replay replays the batch contents.
|
||||
Replay(w KeyValueWriter) error
|
||||
}
|
||||
|
||||
// Batcher wraps the NewBatch method of a backing data store.
|
||||
type Batcher interface {
|
||||
// NewBatch creates a write-only database that buffers changes to its host db
|
||||
// until a final write is called.
|
||||
NewBatch() Batch
|
||||
|
||||
// NewBatchWithSize creates a write-only database batch with pre-allocated buffer.
|
||||
NewBatchWithSize(size int) Batch
|
||||
}
|
||||
|
||||
// HookedBatch wraps an arbitrary batch where each operation may be hooked into
|
||||
// to monitor from black box code.
|
||||
type HookedBatch struct {
|
||||
Batch
|
||||
|
||||
OnPut func(key []byte, value []byte) // Callback if a key is inserted
|
||||
OnDelete func(key []byte) // Callback if a key is deleted
|
||||
}
|
||||
|
||||
// Put inserts the given value into the key-value data store.
|
||||
func (b HookedBatch) Put(key []byte, value []byte) error {
|
||||
if b.OnPut != nil {
|
||||
b.OnPut(key, value)
|
||||
}
|
||||
return b.Batch.Put(key, value)
|
||||
}
|
||||
|
||||
// Delete removes the key from the key-value data store.
|
||||
func (b HookedBatch) Delete(key []byte) error {
|
||||
if b.OnDelete != nil {
|
||||
b.OnDelete(key)
|
||||
}
|
||||
return b.Batch.Delete(key)
|
||||
}
|
||||
|
|
@ -1,192 +0,0 @@
|
|||
// Copyright 2014 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
// Package ethdb defines the interfaces for an Ethereum data store.
|
||||
package ethdb
|
||||
|
||||
import "io"
|
||||
|
||||
// KeyValueReader wraps the Has and Get method of a backing data store.
|
||||
type KeyValueReader interface {
|
||||
// Has retrieves if a key is present in the key-value data store.
|
||||
Has(key []byte) (bool, error)
|
||||
|
||||
// Get retrieves the given key if it's present in the key-value data store.
|
||||
Get(key []byte) ([]byte, error)
|
||||
}
|
||||
|
||||
// KeyValueWriter wraps the Put method of a backing data store.
|
||||
type KeyValueWriter interface {
|
||||
// Put inserts the given value into the key-value data store.
|
||||
Put(key []byte, value []byte) error
|
||||
|
||||
// Delete removes the key from the key-value data store.
|
||||
Delete(key []byte) error
|
||||
}
|
||||
|
||||
// KeyValueStater wraps the Stat method of a backing data store.
|
||||
type KeyValueStater interface {
|
||||
// Stat returns a particular internal stat of the database.
|
||||
Stat(property string) (string, error)
|
||||
}
|
||||
|
||||
// Compacter wraps the Compact method of a backing data store.
|
||||
type Compacter interface {
|
||||
// Compact flattens the underlying data store for the given key range. In essence,
|
||||
// deleted and overwritten versions are discarded, and the data is rearranged to
|
||||
// reduce the cost of operations needed to access them.
|
||||
//
|
||||
// A nil start is treated as a key before all keys in the data store; a nil limit
|
||||
// is treated as a key after all keys in the data store. If both is nil then it
|
||||
// will compact entire data store.
|
||||
Compact(start []byte, limit []byte) error
|
||||
}
|
||||
|
||||
// KeyValueStore contains all the methods required to allow handling different
|
||||
// key-value data stores backing the high level database.
|
||||
type KeyValueStore interface {
|
||||
KeyValueReader
|
||||
KeyValueWriter
|
||||
KeyValueStater
|
||||
Batcher
|
||||
Iteratee
|
||||
Compacter
|
||||
Snapshotter
|
||||
io.Closer
|
||||
}
|
||||
|
||||
// AncientReaderOp contains the methods required to read from immutable ancient data.
|
||||
type AncientReaderOp interface {
|
||||
// HasAncient returns an indicator whether the specified data exists in the
|
||||
// ancient store.
|
||||
HasAncient(kind string, number uint64) (bool, error)
|
||||
|
||||
// Ancient retrieves an ancient binary blob from the append-only immutable files.
|
||||
Ancient(kind string, number uint64) ([]byte, error)
|
||||
|
||||
// AncientRange retrieves multiple items in sequence, starting from the index 'start'.
|
||||
// It will return
|
||||
// - at most 'count' items,
|
||||
// - if maxBytes is specified: at least 1 item (even if exceeding the maxByteSize),
|
||||
// but will otherwise return as many items as fit into maxByteSize.
|
||||
// - if maxBytes is not specified, 'count' items will be returned if they are present
|
||||
AncientRange(kind string, start, count, maxBytes uint64) ([][]byte, error)
|
||||
|
||||
// Ancients returns the ancient item numbers in the ancient store.
|
||||
Ancients() (uint64, error)
|
||||
|
||||
// Tail returns the number of first stored item in the freezer.
|
||||
// This number can also be interpreted as the total deleted item numbers.
|
||||
Tail() (uint64, error)
|
||||
|
||||
// AncientSize returns the ancient size of the specified category.
|
||||
AncientSize(kind string) (uint64, error)
|
||||
}
|
||||
|
||||
// AncientReader is the extended ancient reader interface including 'batched' or 'atomic' reading.
|
||||
type AncientReader interface {
|
||||
AncientReaderOp
|
||||
|
||||
// ReadAncients runs the given read operation while ensuring that no writes take place
|
||||
// on the underlying freezer.
|
||||
ReadAncients(fn func(AncientReaderOp) error) (err error)
|
||||
}
|
||||
|
||||
// AncientWriter contains the methods required to write to immutable ancient data.
|
||||
type AncientWriter interface {
|
||||
// ModifyAncients runs a write operation on the ancient store.
|
||||
// If the function returns an error, any changes to the underlying store are reverted.
|
||||
// The integer return value is the total size of the written data.
|
||||
ModifyAncients(func(AncientWriteOp) error) (int64, error)
|
||||
|
||||
// TruncateHead discards all but the first n ancient data from the ancient store.
|
||||
// After the truncation, the latest item can be accessed it item_n-1(start from 0).
|
||||
TruncateHead(n uint64) (uint64, error)
|
||||
|
||||
// TruncateTail discards the first n ancient data from the ancient store. The already
|
||||
// deleted items are ignored. After the truncation, the earliest item can be accessed
|
||||
// is item_n(start from 0). The deleted items may not be removed from the ancient store
|
||||
// immediately, but only when the accumulated deleted data reach the threshold then
|
||||
// will be removed all together.
|
||||
TruncateTail(n uint64) (uint64, error)
|
||||
|
||||
// Sync flushes all in-memory ancient store data to disk.
|
||||
Sync() error
|
||||
|
||||
// MigrateTable processes and migrates entries of a given table to a new format.
|
||||
// The second argument is a function that takes a raw entry and returns it
|
||||
// in the newest format.
|
||||
MigrateTable(string, func([]byte) ([]byte, error)) error
|
||||
}
|
||||
|
||||
// AncientWriteOp is given to the function argument of ModifyAncients.
|
||||
type AncientWriteOp interface {
|
||||
// Append adds an RLP-encoded item.
|
||||
Append(kind string, number uint64, item interface{}) error
|
||||
|
||||
// AppendRaw adds an item without RLP-encoding it.
|
||||
AppendRaw(kind string, number uint64, item []byte) error
|
||||
}
|
||||
|
||||
// AncientStater wraps the Stat method of a backing data store.
|
||||
type AncientStater interface {
|
||||
// AncientDatadir returns the path of root ancient directory. Empty string
|
||||
// will be returned if ancient store is not enabled at all. The returned
|
||||
// path can be used to construct the path of other freezers.
|
||||
AncientDatadir() (string, error)
|
||||
}
|
||||
|
||||
// Reader contains the methods required to read data from both key-value as well as
|
||||
// immutable ancient data.
|
||||
type Reader interface {
|
||||
KeyValueReader
|
||||
AncientReader
|
||||
}
|
||||
|
||||
// Writer contains the methods required to write data to both key-value as well as
|
||||
// immutable ancient data.
|
||||
type Writer interface {
|
||||
KeyValueWriter
|
||||
AncientWriter
|
||||
}
|
||||
|
||||
// Stater contains the methods required to retrieve states from both key-value as well as
|
||||
// immutable ancient data.
|
||||
type Stater interface {
|
||||
KeyValueStater
|
||||
AncientStater
|
||||
}
|
||||
|
||||
// AncientStore contains all the methods required to allow handling different
|
||||
// ancient data stores backing immutable chain data store.
|
||||
type AncientStore interface {
|
||||
AncientReader
|
||||
AncientWriter
|
||||
io.Closer
|
||||
}
|
||||
|
||||
// Database contains all the methods required by the high level database to not
|
||||
// only access the key-value data store but also the chain freezer.
|
||||
type Database interface {
|
||||
Reader
|
||||
Writer
|
||||
Batcher
|
||||
Iteratee
|
||||
Stater
|
||||
Compacter
|
||||
Snapshotter
|
||||
io.Closer
|
||||
}
|
||||
|
|
@ -1,537 +0,0 @@
|
|||
// Copyright 2019 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package dbtest
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"reflect"
|
||||
"sort"
|
||||
"testing"
|
||||
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"golang.org/x/exp/slices"
|
||||
)
|
||||
|
||||
// TestDatabaseSuite runs a suite of tests against a KeyValueStore database
|
||||
// implementation.
|
||||
func TestDatabaseSuite(t *testing.T, New func() ethdb.KeyValueStore) {
|
||||
t.Run("Iterator", func(t *testing.T) {
|
||||
tests := []struct {
|
||||
content map[string]string
|
||||
prefix string
|
||||
start string
|
||||
order []string
|
||||
}{
|
||||
// Empty databases should be iterable
|
||||
{map[string]string{}, "", "", nil},
|
||||
{map[string]string{}, "non-existent-prefix", "", nil},
|
||||
|
||||
// Single-item databases should be iterable
|
||||
{map[string]string{"key": "val"}, "", "", []string{"key"}},
|
||||
{map[string]string{"key": "val"}, "k", "", []string{"key"}},
|
||||
{map[string]string{"key": "val"}, "l", "", nil},
|
||||
|
||||
// Multi-item databases should be fully iterable
|
||||
{
|
||||
map[string]string{"k1": "v1", "k5": "v5", "k2": "v2", "k4": "v4", "k3": "v3"},
|
||||
"", "",
|
||||
[]string{"k1", "k2", "k3", "k4", "k5"},
|
||||
},
|
||||
{
|
||||
map[string]string{"k1": "v1", "k5": "v5", "k2": "v2", "k4": "v4", "k3": "v3"},
|
||||
"k", "",
|
||||
[]string{"k1", "k2", "k3", "k4", "k5"},
|
||||
},
|
||||
{
|
||||
map[string]string{"k1": "v1", "k5": "v5", "k2": "v2", "k4": "v4", "k3": "v3"},
|
||||
"l", "",
|
||||
nil,
|
||||
},
|
||||
// Multi-item databases should be prefix-iterable
|
||||
{
|
||||
map[string]string{
|
||||
"ka1": "va1", "ka5": "va5", "ka2": "va2", "ka4": "va4", "ka3": "va3",
|
||||
"kb1": "vb1", "kb5": "vb5", "kb2": "vb2", "kb4": "vb4", "kb3": "vb3",
|
||||
},
|
||||
"ka", "",
|
||||
[]string{"ka1", "ka2", "ka3", "ka4", "ka5"},
|
||||
},
|
||||
{
|
||||
map[string]string{
|
||||
"ka1": "va1", "ka5": "va5", "ka2": "va2", "ka4": "va4", "ka3": "va3",
|
||||
"kb1": "vb1", "kb5": "vb5", "kb2": "vb2", "kb4": "vb4", "kb3": "vb3",
|
||||
},
|
||||
"kc", "",
|
||||
nil,
|
||||
},
|
||||
// Multi-item databases should be prefix-iterable with start position
|
||||
{
|
||||
map[string]string{
|
||||
"ka1": "va1", "ka5": "va5", "ka2": "va2", "ka4": "va4", "ka3": "va3",
|
||||
"kb1": "vb1", "kb5": "vb5", "kb2": "vb2", "kb4": "vb4", "kb3": "vb3",
|
||||
},
|
||||
"ka", "3",
|
||||
[]string{"ka3", "ka4", "ka5"},
|
||||
},
|
||||
{
|
||||
map[string]string{
|
||||
"ka1": "va1", "ka5": "va5", "ka2": "va2", "ka4": "va4", "ka3": "va3",
|
||||
"kb1": "vb1", "kb5": "vb5", "kb2": "vb2", "kb4": "vb4", "kb3": "vb3",
|
||||
},
|
||||
"ka", "8",
|
||||
nil,
|
||||
},
|
||||
}
|
||||
for i, tt := range tests {
|
||||
// Create the key-value data store
|
||||
db := New()
|
||||
for key, val := range tt.content {
|
||||
if err := db.Put([]byte(key), []byte(val)); err != nil {
|
||||
t.Fatalf("test %d: failed to insert item %s:%s into database: %v", i, key, val, err)
|
||||
}
|
||||
}
|
||||
// Iterate over the database with the given configs and verify the results
|
||||
it, idx := db.NewIterator([]byte(tt.prefix), []byte(tt.start)), 0
|
||||
for it.Next() {
|
||||
if len(tt.order) <= idx {
|
||||
t.Errorf("test %d: prefix=%q more items than expected: checking idx=%d (key %q), expecting len=%d", i, tt.prefix, idx, it.Key(), len(tt.order))
|
||||
break
|
||||
}
|
||||
if !bytes.Equal(it.Key(), []byte(tt.order[idx])) {
|
||||
t.Errorf("test %d: item %d: key mismatch: have %s, want %s", i, idx, string(it.Key()), tt.order[idx])
|
||||
}
|
||||
if !bytes.Equal(it.Value(), []byte(tt.content[tt.order[idx]])) {
|
||||
t.Errorf("test %d: item %d: value mismatch: have %s, want %s", i, idx, string(it.Value()), tt.content[tt.order[idx]])
|
||||
}
|
||||
idx++
|
||||
}
|
||||
if err := it.Error(); err != nil {
|
||||
t.Errorf("test %d: iteration failed: %v", i, err)
|
||||
}
|
||||
if idx != len(tt.order) {
|
||||
t.Errorf("test %d: iteration terminated prematurely: have %d, want %d", i, idx, len(tt.order))
|
||||
}
|
||||
db.Close()
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("IteratorWith", func(t *testing.T) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
keys := []string{"1", "2", "3", "4", "6", "10", "11", "12", "20", "21", "22"}
|
||||
sort.Strings(keys) // 1, 10, 11, etc
|
||||
|
||||
for _, k := range keys {
|
||||
if err := db.Put([]byte(k), nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator(nil, nil)
|
||||
got, want := iterateKeys(it), keys
|
||||
if err := it.Error(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("Iterator: got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator([]byte("1"), nil)
|
||||
got, want := iterateKeys(it), []string{"1", "10", "11", "12"}
|
||||
if err := it.Error(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("IteratorWith(1,nil): got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator([]byte("5"), nil)
|
||||
got, want := iterateKeys(it), []string{}
|
||||
if err := it.Error(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("IteratorWith(5,nil): got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator(nil, []byte("2"))
|
||||
got, want := iterateKeys(it), []string{"2", "20", "21", "22", "3", "4", "6"}
|
||||
if err := it.Error(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("IteratorWith(nil,2): got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator(nil, []byte("5"))
|
||||
got, want := iterateKeys(it), []string{"6"}
|
||||
if err := it.Error(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("IteratorWith(nil,5): got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("KeyValueOperations", func(t *testing.T) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
key := []byte("foo")
|
||||
|
||||
if got, err := db.Has(key); err != nil {
|
||||
t.Error(err)
|
||||
} else if got {
|
||||
t.Errorf("wrong value: %t", got)
|
||||
}
|
||||
|
||||
value := []byte("hello world")
|
||||
if err := db.Put(key, value); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
if got, err := db.Has(key); err != nil {
|
||||
t.Error(err)
|
||||
} else if !got {
|
||||
t.Errorf("wrong value: %t", got)
|
||||
}
|
||||
|
||||
if got, err := db.Get(key); err != nil {
|
||||
t.Error(err)
|
||||
} else if !bytes.Equal(got, value) {
|
||||
t.Errorf("wrong value: %q", got)
|
||||
}
|
||||
|
||||
if err := db.Delete(key); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
if got, err := db.Has(key); err != nil {
|
||||
t.Error(err)
|
||||
} else if got {
|
||||
t.Errorf("wrong value: %t", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("Batch", func(t *testing.T) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
b := db.NewBatch()
|
||||
for _, k := range []string{"1", "2", "3", "4"} {
|
||||
if err := b.Put([]byte(k), nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
if has, err := db.Has([]byte("1")); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if has {
|
||||
t.Error("db contains element before batch write")
|
||||
}
|
||||
|
||||
if err := b.Write(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator(nil, nil)
|
||||
if got, want := iterateKeys(it), []string{"1", "2", "3", "4"}; !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
b.Reset()
|
||||
|
||||
// Mix writes and deletes in batch
|
||||
b.Put([]byte("5"), nil)
|
||||
b.Delete([]byte("1"))
|
||||
b.Put([]byte("6"), nil)
|
||||
|
||||
b.Delete([]byte("3")) // delete then put
|
||||
b.Put([]byte("3"), nil)
|
||||
|
||||
b.Put([]byte("7"), nil) // put then delete
|
||||
b.Delete([]byte("7"))
|
||||
|
||||
if err := b.Write(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
{
|
||||
it := db.NewIterator(nil, nil)
|
||||
if got, want := iterateKeys(it), []string{"2", "3", "4", "5", "6"}; !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("got: %s; want: %s", got, want)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("BatchReplay", func(t *testing.T) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
want := []string{"1", "2", "3", "4"}
|
||||
b := db.NewBatch()
|
||||
for _, k := range want {
|
||||
if err := b.Put([]byte(k), nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
b2 := db.NewBatch()
|
||||
if err := b.Replay(b2); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := b2.Replay(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
it := db.NewIterator(nil, nil)
|
||||
if got := iterateKeys(it); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("got: %s; want: %s", got, want)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("Snapshot", func(t *testing.T) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
initial := map[string]string{
|
||||
"k1": "v1", "k2": "v2", "k3": "", "k4": "",
|
||||
}
|
||||
for k, v := range initial {
|
||||
db.Put([]byte(k), []byte(v))
|
||||
}
|
||||
snapshot, err := db.NewSnapshot()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for k, v := range initial {
|
||||
got, err := snapshot.Get([]byte(k))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, []byte(v)) {
|
||||
t.Fatalf("Unexpected value want: %v, got %v", v, got)
|
||||
}
|
||||
}
|
||||
|
||||
// Flush more modifications into the database, ensure the snapshot
|
||||
// isn't affected.
|
||||
var (
|
||||
update = map[string]string{"k1": "v1-b", "k3": "v3-b"}
|
||||
insert = map[string]string{"k5": "v5-b"}
|
||||
delete = map[string]string{"k2": ""}
|
||||
)
|
||||
for k, v := range update {
|
||||
db.Put([]byte(k), []byte(v))
|
||||
}
|
||||
for k, v := range insert {
|
||||
db.Put([]byte(k), []byte(v))
|
||||
}
|
||||
for k := range delete {
|
||||
db.Delete([]byte(k))
|
||||
}
|
||||
for k, v := range initial {
|
||||
got, err := snapshot.Get([]byte(k))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, []byte(v)) {
|
||||
t.Fatalf("Unexpected value want: %v, got %v", v, got)
|
||||
}
|
||||
}
|
||||
for k := range insert {
|
||||
got, err := snapshot.Get([]byte(k))
|
||||
if err == nil || len(got) != 0 {
|
||||
t.Fatal("Unexpected value")
|
||||
}
|
||||
}
|
||||
for k := range delete {
|
||||
got, err := snapshot.Get([]byte(k))
|
||||
if err != nil || len(got) == 0 {
|
||||
t.Fatal("Unexpected deletion")
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("OperatonsAfterClose", func(t *testing.T) {
|
||||
db := New()
|
||||
db.Put([]byte("key"), []byte("value"))
|
||||
db.Close()
|
||||
if _, err := db.Get([]byte("key")); err == nil {
|
||||
t.Fatalf("expected error on Get after Close")
|
||||
}
|
||||
if _, err := db.Has([]byte("key")); err == nil {
|
||||
t.Fatalf("expected error on Get after Close")
|
||||
}
|
||||
if err := db.Put([]byte("key2"), []byte("value2")); err == nil {
|
||||
t.Fatalf("expected error on Put after Close")
|
||||
}
|
||||
if err := db.Delete([]byte("key")); err == nil {
|
||||
t.Fatalf("expected error on Delete after Close")
|
||||
}
|
||||
|
||||
b := db.NewBatch()
|
||||
if err := b.Put([]byte("batchkey"), []byte("batchval")); err != nil {
|
||||
t.Fatalf("expected no error on batch.Put after Close, got %v", err)
|
||||
}
|
||||
if err := b.Write(); err == nil {
|
||||
t.Fatalf("expected error on batch.Write after Close")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// BenchDatabaseSuite runs a suite of benchmarks against a KeyValueStore database
|
||||
// implementation.
|
||||
func BenchDatabaseSuite(b *testing.B, New func() ethdb.KeyValueStore) {
|
||||
var (
|
||||
keys, vals = makeDataset(1_000_000, 32, 32, false)
|
||||
sKeys, sVals = makeDataset(1_000_000, 32, 32, true)
|
||||
)
|
||||
// Run benchmarks sequentially
|
||||
b.Run("Write", func(b *testing.B) {
|
||||
benchWrite := func(b *testing.B, keys, vals [][]byte) {
|
||||
b.ResetTimer()
|
||||
b.ReportAllocs()
|
||||
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
for i := 0; i < len(keys); i++ {
|
||||
db.Put(keys[i], vals[i])
|
||||
}
|
||||
}
|
||||
b.Run("WriteSorted", func(b *testing.B) {
|
||||
benchWrite(b, sKeys, sVals)
|
||||
})
|
||||
b.Run("WriteRandom", func(b *testing.B) {
|
||||
benchWrite(b, keys, vals)
|
||||
})
|
||||
})
|
||||
b.Run("Read", func(b *testing.B) {
|
||||
benchRead := func(b *testing.B, keys, vals [][]byte) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
for i := 0; i < len(keys); i++ {
|
||||
db.Put(keys[i], vals[i])
|
||||
}
|
||||
b.ResetTimer()
|
||||
b.ReportAllocs()
|
||||
|
||||
for i := 0; i < len(keys); i++ {
|
||||
db.Get(keys[i])
|
||||
}
|
||||
}
|
||||
b.Run("ReadSorted", func(b *testing.B) {
|
||||
benchRead(b, sKeys, sVals)
|
||||
})
|
||||
b.Run("ReadRandom", func(b *testing.B) {
|
||||
benchRead(b, keys, vals)
|
||||
})
|
||||
})
|
||||
b.Run("Iteration", func(b *testing.B) {
|
||||
benchIteration := func(b *testing.B, keys, vals [][]byte) {
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
for i := 0; i < len(keys); i++ {
|
||||
db.Put(keys[i], vals[i])
|
||||
}
|
||||
b.ResetTimer()
|
||||
b.ReportAllocs()
|
||||
|
||||
it := db.NewIterator(nil, nil)
|
||||
for it.Next() {
|
||||
}
|
||||
it.Release()
|
||||
}
|
||||
b.Run("IterationSorted", func(b *testing.B) {
|
||||
benchIteration(b, sKeys, sVals)
|
||||
})
|
||||
b.Run("IterationRandom", func(b *testing.B) {
|
||||
benchIteration(b, keys, vals)
|
||||
})
|
||||
})
|
||||
b.Run("BatchWrite", func(b *testing.B) {
|
||||
benchBatchWrite := func(b *testing.B, keys, vals [][]byte) {
|
||||
b.ResetTimer()
|
||||
b.ReportAllocs()
|
||||
|
||||
db := New()
|
||||
defer db.Close()
|
||||
|
||||
batch := db.NewBatch()
|
||||
for i := 0; i < len(keys); i++ {
|
||||
batch.Put(keys[i], vals[i])
|
||||
}
|
||||
batch.Write()
|
||||
}
|
||||
b.Run("BenchWriteSorted", func(b *testing.B) {
|
||||
benchBatchWrite(b, sKeys, sVals)
|
||||
})
|
||||
b.Run("BenchWriteRandom", func(b *testing.B) {
|
||||
benchBatchWrite(b, keys, vals)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func iterateKeys(it ethdb.Iterator) []string {
|
||||
keys := []string{}
|
||||
for it.Next() {
|
||||
keys = append(keys, string(it.Key()))
|
||||
}
|
||||
sort.Strings(keys)
|
||||
it.Release()
|
||||
return keys
|
||||
}
|
||||
|
||||
// randomHash generates a random blob of data and returns it as a hash.
|
||||
func randBytes(len int) []byte {
|
||||
buf := make([]byte, len)
|
||||
if n, err := rand.Read(buf); n != len || err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return buf
|
||||
}
|
||||
|
||||
func makeDataset(size, ksize, vsize int, order bool) ([][]byte, [][]byte) {
|
||||
var keys [][]byte
|
||||
var vals [][]byte
|
||||
for i := 0; i < size; i += 1 {
|
||||
keys = append(keys, randBytes(ksize))
|
||||
vals = append(vals, randBytes(vsize))
|
||||
}
|
||||
if order {
|
||||
slices.SortFunc(keys, func(a, b []byte) int { return bytes.Compare(a, b) })
|
||||
}
|
||||
return keys, vals
|
||||
}
|
||||
|
|
@ -1,61 +0,0 @@
|
|||
// Copyright 2018 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package ethdb
|
||||
|
||||
// Iterator iterates over a database's key/value pairs in ascending key order.
|
||||
//
|
||||
// When it encounters an error any seek will return false and will yield no key/
|
||||
// value pairs. The error can be queried by calling the Error method. Calling
|
||||
// Release is still necessary.
|
||||
//
|
||||
// An iterator must be released after use, but it is not necessary to read an
|
||||
// iterator until exhaustion. An iterator is not safe for concurrent use, but it
|
||||
// is safe to use multiple iterators concurrently.
|
||||
type Iterator interface {
|
||||
// Next moves the iterator to the next key/value pair. It returns whether the
|
||||
// iterator is exhausted.
|
||||
Next() bool
|
||||
|
||||
// Error returns any accumulated error. Exhausting all the key/value pairs
|
||||
// is not considered to be an error.
|
||||
Error() error
|
||||
|
||||
// Key returns the key of the current key/value pair, or nil if done. The caller
|
||||
// should not modify the contents of the returned slice, and its contents may
|
||||
// change on the next call to Next.
|
||||
Key() []byte
|
||||
|
||||
// Value returns the value of the current key/value pair, or nil if done. The
|
||||
// caller should not modify the contents of the returned slice, and its contents
|
||||
// may change on the next call to Next.
|
||||
Value() []byte
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
Release()
|
||||
}
|
||||
|
||||
// Iteratee wraps the NewIterator methods of a backing data store.
|
||||
type Iteratee interface {
|
||||
// NewIterator creates a binary-alphabetical iterator over a subset
|
||||
// of database content with a particular key prefix, starting at a particular
|
||||
// initial key (or after, if it does not exist).
|
||||
//
|
||||
// Note: This method assumes that the prefix is NOT part of the start, so there's
|
||||
// no need for the caller to prepend the prefix to the start
|
||||
NewIterator(prefix []byte, start []byte) Iterator
|
||||
}
|
||||
|
|
@ -1,485 +0,0 @@
|
|||
// Copyright 2018 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//go:build !js
|
||||
// +build !js
|
||||
|
||||
// Package leveldb implements the key-value database layer based on LevelDB.
|
||||
package leveldb
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/metrics"
|
||||
"github.com/syndtr/goleveldb/leveldb"
|
||||
"github.com/syndtr/goleveldb/leveldb/errors"
|
||||
"github.com/syndtr/goleveldb/leveldb/filter"
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
"github.com/syndtr/goleveldb/leveldb/util"
|
||||
)
|
||||
|
||||
const (
|
||||
// degradationWarnInterval specifies how often warning should be printed if the
|
||||
// leveldb database cannot keep up with requested writes.
|
||||
degradationWarnInterval = time.Minute
|
||||
|
||||
// minCache is the minimum amount of memory in megabytes to allocate to leveldb
|
||||
// read and write caching, split half and half.
|
||||
minCache = 16
|
||||
|
||||
// minHandles is the minimum number of files handles to allocate to the open
|
||||
// database files.
|
||||
minHandles = 16
|
||||
|
||||
// metricsGatheringInterval specifies the interval to retrieve leveldb database
|
||||
// compaction, io and pause stats to report to the user.
|
||||
metricsGatheringInterval = 3 * time.Second
|
||||
)
|
||||
|
||||
// Database is a persistent key-value store. Apart from basic data storage
|
||||
// functionality it also supports batch writes and iterating over the keyspace in
|
||||
// binary-alphabetical order.
|
||||
type Database struct {
|
||||
fn string // filename for reporting
|
||||
db *leveldb.DB // LevelDB instance
|
||||
|
||||
compTimeMeter metrics.Meter // Meter for measuring the total time spent in database compaction
|
||||
compReadMeter metrics.Meter // Meter for measuring the data read during compaction
|
||||
compWriteMeter metrics.Meter // Meter for measuring the data written during compaction
|
||||
writeDelayNMeter metrics.Meter // Meter for measuring the write delay number due to database compaction
|
||||
writeDelayMeter metrics.Meter // Meter for measuring the write delay duration due to database compaction
|
||||
diskSizeGauge metrics.Gauge // Gauge for tracking the size of all the levels in the database
|
||||
diskReadMeter metrics.Meter // Meter for measuring the effective amount of data read
|
||||
diskWriteMeter metrics.Meter // Meter for measuring the effective amount of data written
|
||||
memCompGauge metrics.Gauge // Gauge for tracking the number of memory compaction
|
||||
level0CompGauge metrics.Gauge // Gauge for tracking the number of table compaction in level0
|
||||
nonlevel0CompGauge metrics.Gauge // Gauge for tracking the number of table compaction in non0 level
|
||||
seekCompGauge metrics.Gauge // Gauge for tracking the number of table compaction caused by read opt
|
||||
manualMemAllocGauge metrics.Gauge // Gauge to track the amount of memory that has been manually allocated (not a part of runtime/GC)
|
||||
|
||||
levelsGauge []metrics.Gauge // Gauge for tracking the number of tables in levels
|
||||
|
||||
quitLock sync.Mutex // Mutex protecting the quit channel access
|
||||
quitChan chan chan error // Quit channel to stop the metrics collection before closing the database
|
||||
|
||||
log log.Logger // Contextual logger tracking the database path
|
||||
}
|
||||
|
||||
// New returns a wrapped LevelDB object. The namespace is the prefix that the
|
||||
// metrics reporting should use for surfacing internal stats.
|
||||
func New(file string, cache int, handles int, namespace string, readonly bool) (*Database, error) {
|
||||
return NewCustom(file, namespace, func(options *opt.Options) {
|
||||
// Ensure we have some minimal caching and file guarantees
|
||||
if cache < minCache {
|
||||
cache = minCache
|
||||
}
|
||||
if handles < minHandles {
|
||||
handles = minHandles
|
||||
}
|
||||
// Set default options
|
||||
options.OpenFilesCacheCapacity = handles
|
||||
options.BlockCacheCapacity = cache / 2 * opt.MiB
|
||||
options.WriteBuffer = cache / 4 * opt.MiB // Two of these are used internally
|
||||
if readonly {
|
||||
options.ReadOnly = true
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// NewCustom returns a wrapped LevelDB object. The namespace is the prefix that the
|
||||
// metrics reporting should use for surfacing internal stats.
|
||||
// The customize function allows the caller to modify the leveldb options.
|
||||
func NewCustom(file string, namespace string, customize func(options *opt.Options)) (*Database, error) {
|
||||
options := configureOptions(customize)
|
||||
logger := log.New("database", file)
|
||||
usedCache := options.GetBlockCacheCapacity() + options.GetWriteBuffer()*2
|
||||
logCtx := []interface{}{"cache", common.StorageSize(usedCache), "handles", options.GetOpenFilesCacheCapacity()}
|
||||
if options.ReadOnly {
|
||||
logCtx = append(logCtx, "readonly", "true")
|
||||
}
|
||||
logger.Info("Allocated cache and file handles", logCtx...)
|
||||
|
||||
// Open the db and recover any potential corruptions
|
||||
db, err := leveldb.OpenFile(file, options)
|
||||
if _, corrupted := err.(*errors.ErrCorrupted); corrupted {
|
||||
db, err = leveldb.RecoverFile(file, nil)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Assemble the wrapper with all the registered metrics
|
||||
ldb := &Database{
|
||||
fn: file,
|
||||
db: db,
|
||||
log: logger,
|
||||
quitChan: make(chan chan error),
|
||||
}
|
||||
ldb.compTimeMeter = metrics.NewRegisteredMeter(namespace+"compact/time", nil)
|
||||
ldb.compReadMeter = metrics.NewRegisteredMeter(namespace+"compact/input", nil)
|
||||
ldb.compWriteMeter = metrics.NewRegisteredMeter(namespace+"compact/output", nil)
|
||||
ldb.diskSizeGauge = metrics.NewRegisteredGauge(namespace+"disk/size", nil)
|
||||
ldb.diskReadMeter = metrics.NewRegisteredMeter(namespace+"disk/read", nil)
|
||||
ldb.diskWriteMeter = metrics.NewRegisteredMeter(namespace+"disk/write", nil)
|
||||
ldb.writeDelayMeter = metrics.NewRegisteredMeter(namespace+"compact/writedelay/duration", nil)
|
||||
ldb.writeDelayNMeter = metrics.NewRegisteredMeter(namespace+"compact/writedelay/counter", nil)
|
||||
ldb.memCompGauge = metrics.NewRegisteredGauge(namespace+"compact/memory", nil)
|
||||
ldb.level0CompGauge = metrics.NewRegisteredGauge(namespace+"compact/level0", nil)
|
||||
ldb.nonlevel0CompGauge = metrics.NewRegisteredGauge(namespace+"compact/nonlevel0", nil)
|
||||
ldb.seekCompGauge = metrics.NewRegisteredGauge(namespace+"compact/seek", nil)
|
||||
ldb.manualMemAllocGauge = metrics.NewRegisteredGauge(namespace+"memory/manualalloc", nil)
|
||||
|
||||
// Start up the metrics gathering and return
|
||||
go ldb.meter(metricsGatheringInterval, namespace)
|
||||
return ldb, nil
|
||||
}
|
||||
|
||||
// configureOptions sets some default options, then runs the provided setter.
|
||||
func configureOptions(customizeFn func(*opt.Options)) *opt.Options {
|
||||
// Set default options
|
||||
options := &opt.Options{
|
||||
Filter: filter.NewBloomFilter(10),
|
||||
DisableSeeksCompaction: true,
|
||||
}
|
||||
// Allow caller to make custom modifications to the options
|
||||
if customizeFn != nil {
|
||||
customizeFn(options)
|
||||
}
|
||||
return options
|
||||
}
|
||||
|
||||
// Close stops the metrics collection, flushes any pending data to disk and closes
|
||||
// all io accesses to the underlying key-value store.
|
||||
func (db *Database) Close() error {
|
||||
db.quitLock.Lock()
|
||||
defer db.quitLock.Unlock()
|
||||
|
||||
if db.quitChan != nil {
|
||||
errc := make(chan error)
|
||||
db.quitChan <- errc
|
||||
if err := <-errc; err != nil {
|
||||
db.log.Error("Metrics collection failed", "err", err)
|
||||
}
|
||||
db.quitChan = nil
|
||||
}
|
||||
return db.db.Close()
|
||||
}
|
||||
|
||||
// Has retrieves if a key is present in the key-value store.
|
||||
func (db *Database) Has(key []byte) (bool, error) {
|
||||
return db.db.Has(key, nil)
|
||||
}
|
||||
|
||||
// Get retrieves the given key if it's present in the key-value store.
|
||||
func (db *Database) Get(key []byte) ([]byte, error) {
|
||||
dat, err := db.db.Get(key, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return dat, nil
|
||||
}
|
||||
|
||||
// Put inserts the given value into the key-value store.
|
||||
func (db *Database) Put(key []byte, value []byte) error {
|
||||
return db.db.Put(key, value, nil)
|
||||
}
|
||||
|
||||
// Delete removes the key from the key-value store.
|
||||
func (db *Database) Delete(key []byte) error {
|
||||
return db.db.Delete(key, nil)
|
||||
}
|
||||
|
||||
// NewBatch creates a write-only key-value store that buffers changes to its host
|
||||
// database until a final write is called.
|
||||
func (db *Database) NewBatch() ethdb.Batch {
|
||||
return &batch{
|
||||
db: db.db,
|
||||
b: new(leveldb.Batch),
|
||||
}
|
||||
}
|
||||
|
||||
// NewBatchWithSize creates a write-only database batch with pre-allocated buffer.
|
||||
func (db *Database) NewBatchWithSize(size int) ethdb.Batch {
|
||||
return &batch{
|
||||
db: db.db,
|
||||
b: leveldb.MakeBatch(size),
|
||||
}
|
||||
}
|
||||
|
||||
// NewIterator creates a binary-alphabetical iterator over a subset
|
||||
// of database content with a particular key prefix, starting at a particular
|
||||
// initial key (or after, if it does not exist).
|
||||
func (db *Database) NewIterator(prefix []byte, start []byte) ethdb.Iterator {
|
||||
return db.db.NewIterator(bytesPrefixRange(prefix, start), nil)
|
||||
}
|
||||
|
||||
// NewSnapshot creates a database snapshot based on the current state.
|
||||
// The created snapshot will not be affected by all following mutations
|
||||
// happened on the database.
|
||||
// Note don't forget to release the snapshot once it's used up, otherwise
|
||||
// the stale data will never be cleaned up by the underlying compactor.
|
||||
func (db *Database) NewSnapshot() (ethdb.Snapshot, error) {
|
||||
snap, err := db.db.GetSnapshot()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &snapshot{db: snap}, nil
|
||||
}
|
||||
|
||||
// Stat returns a particular internal stat of the database.
|
||||
func (db *Database) Stat(property string) (string, error) {
|
||||
if property == "" {
|
||||
property = "leveldb.stats"
|
||||
} else if !strings.HasPrefix(property, "leveldb.") {
|
||||
property = "leveldb." + property
|
||||
}
|
||||
return db.db.GetProperty(property)
|
||||
}
|
||||
|
||||
// Compact flattens the underlying data store for the given key range. In essence,
|
||||
// deleted and overwritten versions are discarded, and the data is rearranged to
|
||||
// reduce the cost of operations needed to access them.
|
||||
//
|
||||
// A nil start is treated as a key before all keys in the data store; a nil limit
|
||||
// is treated as a key after all keys in the data store. If both is nil then it
|
||||
// will compact entire data store.
|
||||
func (db *Database) Compact(start []byte, limit []byte) error {
|
||||
return db.db.CompactRange(util.Range{Start: start, Limit: limit})
|
||||
}
|
||||
|
||||
// Path returns the path to the database directory.
|
||||
func (db *Database) Path() string {
|
||||
return db.fn
|
||||
}
|
||||
|
||||
// meter periodically retrieves internal leveldb counters and reports them to
|
||||
// the metrics subsystem.
|
||||
func (db *Database) meter(refresh time.Duration, namespace string) {
|
||||
// Create the counters to store current and previous compaction values
|
||||
compactions := make([][]int64, 2)
|
||||
for i := 0; i < 2; i++ {
|
||||
compactions[i] = make([]int64, 4)
|
||||
}
|
||||
// Create storages for states and warning log tracer.
|
||||
var (
|
||||
errc chan error
|
||||
merr error
|
||||
|
||||
stats leveldb.DBStats
|
||||
iostats [2]int64
|
||||
delaystats [2]int64
|
||||
lastWritePaused time.Time
|
||||
)
|
||||
timer := time.NewTimer(refresh)
|
||||
defer timer.Stop()
|
||||
|
||||
// Iterate ad infinitum and collect the stats
|
||||
for i := 1; errc == nil && merr == nil; i++ {
|
||||
// Retrieve the database stats
|
||||
// Stats method resets buffers inside therefore it's okay to just pass the struct.
|
||||
err := db.db.Stats(&stats)
|
||||
if err != nil {
|
||||
db.log.Error("Failed to read database stats", "err", err)
|
||||
merr = err
|
||||
continue
|
||||
}
|
||||
// Iterate over all the leveldbTable rows, and accumulate the entries
|
||||
for j := 0; j < len(compactions[i%2]); j++ {
|
||||
compactions[i%2][j] = 0
|
||||
}
|
||||
compactions[i%2][0] = stats.LevelSizes.Sum()
|
||||
for _, t := range stats.LevelDurations {
|
||||
compactions[i%2][1] += t.Nanoseconds()
|
||||
}
|
||||
compactions[i%2][2] = stats.LevelRead.Sum()
|
||||
compactions[i%2][3] = stats.LevelWrite.Sum()
|
||||
// Update all the requested meters
|
||||
if db.diskSizeGauge != nil {
|
||||
db.diskSizeGauge.Update(compactions[i%2][0])
|
||||
}
|
||||
if db.compTimeMeter != nil {
|
||||
db.compTimeMeter.Mark(compactions[i%2][1] - compactions[(i-1)%2][1])
|
||||
}
|
||||
if db.compReadMeter != nil {
|
||||
db.compReadMeter.Mark(compactions[i%2][2] - compactions[(i-1)%2][2])
|
||||
}
|
||||
if db.compWriteMeter != nil {
|
||||
db.compWriteMeter.Mark(compactions[i%2][3] - compactions[(i-1)%2][3])
|
||||
}
|
||||
var (
|
||||
delayN = int64(stats.WriteDelayCount)
|
||||
duration = stats.WriteDelayDuration
|
||||
paused = stats.WritePaused
|
||||
)
|
||||
if db.writeDelayNMeter != nil {
|
||||
db.writeDelayNMeter.Mark(delayN - delaystats[0])
|
||||
}
|
||||
if db.writeDelayMeter != nil {
|
||||
db.writeDelayMeter.Mark(duration.Nanoseconds() - delaystats[1])
|
||||
}
|
||||
// If a warning that db is performing compaction has been displayed, any subsequent
|
||||
// warnings will be withheld for one minute not to overwhelm the user.
|
||||
if paused && delayN-delaystats[0] == 0 && duration.Nanoseconds()-delaystats[1] == 0 &&
|
||||
time.Now().After(lastWritePaused.Add(degradationWarnInterval)) {
|
||||
db.log.Warn("Database compacting, degraded performance")
|
||||
lastWritePaused = time.Now()
|
||||
}
|
||||
delaystats[0], delaystats[1] = delayN, duration.Nanoseconds()
|
||||
|
||||
var (
|
||||
nRead = int64(stats.IORead)
|
||||
nWrite = int64(stats.IOWrite)
|
||||
)
|
||||
if db.diskReadMeter != nil {
|
||||
db.diskReadMeter.Mark(nRead - iostats[0])
|
||||
}
|
||||
if db.diskWriteMeter != nil {
|
||||
db.diskWriteMeter.Mark(nWrite - iostats[1])
|
||||
}
|
||||
iostats[0], iostats[1] = nRead, nWrite
|
||||
|
||||
db.memCompGauge.Update(int64(stats.MemComp))
|
||||
db.level0CompGauge.Update(int64(stats.Level0Comp))
|
||||
db.nonlevel0CompGauge.Update(int64(stats.NonLevel0Comp))
|
||||
db.seekCompGauge.Update(int64(stats.SeekComp))
|
||||
|
||||
for i, tables := range stats.LevelTablesCounts {
|
||||
// Append metrics for additional layers
|
||||
if i >= len(db.levelsGauge) {
|
||||
db.levelsGauge = append(db.levelsGauge, metrics.NewRegisteredGauge(namespace+fmt.Sprintf("tables/level%v", i), nil))
|
||||
}
|
||||
db.levelsGauge[i].Update(int64(tables))
|
||||
}
|
||||
|
||||
// Sleep a bit, then repeat the stats collection
|
||||
select {
|
||||
case errc = <-db.quitChan:
|
||||
// Quit requesting, stop hammering the database
|
||||
case <-timer.C:
|
||||
timer.Reset(refresh)
|
||||
// Timeout, gather a new set of stats
|
||||
}
|
||||
}
|
||||
|
||||
if errc == nil {
|
||||
errc = <-db.quitChan
|
||||
}
|
||||
errc <- merr
|
||||
}
|
||||
|
||||
// batch is a write-only leveldb batch that commits changes to its host database
|
||||
// when Write is called. A batch cannot be used concurrently.
|
||||
type batch struct {
|
||||
db *leveldb.DB
|
||||
b *leveldb.Batch
|
||||
size int
|
||||
}
|
||||
|
||||
// Put inserts the given value into the batch for later committing.
|
||||
func (b *batch) Put(key, value []byte) error {
|
||||
b.b.Put(key, value)
|
||||
b.size += len(key) + len(value)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete inserts the a key removal into the batch for later committing.
|
||||
func (b *batch) Delete(key []byte) error {
|
||||
b.b.Delete(key)
|
||||
b.size += len(key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValueSize retrieves the amount of data queued up for writing.
|
||||
func (b *batch) ValueSize() int {
|
||||
return b.size
|
||||
}
|
||||
|
||||
// Write flushes any accumulated data to disk.
|
||||
func (b *batch) Write() error {
|
||||
return b.db.Write(b.b, nil)
|
||||
}
|
||||
|
||||
// Reset resets the batch for reuse.
|
||||
func (b *batch) Reset() {
|
||||
b.b.Reset()
|
||||
b.size = 0
|
||||
}
|
||||
|
||||
// Replay replays the batch contents.
|
||||
func (b *batch) Replay(w ethdb.KeyValueWriter) error {
|
||||
return b.b.Replay(&replayer{writer: w})
|
||||
}
|
||||
|
||||
// replayer is a small wrapper to implement the correct replay methods.
|
||||
type replayer struct {
|
||||
writer ethdb.KeyValueWriter
|
||||
failure error
|
||||
}
|
||||
|
||||
// Put inserts the given value into the key-value data store.
|
||||
func (r *replayer) Put(key, value []byte) {
|
||||
// If the replay already failed, stop executing ops
|
||||
if r.failure != nil {
|
||||
return
|
||||
}
|
||||
r.failure = r.writer.Put(key, value)
|
||||
}
|
||||
|
||||
// Delete removes the key from the key-value data store.
|
||||
func (r *replayer) Delete(key []byte) {
|
||||
// If the replay already failed, stop executing ops
|
||||
if r.failure != nil {
|
||||
return
|
||||
}
|
||||
r.failure = r.writer.Delete(key)
|
||||
}
|
||||
|
||||
// bytesPrefixRange returns key range that satisfy
|
||||
// - the given prefix, and
|
||||
// - the given seek position
|
||||
func bytesPrefixRange(prefix, start []byte) *util.Range {
|
||||
r := util.BytesPrefix(prefix)
|
||||
r.Start = append(r.Start, start...)
|
||||
return r
|
||||
}
|
||||
|
||||
// snapshot wraps a leveldb snapshot for implementing the Snapshot interface.
|
||||
type snapshot struct {
|
||||
db *leveldb.Snapshot
|
||||
}
|
||||
|
||||
// Has retrieves if a key is present in the snapshot backing by a key-value
|
||||
// data store.
|
||||
func (snap *snapshot) Has(key []byte) (bool, error) {
|
||||
return snap.db.Has(key, nil)
|
||||
}
|
||||
|
||||
// Get retrieves the given key if it's present in the snapshot backing by
|
||||
// key-value data store.
|
||||
func (snap *snapshot) Get(key []byte) ([]byte, error) {
|
||||
return snap.db.Get(key, nil)
|
||||
}
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
func (snap *snapshot) Release() {
|
||||
snap.db.Release()
|
||||
}
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
// Copyright 2019 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package leveldb
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/ethdb/dbtest"
|
||||
"github.com/syndtr/goleveldb/leveldb"
|
||||
"github.com/syndtr/goleveldb/leveldb/storage"
|
||||
)
|
||||
|
||||
func TestLevelDB(t *testing.T) {
|
||||
t.Run("DatabaseSuite", func(t *testing.T) {
|
||||
dbtest.TestDatabaseSuite(t, func() ethdb.KeyValueStore {
|
||||
db, err := leveldb.Open(storage.NewMemStorage(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &Database{
|
||||
db: db,
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkLevelDB(b *testing.B) {
|
||||
dbtest.BenchDatabaseSuite(b, func() ethdb.KeyValueStore {
|
||||
db, err := leveldb.Open(storage.NewMemStorage(), nil)
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
return &Database{
|
||||
db: db,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
@ -1,390 +0,0 @@
|
|||
// Copyright 2018 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
// Package memorydb implements the key-value database layer based on memory maps.
|
||||
package memorydb
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
)
|
||||
|
||||
var (
|
||||
// errMemorydbClosed is returned if a memory database was already closed at the
|
||||
// invocation of a data access operation.
|
||||
errMemorydbClosed = errors.New("database closed")
|
||||
|
||||
// errMemorydbNotFound is returned if a key is requested that is not found in
|
||||
// the provided memory database.
|
||||
errMemorydbNotFound = errors.New("not found")
|
||||
|
||||
// errSnapshotReleased is returned if callers want to retrieve data from a
|
||||
// released snapshot.
|
||||
errSnapshotReleased = errors.New("snapshot released")
|
||||
)
|
||||
|
||||
// Database is an ephemeral key-value store. Apart from basic data storage
|
||||
// functionality it also supports batch writes and iterating over the keyspace in
|
||||
// binary-alphabetical order.
|
||||
type Database struct {
|
||||
db map[string][]byte
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
// New returns a wrapped map with all the required database interface methods
|
||||
// implemented.
|
||||
func New() *Database {
|
||||
return &Database{
|
||||
db: make(map[string][]byte),
|
||||
}
|
||||
}
|
||||
|
||||
// NewWithCap returns a wrapped map pre-allocated to the provided capacity with
|
||||
// all the required database interface methods implemented.
|
||||
func NewWithCap(size int) *Database {
|
||||
return &Database{
|
||||
db: make(map[string][]byte, size),
|
||||
}
|
||||
}
|
||||
|
||||
// Close deallocates the internal map and ensures any consecutive data access op
|
||||
// fails with an error.
|
||||
func (db *Database) Close() error {
|
||||
db.lock.Lock()
|
||||
defer db.lock.Unlock()
|
||||
|
||||
db.db = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// Has retrieves if a key is present in the key-value store.
|
||||
func (db *Database) Has(key []byte) (bool, error) {
|
||||
db.lock.RLock()
|
||||
defer db.lock.RUnlock()
|
||||
|
||||
if db.db == nil {
|
||||
return false, errMemorydbClosed
|
||||
}
|
||||
_, ok := db.db[string(key)]
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
// Get retrieves the given key if it's present in the key-value store.
|
||||
func (db *Database) Get(key []byte) ([]byte, error) {
|
||||
db.lock.RLock()
|
||||
defer db.lock.RUnlock()
|
||||
|
||||
if db.db == nil {
|
||||
return nil, errMemorydbClosed
|
||||
}
|
||||
if entry, ok := db.db[string(key)]; ok {
|
||||
return common.CopyBytes(entry), nil
|
||||
}
|
||||
return nil, errMemorydbNotFound
|
||||
}
|
||||
|
||||
// Put inserts the given value into the key-value store.
|
||||
func (db *Database) Put(key []byte, value []byte) error {
|
||||
db.lock.Lock()
|
||||
defer db.lock.Unlock()
|
||||
|
||||
if db.db == nil {
|
||||
return errMemorydbClosed
|
||||
}
|
||||
db.db[string(key)] = common.CopyBytes(value)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete removes the key from the key-value store.
|
||||
func (db *Database) Delete(key []byte) error {
|
||||
db.lock.Lock()
|
||||
defer db.lock.Unlock()
|
||||
|
||||
if db.db == nil {
|
||||
return errMemorydbClosed
|
||||
}
|
||||
delete(db.db, string(key))
|
||||
return nil
|
||||
}
|
||||
|
||||
// NewBatch creates a write-only key-value store that buffers changes to its host
|
||||
// database until a final write is called.
|
||||
func (db *Database) NewBatch() ethdb.Batch {
|
||||
return &batch{
|
||||
db: db,
|
||||
}
|
||||
}
|
||||
|
||||
// NewBatchWithSize creates a write-only database batch with pre-allocated buffer.
|
||||
func (db *Database) NewBatchWithSize(size int) ethdb.Batch {
|
||||
return &batch{
|
||||
db: db,
|
||||
}
|
||||
}
|
||||
|
||||
// NewIterator creates a binary-alphabetical iterator over a subset
|
||||
// of database content with a particular key prefix, starting at a particular
|
||||
// initial key (or after, if it does not exist).
|
||||
func (db *Database) NewIterator(prefix []byte, start []byte) ethdb.Iterator {
|
||||
db.lock.RLock()
|
||||
defer db.lock.RUnlock()
|
||||
|
||||
var (
|
||||
pr = string(prefix)
|
||||
st = string(append(prefix, start...))
|
||||
keys = make([]string, 0, len(db.db))
|
||||
values = make([][]byte, 0, len(db.db))
|
||||
)
|
||||
// Collect the keys from the memory database corresponding to the given prefix
|
||||
// and start
|
||||
for key := range db.db {
|
||||
if !strings.HasPrefix(key, pr) {
|
||||
continue
|
||||
}
|
||||
if key >= st {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
}
|
||||
// Sort the items and retrieve the associated values
|
||||
sort.Strings(keys)
|
||||
for _, key := range keys {
|
||||
values = append(values, db.db[key])
|
||||
}
|
||||
return &iterator{
|
||||
index: -1,
|
||||
keys: keys,
|
||||
values: values,
|
||||
}
|
||||
}
|
||||
|
||||
// NewSnapshot creates a database snapshot based on the current state.
|
||||
// The created snapshot will not be affected by all following mutations
|
||||
// happened on the database.
|
||||
func (db *Database) NewSnapshot() (ethdb.Snapshot, error) {
|
||||
return newSnapshot(db), nil
|
||||
}
|
||||
|
||||
// Stat returns a particular internal stat of the database.
|
||||
func (db *Database) Stat(property string) (string, error) {
|
||||
return "", errors.New("unknown property")
|
||||
}
|
||||
|
||||
// Compact is not supported on a memory database, but there's no need either as
|
||||
// a memory database doesn't waste space anyway.
|
||||
func (db *Database) Compact(start []byte, limit []byte) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Len returns the number of entries currently present in the memory database.
|
||||
//
|
||||
// Note, this method is only used for testing (i.e. not public in general) and
|
||||
// does not have explicit checks for closed-ness to allow simpler testing code.
|
||||
func (db *Database) Len() int {
|
||||
db.lock.RLock()
|
||||
defer db.lock.RUnlock()
|
||||
|
||||
return len(db.db)
|
||||
}
|
||||
|
||||
// keyvalue is a key-value tuple tagged with a deletion field to allow creating
|
||||
// memory-database write batches.
|
||||
type keyvalue struct {
|
||||
key string
|
||||
value []byte
|
||||
delete bool
|
||||
}
|
||||
|
||||
// batch is a write-only memory batch that commits changes to its host
|
||||
// database when Write is called. A batch cannot be used concurrently.
|
||||
type batch struct {
|
||||
db *Database
|
||||
writes []keyvalue
|
||||
size int
|
||||
}
|
||||
|
||||
// Put inserts the given value into the batch for later committing.
|
||||
func (b *batch) Put(key, value []byte) error {
|
||||
b.writes = append(b.writes, keyvalue{string(key), common.CopyBytes(value), false})
|
||||
b.size += len(key) + len(value)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete inserts the a key removal into the batch for later committing.
|
||||
func (b *batch) Delete(key []byte) error {
|
||||
b.writes = append(b.writes, keyvalue{string(key), nil, true})
|
||||
b.size += len(key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValueSize retrieves the amount of data queued up for writing.
|
||||
func (b *batch) ValueSize() int {
|
||||
return b.size
|
||||
}
|
||||
|
||||
// Write flushes any accumulated data to the memory database.
|
||||
func (b *batch) Write() error {
|
||||
b.db.lock.Lock()
|
||||
defer b.db.lock.Unlock()
|
||||
|
||||
if b.db.db == nil {
|
||||
return errMemorydbClosed
|
||||
}
|
||||
for _, keyvalue := range b.writes {
|
||||
if keyvalue.delete {
|
||||
delete(b.db.db, keyvalue.key)
|
||||
continue
|
||||
}
|
||||
b.db.db[keyvalue.key] = keyvalue.value
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Reset resets the batch for reuse.
|
||||
func (b *batch) Reset() {
|
||||
b.writes = b.writes[:0]
|
||||
b.size = 0
|
||||
}
|
||||
|
||||
// Replay replays the batch contents.
|
||||
func (b *batch) Replay(w ethdb.KeyValueWriter) error {
|
||||
for _, keyvalue := range b.writes {
|
||||
if keyvalue.delete {
|
||||
if err := w.Delete([]byte(keyvalue.key)); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := w.Put([]byte(keyvalue.key), keyvalue.value); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// iterator can walk over the (potentially partial) keyspace of a memory key
|
||||
// value store. Internally it is a deep copy of the entire iterated state,
|
||||
// sorted by keys.
|
||||
type iterator struct {
|
||||
index int
|
||||
keys []string
|
||||
values [][]byte
|
||||
}
|
||||
|
||||
// Next moves the iterator to the next key/value pair. It returns whether the
|
||||
// iterator is exhausted.
|
||||
func (it *iterator) Next() bool {
|
||||
// Short circuit if iterator is already exhausted in the forward direction.
|
||||
if it.index >= len(it.keys) {
|
||||
return false
|
||||
}
|
||||
it.index += 1
|
||||
return it.index < len(it.keys)
|
||||
}
|
||||
|
||||
// Error returns any accumulated error. Exhausting all the key/value pairs
|
||||
// is not considered to be an error. A memory iterator cannot encounter errors.
|
||||
func (it *iterator) Error() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Key returns the key of the current key/value pair, or nil if done. The caller
|
||||
// should not modify the contents of the returned slice, and its contents may
|
||||
// change on the next call to Next.
|
||||
func (it *iterator) Key() []byte {
|
||||
// Short circuit if iterator is not in a valid position
|
||||
if it.index < 0 || it.index >= len(it.keys) {
|
||||
return nil
|
||||
}
|
||||
return []byte(it.keys[it.index])
|
||||
}
|
||||
|
||||
// Value returns the value of the current key/value pair, or nil if done. The
|
||||
// caller should not modify the contents of the returned slice, and its contents
|
||||
// may change on the next call to Next.
|
||||
func (it *iterator) Value() []byte {
|
||||
// Short circuit if iterator is not in a valid position
|
||||
if it.index < 0 || it.index >= len(it.keys) {
|
||||
return nil
|
||||
}
|
||||
return it.values[it.index]
|
||||
}
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
func (it *iterator) Release() {
|
||||
it.index, it.keys, it.values = -1, nil, nil
|
||||
}
|
||||
|
||||
// snapshot wraps a batch of key-value entries deep copied from the in-memory
|
||||
// database for implementing the Snapshot interface.
|
||||
type snapshot struct {
|
||||
db map[string][]byte
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
// newSnapshot initializes the snapshot with the given database instance.
|
||||
func newSnapshot(db *Database) *snapshot {
|
||||
db.lock.RLock()
|
||||
defer db.lock.RUnlock()
|
||||
|
||||
copied := make(map[string][]byte, len(db.db))
|
||||
for key, val := range db.db {
|
||||
copied[key] = common.CopyBytes(val)
|
||||
}
|
||||
return &snapshot{db: copied}
|
||||
}
|
||||
|
||||
// Has retrieves if a key is present in the snapshot backing by a key-value
|
||||
// data store.
|
||||
func (snap *snapshot) Has(key []byte) (bool, error) {
|
||||
snap.lock.RLock()
|
||||
defer snap.lock.RUnlock()
|
||||
|
||||
if snap.db == nil {
|
||||
return false, errSnapshotReleased
|
||||
}
|
||||
_, ok := snap.db[string(key)]
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
// Get retrieves the given key if it's present in the snapshot backing by
|
||||
// key-value data store.
|
||||
func (snap *snapshot) Get(key []byte) ([]byte, error) {
|
||||
snap.lock.RLock()
|
||||
defer snap.lock.RUnlock()
|
||||
|
||||
if snap.db == nil {
|
||||
return nil, errSnapshotReleased
|
||||
}
|
||||
if entry, ok := snap.db[string(key)]; ok {
|
||||
return common.CopyBytes(entry), nil
|
||||
}
|
||||
return nil, errMemorydbNotFound
|
||||
}
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
func (snap *snapshot) Release() {
|
||||
snap.lock.Lock()
|
||||
defer snap.lock.Unlock()
|
||||
|
||||
snap.db = nil
|
||||
}
|
||||
|
|
@ -1,50 +0,0 @@
|
|||
// Copyright 2018 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package memorydb
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"testing"
|
||||
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/ethdb/dbtest"
|
||||
)
|
||||
|
||||
func TestMemoryDB(t *testing.T) {
|
||||
t.Run("DatabaseSuite", func(t *testing.T) {
|
||||
dbtest.TestDatabaseSuite(t, func() ethdb.KeyValueStore {
|
||||
return New()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// BenchmarkBatchAllocs measures the time/allocs for storing 120 kB of data
|
||||
func BenchmarkBatchAllocs(b *testing.B) {
|
||||
b.ReportAllocs()
|
||||
var key = make([]byte, 20)
|
||||
var val = make([]byte, 100)
|
||||
// 120 * 1_000 -> 120_000 == 120kB
|
||||
for i := 0; i < b.N; i++ {
|
||||
batch := New().NewBatch()
|
||||
for j := uint64(0); j < 1000; j++ {
|
||||
binary.BigEndian.PutUint64(key, j)
|
||||
binary.BigEndian.PutUint64(val, j)
|
||||
batch.Put(key, val)
|
||||
}
|
||||
batch.Write()
|
||||
}
|
||||
}
|
||||
|
|
@ -1,668 +0,0 @@
|
|||
// Copyright 2023 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
// Package pebble implements the key-value database layer based on pebble.
|
||||
package pebble
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"runtime"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/cockroachdb/pebble"
|
||||
"github.com/cockroachdb/pebble/bloom"
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/metrics"
|
||||
)
|
||||
|
||||
const (
|
||||
// minCache is the minimum amount of memory in megabytes to allocate to pebble
|
||||
// read and write caching, split half and half.
|
||||
minCache = 16
|
||||
|
||||
// minHandles is the minimum number of files handles to allocate to the open
|
||||
// database files.
|
||||
minHandles = 16
|
||||
|
||||
// metricsGatheringInterval specifies the interval to retrieve pebble database
|
||||
// compaction, io and pause stats to report to the user.
|
||||
metricsGatheringInterval = 3 * time.Second
|
||||
)
|
||||
|
||||
// Database is a persistent key-value store based on the pebble storage engine.
|
||||
// Apart from basic data storage functionality it also supports batch writes and
|
||||
// iterating over the keyspace in binary-alphabetical order.
|
||||
type Database struct {
|
||||
fn string // filename for reporting
|
||||
db *pebble.DB // Underlying pebble storage engine
|
||||
|
||||
compTimeMeter metrics.Meter // Meter for measuring the total time spent in database compaction
|
||||
compReadMeter metrics.Meter // Meter for measuring the data read during compaction
|
||||
compWriteMeter metrics.Meter // Meter for measuring the data written during compaction
|
||||
writeDelayNMeter metrics.Meter // Meter for measuring the write delay number due to database compaction
|
||||
writeDelayMeter metrics.Meter // Meter for measuring the write delay duration due to database compaction
|
||||
diskSizeGauge metrics.Gauge // Gauge for tracking the size of all the levels in the database
|
||||
diskReadMeter metrics.Meter // Meter for measuring the effective amount of data read
|
||||
diskWriteMeter metrics.Meter // Meter for measuring the effective amount of data written
|
||||
memCompGauge metrics.Gauge // Gauge for tracking the number of memory compaction
|
||||
level0CompGauge metrics.Gauge // Gauge for tracking the number of table compaction in level0
|
||||
nonlevel0CompGauge metrics.Gauge // Gauge for tracking the number of table compaction in non0 level
|
||||
seekCompGauge metrics.Gauge // Gauge for tracking the number of table compaction caused by read opt
|
||||
manualMemAllocGauge metrics.Gauge // Gauge for tracking amount of non-managed memory currently allocated
|
||||
|
||||
levelsGauge []metrics.Gauge // Gauge for tracking the number of tables in levels
|
||||
|
||||
quitLock sync.RWMutex // Mutex protecting the quit channel and the closed flag
|
||||
quitChan chan chan error // Quit channel to stop the metrics collection before closing the database
|
||||
closed bool // keep track of whether we're Closed
|
||||
|
||||
log log.Logger // Contextual logger tracking the database path
|
||||
|
||||
activeComp int // Current number of active compactions
|
||||
compStartTime time.Time // The start time of the earliest currently-active compaction
|
||||
compTime atomic.Int64 // Total time spent in compaction in ns
|
||||
level0Comp atomic.Uint32 // Total number of level-zero compactions
|
||||
nonLevel0Comp atomic.Uint32 // Total number of non level-zero compactions
|
||||
writeDelayStartTime time.Time // The start time of the latest write stall
|
||||
writeDelayCount atomic.Int64 // Total number of write stall counts
|
||||
writeDelayTime atomic.Int64 // Total time spent in write stalls
|
||||
|
||||
writeOptions *pebble.WriteOptions
|
||||
}
|
||||
|
||||
func (d *Database) onCompactionBegin(info pebble.CompactionInfo) {
|
||||
if d.activeComp == 0 {
|
||||
d.compStartTime = time.Now()
|
||||
}
|
||||
l0 := info.Input[0]
|
||||
if l0.Level == 0 {
|
||||
d.level0Comp.Add(1)
|
||||
} else {
|
||||
d.nonLevel0Comp.Add(1)
|
||||
}
|
||||
d.activeComp++
|
||||
}
|
||||
|
||||
func (d *Database) onCompactionEnd(info pebble.CompactionInfo) {
|
||||
if d.activeComp == 1 {
|
||||
d.compTime.Add(int64(time.Since(d.compStartTime)))
|
||||
} else if d.activeComp == 0 {
|
||||
panic("should not happen")
|
||||
}
|
||||
d.activeComp--
|
||||
}
|
||||
|
||||
func (d *Database) onWriteStallBegin(b pebble.WriteStallBeginInfo) {
|
||||
d.writeDelayStartTime = time.Now()
|
||||
}
|
||||
|
||||
func (d *Database) onWriteStallEnd() {
|
||||
d.writeDelayTime.Add(int64(time.Since(d.writeDelayStartTime)))
|
||||
}
|
||||
|
||||
// panicLogger is just a noop logger to disable Pebble's internal logger.
|
||||
//
|
||||
// TODO(karalabe): Remove when Pebble sets this as the default.
|
||||
type panicLogger struct{}
|
||||
|
||||
func (l panicLogger) Infof(format string, args ...interface{}) {
|
||||
}
|
||||
|
||||
func (l panicLogger) Errorf(format string, args ...interface{}) {
|
||||
}
|
||||
|
||||
func (l panicLogger) Fatalf(format string, args ...interface{}) {
|
||||
panic(fmt.Errorf("fatal: "+format, args...))
|
||||
}
|
||||
|
||||
// New returns a wrapped pebble DB object. The namespace is the prefix that the
|
||||
// metrics reporting should use for surfacing internal stats.
|
||||
func New(file string, cache int, handles int, namespace string, readonly bool, ephemeral bool) (*Database, error) {
|
||||
// Ensure we have some minimal caching and file guarantees
|
||||
if cache < minCache {
|
||||
cache = minCache
|
||||
}
|
||||
if handles < minHandles {
|
||||
handles = minHandles
|
||||
}
|
||||
logger := log.New("database", file)
|
||||
logger.Info("Allocated cache and file handles", "cache", common.StorageSize(cache*1024*1024), "handles", handles)
|
||||
|
||||
// The max memtable size is limited by the uint32 offsets stored in
|
||||
// internal/arenaskl.node, DeferredBatchOp, and flushableBatchEntry.
|
||||
//
|
||||
// - MaxUint32 on 64-bit platforms;
|
||||
// - MaxInt on 32-bit platforms.
|
||||
//
|
||||
// It is used when slices are limited to Uint32 on 64-bit platforms (the
|
||||
// length limit for slices is naturally MaxInt on 32-bit platforms).
|
||||
//
|
||||
// Taken from https://github.com/cockroachdb/pebble/blob/master/internal/constants/constants.go
|
||||
maxMemTableSize := (1<<31)<<(^uint(0)>>63) - 1
|
||||
|
||||
// Two memory tables is configured which is identical to leveldb,
|
||||
// including a frozen memory table and another live one.
|
||||
memTableLimit := 2
|
||||
memTableSize := cache * 1024 * 1024 / 2 / memTableLimit
|
||||
|
||||
// The memory table size is currently capped at maxMemTableSize-1 due to a
|
||||
// known bug in the pebble where maxMemTableSize is not recognized as a
|
||||
// valid size.
|
||||
//
|
||||
// TODO use the maxMemTableSize as the maximum table size once the issue
|
||||
// in pebble is fixed.
|
||||
if memTableSize >= maxMemTableSize {
|
||||
memTableSize = maxMemTableSize - 1
|
||||
}
|
||||
db := &Database{
|
||||
fn: file,
|
||||
log: logger,
|
||||
quitChan: make(chan chan error),
|
||||
writeOptions: &pebble.WriteOptions{Sync: !ephemeral},
|
||||
}
|
||||
opt := &pebble.Options{
|
||||
// Pebble has a single combined cache area and the write
|
||||
// buffers are taken from this too. Assign all available
|
||||
// memory allowance for cache.
|
||||
Cache: pebble.NewCache(int64(cache * 1024 * 1024)),
|
||||
MaxOpenFiles: handles,
|
||||
|
||||
// The size of memory table(as well as the write buffer).
|
||||
// Note, there may have more than two memory tables in the system.
|
||||
MemTableSize: uint64(memTableSize),
|
||||
|
||||
// MemTableStopWritesThreshold places a hard limit on the size
|
||||
// of the existent MemTables(including the frozen one).
|
||||
// Note, this must be the number of tables not the size of all memtables
|
||||
// according to https://github.com/cockroachdb/pebble/blob/master/options.go#L738-L742
|
||||
// and to https://github.com/cockroachdb/pebble/blob/master/db.go#L1892-L1903.
|
||||
MemTableStopWritesThreshold: memTableLimit,
|
||||
|
||||
// The default compaction concurrency(1 thread),
|
||||
// Here use all available CPUs for faster compaction.
|
||||
MaxConcurrentCompactions: func() int { return runtime.NumCPU() },
|
||||
|
||||
// Per-level options. Options for at least one level must be specified. The
|
||||
// options for the last level are used for all subsequent levels.
|
||||
Levels: []pebble.LevelOptions{
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
{TargetFileSize: 2 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
|
||||
},
|
||||
ReadOnly: readonly,
|
||||
EventListener: &pebble.EventListener{
|
||||
CompactionBegin: db.onCompactionBegin,
|
||||
CompactionEnd: db.onCompactionEnd,
|
||||
WriteStallBegin: db.onWriteStallBegin,
|
||||
WriteStallEnd: db.onWriteStallEnd,
|
||||
},
|
||||
Logger: panicLogger{}, // TODO(karalabe): Delete when this is upstreamed in Pebble
|
||||
}
|
||||
// Disable seek compaction explicitly. Check https://github.com/ethereum/go-ethereum/pull/20130
|
||||
// for more details.
|
||||
opt.Experimental.ReadSamplingMultiplier = -1
|
||||
|
||||
// Open the db and recover any potential corruptions
|
||||
innerDB, err := pebble.Open(file, opt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
db.db = innerDB
|
||||
|
||||
db.compTimeMeter = metrics.NewRegisteredMeter(namespace+"compact/time", nil)
|
||||
db.compReadMeter = metrics.NewRegisteredMeter(namespace+"compact/input", nil)
|
||||
db.compWriteMeter = metrics.NewRegisteredMeter(namespace+"compact/output", nil)
|
||||
db.diskSizeGauge = metrics.NewRegisteredGauge(namespace+"disk/size", nil)
|
||||
db.diskReadMeter = metrics.NewRegisteredMeter(namespace+"disk/read", nil)
|
||||
db.diskWriteMeter = metrics.NewRegisteredMeter(namespace+"disk/write", nil)
|
||||
db.writeDelayMeter = metrics.NewRegisteredMeter(namespace+"compact/writedelay/duration", nil)
|
||||
db.writeDelayNMeter = metrics.NewRegisteredMeter(namespace+"compact/writedelay/counter", nil)
|
||||
db.memCompGauge = metrics.NewRegisteredGauge(namespace+"compact/memory", nil)
|
||||
db.level0CompGauge = metrics.NewRegisteredGauge(namespace+"compact/level0", nil)
|
||||
db.nonlevel0CompGauge = metrics.NewRegisteredGauge(namespace+"compact/nonlevel0", nil)
|
||||
db.seekCompGauge = metrics.NewRegisteredGauge(namespace+"compact/seek", nil)
|
||||
db.manualMemAllocGauge = metrics.NewRegisteredGauge(namespace+"memory/manualalloc", nil)
|
||||
|
||||
// Start up the metrics gathering and return
|
||||
go db.meter(metricsGatheringInterval, namespace)
|
||||
return db, nil
|
||||
}
|
||||
|
||||
// Close stops the metrics collection, flushes any pending data to disk and closes
|
||||
// all io accesses to the underlying key-value store.
|
||||
func (d *Database) Close() error {
|
||||
d.quitLock.Lock()
|
||||
defer d.quitLock.Unlock()
|
||||
// Allow double closing, simplifies things
|
||||
if d.closed {
|
||||
return nil
|
||||
}
|
||||
d.closed = true
|
||||
if d.quitChan != nil {
|
||||
errc := make(chan error)
|
||||
d.quitChan <- errc
|
||||
if err := <-errc; err != nil {
|
||||
d.log.Error("Metrics collection failed", "err", err)
|
||||
}
|
||||
d.quitChan = nil
|
||||
}
|
||||
return d.db.Close()
|
||||
}
|
||||
|
||||
// Has retrieves if a key is present in the key-value store.
|
||||
func (d *Database) Has(key []byte) (bool, error) {
|
||||
d.quitLock.RLock()
|
||||
defer d.quitLock.RUnlock()
|
||||
if d.closed {
|
||||
return false, pebble.ErrClosed
|
||||
}
|
||||
_, closer, err := d.db.Get(key)
|
||||
if err == pebble.ErrNotFound {
|
||||
return false, nil
|
||||
} else if err != nil {
|
||||
return false, err
|
||||
}
|
||||
closer.Close()
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Get retrieves the given key if it's present in the key-value store.
|
||||
func (d *Database) Get(key []byte) ([]byte, error) {
|
||||
d.quitLock.RLock()
|
||||
defer d.quitLock.RUnlock()
|
||||
if d.closed {
|
||||
return nil, pebble.ErrClosed
|
||||
}
|
||||
dat, closer, err := d.db.Get(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ret := make([]byte, len(dat))
|
||||
copy(ret, dat)
|
||||
closer.Close()
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
// Put inserts the given value into the key-value store.
|
||||
func (d *Database) Put(key []byte, value []byte) error {
|
||||
d.quitLock.RLock()
|
||||
defer d.quitLock.RUnlock()
|
||||
if d.closed {
|
||||
return pebble.ErrClosed
|
||||
}
|
||||
return d.db.Set(key, value, d.writeOptions)
|
||||
}
|
||||
|
||||
// Delete removes the key from the key-value store.
|
||||
func (d *Database) Delete(key []byte) error {
|
||||
d.quitLock.RLock()
|
||||
defer d.quitLock.RUnlock()
|
||||
if d.closed {
|
||||
return pebble.ErrClosed
|
||||
}
|
||||
return d.db.Delete(key, nil)
|
||||
}
|
||||
|
||||
// NewBatch creates a write-only key-value store that buffers changes to its host
|
||||
// database until a final write is called.
|
||||
func (d *Database) NewBatch() ethdb.Batch {
|
||||
return &batch{
|
||||
b: d.db.NewBatch(),
|
||||
db: d,
|
||||
}
|
||||
}
|
||||
|
||||
// NewBatchWithSize creates a write-only database batch with pre-allocated buffer.
|
||||
func (d *Database) NewBatchWithSize(size int) ethdb.Batch {
|
||||
return &batch{
|
||||
b: d.db.NewBatchWithSize(size),
|
||||
db: d,
|
||||
}
|
||||
}
|
||||
|
||||
// snapshot wraps a pebble snapshot for implementing the Snapshot interface.
|
||||
type snapshot struct {
|
||||
db *pebble.Snapshot
|
||||
}
|
||||
|
||||
// NewSnapshot creates a database snapshot based on the current state.
|
||||
// The created snapshot will not be affected by all following mutations
|
||||
// happened on the database.
|
||||
// Note don't forget to release the snapshot once it's used up, otherwise
|
||||
// the stale data will never be cleaned up by the underlying compactor.
|
||||
func (d *Database) NewSnapshot() (ethdb.Snapshot, error) {
|
||||
snap := d.db.NewSnapshot()
|
||||
return &snapshot{db: snap}, nil
|
||||
}
|
||||
|
||||
// Has retrieves if a key is present in the snapshot backing by a key-value
|
||||
// data store.
|
||||
func (snap *snapshot) Has(key []byte) (bool, error) {
|
||||
_, closer, err := snap.db.Get(key)
|
||||
if err != nil {
|
||||
if err != pebble.ErrNotFound {
|
||||
return false, err
|
||||
} else {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
closer.Close()
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Get retrieves the given key if it's present in the snapshot backing by
|
||||
// key-value data store.
|
||||
func (snap *snapshot) Get(key []byte) ([]byte, error) {
|
||||
dat, closer, err := snap.db.Get(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ret := make([]byte, len(dat))
|
||||
copy(ret, dat)
|
||||
closer.Close()
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
func (snap *snapshot) Release() {
|
||||
snap.db.Close()
|
||||
}
|
||||
|
||||
// upperBound returns the upper bound for the given prefix
|
||||
func upperBound(prefix []byte) (limit []byte) {
|
||||
for i := len(prefix) - 1; i >= 0; i-- {
|
||||
c := prefix[i]
|
||||
if c == 0xff {
|
||||
continue
|
||||
}
|
||||
limit = make([]byte, i+1)
|
||||
copy(limit, prefix)
|
||||
limit[i] = c + 1
|
||||
break
|
||||
}
|
||||
return limit
|
||||
}
|
||||
|
||||
// Stat returns the internal metrics of Pebble in a text format. It's a developer
|
||||
// method to read everything there is to read independent of Pebble version.
|
||||
//
|
||||
// The property is unused in Pebble as there's only one thing to retrieve.
|
||||
func (d *Database) Stat(property string) (string, error) {
|
||||
return d.db.Metrics().String(), nil
|
||||
}
|
||||
|
||||
// Compact flattens the underlying data store for the given key range. In essence,
|
||||
// deleted and overwritten versions are discarded, and the data is rearranged to
|
||||
// reduce the cost of operations needed to access them.
|
||||
//
|
||||
// A nil start is treated as a key before all keys in the data store; a nil limit
|
||||
// is treated as a key after all keys in the data store. If both is nil then it
|
||||
// will compact entire data store.
|
||||
func (d *Database) Compact(start []byte, limit []byte) error {
|
||||
// There is no special flag to represent the end of key range
|
||||
// in pebble(nil in leveldb). Use an ugly hack to construct a
|
||||
// large key to represent it.
|
||||
// Note any prefixed database entry will be smaller than this
|
||||
// flag, as for trie nodes we need the 32 byte 0xff because
|
||||
// there might be a shared prefix starting with a number of
|
||||
// 0xff-s, so 32 ensures than only a hash collision could touch it.
|
||||
// https://github.com/cockroachdb/pebble/issues/2359#issuecomment-1443995833
|
||||
if limit == nil {
|
||||
limit = bytes.Repeat([]byte{0xff}, 32)
|
||||
}
|
||||
return d.db.Compact(start, limit, true) // Parallelization is preferred
|
||||
}
|
||||
|
||||
// Path returns the path to the database directory.
|
||||
func (d *Database) Path() string {
|
||||
return d.fn
|
||||
}
|
||||
|
||||
// meter periodically retrieves internal pebble counters and reports them to
|
||||
// the metrics subsystem.
|
||||
func (d *Database) meter(refresh time.Duration, namespace string) {
|
||||
var errc chan error
|
||||
timer := time.NewTimer(refresh)
|
||||
defer timer.Stop()
|
||||
|
||||
// Create storage and warning log tracer for write delay.
|
||||
var (
|
||||
compTimes [2]int64
|
||||
writeDelayTimes [2]int64
|
||||
writeDelayCounts [2]int64
|
||||
compWrites [2]int64
|
||||
compReads [2]int64
|
||||
|
||||
nWrites [2]int64
|
||||
)
|
||||
|
||||
// Iterate ad infinitum and collect the stats
|
||||
for i := 1; errc == nil; i++ {
|
||||
var (
|
||||
compWrite int64
|
||||
compRead int64
|
||||
nWrite int64
|
||||
|
||||
stats = d.db.Metrics()
|
||||
compTime = d.compTime.Load()
|
||||
writeDelayCount = d.writeDelayCount.Load()
|
||||
writeDelayTime = d.writeDelayTime.Load()
|
||||
nonLevel0CompCount = int64(d.nonLevel0Comp.Load())
|
||||
level0CompCount = int64(d.level0Comp.Load())
|
||||
)
|
||||
writeDelayTimes[i%2] = writeDelayTime
|
||||
writeDelayCounts[i%2] = writeDelayCount
|
||||
compTimes[i%2] = compTime
|
||||
|
||||
for _, levelMetrics := range stats.Levels {
|
||||
nWrite += int64(levelMetrics.BytesCompacted)
|
||||
nWrite += int64(levelMetrics.BytesFlushed)
|
||||
compWrite += int64(levelMetrics.BytesCompacted)
|
||||
compRead += int64(levelMetrics.BytesRead)
|
||||
}
|
||||
|
||||
nWrite += int64(stats.WAL.BytesWritten)
|
||||
|
||||
compWrites[i%2] = compWrite
|
||||
compReads[i%2] = compRead
|
||||
nWrites[i%2] = nWrite
|
||||
|
||||
if d.writeDelayNMeter != nil {
|
||||
d.writeDelayNMeter.Mark(writeDelayCounts[i%2] - writeDelayCounts[(i-1)%2])
|
||||
}
|
||||
if d.writeDelayMeter != nil {
|
||||
d.writeDelayMeter.Mark(writeDelayTimes[i%2] - writeDelayTimes[(i-1)%2])
|
||||
}
|
||||
if d.compTimeMeter != nil {
|
||||
d.compTimeMeter.Mark(compTimes[i%2] - compTimes[(i-1)%2])
|
||||
}
|
||||
if d.compReadMeter != nil {
|
||||
d.compReadMeter.Mark(compReads[i%2] - compReads[(i-1)%2])
|
||||
}
|
||||
if d.compWriteMeter != nil {
|
||||
d.compWriteMeter.Mark(compWrites[i%2] - compWrites[(i-1)%2])
|
||||
}
|
||||
if d.diskSizeGauge != nil {
|
||||
d.diskSizeGauge.Update(int64(stats.DiskSpaceUsage()))
|
||||
}
|
||||
if d.diskReadMeter != nil {
|
||||
d.diskReadMeter.Mark(0) // pebble doesn't track non-compaction reads
|
||||
}
|
||||
if d.diskWriteMeter != nil {
|
||||
d.diskWriteMeter.Mark(nWrites[i%2] - nWrites[(i-1)%2])
|
||||
}
|
||||
// See https://github.com/cockroachdb/pebble/pull/1628#pullrequestreview-1026664054
|
||||
manuallyAllocated := stats.BlockCache.Size + int64(stats.MemTable.Size) + int64(stats.MemTable.ZombieSize)
|
||||
d.manualMemAllocGauge.Update(manuallyAllocated)
|
||||
d.memCompGauge.Update(stats.Flush.Count)
|
||||
d.nonlevel0CompGauge.Update(nonLevel0CompCount)
|
||||
d.level0CompGauge.Update(level0CompCount)
|
||||
d.seekCompGauge.Update(stats.Compact.ReadCount)
|
||||
|
||||
for i, level := range stats.Levels {
|
||||
// Append metrics for additional layers
|
||||
if i >= len(d.levelsGauge) {
|
||||
d.levelsGauge = append(d.levelsGauge, metrics.NewRegisteredGauge(namespace+fmt.Sprintf("tables/level%v", i), nil))
|
||||
}
|
||||
d.levelsGauge[i].Update(level.NumFiles)
|
||||
}
|
||||
|
||||
// Sleep a bit, then repeat the stats collection
|
||||
select {
|
||||
case errc = <-d.quitChan:
|
||||
// Quit requesting, stop hammering the database
|
||||
case <-timer.C:
|
||||
timer.Reset(refresh)
|
||||
// Timeout, gather a new set of stats
|
||||
}
|
||||
}
|
||||
errc <- nil
|
||||
}
|
||||
|
||||
// batch is a write-only batch that commits changes to its host database
|
||||
// when Write is called. A batch cannot be used concurrently.
|
||||
type batch struct {
|
||||
b *pebble.Batch
|
||||
db *Database
|
||||
size int
|
||||
}
|
||||
|
||||
// Put inserts the given value into the batch for later committing.
|
||||
func (b *batch) Put(key, value []byte) error {
|
||||
b.b.Set(key, value, nil)
|
||||
b.size += len(key) + len(value)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete inserts the a key removal into the batch for later committing.
|
||||
func (b *batch) Delete(key []byte) error {
|
||||
b.b.Delete(key, nil)
|
||||
b.size += len(key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValueSize retrieves the amount of data queued up for writing.
|
||||
func (b *batch) ValueSize() int {
|
||||
return b.size
|
||||
}
|
||||
|
||||
// Write flushes any accumulated data to disk.
|
||||
func (b *batch) Write() error {
|
||||
b.db.quitLock.RLock()
|
||||
defer b.db.quitLock.RUnlock()
|
||||
if b.db.closed {
|
||||
return pebble.ErrClosed
|
||||
}
|
||||
return b.b.Commit(b.db.writeOptions)
|
||||
}
|
||||
|
||||
// Reset resets the batch for reuse.
|
||||
func (b *batch) Reset() {
|
||||
b.b.Reset()
|
||||
b.size = 0
|
||||
}
|
||||
|
||||
// Replay replays the batch contents.
|
||||
func (b *batch) Replay(w ethdb.KeyValueWriter) error {
|
||||
reader := b.b.Reader()
|
||||
for {
|
||||
kind, k, v, ok := reader.Next()
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
// The (k,v) slices might be overwritten if the batch is reset/reused,
|
||||
// and the receiver should copy them if they are to be retained long-term.
|
||||
if kind == pebble.InternalKeyKindSet {
|
||||
w.Put(k, v)
|
||||
} else if kind == pebble.InternalKeyKindDelete {
|
||||
w.Delete(k)
|
||||
} else {
|
||||
return fmt.Errorf("unhandled operation, keytype: %v", kind)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// pebbleIterator is a wrapper of underlying iterator in storage engine.
|
||||
// The purpose of this structure is to implement the missing APIs.
|
||||
//
|
||||
// The pebble iterator is not thread-safe.
|
||||
type pebbleIterator struct {
|
||||
iter *pebble.Iterator
|
||||
moved bool
|
||||
released bool
|
||||
}
|
||||
|
||||
// NewIterator creates a binary-alphabetical iterator over a subset
|
||||
// of database content with a particular key prefix, starting at a particular
|
||||
// initial key (or after, if it does not exist).
|
||||
func (d *Database) NewIterator(prefix []byte, start []byte) ethdb.Iterator {
|
||||
iter, _ := d.db.NewIter(&pebble.IterOptions{
|
||||
LowerBound: append(prefix, start...),
|
||||
UpperBound: upperBound(prefix),
|
||||
})
|
||||
iter.First()
|
||||
return &pebbleIterator{iter: iter, moved: true, released: false}
|
||||
}
|
||||
|
||||
// Next moves the iterator to the next key/value pair. It returns whether the
|
||||
// iterator is exhausted.
|
||||
func (iter *pebbleIterator) Next() bool {
|
||||
if iter.moved {
|
||||
iter.moved = false
|
||||
return iter.iter.Valid()
|
||||
}
|
||||
return iter.iter.Next()
|
||||
}
|
||||
|
||||
// Error returns any accumulated error. Exhausting all the key/value pairs
|
||||
// is not considered to be an error.
|
||||
func (iter *pebbleIterator) Error() error {
|
||||
return iter.iter.Error()
|
||||
}
|
||||
|
||||
// Key returns the key of the current key/value pair, or nil if done. The caller
|
||||
// should not modify the contents of the returned slice, and its contents may
|
||||
// change on the next call to Next.
|
||||
func (iter *pebbleIterator) Key() []byte {
|
||||
return iter.iter.Key()
|
||||
}
|
||||
|
||||
// Value returns the value of the current key/value pair, or nil if done. The
|
||||
// caller should not modify the contents of the returned slice, and its contents
|
||||
// may change on the next call to Next.
|
||||
func (iter *pebbleIterator) Value() []byte {
|
||||
return iter.iter.Value()
|
||||
}
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
func (iter *pebbleIterator) Release() {
|
||||
if !iter.released {
|
||||
iter.iter.Close()
|
||||
iter.released = true
|
||||
}
|
||||
}
|
||||
|
|
@ -1,56 +0,0 @@
|
|||
// Copyright 2023 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package pebble
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/cockroachdb/pebble"
|
||||
"github.com/cockroachdb/pebble/vfs"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/ethdb/dbtest"
|
||||
)
|
||||
|
||||
func TestPebbleDB(t *testing.T) {
|
||||
t.Run("DatabaseSuite", func(t *testing.T) {
|
||||
dbtest.TestDatabaseSuite(t, func() ethdb.KeyValueStore {
|
||||
db, err := pebble.Open("", &pebble.Options{
|
||||
FS: vfs.NewMem(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &Database{
|
||||
db: db,
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkPebbleDB(b *testing.B) {
|
||||
dbtest.BenchDatabaseSuite(b, func() ethdb.KeyValueStore {
|
||||
db, err := pebble.Open("", &pebble.Options{
|
||||
FS: vfs.NewMem(),
|
||||
})
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
return &Database{
|
||||
db: db,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
@ -1,154 +0,0 @@
|
|||
// Copyright 2022 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
// Package remotedb implements the key-value database layer based on a remote geth
|
||||
// node. Under the hood, it utilises the `debug_dbGet` method to implement a
|
||||
// read-only database.
|
||||
// There really are no guarantees in this database, since the local geth does not
|
||||
// exclusive access, but it can be used for basic diagnostics of a remote node.
|
||||
package remotedb
|
||||
|
||||
import (
|
||||
"github.com/ethereum/go-ethereum/common/hexutil"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
)
|
||||
|
||||
// Database is a key-value lookup for a remote database via debug_dbGet.
|
||||
type Database struct {
|
||||
remote *rpc.Client
|
||||
}
|
||||
|
||||
func (db *Database) Has(key []byte) (bool, error) {
|
||||
if _, err := db.Get(key); err != nil {
|
||||
return false, nil
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (db *Database) Get(key []byte) ([]byte, error) {
|
||||
var resp hexutil.Bytes
|
||||
err := db.remote.Call(&resp, "debug_dbGet", hexutil.Bytes(key))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (db *Database) HasAncient(kind string, number uint64) (bool, error) {
|
||||
if _, err := db.Ancient(kind, number); err != nil {
|
||||
return false, nil
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (db *Database) Ancient(kind string, number uint64) ([]byte, error) {
|
||||
var resp hexutil.Bytes
|
||||
err := db.remote.Call(&resp, "debug_dbAncient", kind, number)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (db *Database) AncientRange(kind string, start, count, maxBytes uint64) ([][]byte, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) Ancients() (uint64, error) {
|
||||
var resp uint64
|
||||
err := db.remote.Call(&resp, "debug_dbAncients")
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func (db *Database) Tail() (uint64, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) AncientSize(kind string) (uint64, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) ReadAncients(fn func(op ethdb.AncientReaderOp) error) (err error) {
|
||||
return fn(db)
|
||||
}
|
||||
|
||||
func (db *Database) Put(key []byte, value []byte) error {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) Delete(key []byte) error {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) ModifyAncients(f func(ethdb.AncientWriteOp) error) (int64, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) TruncateHead(n uint64) (uint64, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) TruncateTail(n uint64) (uint64, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) Sync() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (db *Database) MigrateTable(s string, f func([]byte) ([]byte, error)) error {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) NewBatch() ethdb.Batch {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) NewBatchWithSize(size int) ethdb.Batch {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) NewIterator(prefix []byte, start []byte) ethdb.Iterator {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) Stat(property string) (string, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) AncientDatadir() (string, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) Compact(start []byte, limit []byte) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (db *Database) NewSnapshot() (ethdb.Snapshot, error) {
|
||||
panic("not supported")
|
||||
}
|
||||
|
||||
func (db *Database) Close() error {
|
||||
db.remote.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
func New(client *rpc.Client) ethdb.Database {
|
||||
return &Database{
|
||||
remote: client,
|
||||
}
|
||||
}
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
// Copyright 2022 The go-ethereum Authors
|
||||
// This file is part of the go-ethereum library.
|
||||
//
|
||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Lesser General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// The go-ethereum library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Lesser General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package ethdb
|
||||
|
||||
type Snapshot interface {
|
||||
// Has retrieves if a key is present in the snapshot backing by a key-value
|
||||
// data store.
|
||||
Has(key []byte) (bool, error)
|
||||
|
||||
// Get retrieves the given key if it's present in the snapshot backing by
|
||||
// key-value data store.
|
||||
Get(key []byte) ([]byte, error)
|
||||
|
||||
// Release releases associated resources. Release should always succeed and can
|
||||
// be called multiple times without causing error.
|
||||
Release()
|
||||
}
|
||||
|
||||
// Snapshotter wraps the Snapshot method of a backing data store.
|
||||
type Snapshotter interface {
|
||||
// NewSnapshot creates a database snapshot based on the current state.
|
||||
// The created snapshot will not be affected by all following mutations
|
||||
// happened on the database.
|
||||
// Note don't forget to release the snapshot once it's used up, otherwise
|
||||
// the stale data will never be cleaned up by the underlying compactor.
|
||||
NewSnapshot() (Snapshot, error)
|
||||
}
|
||||
Loading…
Reference in a new issue