forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-28 07:36:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refine the implementation

Browse source
This commit is contained in:
user 2022-06-30 22:54:13 +08:00
parent 81157fc4a8
commit 978827202f
6 changed files with 24 additions and 458 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
core/blockchain.go
View file

@ -941,7 +941,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
defer bc.wg.Done()
if bc.cacheConfig.AncientRecentLimit != 0 {
log.Info("throwing ancient turned on, set ancientLimit to 0 during snap sync")
ancientLimit = 0
}
@ -2303,7 +2302,7 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) {
rawdb.WriteTxIndexTail(bc.db, 0)
} else {
// Prune all stale tx indices and record the tx index tail
log.Info("Scheduled transactions unindexing", "from block", 0, "to", head-bc.txLookupLimit+1)
log.Info("Scheduled blocks & transactions unindexing", "from block", 0, "to", head-bc.txLookupLimit+1)
rawdb.UnindexTransactions(bc.db, 0, head-bc.txLookupLimit+1, bc.quit)
}
return
@ -2328,7 +2327,7 @@ func (bc *BlockChain) maintainTxIndex(ancients uint64) {
rawdb.IndexTransactions(bc.db, head-bc.txLookupLimit+1, *tail, bc.quit)
} else {
// Unindex a part of stale indices and forward index tail to HEAD-limit
log.Info("Scheduled transactions unindexing", "from block", *tail, "to", head-bc.txLookupLimit+1)
log.Info("Scheduled blocks & transactions unindexing", "from block", *tail, "to", head-bc.txLookupLimit+1)
rawdb.UnindexTransactions(bc.db, *tail, head-bc.txLookupLimit+1, bc.quit)
}
}

18
core/rawdb/chain_iterator.go
View file

