mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-23 05:06:43 +00:00
triedb/pathdb: add history index config
This commit is contained in:
Gary Rong
parent
659342a523
commit
b6da120f9b
5 changed files with 123 additions and 61 deletions
|
|
@ -107,10 +107,8 @@ func (r *indexReader) refresh() error {
|
|||
// may have been modified by additional elements written to the disk.
|
||||
if len(r.descList) != 0 {
|
||||
last := r.descList[len(r.descList)-1]
|
||||
if !last.full() {
|
||||
delete(r.readers, last.id)
|
||||
}
|
||||
}
|
||||
descList, err := loadIndexData(r.db, r.state)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -158,19 +156,23 @@ type indexWriter struct {
|
|||
lastID uint64 // The ID of the latest tracked history
|
||||
state stateIdent
|
||||
db ethdb.KeyValueReader
|
||||
|
||||
// configs
|
||||
blockCap uint16 // Maximum number of entries grouped into a single index block
|
||||
}
|
||||
|
||||
// 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, blockCap uint16) (*indexWriter, error) {
|
||||
blob := readStateIndex(state, db)
|
||||
if len(blob) == 0 {
|
||||
desc := newIndexBlockDesc(0)
|
||||
bw, _ := newBlockWriter(nil, desc)
|
||||
bw, _ := newBlockWriter(nil, desc, blockCap)
|
||||
return &indexWriter{
|
||||
descList: []*indexBlockDesc{desc},
|
||||
bw: bw,
|
||||
state: state,
|
||||
db: db,
|
||||
blockCap: blockCap,
|
||||
}, nil
|
||||
}
|
||||
descList, err := parseIndex(blob)
|
||||
|
|
@ -179,7 +181,7 @@ func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, er
|
|||
}
|
||||
lastDesc := descList[len(descList)-1]
|
||||
indexBlock := readStateIndexBlock(state, db, lastDesc.id)
|
||||
bw, err := newBlockWriter(indexBlock, lastDesc)
|
||||
bw, err := newBlockWriter(indexBlock, lastDesc, blockCap)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -189,6 +191,7 @@ func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, er
|
|||
bw: bw,
|
||||
state: state,
|
||||
db: db,
|
||||
blockCap: blockCap,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
|
@ -218,7 +221,7 @@ func (w *indexWriter) rotate() error {
|
|||
desc = newIndexBlockDesc(w.bw.desc.id + 1)
|
||||
)
|
||||
w.frozen = append(w.frozen, w.bw)
|
||||
w.bw, err = newBlockWriter(nil, desc)
|
||||
w.bw, err = newBlockWriter(nil, desc, w.blockCap)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -265,21 +268,25 @@ type indexDeleter struct {
|
|||
lastID uint64 // The ID of the latest tracked history
|
||||
state stateIdent
|
||||
db ethdb.KeyValueReader
|
||||
|
||||
// configs
|
||||
blockCap uint16 // Maximum number of entries grouped into a single index block
|
||||
}
|
||||
|
||||
// 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, blockCap uint16) (*indexDeleter, error) {
|
||||
blob := readStateIndex(state, db)
|
||||
if len(blob) == 0 {
|
||||
// TODO(rjl493456442) we can probably return an error here,
|
||||
// deleter with no data is meaningless.
|
||||
desc := newIndexBlockDesc(0)
|
||||
bw, _ := newBlockWriter(nil, desc)
|
||||
bw, _ := newBlockWriter(nil, desc, blockCap)
|
||||
return &indexDeleter{
|
||||
descList: []*indexBlockDesc{desc},
|
||||
bw: bw,
|
||||
state: state,
|
||||
db: db,
|
||||
blockCap: blockCap,
|
||||
}, nil
|
||||
}
|
||||
descList, err := parseIndex(blob)
|
||||
|
|
@ -288,7 +295,7 @@ func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter,
|
|||
}
|
||||
lastDesc := descList[len(descList)-1]
|
||||
indexBlock := readStateIndexBlock(state, db, lastDesc.id)
|
||||
bw, err := newBlockWriter(indexBlock, lastDesc)
|
||||
bw, err := newBlockWriter(indexBlock, lastDesc, blockCap)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -334,7 +341,7 @@ func (d *indexDeleter) pop(id uint64) error {
|
|||
// Open the previous block writer for deleting
|
||||
lastDesc := d.descList[len(d.descList)-1]
|
||||
indexBlock := readStateIndexBlock(d.state, d.db, lastDesc.id)
|
||||
bw, err := newBlockWriter(indexBlock, lastDesc)
|
||||
bw, err := newBlockWriter(indexBlock, lastDesc, d.blockCap)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,9 +26,7 @@ import (
|
|||
|
||||
const (
|
||||
indexBlockDescSize = 14 // The size of index block descriptor
|
||||
indexBlockEntriesCap = 4096 // The maximum number of entries can be grouped in a block
|
||||
indexBlockRestartLen = 256 // The restart interval length of index block
|
||||
historyIndexBatch = 1_000_000 // The number of state history indexes for constructing or deleting as batch
|
||||
)
|
||||
|
||||
// indexBlockDesc represents a descriptor for an index block, which contains a
|
||||
|
|
@ -51,8 +49,8 @@ func (d *indexBlockDesc) empty() bool {
|
|||
|
||||
// full indicates whether the number of elements in the block exceeds the
|
||||
// preconfigured limit.
|
||||
func (d *indexBlockDesc) full() bool {
|
||||
return d.entries >= indexBlockEntriesCap
|
||||
func (d *indexBlockDesc) full(capacity uint16) bool {
|
||||
return d.entries >= capacity
|
||||
}
|
||||
|
||||
// encode packs index block descriptor into byte stream.
|
||||
|
|
@ -222,13 +220,15 @@ type blockWriter struct {
|
|||
desc *indexBlockDesc // Descriptor of the block
|
||||
restarts []uint16 // Offsets into the data slice, marking the start of each section
|
||||
data []byte // Aggregated encoded data slice
|
||||
capacity uint16 // Maximum number of entries grouped into a single block
|
||||
}
|
||||
|
||||
func newBlockWriter(blob []byte, desc *indexBlockDesc) (*blockWriter, error) {
|
||||
func newBlockWriter(blob []byte, desc *indexBlockDesc, capacity uint16) (*blockWriter, error) {
|
||||
if len(blob) == 0 {
|
||||
return &blockWriter{
|
||||
desc: desc,
|
||||
data: make([]byte, 0, 1024),
|
||||
capacity: capacity,
|
||||
}, nil
|
||||
}
|
||||
restarts, data, err := parseIndexBlock(blob)
|
||||
|
|
@ -239,6 +239,7 @@ func newBlockWriter(blob []byte, desc *indexBlockDesc) (*blockWriter, error) {
|
|||
desc: desc,
|
||||
restarts: restarts,
|
||||
data: data, // safe to own the slice
|
||||
capacity: capacity,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
|
@ -372,7 +373,7 @@ func (b *blockWriter) empty() bool {
|
|||
}
|
||||
|
||||
func (b *blockWriter) full() bool {
|
||||
return b.desc.full()
|
||||
return b.desc.full(b.capacity)
|
||||
}
|
||||
|
||||
// finish finalizes the index block encoding by appending the encoded restart points
|
||||
|
|
|
|||
|
|
@ -25,10 +25,11 @@ import (
|
|||
)
|
||||
|
||||
func TestBlockReaderBasic(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
elements := []uint64{
|
||||
1, 5, 10, 11, 20,
|
||||
}
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
for i := 0; i < len(elements); i++ {
|
||||
bw.append(elements[i])
|
||||
}
|
||||
|
|
@ -60,13 +61,14 @@ func TestBlockReaderBasic(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestBlockReaderLarge(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
var elements []uint64
|
||||
for i := 0; i < 1000; i++ {
|
||||
elements = append(elements, rand.Uint64())
|
||||
}
|
||||
slices.Sort(elements)
|
||||
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
for i := 0; i < len(elements); i++ {
|
||||
bw.append(elements[i])
|
||||
}
|
||||
|
|
@ -95,7 +97,8 @@ func TestBlockReaderLarge(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestBlockWriterBasic(t *testing.T) {
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
blockCap := uint16(4096)
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
if !bw.empty() {
|
||||
t.Fatal("expected empty block")
|
||||
}
|
||||
|
|
@ -107,7 +110,7 @@ func TestBlockWriterBasic(t *testing.T) {
|
|||
bw.append(uint64(i + 3))
|
||||
}
|
||||
|
||||
bw, err := newBlockWriter(bw.finish(), newIndexBlockDesc(0))
|
||||
bw, err := newBlockWriter(bw.finish(), newIndexBlockDesc(0), blockCap)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to construct the block writer, %v", err)
|
||||
}
|
||||
|
|
@ -120,7 +123,8 @@ func TestBlockWriterBasic(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestBlockWriterDelete(t *testing.T) {
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
blockCap := uint16(4096)
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
for i := 0; i < 10; i++ {
|
||||
bw.append(uint64(i + 1))
|
||||
}
|
||||
|
|
@ -144,10 +148,11 @@ func TestBlockWriterDelete(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestBlcokWriterDeleteWithData(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
elements := []uint64{
|
||||
1, 5, 10, 11, 20,
|
||||
}
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
for i := 0; i < len(elements); i++ {
|
||||
bw.append(elements[i])
|
||||
}
|
||||
|
|
@ -158,7 +163,7 @@ func TestBlcokWriterDeleteWithData(t *testing.T) {
|
|||
max: 20,
|
||||
entries: 5,
|
||||
}
|
||||
bw, err := newBlockWriter(bw.finish(), desc)
|
||||
bw, err := newBlockWriter(bw.finish(), desc, blockCap)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to construct block writer %v", err)
|
||||
}
|
||||
|
|
@ -201,7 +206,8 @@ func TestBlcokWriterDeleteWithData(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestCorruptedIndexBlock(t *testing.T) {
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
blockCap := uint16(4096)
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
for i := 0; i < 10; i++ {
|
||||
bw.append(uint64(i + 1))
|
||||
}
|
||||
|
|
@ -209,7 +215,7 @@ func TestCorruptedIndexBlock(t *testing.T) {
|
|||
|
||||
// Mutate the buffer manually
|
||||
buf[len(buf)-1]++
|
||||
_, err := newBlockWriter(buf, newIndexBlockDesc(0))
|
||||
_, err := newBlockWriter(buf, newIndexBlockDesc(0), blockCap)
|
||||
if err == nil {
|
||||
t.Fatal("Corrupted index block data is not detected")
|
||||
}
|
||||
|
|
@ -218,8 +224,9 @@ func TestCorruptedIndexBlock(t *testing.T) {
|
|||
// BenchmarkParseIndexBlock benchmarks the performance of parseIndexBlock.
|
||||
func BenchmarkParseIndexBlock(b *testing.B) {
|
||||
// Generate a realistic index block blob
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0))
|
||||
for i := 0; i < 4096; i++ {
|
||||
blockCap := uint16(4096)
|
||||
bw, _ := newBlockWriter(nil, newIndexBlockDesc(0), blockCap)
|
||||
for i := 0; i < int(blockCap); i++ {
|
||||
bw.append(uint64(i * 2))
|
||||
}
|
||||
blob := bw.finish()
|
||||
|
|
@ -238,13 +245,14 @@ func BenchmarkBlockWriterAppend(b *testing.B) {
|
|||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
|
||||
blockCap := uint16(4096)
|
||||
desc := newIndexBlockDesc(0)
|
||||
writer, _ := newBlockWriter(nil, desc)
|
||||
writer, _ := newBlockWriter(nil, desc, blockCap)
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
if writer.full() {
|
||||
desc = newIndexBlockDesc(0)
|
||||
writer, _ = newBlockWriter(nil, desc)
|
||||
writer, _ = newBlockWriter(nil, desc, blockCap)
|
||||
}
|
||||
if err := writer.append(writer.desc.max + 1); err != nil {
|
||||
b.Error(err)
|
||||
|
|
|
|||
|
|
@ -29,11 +29,12 @@ import (
|
|||
)
|
||||
|
||||
func TestIndexReaderBasic(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
elements := []uint64{
|
||||
1, 5, 10, 11, 20,
|
||||
}
|
||||
db := rawdb.NewMemoryDatabase()
|
||||
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
||||
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), blockCap)
|
||||
for i := 0; i < len(elements); i++ {
|
||||
bw.append(elements[i])
|
||||
}
|
||||
|
|
@ -68,14 +69,15 @@ func TestIndexReaderBasic(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestIndexReaderLarge(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
var elements []uint64
|
||||
for i := 0; i < 10*indexBlockEntriesCap; i++ {
|
||||
for i := 0; i < 10*int(blockCap); i++ {
|
||||
elements = append(elements, rand.Uint64())
|
||||
}
|
||||
slices.Sort(elements)
|
||||
|
||||
db := rawdb.NewMemoryDatabase()
|
||||
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
||||
bw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), blockCap)
|
||||
for i := 0; i < len(elements); i++ {
|
||||
bw.append(elements[i])
|
||||
}
|
||||
|
|
@ -121,8 +123,9 @@ func TestEmptyIndexReader(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestIndexWriterBasic(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
db := rawdb.NewMemoryDatabase()
|
||||
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
||||
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), blockCap)
|
||||
iw.append(2)
|
||||
if err := iw.append(1); err == nil {
|
||||
t.Fatal("out-of-order insertion is not expected")
|
||||
|
|
@ -134,7 +137,7 @@ func TestIndexWriterBasic(t *testing.T) {
|
|||
iw.finish(batch)
|
||||
batch.Write()
|
||||
|
||||
iw, err := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
||||
iw, err := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), blockCap)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to construct the block writer, %v", err)
|
||||
}
|
||||
|
|
@ -147,9 +150,10 @@ func TestIndexWriterBasic(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestIndexWriterDelete(t *testing.T) {
|
||||
blockCap := uint16(4096)
|
||||
db := rawdb.NewMemoryDatabase()
|
||||
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}))
|
||||
for i := 0; i < indexBlockEntriesCap*4; i++ {
|
||||
iw, _ := newIndexWriter(db, newAccountIdent(common.Hash{0xa}), blockCap)
|
||||
for i := 0; i < int(blockCap)*4; i++ {
|
||||
iw.append(uint64(i + 1))
|
||||
}
|
||||
batch := db.NewBatch()
|
||||
|
|
@ -157,11 +161,11 @@ func TestIndexWriterDelete(t *testing.T) {
|
|||
batch.Write()
|
||||
|
||||
// Delete unknown id, the request should be rejected
|
||||
id, _ := newIndexDeleter(db, newAccountIdent(common.Hash{0xa}))
|
||||
if err := id.pop(indexBlockEntriesCap * 5); err == nil {
|
||||
id, _ := newIndexDeleter(db, newAccountIdent(common.Hash{0xa}), blockCap)
|
||||
if err := id.pop(uint64(blockCap) * 5); err == nil {
|
||||
t.Fatal("Expect error to occur for unknown id")
|
||||
}
|
||||
for i := indexBlockEntriesCap * 4; i >= 1; i-- {
|
||||
for i := int(blockCap) * 4; i >= 1; i-- {
|
||||
if err := id.pop(uint64(i)); err != nil {
|
||||
t.Fatalf("Unexpected error for element popping, %v", err)
|
||||
}
|
||||
|
|
@ -179,7 +183,7 @@ func TestIndexWriterDelete(t *testing.T) {
|
|||
func TestBatchIndexerWrite(t *testing.T) {
|
||||
var (
|
||||
db = rawdb.NewMemoryDatabase()
|
||||
batch = newBatchIndexer(db, false, typeStateHistory)
|
||||
batch = newBatchIndexer(indexerConfigs[typeStateHistory], db, false, typeStateHistory)
|
||||
histories = makeStateHistories(10)
|
||||
)
|
||||
for i, h := range histories {
|
||||
|
|
@ -256,7 +260,7 @@ func TestBatchIndexerWrite(t *testing.T) {
|
|||
func TestBatchIndexerDelete(t *testing.T) {
|
||||
var (
|
||||
db = rawdb.NewMemoryDatabase()
|
||||
bw = newBatchIndexer(db, false, typeStateHistory)
|
||||
bw = newBatchIndexer(indexerConfigs[typeStateHistory], db, false, typeStateHistory)
|
||||
histories = makeStateHistories(10)
|
||||
)
|
||||
// Index histories
|
||||
|
|
@ -270,7 +274,7 @@ func TestBatchIndexerDelete(t *testing.T) {
|
|||
}
|
||||
|
||||
// Unindex histories
|
||||
bd := newBatchIndexer(db, true, typeStateHistory)
|
||||
bd := newBatchIndexer(indexerConfigs[typeStateHistory], db, true, typeStateHistory)
|
||||
for i := len(histories) - 1; i >= 0; i-- {
|
||||
if err := bd.process(histories[i], uint64(i+1)); err != nil {
|
||||
t.Fatalf("Failed to process history, %v", err)
|
||||
|
|
|
|||
|
|
@ -120,6 +120,7 @@ func deleteIndexMetadata(db ethdb.KeyValueWriter, typ historyType) {
|
|||
// batchIndexer is responsible for performing batch indexing or unindexing
|
||||
// of historical data (e.g., state or trie node changes) atomically.
|
||||
type batchIndexer struct {
|
||||
config historyIndexerConfig // Configs for history index
|
||||
index map[stateIdent][]uint64 // List of history IDs for tracked state entry
|
||||
pending int // Number of entries processed in the current batch.
|
||||
delete bool // Operation mode: true for unindex, false for index.
|
||||
|
|
@ -129,8 +130,9 @@ type batchIndexer struct {
|
|||
}
|
||||
|
||||
// newBatchIndexer constructs the batch indexer with the supplied mode.
|
||||
func newBatchIndexer(db ethdb.KeyValueStore, delete bool, typ historyType) *batchIndexer {
|
||||
func newBatchIndexer(config historyIndexerConfig, db ethdb.KeyValueStore, delete bool, typ historyType) *batchIndexer {
|
||||
return &batchIndexer{
|
||||
config: config,
|
||||
index: make(map[stateIdent][]uint64),
|
||||
delete: delete,
|
||||
typ: typ,
|
||||
|
|
@ -156,7 +158,7 @@ func (b *batchIndexer) finish(force bool) error {
|
|||
if b.pending == 0 {
|
||||
return nil
|
||||
}
|
||||
if !force && b.pending < historyIndexBatch {
|
||||
if !force && b.pending < b.config.indexBatchSize {
|
||||
return nil
|
||||
}
|
||||
var (
|
||||
|
|
@ -170,7 +172,7 @@ func (b *batchIndexer) finish(force bool) error {
|
|||
for ident, list := range b.index {
|
||||
eg.Go(func() error {
|
||||
if !b.delete {
|
||||
iw, err := newIndexWriter(b.db, ident)
|
||||
iw, err := newIndexWriter(b.db, ident, b.config.maxEntriesPerBlock)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -183,7 +185,7 @@ func (b *batchIndexer) finish(force bool) error {
|
|||
iw.finish(batch)
|
||||
batchMu.Unlock()
|
||||
} else {
|
||||
id, err := newIndexDeleter(b.db, ident)
|
||||
id, err := newIndexDeleter(b.db, ident, b.config.maxEntriesPerBlock)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -222,7 +224,7 @@ func (b *batchIndexer) finish(force bool) error {
|
|||
}
|
||||
|
||||
// indexSingle processes the state history with the specified ID for indexing.
|
||||
func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, typ historyType) error {
|
||||
func indexSingle(config historyIndexerConfig, historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, typ historyType) error {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
if typ == typeStateHistory {
|
||||
|
|
@ -243,7 +245,7 @@ func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancient
|
|||
var (
|
||||
err error
|
||||
h history
|
||||
b = newBatchIndexer(db, false, typ)
|
||||
b = newBatchIndexer(config, db, false, typ)
|
||||
)
|
||||
if typ == typeStateHistory {
|
||||
h, err = readStateHistory(freezer, historyID)
|
||||
|
|
@ -264,7 +266,7 @@ func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancient
|
|||
}
|
||||
|
||||
// unindexSingle processes the state history with the specified ID for unindexing.
|
||||
func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, typ historyType) error {
|
||||
func unindexSingle(config historyIndexerConfig, historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, typ historyType) error {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
if typ == typeStateHistory {
|
||||
|
|
@ -286,7 +288,7 @@ func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancie
|
|||
err error
|
||||
h history
|
||||
)
|
||||
b := newBatchIndexer(db, true, typ)
|
||||
b := newBatchIndexer(config, db, true, typ)
|
||||
if typ == typeStateHistory {
|
||||
h, err = readStateHistory(freezer, historyID)
|
||||
} else {
|
||||
|
|
@ -319,6 +321,7 @@ type interruptSignal struct {
|
|||
// If a state history is removed due to a rollback, the associated indexes should
|
||||
// be unmarked accordingly.
|
||||
type indexIniter struct {
|
||||
config historyIndexerConfig
|
||||
disk ethdb.KeyValueStore
|
||||
freezer ethdb.AncientStore
|
||||
interrupt chan *interruptSignal
|
||||
|
|
@ -334,8 +337,9 @@ type indexIniter struct {
|
|||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func newIndexIniter(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, typ historyType, lastID uint64) *indexIniter {
|
||||
func newIndexIniter(config historyIndexerConfig, disk ethdb.KeyValueStore, freezer ethdb.AncientStore, typ historyType, lastID uint64) *indexIniter {
|
||||
initer := &indexIniter{
|
||||
config: config,
|
||||
disk: disk,
|
||||
freezer: freezer,
|
||||
interrupt: make(chan *interruptSignal),
|
||||
|
|
@ -446,7 +450,7 @@ func (i *indexIniter) run(lastID uint64) {
|
|||
// been fully indexed, unindex it here and shut down the initializer.
|
||||
if checkDone() {
|
||||
i.log.Info("Truncate the extra history", "id", lastID)
|
||||
if err := unindexSingle(lastID, i.disk, i.freezer, i.typ); err != nil {
|
||||
if err := unindexSingle(i.config, lastID, i.disk, i.freezer, i.typ); err != nil {
|
||||
signal.result <- err
|
||||
return
|
||||
}
|
||||
|
|
@ -547,7 +551,7 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
|||
current = beginID
|
||||
start = time.Now()
|
||||
logged = time.Now()
|
||||
batch = newBatchIndexer(i.disk, false, i.typ)
|
||||
batch = newBatchIndexer(i.config, i.disk, false, i.typ)
|
||||
)
|
||||
for current <= lastID {
|
||||
count := lastID - current + 1
|
||||
|
|
@ -653,6 +657,37 @@ func (i *indexIniter) recover(lastID uint64) {
|
|||
}
|
||||
}
|
||||
|
||||
// historyIndexerConfig defines the configuration parameters used for building
|
||||
// history indexes.
|
||||
//
|
||||
// Each state index maintains a set of index blocks, each grouping multiple history
|
||||
// entries. The configuration below controls how these entries are organized,
|
||||
// batched, and stored for different types of history data.
|
||||
type historyIndexerConfig struct {
|
||||
maxEntriesPerBlock uint16 // Maximum number of entries grouped into a single index block
|
||||
indexBatchSize int // Number of history entries processed in a single operation
|
||||
}
|
||||
|
||||
// indexerConfigs defines the configs for various history indexer.
|
||||
var indexerConfigs = map[historyType]historyIndexerConfig{
|
||||
typeStateHistory: {
|
||||
// The average size of each entry is approximately 1–2 bytes. Therefore, a block
|
||||
// size of 2K entries is chosen to roughly limit the storage size to around 4KB.
|
||||
//
|
||||
// Note, this block size was chosen as 4K previously. Changing the block size
|
||||
// won't break the backward compatibility.
|
||||
maxEntriesPerBlock: 2048,
|
||||
indexBatchSize: 512 * 1024,
|
||||
},
|
||||
typeTrienodeHistory: {
|
||||
// The average size of each entry is approximately 1–2 bytes, with an additional
|
||||
// byte reserved for extension. Therefore, a block size of 1K entries is chosen
|
||||
// to roughly limit the storage size to around 4KB.
|
||||
maxEntriesPerBlock: 1024,
|
||||
indexBatchSize: 512 * 1024,
|
||||
},
|
||||
}
|
||||
|
||||
// historyIndexer manages the indexing and unindexing of state histories,
|
||||
// providing access to historical states.
|
||||
//
|
||||
|
|
@ -664,6 +699,7 @@ func (i *indexIniter) recover(lastID uint64) {
|
|||
// the history index is created or removed along with the corresponding
|
||||
// state history.
|
||||
type historyIndexer struct {
|
||||
config historyIndexerConfig
|
||||
initer *indexIniter
|
||||
typ historyType
|
||||
disk ethdb.KeyValueStore
|
||||
|
|
@ -718,8 +754,14 @@ func checkVersion(disk ethdb.KeyValueStore, typ historyType) {
|
|||
// initer to complete the indexing of any remaining state histories.
|
||||
func newHistoryIndexer(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, lastHistoryID uint64, typ historyType) *historyIndexer {
|
||||
checkVersion(disk, typ)
|
||||
|
||||
config, exists := indexerConfigs[typ]
|
||||
if !exists {
|
||||
panic(fmt.Errorf("unknown history type: %v", typ))
|
||||
}
|
||||
return &historyIndexer{
|
||||
initer: newIndexIniter(disk, freezer, typ, lastHistoryID),
|
||||
config: config,
|
||||
initer: newIndexIniter(config, disk, freezer, typ, lastHistoryID),
|
||||
typ: typ,
|
||||
disk: disk,
|
||||
freezer: freezer,
|
||||
|
|
@ -748,7 +790,7 @@ func (i *historyIndexer) extend(historyID uint64) error {
|
|||
case <-i.initer.closed:
|
||||
return errors.New("indexer is closed")
|
||||
case <-i.initer.done:
|
||||
return indexSingle(historyID, i.disk, i.freezer, i.typ)
|
||||
return indexSingle(i.config, historyID, i.disk, i.freezer, i.typ)
|
||||
case i.initer.interrupt <- signal:
|
||||
return <-signal.result
|
||||
}
|
||||
|
|
@ -765,7 +807,7 @@ func (i *historyIndexer) shorten(historyID uint64) error {
|
|||
case <-i.initer.closed:
|
||||
return errors.New("indexer is closed")
|
||||
case <-i.initer.done:
|
||||
return unindexSingle(historyID, i.disk, i.freezer, i.typ)
|
||||
return unindexSingle(i.config, historyID, i.disk, i.freezer, i.typ)
|
||||
case i.initer.interrupt <- signal:
|
||||
return <-signal.result
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue