core: use finalized block as the chain freeze indicator

This commit is contained in:
Gary Rong 2023-12-15 11:49:36 +08:00
parent 5c67066a05
commit 33c4b23108
4 changed files with 43 additions and 56 deletions

View file

@ -1757,7 +1757,7 @@ func testRepair(t *testing.T, tt *rewindTest, snapshots bool) {
func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme string) { func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme string) {
// It's hard to follow the test case, visualize the input // It's hard to follow the test case, visualize the input
//log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))) // log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true))))
// fmt.Println(tt.dump(true)) // fmt.Println(tt.dump(true))
// Create a temporary persistent database // Create a temporary persistent database
@ -1830,10 +1830,14 @@ func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme s
} }
// Force run a freeze cycle // Force run a freeze cycle
type freezer interface { type freezer interface {
Freeze(threshold uint64) error Freeze() error
Ancients() (uint64, error) Ancients() (uint64, error)
} }
db.(freezer).Freeze(tt.freezeThreshold) if tt.freezeThreshold < uint64(tt.canonicalBlocks) {
final := uint64(tt.canonicalBlocks) - tt.freezeThreshold
chain.SetFinalized(canonblocks[int(final)-1].Header())
}
db.(freezer).Freeze()
// Set the simulated pivot block // Set the simulated pivot block
if tt.pivotBlock != nil { if tt.pivotBlock != nil {

View file

@ -2044,10 +2044,14 @@ func testSetHeadWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme
// Force run a freeze cycle // Force run a freeze cycle
type freezer interface { type freezer interface {
Freeze(threshold uint64) error Freeze() error
Ancients() (uint64, error) Ancients() (uint64, error)
} }
db.(freezer).Freeze(tt.freezeThreshold) if tt.freezeThreshold < uint64(tt.canonicalBlocks) {
final := uint64(tt.canonicalBlocks) - tt.freezeThreshold
chain.SetFinalized(canonblocks[int(final)-1].Header())
}
db.(freezer).Freeze()
// Set the simulated pivot block // Set the simulated pivot block
if tt.pivotBlock != nil { if tt.pivotBlock != nil {

View file

@ -17,15 +17,14 @@
package rawdb package rawdb
import ( import (
"errors"
"fmt" "fmt"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/params"
) )
const ( const (
@ -43,8 +42,6 @@ const (
// The background thread will keep moving ancient chain segments from key-value // The background thread will keep moving ancient chain segments from key-value
// database to flat files for saving space on live database. // database to flat files for saving space on live database.
type chainFreezer struct { type chainFreezer struct {
threshold atomic.Uint64 // Number of recent blocks not to freeze (params.FullImmutabilityThreshold apart from tests)
*Freezer *Freezer
quit chan struct{} quit chan struct{}
wg sync.WaitGroup wg sync.WaitGroup
@ -57,13 +54,11 @@ func newChainFreezer(datadir string, namespace string, readonly bool) (*chainFre
if err != nil { if err != nil {
return nil, err return nil, err
} }
cf := chainFreezer{ return &chainFreezer{
Freezer: freezer, Freezer: freezer,
quit: make(chan struct{}), quit: make(chan struct{}),
trigger: make(chan chan struct{}), trigger: make(chan chan struct{}),
} }, nil
cf.threshold.Store(params.FullImmutabilityThreshold)
return &cf, nil
} }
// Close closes the chain freezer instance and terminates the background thread. // Close closes the chain freezer instance and terminates the background thread.
@ -77,6 +72,19 @@ func (f *chainFreezer) Close() error {
return f.Freezer.Close() return f.Freezer.Close()
} }
// readFinalized loads the finalized block from database.
func (f *chainFreezer) readFinalized(db ethdb.KeyValueReader) (uint64, common.Hash, error) {
hash := ReadFinalizedBlockHash(db)
if hash == (common.Hash{}) {
return 0, common.Hash{}, errors.New("finalized block is not available")
}
number := ReadHeaderNumber(db, hash)
if number == nil {
return 0, common.Hash{}, errors.New("finalized block number is not available")
}
return *number, hash, nil
}
// freeze is a background thread that periodically checks the blockchain for any // freeze is a background thread that periodically checks the blockchain for any
// import progress and moves ancient data from the fast database into the freezer. // import progress and moves ancient data from the fast database into the freezer.
// //
@ -114,47 +122,28 @@ func (f *chainFreezer) freeze(db ethdb.KeyValueStore) {
return return
} }
} }
// Retrieve the freezing threshold. finalNumber, finalHash, err := f.readFinalized(nfdb)
hash := ReadHeadBlockHash(nfdb) if err != nil {
if hash == (common.Hash{}) { backoff = true // chain is not finalized yet
log.Debug("Current full block hash unavailable") // new chain, empty database log.Debug("Finalized block is not available yet")
backoff = true
continue continue
} }
number := ReadHeaderNumber(nfdb, hash)
threshold := f.threshold.Load()
frozen := f.frozen.Load() frozen := f.frozen.Load()
switch {
case number == nil:
log.Error("Current full block number unavailable", "hash", hash)
backoff = true
continue
case *number < threshold: // Short circuit if finalized blocks are already frozen.
log.Debug("Current full block not old enough to freeze", "number", *number, "hash", hash, "delay", threshold) if frozen != 0 && frozen-1 >= finalNumber {
backoff = true
continue
case *number-threshold <= frozen:
log.Debug("Ancient blocks frozen already", "number", *number, "hash", hash, "frozen", frozen)
backoff = true backoff = true
log.Debug("Ancient blocks frozen already", "number", finalNumber, "hash", finalHash, "frozen", frozen)
continue continue
} }
head := ReadHeader(nfdb, hash, *number)
if head == nil {
log.Error("Current full block unavailable", "number", *number, "hash", hash)
backoff = true
continue
}
// Seems we have data ready to be frozen, process in usable batches // Seems we have data ready to be frozen, process in usable batches
var ( var (
start = time.Now() start = time.Now()
first, _ = f.Ancients() first = frozen
limit = *number - threshold limit = finalNumber
) )
if limit-first > freezerBatchLimit { if limit-first+1 > freezerBatchLimit {
limit = first + freezerBatchLimit limit = freezerBatchLimit + first - 1
} }
ancients, err := f.freezeRange(nfdb, first, limit) ancients, err := f.freezeRange(nfdb, first, limit)
if err != nil { if err != nil {
@ -162,12 +151,10 @@ func (f *chainFreezer) freeze(db ethdb.KeyValueStore) {
backoff = true backoff = true
continue continue
} }
// Batch of blocks have been frozen, flush them before wiping from leveldb // Batch of blocks have been frozen, flush them before wiping from leveldb
if err := f.Sync(); err != nil { if err := f.Sync(); err != nil {
log.Crit("Failed to flush frozen tables", "err", err) log.Crit("Failed to flush frozen tables", "err", err)
} }
// Wipe out all data from the active database // Wipe out all data from the active database
batch := db.NewBatch() batch := db.NewBatch()
for i := 0; i < len(ancients); i++ { for i := 0; i < len(ancients); i++ {
@ -251,7 +238,7 @@ func (f *chainFreezer) freeze(db ethdb.KeyValueStore) {
} }
func (f *chainFreezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hashes []common.Hash, err error) { func (f *chainFreezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hashes []common.Hash, err error) {
hashes = make([]common.Hash, 0, limit-number) hashes = make([]common.Hash, 0, limit-number+1)
_, err = f.ModifyAncients(func(op ethdb.AncientWriteOp) error { _, err = f.ModifyAncients(func(op ethdb.AncientWriteOp) error {
for ; number <= limit; number++ { for ; number <= limit; number++ {
@ -293,11 +280,9 @@ func (f *chainFreezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hash
if err := op.AppendRaw(ChainFreezerDifficultyTable, number, td); err != nil { if err := op.AppendRaw(ChainFreezerDifficultyTable, number, td); err != nil {
return fmt.Errorf("can't write td to Freezer: %v", err) return fmt.Errorf("can't write td to Freezer: %v", err)
} }
hashes = append(hashes, hash) hashes = append(hashes, hash)
} }
return nil return nil
}) })
return hashes, err return hashes, err
} }

View file

@ -66,16 +66,10 @@ func (frdb *freezerdb) Close() error {
// Freeze is a helper method used for external testing to trigger and block until // Freeze is a helper method used for external testing to trigger and block until
// a freeze cycle completes, without having to sleep for a minute to trigger the // a freeze cycle completes, without having to sleep for a minute to trigger the
// automatic background run. // automatic background run.
func (frdb *freezerdb) Freeze(threshold uint64) error { func (frdb *freezerdb) Freeze() error {
if frdb.AncientStore.(*chainFreezer).readonly { if frdb.AncientStore.(*chainFreezer).readonly {
return errReadOnly return errReadOnly
} }
// Set the freezer threshold to a temporary value
defer func(old uint64) {
frdb.AncientStore.(*chainFreezer).threshold.Store(old)
}(frdb.AncientStore.(*chainFreezer).threshold.Load())
frdb.AncientStore.(*chainFreezer).threshold.Store(threshold)
// Trigger a freeze cycle and block until it's done // Trigger a freeze cycle and block until it's done
trigger := make(chan struct{}, 1) trigger := make(chan struct{}, 1)
frdb.AncientStore.(*chainFreezer).trigger <- trigger frdb.AncientStore.(*chainFreezer).trigger <- trigger