@ -84,8 +84,9 @@ func InitDatabaseFromFreezer(db ethdb.Database) {
}
type blockTxHashes struct {
number uint64
hashes []common.Hash
number uint64
blockHash common.Hash
hashes []common.Hash
}
// iterateTransactions iterates over all transactions in the (canon) block
@ -96,6 +97,7 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool
// One thread sequentially reads data from db
type numberRlp struct {
number uint64
hash common.Hash
rlp rlp.RawValue
}
if to == from {
@ -118,9 +120,10 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool
defer close(rlpCh)
for n != end {
data := ReadCanonicalBodyRLP(db, n)
hash := ReadCanonicalHash(db, n)
// Feed the block to the aggregator, or abort on interrupt
select {
case rlpCh <- &numberRlp{n, data}:
case rlpCh <- &numberRlp{n, hash, data}:
case <-interrupt:
return
}
@ -151,8 +154,9 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool
hashes = append(hashes, tx.Hash())
}
result := &blockTxHashes{
hashes: hashes,
number: data.number,
hashes: hashes,
blockHash: data.hash,
number: data.number,
}
// Feed the block to the aggregator, or abort on interrupt
select {
@ -303,6 +307,10 @@ func unindexTransactions(db ethdb.Database, from uint64, to uint64, interrupt ch
nextNum = delivery.number + 1
DeleteTxLookupEntries(batch, delivery.hashes)
txs += len(delivery.hashes)
// Delete all about the block
if delivery.number != 0 { // never delete the genesis block
DeleteBlock(batch, delivery.blockHash, delivery.number)
}
blocks++
// If enough data was accumulated in memory or we're at the last block, dump to disk

372
core/rawdb/chain_thrower.go
View file

@ -1,372 +0,0 @@
// Copyright 2022 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package rawdb
import (
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/params"
)
const (
// throwerRecheckInterval is the frequency to check the key-value database for
// chain progression that might permit new blocks to be frozen into immutable
// storage.
throwerRecheckInterval = time.Minute
// throwerBatchLimit is the maximum number of blocks to freeze in one batch
// before doing an fsync and deleting it from the key-value store.
throwerBatchLimit = 30000
)
// chainThrower is a wrapper of freezer with additional chain freezing feature.
// The background thread will keep moving ancient chain segments from key-value
// database to flat files for saving space on live database.
type chainThrower struct {
throwdb
// WARNING: The `threshold` field is accessed atomically. On 32 bit platforms, only
// 64-bit aligned fields can be atomic. The struct is guaranteed to be so aligned,
// so take advantage of that (https://golang.org/pkg/sync/atomic/#pkg-note-BUG).
threshold uint64 // Number of recent blocks not to freeze (params.FullImmutabilityThreshold apart from tests)
quit chan struct{}
wg sync.WaitGroup
trigger chan chan struct{} // Manual blocking freeze trigger, test determinism
}
// throwdb is a database wrapper that disables freezer data retrievals.
type throwdb struct {
ethdb.KeyValueStore
}
// HasAncient always returns false as `throwdb` has thrown everything written to it
func (db *throwdb) HasAncient(kind string, number uint64) (bool, error) {
return false, nil
}
// Ancient returns an empty result as `throwdb` has thrown everything written to it
func (db *throwdb) Ancient(kind string, number uint64) ([]byte, error) {
return []byte{}, nil
}
// AncientRange returns an empty result as `throwdb` has thrown everything written to it
func (db *throwdb) AncientRange(kind string, start, max, maxByteSize uint64) ([][]byte, error) {
return [][]byte{}, nil
}
// Ancients returns 0 as we don't have a backing chain freezer.
func (db *throwdb) Ancients() (uint64, error) {
return 0, nil
}
// Tail returns 0 as we don't have a backing chain freezer.
func (db *throwdb) Tail() (uint64, error) {
return 0, errNotSupported
}
// AncientSize returns an error as we don't have a backing chain freezer.
func (db *throwdb) AncientSize(kind string) (uint64, error) {
return 0, nil
}
// ModifyAncients is not supported.
func (db *throwdb) ModifyAncients(func(ethdb.AncientWriteOp) error) (int64, error) {
return 0, nil
}
// TruncateHead returns an error as we don't have a backing chain freezer.
func (db *throwdb) TruncateHead(items uint64) error {
return nil
}
// TruncateTail returns an error as we don't have a backing chain freezer.
func (db *throwdb) TruncateTail(items uint64) error {
return nil
}
// Sync returns an error as we don't have a backing chain freezer.
func (db *throwdb) Sync() error {
return nil
}
func (db *throwdb) ReadAncients(fn func(reader ethdb.AncientReaderOp) error) (err error) {
// Unlike other ancient-related methods, this method does not return
// errNotSupported when invoked.
// The reason for this is that the caller might want to do several things:
// 1. Check if something is in freezer,
// 2. If not, check leveldb.
//
// This will work, since the ancient-checks inside 'fn' will return errors,
// and the leveldb work will continue.
//
// If we instead were to return errNotSupported here, then the caller would
// have to explicitly check for that, having an extra clause to do the
// non-ancient operations.
return fn(db)
}
// MigrateTable processes the entries in a given table in sequence
// converting them to a new format if they're of an old format.
func (db *throwdb) MigrateTable(kind string, convert convertLegacyFn) error {
return errNotSupported
}
// AncientDatadir returns an error as we don't have a backing chain freezer.
func (db *throwdb) AncientDatadir() (string, error) {
return "", errNotSupported
}
// newChainFreezer initializes the freezer for ancient chain data.
func newChainThrower(datadir string, namespace string, readonly bool) (*chainThrower, error) {
return &chainThrower{
threshold: params.FullImmutabilityThreshold,
quit: make(chan struct{}),
trigger: make(chan chan struct{}),
}, nil
}
// Close closes the chain freezer instance and terminates the background thread.
func (f *chainThrower) Close() error {
select {
case <-f.quit:
default:
close(f.quit)
}
f.wg.Wait()
return nil
}
// 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.
//
// This functionality is deliberately broken off from block importing to avoid
// incurring additional data shuffling delays on block propagation.
func (f *chainThrower) throw(db ethdb.KeyValueStore) {
nfdb := &throwdb{KeyValueStore: db}
var (
backoff bool
triggered chan struct{} // Used in tests
)
for {
select {
case <-f.quit:
log.Info("Thrower shutting down")
return
default:
}
if backoff {
// If we were doing a manual trigger, notify it
if triggered != nil {
triggered <- struct{}{}
triggered = nil
}
select {
case <-time.NewTimer(throwerRecheckInterval).C:
backoff = false
case triggered = <-f.trigger:
backoff = false
case <-f.quit:
return
}
}
// Retrieve the freezing threshold.
hash := ReadHeadBlockHash(nfdb)
if hash == (common.Hash{}) {
log.Debug("Current full block hash unavailable") // new chain, empty database
backoff = true
continue
}
number := ReadHeaderNumber(nfdb, hash)
threshold := atomic.LoadUint64(&f.threshold)
last := ReadAncientThrowLastBlockNumber(db)
if last == nil {
last = new(uint64)
}
switch {
case number == nil:
log.Error("Current full block number unavailable", "hash", hash)
backoff = true
continue
case *number < threshold:
log.Debug("Current full block not old enough", "number", *number, "hash", hash, "delay", threshold)
backoff = true
continue
case *number-threshold <= *last:
log.Debug("Ancient blocks frozen already", "number", *number, "hash", hash, "last", last)
backoff = true
continue
}
storedSections := ReadStoredBloomSections(nfdb)
if storedSections*params.BloomBitsBlocks-1 < *last {
log.Warn("Attempt to prune the ancient blocks that bloom filter haven't finished yet, postpone to next round", "storedSections", storedSections, "pruneTo", *last)
return
}
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
var (
start = time.Now()
first = *last
limit = *number - threshold
)
if limit-first > freezerBatchLimit {
limit = first + freezerBatchLimit
}
log.Info("schedule throwing blocks", "from", first, "to", limit)
ancients, err := f.throwRange(nfdb, first, limit)
if err != nil {
log.Error("Error in block freeze operation", "err", err)
backoff = true
continue
}
// Wipe out all data from the active database
batch := db.NewBatch()
for i := 0; i < len(ancients); i++ {
// Always keep the genesis block in active database
if first+uint64(i) != 0 {
DeleteBlockWithoutNumber(batch, ancients[i], first+uint64(i))
DeleteCanonicalHash(batch, first+uint64(i))
}
}
if err := batch.Write(); err != nil {
log.Crit("Failed to throw ancient blocks", "err", err)
}
batch.Reset()
// record
WriteAncientThrowLastBlockNumber(db, limit)
// Wipe out side chains also and track dangling side chains
var dangling []common.Hash
for number := first; number < limit; number++ {
// Always keep the genesis block in active database
if number != 0 {
dangling = ReadAllHashes(db, number)
for _, hash := range dangling {
log.Trace("Deleting side chain", "number", number, "hash", hash)
DeleteBlock(batch, hash, number)
}
}
}
if err := batch.Write(); err != nil {
log.Crit("Failed to delete frozen side blocks", "err", err)
}
batch.Reset()
// Step into the future and delete and dangling side chains
if limit > 0 {
tip := limit
for len(dangling) > 0 {
drop := make(map[common.Hash]struct{})
for _, hash := range dangling {
log.Debug("Dangling parent from Freezer", "number", tip-1, "hash", hash)
drop[hash] = struct{}{}
}
children := ReadAllHashes(db, tip)
for i := 0; i < len(children); i++ {
// Dig up the child and ensure it's dangling
child := ReadHeader(nfdb, children[i], tip)
if child == nil {
log.Error("Missing dangling header", "number", tip, "hash", children[i])
continue
}
if _, ok := drop[child.ParentHash]; !ok {
children = append(children[:i], children[i+1:]...)
i--
continue
}
// Delete all block data associated with the child
log.Debug("Deleting dangling block", "number", tip, "hash", children[i], "parent", child.ParentHash)
DeleteBlock(batch, children[i], tip)
}
dangling = children
tip++
}
if err := batch.Write(); err != nil {
log.Crit("Failed to delete dangling side blocks", "err", err)
}
}
// Log something friendly for the user
context := []interface{}{
"blocks", limit - first, "elapsed", common.PrettyDuration(time.Since(start)), "number", limit - 1,
}
if n := len(ancients); n > 0 {
context = append(context, []interface{}{"hash", ancients[n-1]}...)
}
log.Info("Deep throw chain segment", context...)
// Avoid database thrashing with tiny writes
if limit-first < freezerBatchLimit {
backoff = true
}
}
}
func (f *chainThrower) throwRange(nfdb *throwdb, number, limit uint64) (hashes []common.Hash, err error) {
hashes = make([]common.Hash, 0, limit-number)
for ; number <= limit; number++ {
// Retrieve all the components of the canonical block.
hash := ReadCanonicalHash(nfdb, number)
if hash == (common.Hash{}) {
log.Error("canonical hash missing, can't freeze", "block %d", number)
continue
}
header := ReadHeaderRLP(nfdb, hash, number)
if len(header) == 0 {
log.Error("block header missing, can't freeze", "block %d", number)
continue
}
body := ReadBodyRLP(nfdb, hash, number)
if len(body) == 0 {
log.Error("block body missing, can't freeze", "block %d", number)
continue
}
receipts := ReadReceiptsRLP(nfdb, hash, number)
if len(receipts) == 0 {
log.Error("block receipts missing, can't freeze", "block %d", number)
continue
}
td := ReadTdRLP(nfdb, hash, number)
if len(td) == 0 {
log.Error("total difficulty missing, can't freeze", "block %d", number)
continue
}
hashes = append(hashes, hash)
}
return hashes, err
}

47
core/rawdb/database.go
View file

@ -165,7 +165,7 @@ func NewDatabase(db ethdb.KeyValueStore) ethdb.Database {
// NewDatabaseWithFreezer creates a high level database on top of a given key-
// value data store with a freezer moving immutable chain segments into cold
// storage.
func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace string, readonly bool) (ethdb.Database, error) {
func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace string, readonly, discardAncient bool) (ethdb.Database, error) {
// Create the idle freezer instance
frdb, err := newChainFreezer(freezer, namespace, readonly, freezerTableSize, FreezerNoSnappy)
if err != nil {
@ -237,7 +237,7 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st
}
}
// Freezer is consistent with the key-value database, permit combining the two
if !frdb.readonly {
if !frdb.readonly && !discardAncient {
frdb.wg.Add(1)
go func() {
frdb.freeze(db)
@ -250,29 +250,6 @@ func NewDatabaseWithFreezer(db ethdb.KeyValueStore, freezer string, namespace st
}, nil
}
// NewDatabaseWithThrower creates a high level database on top of a given key-
// value data store with a freezer moving immutable chain segments into cold
// storage.
func NewDatabaseWithThrower(db ethdb.KeyValueStore, freezer string, namespace string, readonly bool) (ethdb.Database, error) {
// Create the idle freezer instance
throwdb, err := newChainThrower(freezer, namespace, readonly)
if err != nil {
return nil, err
}
// Thrower is consistent with the key-value database, permit combining the two
throwdb.wg.Add(1)
go func() {
throwdb.throw(db)
throwdb.wg.Done()
}()
return &freezerdb{
KeyValueStore: db,
AncientStore: throwdb,
}, nil
}
// NewMemoryDatabase creates an ephemeral in-memory key-value database without a
// freezer moving immutable chain segments into cold storage.
func NewMemoryDatabase() ethdb.Database {
@ -298,29 +275,13 @@ func NewLevelDBDatabase(file string, cache int, handles int, namespace string, r
// NewLevelDBDatabaseWithFreezer creates a persistent key-value database with a
// freezer moving immutable chain segments into cold storage.
func NewLevelDBDatabaseWithFreezer(file string, cache int, handles int, freezer string, namespace string, readonly bool) (ethdb.Database, error) {
func NewLevelDBDatabaseWithFreezer(file string, cache int, handles int, freezer string, namespace string, readonly bool, ancientRecentLimit uint64) (ethdb.Database, error) {
kvdb, err := leveldb.New(file, cache, handles, namespace, readonly)
if err != nil {
return nil, err
}
frdb, err := NewDatabaseWithFreezer(kvdb, freezer, namespace, readonly)
if err != nil {
kvdb.Close()
return nil, err
}
return frdb, nil
}
// NewLevelDBDatabaseWithFreezer creates a persistent key-value database with a
// freezer moving immutable chain segments into cold storage.
func NewLevelDBDatabaseWithThrower(file string, cache int, handles int, freezer string, namespace string, readonly bool) (ethdb.Database, error) {
kvdb, err := leveldb.New(file, cache, handles, namespace, readonly)
if err != nil {
return nil, err
}
frdb, err := NewDatabaseWithThrower(kvdb, freezer, namespace, readonly)
frdb, err := NewDatabaseWithFreezer(kvdb, freezer, namespace, readonly, ancientRecentLimit != 0)
if err != nil {
kvdb.Close()
return nil, err

6
eth/backend.go
View file

@ -133,7 +133,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
ethashConfig.NotifyFull = config.Miner.NotifyFull
// Assemble the Ethereum object
chainDb, err := stack.OpenDatabaseWrapper("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit != 0)
chainDb, err := stack.OpenDatabaseWithFreezer("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false, config.AncientRecentLimit)
if err != nil {
return nil, err
}
@ -215,7 +215,9 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
eth.blockchain.SetHead(compat.RewindTo)
rawdb.WriteChainConfig(chainDb, genesisHash, chainConfig)
}
eth.bloomIndexer.Start(eth.blockchain)
if config.AncientRecentLimit == 0 {
eth.bloomIndexer.Start(eth.blockchain)
}
if config.TxPool.Journal != "" {
config.TxPool.Journal = stack.ResolvePath(config.TxPool.Journal)

34
node/node.go
View file

@ -712,38 +712,6 @@ func (n *Node) OpenDatabase(name string, cache, handles int, namespace string, r
return db, err
}
// OpenDatabaseWrapper ...
func (n *Node) OpenDatabaseWrapper(name string, cache, handles int, freezer, namespace string, readonly, throwAncient bool) (ethdb.Database, error) {
n.lock.Lock()
defer n.lock.Unlock()
if n.state == closedState {
return nil, ErrNodeStopped
}
var db ethdb.Database
var err error
if n.config.DataDir == "" {
db = rawdb.NewMemoryDatabase()
} else if !throwAncient {
root := n.ResolvePath(name)
switch {
case freezer == "":
freezer = filepath.Join(root, "ancient")
case !filepath.IsAbs(freezer):
freezer = n.ResolvePath(freezer)
}
db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly)
} else {
root := n.ResolvePath(name)
db, err = rawdb.NewLevelDBDatabaseWithThrower(root, cache, handles, freezer, namespace, readonly)
}
if err == nil {
db = n.wrapDatabase(db)
}
return db, err
}
// OpenDatabaseWithFreezer opens an existing database with the given name (or
// creates one if no previous can be found) from within the node's data directory,
// also attaching a chain freezer to it that moves ancient chain data from the
@ -768,7 +736,7 @@ func (n *Node) OpenDatabaseWithFreezer(name string, cache, handles int, freezer,
case !filepath.IsAbs(freezer):
freezer = n.ResolvePath(freezer)
}
db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly)
db, err = rawdb.NewLevelDBDatabaseWithFreezer(root, cache, handles, freezer, namespace, readonly, ancientRecentLimit)
}
if err == nil {