forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Delete ethdb directory

Browse source 
Signed-off-by: Isabel Schöps Thiel @IsabelSchoepd <155141998+IST-Github@users.noreply.github.com>
This commit is contained in:
Isabel Schöps Thiel @IsabelSchoepd 2024-01-04 04:33:23 +01:00 committed by GitHub
parent d9efc41814
commit bf02be3d0c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 0 additions and 2760 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
ethdb/batch.go
View file

@ -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)
}

192
ethdb/database.go
View file

@ -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
}

537
ethdb/dbtest/testsuite.go
View file

@ -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
}

61
ethdb/iterator.go
View file

@ -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
}

485
ethdb/leveldb/leveldb.go
View file

@ -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()
}

52
ethdb/leveldb/leveldb_test.go
View file

@ -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,
}
})
}

390
ethdb/memorydb/memorydb.go
View file

@ -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
}

50
ethdb/memorydb/memorydb_test.go
View file

@ -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()
}
}

668
ethdb/pebble/pebble.go
View file

@ -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
}
}

56
ethdb/pebble/pebble_test.go
View file

@ -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,
}
})
}

154
ethdb/remotedb/remotedb.go
View file

@ -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,
}
}

41
ethdb/snapshot.go
View file

@ -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)
}