mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-24 13:46:43 +00:00
no global cache
Signed-off-by: Delweng <delweng@gmail.com>
This commit is contained in:
parent
faca680a13
commit
2b64c5c938
7 changed files with 114 additions and 71 deletions
|
|
@ -60,6 +60,9 @@ const (
|
||||||
var (
|
var (
|
||||||
// maxDiffLayers is the maximum diff layers allowed in the layer tree.
|
// maxDiffLayers is the maximum diff layers allowed in the layer tree.
|
||||||
maxDiffLayers = 128
|
maxDiffLayers = 128
|
||||||
|
|
||||||
|
// historyCacheSize is the maximum size of history cache.
|
||||||
|
historyCacheSize = 4096
|
||||||
)
|
)
|
||||||
|
|
||||||
// layer is the interface implemented by all state layers which includes some
|
// layer is the interface implemented by all state layers which includes some
|
||||||
|
|
@ -225,6 +228,7 @@ type Database struct {
|
||||||
freezer ethdb.ResettableAncientStore // Freezer for storing trie histories, nil possible in tests
|
freezer ethdb.ResettableAncientStore // Freezer for storing trie histories, nil possible in tests
|
||||||
lock sync.RWMutex // Lock to prevent mutations from happening at the same time
|
lock sync.RWMutex // Lock to prevent mutations from happening at the same time
|
||||||
indexer *historyIndexer // History indexer
|
indexer *historyIndexer // History indexer
|
||||||
|
cacher *historyCacher // History cacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// New attempts to load an already existing layer from a persistent key-value
|
// New attempts to load an already existing layer from a persistent key-value
|
||||||
|
|
@ -242,6 +246,7 @@ func New(diskdb ethdb.Database, config *Config, isVerkle bool) *Database {
|
||||||
config: config,
|
config: config,
|
||||||
diskdb: diskdb,
|
diskdb: diskdb,
|
||||||
hasher: merkleNodeHasher,
|
hasher: merkleNodeHasher,
|
||||||
|
cacher: newHistoryCacher(historyCacheSize),
|
||||||
}
|
}
|
||||||
// Establish a dedicated database namespace tailored for verkle-specific
|
// Establish a dedicated database namespace tailored for verkle-specific
|
||||||
// data, ensuring the isolation of both verkle and merkle tree data. It's
|
// data, ensuring the isolation of both verkle and merkle tree data. It's
|
||||||
|
|
@ -276,7 +281,7 @@ func New(diskdb ethdb.Database, config *Config, isVerkle bool) *Database {
|
||||||
}
|
}
|
||||||
// TODO (rjl493456442) disable the background indexing in read-only mode
|
// TODO (rjl493456442) disable the background indexing in read-only mode
|
||||||
if db.freezer != nil && db.config.EnableStateIndexing {
|
if db.freezer != nil && db.config.EnableStateIndexing {
|
||||||
db.indexer = newHistoryIndexer(db.diskdb, db.freezer, db.tree.bottom().stateID())
|
db.indexer = newHistoryIndexer(db.diskdb, db.freezer, db.tree.bottom().stateID(), db.cacher)
|
||||||
log.Info("Enabled state history indexing")
|
log.Info("Enabled state history indexing")
|
||||||
}
|
}
|
||||||
fields := config.fields()
|
fields := config.fields()
|
||||||
|
|
@ -531,7 +536,7 @@ func (db *Database) Enable(root common.Hash) error {
|
||||||
// 2. Re-initialize the indexer so it starts indexing from the new state root.
|
// 2. Re-initialize the indexer so it starts indexing from the new state root.
|
||||||
if db.indexer != nil && db.freezer != nil && db.config.EnableStateIndexing {
|
if db.indexer != nil && db.freezer != nil && db.config.EnableStateIndexing {
|
||||||
db.indexer.close()
|
db.indexer.close()
|
||||||
db.indexer = newHistoryIndexer(db.diskdb, db.freezer, db.tree.bottom().stateID())
|
db.indexer = newHistoryIndexer(db.diskdb, db.freezer, db.tree.bottom().stateID(), db.cacher)
|
||||||
log.Info("Re-enabled state history indexing")
|
log.Info("Re-enabled state history indexing")
|
||||||
}
|
}
|
||||||
log.Info("Rebuilt trie database", "root", root)
|
log.Info("Rebuilt trie database", "root", root)
|
||||||
|
|
|
||||||
|
|
@ -76,15 +76,16 @@ type indexReader struct {
|
||||||
readers map[uint32]*blockReader
|
readers map[uint32]*blockReader
|
||||||
state stateIdent
|
state stateIdent
|
||||||
timings *readTimings
|
timings *readTimings
|
||||||
|
cacher *historyCacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// loadIndexData loads the index data associated with the specified state.
|
// loadIndexData loads the index data associated with the specified state.
|
||||||
func loadIndexData(db ethdb.KeyValueReader, state stateIdent, timings *readTimings, cacheRead bool) ([]*indexBlockDesc, error) {
|
func loadIndexData(db ethdb.KeyValueReader, state stateIdent, timings *readTimings, cacheRead bool, cacher *historyCacher) ([]*indexBlockDesc, error) {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
key := state.String()
|
key := state.String()
|
||||||
var blob []byte
|
var blob []byte
|
||||||
if cacheRead && historyIndexCache.Contains(key) {
|
if cacheRead && cacher != nil && cacher.index.Contains(key) {
|
||||||
blob, _ = historyIndexCache.Get(key)
|
blob, _ = cacher.index.Get(key)
|
||||||
} else {
|
} else {
|
||||||
if state.account {
|
if state.account {
|
||||||
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
||||||
|
|
@ -98,19 +99,21 @@ func loadIndexData(db ethdb.KeyValueReader, state stateIdent, timings *readTimin
|
||||||
if len(blob) == 0 {
|
if len(blob) == 0 {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
historyIndexCache.Add(key, blob)
|
if cacher != nil {
|
||||||
|
cacher.index.Add(key, blob)
|
||||||
|
}
|
||||||
return parseIndex(blob)
|
return parseIndex(blob)
|
||||||
}
|
}
|
||||||
|
|
||||||
// newIndexReader constructs a index reader for the specified state. Reader with
|
// newIndexReader constructs a index reader for the specified state. Reader with
|
||||||
// empty data is allowed.
|
// empty data is allowed.
|
||||||
func newIndexReader(db ethdb.KeyValueReader, state stateIdent) (*indexReader, error) {
|
func newIndexReader(db ethdb.KeyValueReader, state stateIdent, cacher *historyCacher) (*indexReader, error) {
|
||||||
return newIndexReaderWithTimings(db, state, nil)
|
return newIndexReaderWithTimings(db, state, nil, cacher)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Helper to allow passing timings
|
// Helper to allow passing timings
|
||||||
func newIndexReaderWithTimings(db ethdb.KeyValueReader, state stateIdent, timings *readTimings) (*indexReader, error) {
|
func newIndexReaderWithTimings(db ethdb.KeyValueReader, state stateIdent, timings *readTimings, cacher *historyCacher) (*indexReader, error) {
|
||||||
descList, err := loadIndexData(db, state, timings, true)
|
descList, err := loadIndexData(db, state, timings, true, cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
@ -120,6 +123,7 @@ func newIndexReaderWithTimings(db ethdb.KeyValueReader, state stateIdent, timing
|
||||||
db: db,
|
db: db,
|
||||||
state: state,
|
state: state,
|
||||||
timings: timings,
|
timings: timings,
|
||||||
|
cacher: cacher,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -134,7 +138,7 @@ func (r *indexReader) refresh() error {
|
||||||
delete(r.readers, last.id)
|
delete(r.readers, last.id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
descList, err := loadIndexData(r.db, r.state, r.timings, false)
|
descList, err := loadIndexData(r.db, r.state, r.timings, false, r.cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -161,16 +165,17 @@ func (r *indexReader) readGreaterThan(id uint64) (uint64, error) {
|
||||||
)
|
)
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
key := fmt.Sprintf("%s:%d", r.state.String(), desc.id)
|
key := fmt.Sprintf("%s:%d", r.state.String(), desc.id)
|
||||||
if val, ok := historyBlockCache.Get(key); ok {
|
if r.cacher != nil && r.cacher.block.Contains(key) {
|
||||||
blob = val
|
blob, _ = r.cacher.block.Get(key)
|
||||||
} else {
|
} else {
|
||||||
if r.state.account {
|
if r.state.account {
|
||||||
blob = rawdb.ReadAccountHistoryIndexBlock(r.db, r.state.addressHash, desc.id)
|
blob = rawdb.ReadAccountHistoryIndexBlock(r.db, r.state.addressHash, desc.id)
|
||||||
} else {
|
} else {
|
||||||
blob = rawdb.ReadStorageHistoryIndexBlock(r.db, r.state.addressHash, r.state.storageHash, desc.id)
|
blob = rawdb.ReadStorageHistoryIndexBlock(r.db, r.state.addressHash, r.state.storageHash, desc.id)
|
||||||
}
|
}
|
||||||
if len(blob) > 0 {
|
|
||||||
historyBlockCache.Add(key, blob)
|
if r.cacher != nil && len(blob) > 0 {
|
||||||
|
r.cacher.block.Add(key, blob)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if r.timings != nil {
|
if r.timings != nil {
|
||||||
|
|
@ -200,10 +205,11 @@ type indexWriter struct {
|
||||||
lastID uint64 // The ID of the latest tracked history
|
lastID uint64 // The ID of the latest tracked history
|
||||||
state stateIdent
|
state stateIdent
|
||||||
db ethdb.KeyValueReader
|
db ethdb.KeyValueReader
|
||||||
|
cacher *historyCacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// newIndexWriter constructs the index writer for the specified state.
|
// newIndexWriter constructs the index writer for the specified state.
|
||||||
func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, error) {
|
func newIndexWriter(db ethdb.KeyValueReader, state stateIdent, cacher *historyCacher) (*indexWriter, error) {
|
||||||
var blob []byte
|
var blob []byte
|
||||||
if state.account {
|
if state.account {
|
||||||
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
||||||
|
|
@ -218,6 +224,7 @@ func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, er
|
||||||
bw: bw,
|
bw: bw,
|
||||||
state: state,
|
state: state,
|
||||||
db: db,
|
db: db,
|
||||||
|
cacher: cacher,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
descList, err := parseIndex(blob)
|
descList, err := parseIndex(blob)
|
||||||
|
|
@ -243,6 +250,7 @@ func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, er
|
||||||
bw: bw,
|
bw: bw,
|
||||||
state: state,
|
state: state,
|
||||||
db: db,
|
db: db,
|
||||||
|
cacher: cacher,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -301,8 +309,10 @@ func (w *indexWriter) finish(batch ethdb.Batch) {
|
||||||
}
|
}
|
||||||
for _, bw := range writers {
|
for _, bw := range writers {
|
||||||
buf := bw.finish()
|
buf := bw.finish()
|
||||||
if key := fmt.Sprintf("%s:%d", w.state.String(), bw.desc.id); historyBlockCache.Contains(key) {
|
if w.cacher != nil {
|
||||||
historyBlockCache.Add(key, buf)
|
if key := fmt.Sprintf("%s:%d", w.state.String(), bw.desc.id); w.cacher.block.Contains(key) {
|
||||||
|
w.cacher.block.Add(key, buf)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if w.state.account {
|
if w.state.account {
|
||||||
rawdb.WriteAccountHistoryIndexBlock(batch, w.state.addressHash, bw.desc.id, buf)
|
rawdb.WriteAccountHistoryIndexBlock(batch, w.state.addressHash, bw.desc.id, buf)
|
||||||
|
|
@ -316,8 +326,10 @@ func (w *indexWriter) finish(batch ethdb.Batch) {
|
||||||
for _, desc := range descList {
|
for _, desc := range descList {
|
||||||
buf = append(buf, desc.encode()...)
|
buf = append(buf, desc.encode()...)
|
||||||
}
|
}
|
||||||
if key := w.state.String(); historyIndexCache.Contains(key) {
|
if w.cacher != nil {
|
||||||
historyIndexCache.Add(key, buf)
|
if key := w.state.String(); w.cacher.index.Contains(key) {
|
||||||
|
w.cacher.index.Add(key, buf)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if w.state.account {
|
if w.state.account {
|
||||||
rawdb.WriteAccountHistoryIndex(batch, w.state.addressHash, buf)
|
rawdb.WriteAccountHistoryIndex(batch, w.state.addressHash, buf)
|
||||||
|
|
@ -334,10 +346,11 @@ type indexDeleter struct {
|
||||||
lastID uint64 // The ID of the latest tracked history
|
lastID uint64 // The ID of the latest tracked history
|
||||||
state stateIdent
|
state stateIdent
|
||||||
db ethdb.KeyValueReader
|
db ethdb.KeyValueReader
|
||||||
|
cacher *historyCacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// newIndexDeleter constructs the index deleter for the specified state.
|
// newIndexDeleter constructs the index deleter for the specified state.
|
||||||
func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter, error) {
|
func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent, cacher *historyCacher) (*indexDeleter, error) {
|
||||||
var blob []byte
|
var blob []byte
|
||||||
if state.account {
|
if state.account {
|
||||||
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
||||||
|
|
@ -354,6 +367,7 @@ func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter,
|
||||||
bw: bw,
|
bw: bw,
|
||||||
state: state,
|
state: state,
|
||||||
db: db,
|
db: db,
|
||||||
|
cacher: cacher,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
descList, err := parseIndex(blob)
|
descList, err := parseIndex(blob)
|
||||||
|
|
@ -379,6 +393,7 @@ func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter,
|
||||||
bw: bw,
|
bw: bw,
|
||||||
state: state,
|
state: state,
|
||||||
db: db,
|
db: db,
|
||||||
|
cacher: cacher,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -436,9 +451,10 @@ func (d *indexDeleter) pop(id uint64) error {
|
||||||
// This function is safe to be called multiple times.
|
// This function is safe to be called multiple times.
|
||||||
func (d *indexDeleter) finish(batch ethdb.Batch) {
|
func (d *indexDeleter) finish(batch ethdb.Batch) {
|
||||||
for _, id := range d.dropped {
|
for _, id := range d.dropped {
|
||||||
key := fmt.Sprintf("%s:%d", d.state.String(), id)
|
if d.cacher != nil {
|
||||||
if historyBlockCache.Contains(key) {
|
if key := fmt.Sprintf("%s:%d", d.state.String(), id); d.cacher.block.Contains(key) {
|
||||||
historyBlockCache.Remove(key)
|
d.cacher.block.Remove(key)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if d.state.account {
|
if d.state.account {
|
||||||
rawdb.DeleteAccountHistoryIndexBlock(batch, d.state.addressHash, id)
|
rawdb.DeleteAccountHistoryIndexBlock(batch, d.state.addressHash, id)
|
||||||
|
|
@ -451,9 +467,10 @@ func (d *indexDeleter) finish(batch ethdb.Batch) {
|
||||||
// Flush the content of last block writer, regardless it's dirty or not
|
// Flush the content of last block writer, regardless it's dirty or not
|
||||||
if !d.bw.empty() {
|
if !d.bw.empty() {
|
||||||
buf := d.bw.finish()
|
buf := d.bw.finish()
|
||||||
key := fmt.Sprintf("%s:%d", d.state.String(), d.bw.desc.id)
|
if d.cacher != nil {
|
||||||
if historyBlockCache.Contains(key) {
|
if key := fmt.Sprintf("%s:%d", d.state.String(), d.bw.desc.id); d.cacher.block.Contains(key) {
|
||||||
historyBlockCache.Add(key, buf)
|
d.cacher.block.Add(key, buf)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if d.state.account {
|
if d.state.account {
|
||||||
|
|
@ -464,8 +481,10 @@ func (d *indexDeleter) finish(batch ethdb.Batch) {
|
||||||
}
|
}
|
||||||
// Flush the index metadata into the supplied batch
|
// Flush the index metadata into the supplied batch
|
||||||
if d.empty() {
|
if d.empty() {
|
||||||
if key := d.state.String(); historyIndexCache.Contains(key) {
|
if d.cacher != nil {
|
||||||
historyIndexCache.Remove(key)
|
if key := d.state.String(); d.cacher.index.Contains(key) {
|
||||||
|
d.cacher.index.Remove(key)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if d.state.account {
|
if d.state.account {
|
||||||
rawdb.DeleteAccountHistoryIndex(batch, d.state.addressHash)
|
rawdb.DeleteAccountHistoryIndex(batch, d.state.addressHash)
|
||||||
|
|
@ -477,8 +496,11 @@ func (d *indexDeleter) finish(batch ethdb.Batch) {
|
||||||
for _, desc := range d.descList {
|
for _, desc := range d.descList {
|
||||||
buf = append(buf, desc.encode()...)
|
buf = append(buf, desc.encode()...)
|
||||||
}
|
}
|
||||||
if key := d.state.String(); historyIndexCache.Contains(key) {
|
|
||||||
historyIndexCache.Add(key, buf)
|
if d.cacher != nil {
|
||||||
|
if key := d.state.String(); d.cacher.index.Contains(key) {
|
||||||
|
d.cacher.index.Add(key, buf)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if d.state.account {
|
if d.state.account {
|
||||||
|
|
|
||||||
|
|
@ -33,7 +33,7 @@ func TestIndexReaderBasic(t *testing.T) {
|
||||||
1, 5, 10, 11, 20,
|
1, 5, 10, 11, 20,
|
||||||
}
|
}
|
||||||
db := rawdb.NewMemoryDatabase()
|
db := rawdb.NewMemoryDatabase()
|
||||||
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
for i := 0; i < len(elements); i++ {
|
for i := 0; i < len(elements); i++ {
|
||||||
bw.append(elements[i])
|
bw.append(elements[i])
|
||||||
}
|
}
|
||||||
|
|
@ -41,7 +41,7 @@ func TestIndexReaderBasic(t *testing.T) {
|
||||||
bw.finish(batch)
|
bw.finish(batch)
|
||||||
batch.Write()
|
batch.Write()
|
||||||
|
|
||||||
br, err := newIndexReader(db, newAccountIdent(common.Hash{0xa}))
|
br, err := newIndexReader(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to construct the index reader, %v", err)
|
t.Fatalf("Failed to construct the index reader, %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -75,7 +75,7 @@ func TestIndexReaderLarge(t *testing.T) {
|
||||||
slices.Sort(elements)
|
slices.Sort(elements)
|
||||||
|
|
||||||
db := rawdb.NewMemoryDatabase()
|
db := rawdb.NewMemoryDatabase()
|
||||||
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
for i := 0; i < len(elements); i++ {
|
for i := 0; i < len(elements); i++ {
|
||||||
bw.append(elements[i])
|
bw.append(elements[i])
|
||||||
}
|
}
|
||||||
|
|
@ -83,7 +83,7 @@ func TestIndexReaderLarge(t *testing.T) {
|
||||||
bw.finish(batch)
|
bw.finish(batch)
|
||||||
batch.Write()
|
batch.Write()
|
||||||
|
|
||||||
br, err := newIndexReader(db, newAccountIdent(common.Hash{0xa}))
|
br, err := newIndexReader(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to construct the index reader, %v", err)
|
t.Fatalf("Failed to construct the index reader, %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -107,7 +107,7 @@ func TestIndexReaderLarge(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestEmptyIndexReader(t *testing.T) {
|
func TestEmptyIndexReader(t *testing.T) {
|
||||||
br, err := newIndexReader(rawdb.NewMemoryDatabase(), newAccountIdent(common.Hash{0xa}))
|
br, err := newIndexReader(rawdb.NewMemoryDatabase(), newAccountIdent(common.Hash{0xa}), nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to construct the index reader, %v", err)
|
t.Fatalf("Failed to construct the index reader, %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -122,7 +122,7 @@ func TestEmptyIndexReader(t *testing.T) {
|
||||||
|
|
||||||
func TestIndexWriterBasic(t *testing.T) {
|
func TestIndexWriterBasic(t *testing.T) {
|
||||||
db := rawdb.NewMemoryDatabase()
|
db := rawdb.NewMemoryDatabase()
|
||||||
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
iw.append(2)
|
iw.append(2)
|
||||||
if err := iw.append(1); err == nil {
|
if err := iw.append(1); err == nil {
|
||||||
t.Fatal("out-of-order insertion is not expected")
|
t.Fatal("out-of-order insertion is not expected")
|
||||||
|
|
@ -134,7 +134,7 @@ func TestIndexWriterBasic(t *testing.T) {
|
||||||
iw.finish(batch)
|
iw.finish(batch)
|
||||||
batch.Write()
|
batch.Write()
|
||||||
|
|
||||||
iw, err := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
iw, err := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to construct the block writer, %v", err)
|
t.Fatalf("Failed to construct the block writer, %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -148,7 +148,7 @@ func TestIndexWriterBasic(t *testing.T) {
|
||||||
|
|
||||||
func TestIndexWriterDelete(t *testing.T) {
|
func TestIndexWriterDelete(t *testing.T) {
|
||||||
db := rawdb.NewMemoryDatabase()
|
db := rawdb.NewMemoryDatabase()
|
||||||
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
for i := 0; i < indexBlockEntriesCap*4; i++ {
|
for i := 0; i < indexBlockEntriesCap*4; i++ {
|
||||||
iw.append(uint64(i + 1))
|
iw.append(uint64(i + 1))
|
||||||
}
|
}
|
||||||
|
|
@ -157,7 +157,7 @@ func TestIndexWriterDelete(t *testing.T) {
|
||||||
batch.Write()
|
batch.Write()
|
||||||
|
|
||||||
// Delete unknown id, the request should be rejected
|
// Delete unknown id, the request should be rejected
|
||||||
id, _ := newIndexDeleter(db, newAccountIdent(common.Hash{0xa}))
|
id, _ := newIndexDeleter(db, newAccountIdent(common.Hash{0xa}), nil)
|
||||||
if err := id.pop(indexBlockEntriesCap * 5); err == nil {
|
if err := id.pop(indexBlockEntriesCap * 5); err == nil {
|
||||||
t.Fatal("Expect error to occur for unknown id")
|
t.Fatal("Expect error to occur for unknown id")
|
||||||
}
|
}
|
||||||
|
|
@ -179,7 +179,7 @@ func TestIndexWriterDelete(t *testing.T) {
|
||||||
func TestBatchIndexerWrite(t *testing.T) {
|
func TestBatchIndexerWrite(t *testing.T) {
|
||||||
var (
|
var (
|
||||||
db = rawdb.NewMemoryDatabase()
|
db = rawdb.NewMemoryDatabase()
|
||||||
batch = newBatchIndexer(db, false)
|
batch = newBatchIndexer(db, nil, false)
|
||||||
histories = makeHistories(10)
|
histories = makeHistories(10)
|
||||||
)
|
)
|
||||||
for i, h := range histories {
|
for i, h := range histories {
|
||||||
|
|
@ -212,7 +212,7 @@ func TestBatchIndexerWrite(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for addrHash, indexes := range accounts {
|
for addrHash, indexes := range accounts {
|
||||||
ir, _ := newIndexReader(db, newAccountIdent(addrHash))
|
ir, _ := newIndexReader(db, newAccountIdent(addrHash), nil)
|
||||||
for i := 0; i < len(indexes)-1; i++ {
|
for i := 0; i < len(indexes)-1; i++ {
|
||||||
n, err := ir.readGreaterThan(indexes[i])
|
n, err := ir.readGreaterThan(indexes[i])
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -232,7 +232,7 @@ func TestBatchIndexerWrite(t *testing.T) {
|
||||||
}
|
}
|
||||||
for addrHash, slots := range storages {
|
for addrHash, slots := range storages {
|
||||||
for slotHash, indexes := range slots {
|
for slotHash, indexes := range slots {
|
||||||
ir, _ := newIndexReader(db, newStorageIdent(addrHash, slotHash))
|
ir, _ := newIndexReader(db, newStorageIdent(addrHash, slotHash), nil)
|
||||||
for i := 0; i < len(indexes)-1; i++ {
|
for i := 0; i < len(indexes)-1; i++ {
|
||||||
n, err := ir.readGreaterThan(indexes[i])
|
n, err := ir.readGreaterThan(indexes[i])
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -256,7 +256,7 @@ func TestBatchIndexerWrite(t *testing.T) {
|
||||||
func TestBatchIndexerDelete(t *testing.T) {
|
func TestBatchIndexerDelete(t *testing.T) {
|
||||||
var (
|
var (
|
||||||
db = rawdb.NewMemoryDatabase()
|
db = rawdb.NewMemoryDatabase()
|
||||||
bw = newBatchIndexer(db, false)
|
bw = newBatchIndexer(db, nil, false)
|
||||||
histories = makeHistories(10)
|
histories = makeHistories(10)
|
||||||
)
|
)
|
||||||
// Index histories
|
// Index histories
|
||||||
|
|
@ -270,7 +270,7 @@ func TestBatchIndexerDelete(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Unindex histories
|
// Unindex histories
|
||||||
bd := newBatchIndexer(db, true)
|
bd := newBatchIndexer(db, nil, true)
|
||||||
for i := len(histories) - 1; i >= 0; i-- {
|
for i := len(histories) - 1; i >= 0; i-- {
|
||||||
if err := bd.process(histories[i], uint64(i+1)); err != nil {
|
if err := bd.process(histories[i], uint64(i+1)); err != nil {
|
||||||
t.Fatalf("Failed to process history, %v", err)
|
t.Fatalf("Failed to process history, %v", err)
|
||||||
|
|
|
||||||
|
|
@ -79,15 +79,17 @@ type batchIndexer struct {
|
||||||
delete bool // Index or unindex mode
|
delete bool // Index or unindex mode
|
||||||
lastID uint64 // The ID of latest processed history
|
lastID uint64 // The ID of latest processed history
|
||||||
db ethdb.KeyValueStore
|
db ethdb.KeyValueStore
|
||||||
|
cacher *historyCacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// newBatchIndexer constructs the batch indexer with the supplied mode.
|
// newBatchIndexer constructs the batch indexer with the supplied mode.
|
||||||
func newBatchIndexer(db ethdb.KeyValueStore, delete bool) *batchIndexer {
|
func newBatchIndexer(db ethdb.KeyValueStore, cacher *historyCacher, delete bool) *batchIndexer {
|
||||||
return &batchIndexer{
|
return &batchIndexer{
|
||||||
accounts: make(map[common.Hash][]uint64),
|
accounts: make(map[common.Hash][]uint64),
|
||||||
storages: make(map[common.Hash]map[common.Hash][]uint64),
|
storages: make(map[common.Hash]map[common.Hash][]uint64),
|
||||||
delete: delete,
|
delete: delete,
|
||||||
db: db,
|
db: db,
|
||||||
|
cacher: cacher,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -139,7 +141,7 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
for addrHash, idList := range b.accounts {
|
for addrHash, idList := range b.accounts {
|
||||||
eg.Go(func() error {
|
eg.Go(func() error {
|
||||||
if !b.delete {
|
if !b.delete {
|
||||||
iw, err := newIndexWriter(b.db, newAccountIdent(addrHash))
|
iw, err := newIndexWriter(b.db, newAccountIdent(addrHash), b.cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -152,7 +154,7 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
iw.finish(batch)
|
iw.finish(batch)
|
||||||
batchMu.Unlock()
|
batchMu.Unlock()
|
||||||
} else {
|
} else {
|
||||||
id, err := newIndexDeleter(b.db, newAccountIdent(addrHash))
|
id, err := newIndexDeleter(b.db, newAccountIdent(addrHash), b.cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -173,7 +175,7 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
for storageHash, idList := range slots {
|
for storageHash, idList := range slots {
|
||||||
eg.Go(func() error {
|
eg.Go(func() error {
|
||||||
if !b.delete {
|
if !b.delete {
|
||||||
iw, err := newIndexWriter(b.db, newStorageIdent(addrHash, storageHash))
|
iw, err := newIndexWriter(b.db, newStorageIdent(addrHash, storageHash), b.cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -186,7 +188,7 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
iw.finish(batch)
|
iw.finish(batch)
|
||||||
batchMu.Unlock()
|
batchMu.Unlock()
|
||||||
} else {
|
} else {
|
||||||
id, err := newIndexDeleter(b.db, newStorageIdent(addrHash, storageHash))
|
id, err := newIndexDeleter(b.db, newStorageIdent(addrHash, storageHash), b.cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -227,7 +229,7 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
// indexSingle processes the state history with the specified ID for indexing.
|
// indexSingle processes the state history with the specified ID for indexing.
|
||||||
func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader) error {
|
func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, cacher *historyCacher) error {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
defer func() {
|
defer func() {
|
||||||
indexHistoryTimer.UpdateSince(start)
|
indexHistoryTimer.UpdateSince(start)
|
||||||
|
|
@ -245,7 +247,7 @@ func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancient
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
b := newBatchIndexer(db, false)
|
b := newBatchIndexer(db, cacher, false)
|
||||||
if err := b.process(h, historyID); err != nil {
|
if err := b.process(h, historyID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -257,7 +259,7 @@ func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancient
|
||||||
}
|
}
|
||||||
|
|
||||||
// unindexSingle processes the state history with the specified ID for unindexing.
|
// unindexSingle processes the state history with the specified ID for unindexing.
|
||||||
func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader) error {
|
func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, cacher *historyCacher) error {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
defer func() {
|
defer func() {
|
||||||
unindexHistoryTimer.UpdateSince(start)
|
unindexHistoryTimer.UpdateSince(start)
|
||||||
|
|
@ -275,7 +277,7 @@ func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancie
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
b := newBatchIndexer(db, true)
|
b := newBatchIndexer(db, cacher, true)
|
||||||
if err := b.process(h, historyID); err != nil {
|
if err := b.process(h, historyID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -302,6 +304,7 @@ type interruptSignal struct {
|
||||||
type indexIniter struct {
|
type indexIniter struct {
|
||||||
disk ethdb.KeyValueStore
|
disk ethdb.KeyValueStore
|
||||||
freezer ethdb.AncientStore
|
freezer ethdb.AncientStore
|
||||||
|
cacher *historyCacher
|
||||||
interrupt chan *interruptSignal
|
interrupt chan *interruptSignal
|
||||||
done chan struct{}
|
done chan struct{}
|
||||||
closed chan struct{}
|
closed chan struct{}
|
||||||
|
|
@ -416,7 +419,7 @@ func (i *indexIniter) run(lastID uint64) {
|
||||||
// been fully indexed, unindex it here and shut down the initializer.
|
// been fully indexed, unindex it here and shut down the initializer.
|
||||||
if checkDone() {
|
if checkDone() {
|
||||||
log.Info("Truncate the extra history", "id", lastID)
|
log.Info("Truncate the extra history", "id", lastID)
|
||||||
if err := unindexSingle(lastID, i.disk, i.freezer); err != nil {
|
if err := unindexSingle(lastID, i.disk, i.freezer, i.cacher); err != nil {
|
||||||
signal.result <- err
|
signal.result <- err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
@ -517,7 +520,7 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
||||||
current = beginID
|
current = beginID
|
||||||
start = time.Now()
|
start = time.Now()
|
||||||
logged = time.Now()
|
logged = time.Now()
|
||||||
batch = newBatchIndexer(i.disk, false)
|
batch = newBatchIndexer(i.disk, i.cacher, false)
|
||||||
)
|
)
|
||||||
for current <= lastID {
|
for current <= lastID {
|
||||||
count := lastID - current + 1
|
count := lastID - current + 1
|
||||||
|
|
@ -585,6 +588,7 @@ type historyIndexer struct {
|
||||||
initer *indexIniter
|
initer *indexIniter
|
||||||
disk ethdb.KeyValueStore
|
disk ethdb.KeyValueStore
|
||||||
freezer ethdb.AncientStore
|
freezer ethdb.AncientStore
|
||||||
|
cacher *historyCacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkVersion checks whether the index data in the database matches the version.
|
// checkVersion checks whether the index data in the database matches the version.
|
||||||
|
|
@ -611,12 +615,13 @@ func checkVersion(disk ethdb.KeyValueStore) {
|
||||||
|
|
||||||
// newHistoryIndexer constructs the history indexer and launches the background
|
// newHistoryIndexer constructs the history indexer and launches the background
|
||||||
// initer to complete the indexing of any remaining state histories.
|
// initer to complete the indexing of any remaining state histories.
|
||||||
func newHistoryIndexer(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, lastHistoryID uint64) *historyIndexer {
|
func newHistoryIndexer(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, lastHistoryID uint64, cacher *historyCacher) *historyIndexer {
|
||||||
checkVersion(disk)
|
checkVersion(disk)
|
||||||
return &historyIndexer{
|
return &historyIndexer{
|
||||||
initer: newIndexIniter(disk, freezer, lastHistoryID),
|
initer: newIndexIniter(disk, freezer, lastHistoryID),
|
||||||
disk: disk,
|
disk: disk,
|
||||||
freezer: freezer,
|
freezer: freezer,
|
||||||
|
cacher: cacher,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -642,7 +647,7 @@ func (i *historyIndexer) extend(historyID uint64) error {
|
||||||
case <-i.initer.closed:
|
case <-i.initer.closed:
|
||||||
return errors.New("indexer is closed")
|
return errors.New("indexer is closed")
|
||||||
case <-i.initer.done:
|
case <-i.initer.done:
|
||||||
return indexSingle(historyID, i.disk, i.freezer)
|
return indexSingle(historyID, i.disk, i.freezer, i.cacher)
|
||||||
case i.initer.interrupt <- signal:
|
case i.initer.interrupt <- signal:
|
||||||
return <-signal.result
|
return <-signal.result
|
||||||
}
|
}
|
||||||
|
|
@ -659,7 +664,7 @@ func (i *historyIndexer) shorten(historyID uint64) error {
|
||||||
case <-i.initer.closed:
|
case <-i.initer.closed:
|
||||||
return errors.New("indexer is closed")
|
return errors.New("indexer is closed")
|
||||||
case <-i.initer.done:
|
case <-i.initer.done:
|
||||||
return unindexSingle(historyID, i.disk, i.freezer)
|
return unindexSingle(historyID, i.disk, i.freezer, i.cacher)
|
||||||
case i.initer.interrupt <- signal:
|
case i.initer.interrupt <- signal:
|
||||||
return <-signal.result
|
return <-signal.result
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -32,11 +32,6 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
|
||||||
historyIndexCache = lru.NewCache[string, []byte](10000)
|
|
||||||
historyBlockCache = lru.NewCache[string, []byte](10000)
|
|
||||||
)
|
|
||||||
|
|
||||||
// stateIdent represents the identifier of a state element, which can be
|
// stateIdent represents the identifier of a state element, which can be
|
||||||
// either an account or a storage slot.
|
// either an account or a storage slot.
|
||||||
type stateIdent struct {
|
type stateIdent struct {
|
||||||
|
|
@ -131,8 +126,8 @@ type indexReaderWithLimitTag struct {
|
||||||
|
|
||||||
// newIndexReaderWithLimitTag constructs a index reader with indexing position.
|
// newIndexReaderWithLimitTag constructs a index reader with indexing position.
|
||||||
// Add timings parameter to pass through
|
// Add timings parameter to pass through
|
||||||
func newIndexReaderWithLimitTag(db ethdb.KeyValueReader, state stateIdent, limit uint64, timings *readTimings) (*indexReaderWithLimitTag, error) {
|
func newIndexReaderWithLimitTag(db ethdb.KeyValueReader, state stateIdent, limit uint64, timings *readTimings, cacher *historyCacher) (*indexReaderWithLimitTag, error) {
|
||||||
r, err := newIndexReaderWithTimings(db, state, timings)
|
r, err := newIndexReaderWithTimings(db, state, timings, cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
@ -190,19 +185,35 @@ func (r *indexReaderWithLimitTag) readGreaterThan(id uint64, lastID uint64) (uin
|
||||||
return r.reader.readGreaterThan(id)
|
return r.reader.readGreaterThan(id)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// historyCacher is responsible for caching history objects in memory to speed up access.
|
||||||
|
type historyCacher struct {
|
||||||
|
index *lru.Cache[string, []byte]
|
||||||
|
block *lru.Cache[string, []byte]
|
||||||
|
}
|
||||||
|
|
||||||
|
// newHistoryCacher creates a new historyCacher instance.
|
||||||
|
func newHistoryCacher(size int) *historyCacher {
|
||||||
|
return &historyCacher{
|
||||||
|
index: lru.NewCache[string, []byte](size),
|
||||||
|
block: lru.NewCache[string, []byte](size),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// historyReader is the structure to access historic state data.
|
// historyReader is the structure to access historic state data.
|
||||||
type historyReader struct {
|
type historyReader struct {
|
||||||
disk ethdb.KeyValueReader
|
disk ethdb.KeyValueReader
|
||||||
freezer ethdb.AncientReader
|
freezer ethdb.AncientReader
|
||||||
readers map[string]*indexReaderWithLimitTag
|
readers map[string]*indexReaderWithLimitTag
|
||||||
|
cacher *historyCacher
|
||||||
}
|
}
|
||||||
|
|
||||||
// newHistoryReader constructs the history reader with the supplied db.
|
// newHistoryReader constructs the history reader with the supplied db and cacher.
|
||||||
func newHistoryReader(disk ethdb.KeyValueReader, freezer ethdb.AncientReader) *historyReader {
|
func newHistoryReader(disk ethdb.KeyValueReader, freezer ethdb.AncientReader, cacher *historyCacher) *historyReader {
|
||||||
return &historyReader{
|
return &historyReader{
|
||||||
disk: disk,
|
disk: disk,
|
||||||
freezer: freezer,
|
freezer: freezer,
|
||||||
readers: make(map[string]*indexReaderWithLimitTag),
|
readers: make(map[string]*indexReaderWithLimitTag),
|
||||||
|
cacher: cacher,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -386,7 +397,7 @@ func (r *historyReader) read(state stateIdentQuery, stateID uint64, lastID uint6
|
||||||
// state retrieval
|
// state retrieval
|
||||||
ir, ok := r.readers[state.String()]
|
ir, ok := r.readers[state.String()]
|
||||||
if !ok {
|
if !ok {
|
||||||
ir, err = newIndexReaderWithLimitTag(r.disk, state.stateIdent, metadata.Last, timings)
|
ir, err = newIndexReaderWithLimitTag(r.disk, state.stateIdent, metadata.Last, timings, r.cacher)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -133,7 +133,7 @@ func testHistoryReader(t *testing.T, historyLimit uint64) {
|
||||||
var (
|
var (
|
||||||
roots = env.roots
|
roots = env.roots
|
||||||
dRoot = env.db.tree.bottom().rootHash()
|
dRoot = env.db.tree.bottom().rootHash()
|
||||||
hr = newHistoryReader(env.db.diskdb, env.db.freezer)
|
hr = newHistoryReader(env.db.diskdb, env.db.freezer, env.db.cacher)
|
||||||
)
|
)
|
||||||
for _, root := range roots {
|
for _, root := range roots {
|
||||||
if root == dRoot {
|
if root == dRoot {
|
||||||
|
|
|
||||||
|
|
@ -230,7 +230,7 @@ func (db *Database) HistoricReader(root common.Hash) (*HistoricalStateReader, er
|
||||||
return &HistoricalStateReader{
|
return &HistoricalStateReader{
|
||||||
id: *id,
|
id: *id,
|
||||||
db: db,
|
db: db,
|
||||||
reader: newHistoryReader(db.diskdb, db.freezer),
|
reader: newHistoryReader(db.diskdb, db.freezer, db.cacher),
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue