mirror of https://github.com/ethereum/go-ethereum
core/rawdb: implement in-memory freezer (#29135)
parent
c04b8e6d74
commit
f46c878441
@ -0,0 +1,325 @@ |
|||||||
|
// Copyright 2024 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 ancienttest |
||||||
|
|
||||||
|
import ( |
||||||
|
"bytes" |
||||||
|
"reflect" |
||||||
|
"testing" |
||||||
|
|
||||||
|
"github.com/ethereum/go-ethereum/ethdb" |
||||||
|
"github.com/ethereum/go-ethereum/internal/testrand" |
||||||
|
) |
||||||
|
|
||||||
|
// TestAncientSuite runs a suite of tests against an ancient database
|
||||||
|
// implementation.
|
||||||
|
func TestAncientSuite(t *testing.T, newFn func(kinds []string) ethdb.AncientStore) { |
||||||
|
// Test basic read methods
|
||||||
|
t.Run("BasicRead", func(t *testing.T) { basicRead(t, newFn) }) |
||||||
|
|
||||||
|
// Test batch read method
|
||||||
|
t.Run("BatchRead", func(t *testing.T) { batchRead(t, newFn) }) |
||||||
|
|
||||||
|
// Test basic write methods
|
||||||
|
t.Run("BasicWrite", func(t *testing.T) { basicWrite(t, newFn) }) |
||||||
|
|
||||||
|
// Test if data mutation is allowed after db write
|
||||||
|
t.Run("nonMutable", func(t *testing.T) { nonMutable(t, newFn) }) |
||||||
|
} |
||||||
|
|
||||||
|
func basicRead(t *testing.T, newFn func(kinds []string) ethdb.AncientStore) { |
||||||
|
var ( |
||||||
|
db = newFn([]string{"a"}) |
||||||
|
data = makeDataset(100, 32) |
||||||
|
) |
||||||
|
defer db.Close() |
||||||
|
|
||||||
|
db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < len(data); i++ { |
||||||
|
op.AppendRaw("a", uint64(i), data[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
db.TruncateTail(10) |
||||||
|
db.TruncateHead(90) |
||||||
|
|
||||||
|
// Test basic tail and head retrievals
|
||||||
|
tail, err := db.Tail() |
||||||
|
if err != nil || tail != 10 { |
||||||
|
t.Fatal("Failed to retrieve tail") |
||||||
|
} |
||||||
|
ancient, err := db.Ancients() |
||||||
|
if err != nil || ancient != 90 { |
||||||
|
t.Fatal("Failed to retrieve ancient") |
||||||
|
} |
||||||
|
|
||||||
|
// Test the deleted items shouldn't be reachable
|
||||||
|
var cases = []struct { |
||||||
|
start int |
||||||
|
limit int |
||||||
|
}{ |
||||||
|
{0, 10}, |
||||||
|
{90, 100}, |
||||||
|
} |
||||||
|
for _, c := range cases { |
||||||
|
for i := c.start; i < c.limit; i++ { |
||||||
|
exist, err := db.HasAncient("a", uint64(i)) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to check presence, %v", err) |
||||||
|
} |
||||||
|
if exist { |
||||||
|
t.Fatalf("Item %d is already truncated", uint64(i)) |
||||||
|
} |
||||||
|
_, err = db.Ancient("a", uint64(i)) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Error is expected for non-existent item") |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// Test the items in range should be reachable
|
||||||
|
for i := 10; i < 90; i++ { |
||||||
|
exist, err := db.HasAncient("a", uint64(i)) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to check presence, %v", err) |
||||||
|
} |
||||||
|
if !exist { |
||||||
|
t.Fatalf("Item %d is missing", uint64(i)) |
||||||
|
} |
||||||
|
blob, err := db.Ancient("a", uint64(i)) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to retrieve item, %v", err) |
||||||
|
} |
||||||
|
if !bytes.Equal(blob, data[i]) { |
||||||
|
t.Fatalf("Unexpected item content, want: %v, got: %v", data[i], blob) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// Test the items in unknown table shouldn't be reachable
|
||||||
|
exist, err := db.HasAncient("b", uint64(0)) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to check presence, %v", err) |
||||||
|
} |
||||||
|
if exist { |
||||||
|
t.Fatal("Item in unknown table shouldn't be found") |
||||||
|
} |
||||||
|
_, err = db.Ancient("b", uint64(0)) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Error is expected for unknown table") |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func batchRead(t *testing.T, newFn func(kinds []string) ethdb.AncientStore) { |
||||||
|
var ( |
||||||
|
db = newFn([]string{"a"}) |
||||||
|
data = makeDataset(100, 32) |
||||||
|
) |
||||||
|
defer db.Close() |
||||||
|
|
||||||
|
db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), data[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
db.TruncateTail(10) |
||||||
|
db.TruncateHead(90) |
||||||
|
|
||||||
|
// Test the items in range should be reachable
|
||||||
|
var cases = []struct { |
||||||
|
start uint64 |
||||||
|
count uint64 |
||||||
|
maxSize uint64 |
||||||
|
expStart int |
||||||
|
expLimit int |
||||||
|
}{ |
||||||
|
// Items in range [10, 90) with no size limitation
|
||||||
|
{ |
||||||
|
10, 80, 0, 10, 90, |
||||||
|
}, |
||||||
|
// Items in range [10, 90) with 32 size cap, single item is expected
|
||||||
|
{ |
||||||
|
10, 80, 32, 10, 11, |
||||||
|
}, |
||||||
|
// Items in range [10, 90) with 31 size cap, single item is expected
|
||||||
|
{ |
||||||
|
10, 80, 31, 10, 11, |
||||||
|
}, |
||||||
|
// Items in range [10, 90) with 32*80 size cap, all items are expected
|
||||||
|
{ |
||||||
|
10, 80, 32 * 80, 10, 90, |
||||||
|
}, |
||||||
|
// Extra items above the last item are not returned
|
||||||
|
{ |
||||||
|
10, 90, 0, 10, 90, |
||||||
|
}, |
||||||
|
} |
||||||
|
for i, c := range cases { |
||||||
|
batch, err := db.AncientRange("a", c.start, c.count, c.maxSize) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to retrieve item in range, %v", err) |
||||||
|
} |
||||||
|
if !reflect.DeepEqual(batch, data[c.expStart:c.expLimit]) { |
||||||
|
t.Fatalf("Case %d, Batch content is not matched", i) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// Test out-of-range / zero-size retrieval should be rejected
|
||||||
|
_, err := db.AncientRange("a", 0, 1, 0) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Out-of-range retrieval should be rejected") |
||||||
|
} |
||||||
|
_, err = db.AncientRange("a", 90, 1, 0) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Out-of-range retrieval should be rejected") |
||||||
|
} |
||||||
|
_, err = db.AncientRange("a", 10, 0, 0) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Zero-size retrieval should be rejected") |
||||||
|
} |
||||||
|
|
||||||
|
// Test item in unknown table shouldn't be reachable
|
||||||
|
_, err = db.AncientRange("b", 10, 1, 0) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Item in unknown table shouldn't be found") |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func basicWrite(t *testing.T, newFn func(kinds []string) ethdb.AncientStore) { |
||||||
|
var ( |
||||||
|
db = newFn([]string{"a", "b"}) |
||||||
|
dataA = makeDataset(100, 32) |
||||||
|
dataB = makeDataset(100, 32) |
||||||
|
) |
||||||
|
defer db.Close() |
||||||
|
|
||||||
|
// The ancient write to tables should be aligned
|
||||||
|
_, err := db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), dataA[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
if err == nil { |
||||||
|
t.Fatal("Unaligned ancient write should be rejected") |
||||||
|
} |
||||||
|
|
||||||
|
// Test normal ancient write
|
||||||
|
size, err := db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), dataA[i]) |
||||||
|
op.AppendRaw("b", uint64(i), dataB[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to write ancient data %v", err) |
||||||
|
} |
||||||
|
wantSize := int64(6400) |
||||||
|
if size != wantSize { |
||||||
|
t.Fatalf("Ancient write size is not expected, want: %d, got: %d", wantSize, size) |
||||||
|
} |
||||||
|
|
||||||
|
// Write should work after head truncating
|
||||||
|
db.TruncateHead(90) |
||||||
|
_, err = db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 90; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), dataA[i]) |
||||||
|
op.AppendRaw("b", uint64(i), dataB[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to write ancient data %v", err) |
||||||
|
} |
||||||
|
|
||||||
|
// Write should work after truncating everything
|
||||||
|
db.TruncateTail(0) |
||||||
|
_, err = db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), dataA[i]) |
||||||
|
op.AppendRaw("b", uint64(i), dataB[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
if err != nil { |
||||||
|
t.Fatalf("Failed to write ancient data %v", err) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func nonMutable(t *testing.T, newFn func(kinds []string) ethdb.AncientStore) { |
||||||
|
db := newFn([]string{"a"}) |
||||||
|
defer db.Close() |
||||||
|
|
||||||
|
// We write 100 zero-bytes to the freezer and immediately mutate the slice
|
||||||
|
db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
data := make([]byte, 100) |
||||||
|
op.AppendRaw("a", uint64(0), data) |
||||||
|
for i := range data { |
||||||
|
data[i] = 0xff |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
// Now read it.
|
||||||
|
data, err := db.Ancient("a", uint64(0)) |
||||||
|
if err != nil { |
||||||
|
t.Fatal(err) |
||||||
|
} |
||||||
|
for k, v := range data { |
||||||
|
if v != 0 { |
||||||
|
t.Fatalf("byte %d != 0: %x", k, v) |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// TestResettableAncientSuite runs a suite of tests against a resettable ancient
|
||||||
|
// database implementation.
|
||||||
|
func TestResettableAncientSuite(t *testing.T, newFn func(kinds []string) ethdb.ResettableAncientStore) { |
||||||
|
t.Run("Reset", func(t *testing.T) { |
||||||
|
var ( |
||||||
|
db = newFn([]string{"a"}) |
||||||
|
data = makeDataset(100, 32) |
||||||
|
) |
||||||
|
defer db.Close() |
||||||
|
|
||||||
|
db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), data[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
db.TruncateTail(10) |
||||||
|
db.TruncateHead(90) |
||||||
|
|
||||||
|
// Ancient write should work after resetting
|
||||||
|
db.Reset() |
||||||
|
db.ModifyAncients(func(op ethdb.AncientWriteOp) error { |
||||||
|
for i := 0; i < 100; i++ { |
||||||
|
op.AppendRaw("a", uint64(i), data[i]) |
||||||
|
} |
||||||
|
return nil |
||||||
|
}) |
||||||
|
}) |
||||||
|
} |
||||||
|
|
||||||
|
func makeDataset(size, value int) [][]byte { |
||||||
|
var vals [][]byte |
||||||
|
for i := 0; i < size; i += 1 { |
||||||
|
vals = append(vals, testrand.Bytes(value)) |
||||||
|
} |
||||||
|
return vals |
||||||
|
} |
@ -0,0 +1,428 @@ |
|||||||
|
// Copyright 2024 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 rawdb |
||||||
|
|
||||||
|
import ( |
||||||
|
"errors" |
||||||
|
"fmt" |
||||||
|
"sync" |
||||||
|
|
||||||
|
"github.com/ethereum/go-ethereum/common" |
||||||
|
"github.com/ethereum/go-ethereum/common/math" |
||||||
|
"github.com/ethereum/go-ethereum/ethdb" |
||||||
|
"github.com/ethereum/go-ethereum/log" |
||||||
|
"github.com/ethereum/go-ethereum/rlp" |
||||||
|
) |
||||||
|
|
||||||
|
// memoryTable is used to store a list of sequential items in memory.
|
||||||
|
type memoryTable struct { |
||||||
|
name string // Table name
|
||||||
|
items uint64 // Number of stored items in the table, including the deleted ones
|
||||||
|
offset uint64 // Number of deleted items from the table
|
||||||
|
data [][]byte // List of rlp-encoded items, sort in order
|
||||||
|
size uint64 // Total memory size occupied by the table
|
||||||
|
lock sync.RWMutex |
||||||
|
} |
||||||
|
|
||||||
|
// newMemoryTable initializes the memory table.
|
||||||
|
func newMemoryTable(name string) *memoryTable { |
||||||
|
return &memoryTable{name: name} |
||||||
|
} |
||||||
|
|
||||||
|
// has returns an indicator whether the specified data exists.
|
||||||
|
func (t *memoryTable) has(number uint64) bool { |
||||||
|
t.lock.RLock() |
||||||
|
defer t.lock.RUnlock() |
||||||
|
|
||||||
|
return number >= t.offset && number < t.items |
||||||
|
} |
||||||
|
|
||||||
|
// retrieve 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
|
||||||
|
func (t *memoryTable) retrieve(start uint64, count, maxBytes uint64) ([][]byte, error) { |
||||||
|
t.lock.RLock() |
||||||
|
defer t.lock.RUnlock() |
||||||
|
|
||||||
|
var ( |
||||||
|
size uint64 |
||||||
|
batch [][]byte |
||||||
|
) |
||||||
|
// Ensure the start is written, not deleted from the tail, and that the
|
||||||
|
// caller actually wants something.
|
||||||
|
if t.items <= start || t.offset > start || count == 0 { |
||||||
|
return nil, errOutOfBounds |
||||||
|
} |
||||||
|
// Cap the item count if the retrieval is out of bound.
|
||||||
|
if start+count > t.items { |
||||||
|
count = t.items - start |
||||||
|
} |
||||||
|
for n := start; n < start+count; n++ { |
||||||
|
index := n - t.offset |
||||||
|
if len(batch) != 0 && maxBytes != 0 && size+uint64(len(t.data[index])) > maxBytes { |
||||||
|
return batch, nil |
||||||
|
} |
||||||
|
batch = append(batch, t.data[index]) |
||||||
|
size += uint64(len(t.data[index])) |
||||||
|
} |
||||||
|
return batch, nil |
||||||
|
} |
||||||
|
|
||||||
|
// truncateHead discards any recent data above the provided threshold number.
|
||||||
|
func (t *memoryTable) truncateHead(items uint64) error { |
||||||
|
t.lock.Lock() |
||||||
|
defer t.lock.Unlock() |
||||||
|
|
||||||
|
// Short circuit if nothing to delete.
|
||||||
|
if t.items <= items { |
||||||
|
return nil |
||||||
|
} |
||||||
|
if items < t.offset { |
||||||
|
return errors.New("truncation below tail") |
||||||
|
} |
||||||
|
t.data = t.data[:items-t.offset] |
||||||
|
t.items = items |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// truncateTail discards any recent data before the provided threshold number.
|
||||||
|
func (t *memoryTable) truncateTail(items uint64) error { |
||||||
|
t.lock.Lock() |
||||||
|
defer t.lock.Unlock() |
||||||
|
|
||||||
|
// Short circuit if nothing to delete.
|
||||||
|
if t.offset >= items { |
||||||
|
return nil |
||||||
|
} |
||||||
|
if t.items < items { |
||||||
|
return errors.New("truncation above head") |
||||||
|
} |
||||||
|
t.data = t.data[items-t.offset:] |
||||||
|
t.offset = items |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// commit merges the given item batch into table. It's presumed that the
|
||||||
|
// batch is ordered and continuous with table.
|
||||||
|
func (t *memoryTable) commit(batch [][]byte) error { |
||||||
|
t.lock.Lock() |
||||||
|
defer t.lock.Unlock() |
||||||
|
|
||||||
|
for _, item := range batch { |
||||||
|
t.size += uint64(len(item)) |
||||||
|
} |
||||||
|
t.data = append(t.data, batch...) |
||||||
|
t.items += uint64(len(batch)) |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// memoryBatch is the singleton batch used for ancient write.
|
||||||
|
type memoryBatch struct { |
||||||
|
data map[string][][]byte |
||||||
|
next map[string]uint64 |
||||||
|
size map[string]int64 |
||||||
|
} |
||||||
|
|
||||||
|
func newMemoryBatch() *memoryBatch { |
||||||
|
return &memoryBatch{ |
||||||
|
data: make(map[string][][]byte), |
||||||
|
next: make(map[string]uint64), |
||||||
|
size: make(map[string]int64), |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func (b *memoryBatch) reset(freezer *MemoryFreezer) { |
||||||
|
b.data = make(map[string][][]byte) |
||||||
|
b.next = make(map[string]uint64) |
||||||
|
b.size = make(map[string]int64) |
||||||
|
|
||||||
|
for name, table := range freezer.tables { |
||||||
|
b.next[name] = table.items |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// Append adds an RLP-encoded item.
|
||||||
|
func (b *memoryBatch) Append(kind string, number uint64, item interface{}) error { |
||||||
|
if b.next[kind] != number { |
||||||
|
return errOutOrderInsertion |
||||||
|
} |
||||||
|
blob, err := rlp.EncodeToBytes(item) |
||||||
|
if err != nil { |
||||||
|
return err |
||||||
|
} |
||||||
|
b.data[kind] = append(b.data[kind], blob) |
||||||
|
b.next[kind]++ |
||||||
|
b.size[kind] += int64(len(blob)) |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// AppendRaw adds an item without RLP-encoding it.
|
||||||
|
func (b *memoryBatch) AppendRaw(kind string, number uint64, blob []byte) error { |
||||||
|
if b.next[kind] != number { |
||||||
|
return errOutOrderInsertion |
||||||
|
} |
||||||
|
b.data[kind] = append(b.data[kind], common.CopyBytes(blob)) |
||||||
|
b.next[kind]++ |
||||||
|
b.size[kind] += int64(len(blob)) |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// commit is called at the end of a write operation and writes all remaining
|
||||||
|
// data to tables.
|
||||||
|
func (b *memoryBatch) commit(freezer *MemoryFreezer) (items uint64, writeSize int64, err error) { |
||||||
|
// Check that count agrees on all batches.
|
||||||
|
items = math.MaxUint64 |
||||||
|
for name, next := range b.next { |
||||||
|
if items < math.MaxUint64 && next != items { |
||||||
|
return 0, 0, fmt.Errorf("table %s is at item %d, want %d", name, next, items) |
||||||
|
} |
||||||
|
items = next |
||||||
|
} |
||||||
|
// Commit all table batches.
|
||||||
|
for name, batch := range b.data { |
||||||
|
table := freezer.tables[name] |
||||||
|
if err := table.commit(batch); err != nil { |
||||||
|
return 0, 0, err |
||||||
|
} |
||||||
|
writeSize += b.size[name] |
||||||
|
} |
||||||
|
return items, writeSize, nil |
||||||
|
} |
||||||
|
|
||||||
|
// MemoryFreezer is an ephemeral ancient store. It implements the ethdb.AncientStore
|
||||||
|
// interface and can be used along with ephemeral key-value store.
|
||||||
|
type MemoryFreezer struct { |
||||||
|
items uint64 // Number of items stored
|
||||||
|
tail uint64 // Number of the first stored item in the freezer
|
||||||
|
readonly bool // Flag if the freezer is only for reading
|
||||||
|
lock sync.RWMutex // Lock to protect fields
|
||||||
|
tables map[string]*memoryTable // Tables for storing everything
|
||||||
|
writeBatch *memoryBatch // Pre-allocated write batch
|
||||||
|
} |
||||||
|
|
||||||
|
// NewMemoryFreezer initializes an in-memory freezer instance.
|
||||||
|
func NewMemoryFreezer(readonly bool, tableName map[string]bool) *MemoryFreezer { |
||||||
|
tables := make(map[string]*memoryTable) |
||||||
|
for name := range tableName { |
||||||
|
tables[name] = newMemoryTable(name) |
||||||
|
} |
||||||
|
return &MemoryFreezer{ |
||||||
|
writeBatch: newMemoryBatch(), |
||||||
|
readonly: readonly, |
||||||
|
tables: tables, |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// HasAncient returns an indicator whether the specified data exists.
|
||||||
|
func (f *MemoryFreezer) HasAncient(kind string, number uint64) (bool, error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
if table := f.tables[kind]; table != nil { |
||||||
|
return table.has(number), nil |
||||||
|
} |
||||||
|
return false, nil |
||||||
|
} |
||||||
|
|
||||||
|
// Ancient retrieves an ancient binary blob from the in-memory freezer.
|
||||||
|
func (f *MemoryFreezer) Ancient(kind string, number uint64) ([]byte, error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
t := f.tables[kind] |
||||||
|
if t == nil { |
||||||
|
return nil, errUnknownTable |
||||||
|
} |
||||||
|
data, err := t.retrieve(number, 1, 0) |
||||||
|
if err != nil { |
||||||
|
return nil, err |
||||||
|
} |
||||||
|
return data[0], nil |
||||||
|
} |
||||||
|
|
||||||
|
// 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
|
||||||
|
func (f *MemoryFreezer) AncientRange(kind string, start, count, maxBytes uint64) ([][]byte, error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
t := f.tables[kind] |
||||||
|
if t == nil { |
||||||
|
return nil, errUnknownTable |
||||||
|
} |
||||||
|
return t.retrieve(start, count, maxBytes) |
||||||
|
} |
||||||
|
|
||||||
|
// Ancients returns the ancient item numbers in the freezer.
|
||||||
|
func (f *MemoryFreezer) Ancients() (uint64, error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
return f.items, nil |
||||||
|
} |
||||||
|
|
||||||
|
// Tail returns the number of first stored item in the freezer.
|
||||||
|
// This number can also be interpreted as the total deleted item numbers.
|
||||||
|
func (f *MemoryFreezer) Tail() (uint64, error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
return f.tail, nil |
||||||
|
} |
||||||
|
|
||||||
|
// AncientSize returns the ancient size of the specified category.
|
||||||
|
func (f *MemoryFreezer) AncientSize(kind string) (uint64, error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
if table := f.tables[kind]; table != nil { |
||||||
|
return table.size, nil |
||||||
|
} |
||||||
|
return 0, errUnknownTable |
||||||
|
} |
||||||
|
|
||||||
|
// ReadAncients runs the given read operation while ensuring that no writes take place
|
||||||
|
// on the underlying freezer.
|
||||||
|
func (f *MemoryFreezer) ReadAncients(fn func(ethdb.AncientReaderOp) error) (err error) { |
||||||
|
f.lock.RLock() |
||||||
|
defer f.lock.RUnlock() |
||||||
|
|
||||||
|
return fn(f) |
||||||
|
} |
||||||
|
|
||||||
|
// ModifyAncients runs the given write operation.
|
||||||
|
func (f *MemoryFreezer) ModifyAncients(fn func(ethdb.AncientWriteOp) error) (writeSize int64, err error) { |
||||||
|
f.lock.Lock() |
||||||
|
defer f.lock.Unlock() |
||||||
|
|
||||||
|
if f.readonly { |
||||||
|
return 0, errReadOnly |
||||||
|
} |
||||||
|
// Roll back all tables to the starting position in case of error.
|
||||||
|
defer func(old uint64) { |
||||||
|
if err == nil { |
||||||
|
return |
||||||
|
} |
||||||
|
// The write operation has failed. Go back to the previous item position.
|
||||||
|
for name, table := range f.tables { |
||||||
|
err := table.truncateHead(old) |
||||||
|
if err != nil { |
||||||
|
log.Error("Freezer table roll-back failed", "table", name, "index", old, "err", err) |
||||||
|
} |
||||||
|
} |
||||||
|
}(f.items) |
||||||
|
|
||||||
|
// Modify the ancients in batch.
|
||||||
|
f.writeBatch.reset(f) |
||||||
|
if err := fn(f.writeBatch); err != nil { |
||||||
|
return 0, err |
||||||
|
} |
||||||
|
item, writeSize, err := f.writeBatch.commit(f) |
||||||
|
if err != nil { |
||||||
|
return 0, err |
||||||
|
} |
||||||
|
f.items = item |
||||||
|
return writeSize, nil |
||||||
|
} |
||||||
|
|
||||||
|
// TruncateHead discards any recent data above the provided threshold number.
|
||||||
|
// It returns the previous head number.
|
||||||
|
func (f *MemoryFreezer) TruncateHead(items uint64) (uint64, error) { |
||||||
|
f.lock.Lock() |
||||||
|
defer f.lock.Unlock() |
||||||
|
|
||||||
|
if f.readonly { |
||||||
|
return 0, errReadOnly |
||||||
|
} |
||||||
|
old := f.items |
||||||
|
if old <= items { |
||||||
|
return old, nil |
||||||
|
} |
||||||
|
for _, table := range f.tables { |
||||||
|
if err := table.truncateHead(items); err != nil { |
||||||
|
return 0, err |
||||||
|
} |
||||||
|
} |
||||||
|
f.items = items |
||||||
|
return old, nil |
||||||
|
} |
||||||
|
|
||||||
|
// TruncateTail discards any recent data below the provided threshold number.
|
||||||
|
func (f *MemoryFreezer) TruncateTail(tail uint64) (uint64, error) { |
||||||
|
f.lock.Lock() |
||||||
|
defer f.lock.Unlock() |
||||||
|
|
||||||
|
if f.readonly { |
||||||
|
return 0, errReadOnly |
||||||
|
} |
||||||
|
old := f.tail |
||||||
|
if old >= tail { |
||||||
|
return old, nil |
||||||
|
} |
||||||
|
for _, table := range f.tables { |
||||||
|
if err := table.truncateTail(tail); err != nil { |
||||||
|
return 0, err |
||||||
|
} |
||||||
|
} |
||||||
|
f.tail = tail |
||||||
|
return old, nil |
||||||
|
} |
||||||
|
|
||||||
|
// Sync flushes all data tables to disk.
|
||||||
|
func (f *MemoryFreezer) Sync() error { |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// 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.
|
||||||
|
func (f *MemoryFreezer) MigrateTable(string, func([]byte) ([]byte, error)) error { |
||||||
|
return errors.New("not implemented") |
||||||
|
} |
||||||
|
|
||||||
|
// Close releases all the sources held by the memory freezer. It will panic if
|
||||||
|
// any following invocation is made to a closed freezer.
|
||||||
|
func (f *MemoryFreezer) Close() error { |
||||||
|
f.lock.Lock() |
||||||
|
defer f.lock.Unlock() |
||||||
|
|
||||||
|
f.tables = nil |
||||||
|
f.writeBatch = nil |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// Reset drops all the data cached in the memory freezer and reset itself
|
||||||
|
// back to default state.
|
||||||
|
func (f *MemoryFreezer) Reset() error { |
||||||
|
f.lock.Lock() |
||||||
|
defer f.lock.Unlock() |
||||||
|
|
||||||
|
tables := make(map[string]*memoryTable) |
||||||
|
for name := range f.tables { |
||||||
|
tables[name] = newMemoryTable(name) |
||||||
|
} |
||||||
|
f.tables = tables |
||||||
|
f.items, f.tail = 0, 0 |
||||||
|
return nil |
||||||
|
} |
@ -0,0 +1,41 @@ |
|||||||
|
// Copyright 2024 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 rawdb |
||||||
|
|
||||||
|
import ( |
||||||
|
"testing" |
||||||
|
|
||||||
|
"github.com/ethereum/go-ethereum/core/rawdb/ancienttest" |
||||||
|
"github.com/ethereum/go-ethereum/ethdb" |
||||||
|
) |
||||||
|
|
||||||
|
func TestMemoryFreezer(t *testing.T) { |
||||||
|
ancienttest.TestAncientSuite(t, func(kinds []string) ethdb.AncientStore { |
||||||
|
tables := make(map[string]bool) |
||||||
|
for _, kind := range kinds { |
||||||
|
tables[kind] = true |
||||||
|
} |
||||||
|
return NewMemoryFreezer(false, tables) |
||||||
|
}) |
||||||
|
ancienttest.TestResettableAncientSuite(t, func(kinds []string) ethdb.ResettableAncientStore { |
||||||
|
tables := make(map[string]bool) |
||||||
|
for _, kind := range kinds { |
||||||
|
tables[kind] = true |
||||||
|
} |
||||||
|
return NewMemoryFreezer(false, tables) |
||||||
|
}) |
||||||
|
} |
Loading…
Reference in new issue