|
@@ -80,14 +80,22 @@ var (
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
const (
|
|
const (
|
|
|
- bodyCacheLimit = 256
|
|
|
|
|
- blockCacheLimit = 256
|
|
|
|
|
- receiptsCacheLimit = 10000
|
|
|
|
|
- txLookupCacheLimit = 1024
|
|
|
|
|
- maxFutureBlocks = 256
|
|
|
|
|
- maxTimeFutureBlocks = 30
|
|
|
|
|
- badBlockLimit = 10
|
|
|
|
|
- maxBeyondBlocks = 2048
|
|
|
|
|
|
|
+ bodyCacheLimit = 256
|
|
|
|
|
+ blockCacheLimit = 256
|
|
|
|
|
+ diffLayerCacheLimit = 1024
|
|
|
|
|
+ diffLayerRLPCacheLimit = 256
|
|
|
|
|
+ receiptsCacheLimit = 10000
|
|
|
|
|
+ txLookupCacheLimit = 1024
|
|
|
|
|
+ maxFutureBlocks = 256
|
|
|
|
|
+ maxTimeFutureBlocks = 30
|
|
|
|
|
+ maxBeyondBlocks = 2048
|
|
|
|
|
+
|
|
|
|
|
+ diffLayerFreezerRecheckInterval = 3 * time.Second
|
|
|
|
|
+ diffLayerPruneRecheckInterval = 1 * time.Second // The interval to prune unverified diff layers
|
|
|
|
|
+ maxDiffQueueDist = 2048 // Maximum allowed distance from the chain head to queue diffLayers
|
|
|
|
|
+ maxDiffLimit = 2048 // Maximum number of unique diff layers a peer may have responded
|
|
|
|
|
+ maxDiffForkDist = 11 // Maximum allowed backward distance from the chain head
|
|
|
|
|
+ maxDiffLimitForBroadcast = 128 // Maximum number of unique diff layers a peer may have broadcasted
|
|
|
|
|
|
|
|
// BlockChainVersion ensures that an incompatible database forces a resync from scratch.
|
|
// BlockChainVersion ensures that an incompatible database forces a resync from scratch.
|
|
|
//
|
|
//
|
|
@@ -131,6 +139,11 @@ type CacheConfig struct {
|
|
|
SnapshotWait bool // Wait for snapshot construction on startup. TODO(karalabe): This is a dirty hack for testing, nuke it
|
|
SnapshotWait bool // Wait for snapshot construction on startup. TODO(karalabe): This is a dirty hack for testing, nuke it
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// To avoid cycle import
|
|
|
|
|
+type PeerIDer interface {
|
|
|
|
|
+ ID() string
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// defaultCacheConfig are the default caching values if none are specified by the
|
|
// defaultCacheConfig are the default caching values if none are specified by the
|
|
|
// user (also used during testing).
|
|
// user (also used during testing).
|
|
|
var defaultCacheConfig = &CacheConfig{
|
|
var defaultCacheConfig = &CacheConfig{
|
|
@@ -142,6 +155,8 @@ var defaultCacheConfig = &CacheConfig{
|
|
|
SnapshotWait: true,
|
|
SnapshotWait: true,
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+type BlockChainOption func(*BlockChain) *BlockChain
|
|
|
|
|
+
|
|
|
// BlockChain represents the canonical chain given a database with a genesis
|
|
// BlockChain represents the canonical chain given a database with a genesis
|
|
|
// block. The Blockchain manages chain imports, reverts, chain reorganisations.
|
|
// block. The Blockchain manages chain imports, reverts, chain reorganisations.
|
|
|
//
|
|
//
|
|
@@ -196,6 +211,21 @@ type BlockChain struct {
|
|
|
txLookupCache *lru.Cache // Cache for the most recent transaction lookup data.
|
|
txLookupCache *lru.Cache // Cache for the most recent transaction lookup data.
|
|
|
futureBlocks *lru.Cache // future blocks are blocks added for later processing
|
|
futureBlocks *lru.Cache // future blocks are blocks added for later processing
|
|
|
|
|
|
|
|
|
|
+ // trusted diff layers
|
|
|
|
|
+ diffLayerCache *lru.Cache // Cache for the diffLayers
|
|
|
|
|
+ diffLayerRLPCache *lru.Cache // Cache for the rlp encoded diffLayers
|
|
|
|
|
+ diffQueue *prque.Prque // A Priority queue to store recent diff layer
|
|
|
|
|
+ diffQueueBuffer chan *types.DiffLayer
|
|
|
|
|
+ diffLayerFreezerBlockLimit uint64
|
|
|
|
|
+
|
|
|
|
|
+ // untrusted diff layers
|
|
|
|
|
+ diffMux sync.RWMutex
|
|
|
|
|
+ blockHashToDiffLayers map[common.Hash]map[common.Hash]*types.DiffLayer // map[blockHash] map[DiffHash]Diff
|
|
|
|
|
+ diffHashToBlockHash map[common.Hash]common.Hash // map[diffHash]blockHash
|
|
|
|
|
+ diffHashToPeers map[common.Hash]map[string]struct{} // map[diffHash]map[pid]
|
|
|
|
|
+ diffNumToBlockHashes map[uint64]map[common.Hash]struct{} // map[number]map[blockHash]
|
|
|
|
|
+ diffPeersToDiffHashes map[string]map[common.Hash]struct{} // map[pid]map[diffHash]
|
|
|
|
|
+
|
|
|
quit chan struct{} // blockchain quit channel
|
|
quit chan struct{} // blockchain quit channel
|
|
|
wg sync.WaitGroup // chain processing wait group for shutting down
|
|
wg sync.WaitGroup // chain processing wait group for shutting down
|
|
|
running int32 // 0 if chain is running, 1 when stopped
|
|
running int32 // 0 if chain is running, 1 when stopped
|
|
@@ -213,12 +243,15 @@ type BlockChain struct {
|
|
|
// NewBlockChain returns a fully initialised block chain using information
|
|
// NewBlockChain returns a fully initialised block chain using information
|
|
|
// available in the database. It initialises the default Ethereum Validator and
|
|
// available in the database. It initialises the default Ethereum Validator and
|
|
|
// Processor.
|
|
// Processor.
|
|
|
-func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool, txLookupLimit *uint64) (*BlockChain, error) {
|
|
|
|
|
|
|
+func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine,
|
|
|
|
|
+ vmConfig vm.Config, shouldPreserve func(block *types.Block) bool, txLookupLimit *uint64,
|
|
|
|
|
+ options ...BlockChainOption) (*BlockChain, error) {
|
|
|
if cacheConfig == nil {
|
|
if cacheConfig == nil {
|
|
|
cacheConfig = defaultCacheConfig
|
|
cacheConfig = defaultCacheConfig
|
|
|
}
|
|
}
|
|
|
if cacheConfig.TriesInMemory != 128 {
|
|
if cacheConfig.TriesInMemory != 128 {
|
|
|
- log.Warn("TriesInMemory isn't the default value(128), you need specify exact same TriesInMemory when prune data", "triesInMemory", cacheConfig.TriesInMemory)
|
|
|
|
|
|
|
+ log.Warn("TriesInMemory isn't the default value(128), you need specify exact same TriesInMemory when prune data",
|
|
|
|
|
+ "triesInMemory", cacheConfig.TriesInMemory)
|
|
|
}
|
|
}
|
|
|
bodyCache, _ := lru.New(bodyCacheLimit)
|
|
bodyCache, _ := lru.New(bodyCacheLimit)
|
|
|
bodyRLPCache, _ := lru.New(bodyCacheLimit)
|
|
bodyRLPCache, _ := lru.New(bodyCacheLimit)
|
|
@@ -226,6 +259,8 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
|
|
|
blockCache, _ := lru.New(blockCacheLimit)
|
|
blockCache, _ := lru.New(blockCacheLimit)
|
|
|
txLookupCache, _ := lru.New(txLookupCacheLimit)
|
|
txLookupCache, _ := lru.New(txLookupCacheLimit)
|
|
|
futureBlocks, _ := lru.New(maxFutureBlocks)
|
|
futureBlocks, _ := lru.New(maxFutureBlocks)
|
|
|
|
|
+ diffLayerCache, _ := lru.New(diffLayerCacheLimit)
|
|
|
|
|
+ diffLayerRLPCache, _ := lru.New(diffLayerRLPCacheLimit)
|
|
|
|
|
|
|
|
bc := &BlockChain{
|
|
bc := &BlockChain{
|
|
|
chainConfig: chainConfig,
|
|
chainConfig: chainConfig,
|
|
@@ -237,17 +272,26 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
|
|
|
Journal: cacheConfig.TrieCleanJournal,
|
|
Journal: cacheConfig.TrieCleanJournal,
|
|
|
Preimages: cacheConfig.Preimages,
|
|
Preimages: cacheConfig.Preimages,
|
|
|
}),
|
|
}),
|
|
|
- triesInMemory: cacheConfig.TriesInMemory,
|
|
|
|
|
- quit: make(chan struct{}),
|
|
|
|
|
- shouldPreserve: shouldPreserve,
|
|
|
|
|
- bodyCache: bodyCache,
|
|
|
|
|
- bodyRLPCache: bodyRLPCache,
|
|
|
|
|
- receiptsCache: receiptsCache,
|
|
|
|
|
- blockCache: blockCache,
|
|
|
|
|
- txLookupCache: txLookupCache,
|
|
|
|
|
- futureBlocks: futureBlocks,
|
|
|
|
|
- engine: engine,
|
|
|
|
|
- vmConfig: vmConfig,
|
|
|
|
|
|
|
+ triesInMemory: cacheConfig.TriesInMemory,
|
|
|
|
|
+ quit: make(chan struct{}),
|
|
|
|
|
+ shouldPreserve: shouldPreserve,
|
|
|
|
|
+ bodyCache: bodyCache,
|
|
|
|
|
+ bodyRLPCache: bodyRLPCache,
|
|
|
|
|
+ receiptsCache: receiptsCache,
|
|
|
|
|
+ blockCache: blockCache,
|
|
|
|
|
+ diffLayerCache: diffLayerCache,
|
|
|
|
|
+ diffLayerRLPCache: diffLayerRLPCache,
|
|
|
|
|
+ txLookupCache: txLookupCache,
|
|
|
|
|
+ futureBlocks: futureBlocks,
|
|
|
|
|
+ engine: engine,
|
|
|
|
|
+ vmConfig: vmConfig,
|
|
|
|
|
+ diffQueue: prque.New(nil),
|
|
|
|
|
+ diffQueueBuffer: make(chan *types.DiffLayer),
|
|
|
|
|
+ blockHashToDiffLayers: make(map[common.Hash]map[common.Hash]*types.DiffLayer),
|
|
|
|
|
+ diffHashToBlockHash: make(map[common.Hash]common.Hash),
|
|
|
|
|
+ diffHashToPeers: make(map[common.Hash]map[string]struct{}),
|
|
|
|
|
+ diffNumToBlockHashes: make(map[uint64]map[common.Hash]struct{}),
|
|
|
|
|
+ diffPeersToDiffHashes: make(map[string]map[common.Hash]struct{}),
|
|
|
}
|
|
}
|
|
|
bc.validator = NewBlockValidator(chainConfig, bc, engine)
|
|
bc.validator = NewBlockValidator(chainConfig, bc, engine)
|
|
|
bc.processor = NewStateProcessor(chainConfig, bc, engine)
|
|
bc.processor = NewStateProcessor(chainConfig, bc, engine)
|
|
@@ -375,6 +419,10 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
|
|
|
}
|
|
}
|
|
|
bc.snaps, _ = snapshot.New(bc.db, bc.stateCache.TrieDB(), bc.cacheConfig.SnapshotLimit, int(bc.cacheConfig.TriesInMemory), head.Root(), !bc.cacheConfig.SnapshotWait, true, recover)
|
|
bc.snaps, _ = snapshot.New(bc.db, bc.stateCache.TrieDB(), bc.cacheConfig.SnapshotLimit, int(bc.cacheConfig.TriesInMemory), head.Root(), !bc.cacheConfig.SnapshotWait, true, recover)
|
|
|
}
|
|
}
|
|
|
|
|
+ // do options before start any routine
|
|
|
|
|
+ for _, option := range options {
|
|
|
|
|
+ bc = option(bc)
|
|
|
|
|
+ }
|
|
|
// Take ownership of this particular state
|
|
// Take ownership of this particular state
|
|
|
go bc.update()
|
|
go bc.update()
|
|
|
if txLookupLimit != nil {
|
|
if txLookupLimit != nil {
|
|
@@ -396,6 +444,12 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
|
|
|
triedb.SaveCachePeriodically(bc.cacheConfig.TrieCleanJournal, bc.cacheConfig.TrieCleanRejournal, bc.quit)
|
|
triedb.SaveCachePeriodically(bc.cacheConfig.TrieCleanJournal, bc.cacheConfig.TrieCleanRejournal, bc.quit)
|
|
|
}()
|
|
}()
|
|
|
}
|
|
}
|
|
|
|
|
+ // Need persist and prune diff layer
|
|
|
|
|
+ if bc.db.DiffStore() != nil {
|
|
|
|
|
+ go bc.trustedDiffLayerLoop()
|
|
|
|
|
+ }
|
|
|
|
|
+ go bc.untrustedDiffLayerPruneLoop()
|
|
|
|
|
+
|
|
|
return bc, nil
|
|
return bc, nil
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -404,11 +458,19 @@ func (bc *BlockChain) GetVMConfig() *vm.Config {
|
|
|
return &bc.vmConfig
|
|
return &bc.vmConfig
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-func (bc *BlockChain) CacheReceipts(hash common.Hash, receipts types.Receipts) {
|
|
|
|
|
|
|
+func (bc *BlockChain) cacheReceipts(hash common.Hash, receipts types.Receipts) {
|
|
|
bc.receiptsCache.Add(hash, receipts)
|
|
bc.receiptsCache.Add(hash, receipts)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-func (bc *BlockChain) CacheBlock(hash common.Hash, block *types.Block) {
|
|
|
|
|
|
|
+func (bc *BlockChain) cacheDiffLayer(diffLayer *types.DiffLayer) {
|
|
|
|
|
+ bc.diffLayerCache.Add(diffLayer.BlockHash, diffLayer)
|
|
|
|
|
+ if bc.db.DiffStore() != nil {
|
|
|
|
|
+ // push to priority queue before persisting
|
|
|
|
|
+ bc.diffQueueBuffer <- diffLayer
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (bc *BlockChain) cacheBlock(hash common.Hash, block *types.Block) {
|
|
|
bc.blockCache.Add(hash, block)
|
|
bc.blockCache.Add(hash, block)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -873,6 +935,45 @@ func (bc *BlockChain) GetBodyRLP(hash common.Hash) rlp.RawValue {
|
|
|
return body
|
|
return body
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// GetDiffLayerRLP retrieves a diff layer in RLP encoding from the cache or database by blockHash
|
|
|
|
|
+func (bc *BlockChain) GetDiffLayerRLP(blockHash common.Hash) rlp.RawValue {
|
|
|
|
|
+ // Short circuit if the diffLayer's already in the cache, retrieve otherwise
|
|
|
|
|
+ if cached, ok := bc.diffLayerRLPCache.Get(blockHash); ok {
|
|
|
|
|
+ return cached.(rlp.RawValue)
|
|
|
|
|
+ }
|
|
|
|
|
+ if cached, ok := bc.diffLayerCache.Get(blockHash); ok {
|
|
|
|
|
+ diff := cached.(*types.DiffLayer)
|
|
|
|
|
+ bz, err := rlp.EncodeToBytes(diff)
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ bc.diffLayerRLPCache.Add(blockHash, rlp.RawValue(bz))
|
|
|
|
|
+ return bz
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // fallback to untrusted sources.
|
|
|
|
|
+ diff := bc.GetUnTrustedDiffLayer(blockHash, "")
|
|
|
|
|
+ if diff != nil {
|
|
|
|
|
+ bz, err := rlp.EncodeToBytes(diff)
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ // No need to cache untrusted data
|
|
|
|
|
+ return bz
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // fallback to disk
|
|
|
|
|
+ diffStore := bc.db.DiffStore()
|
|
|
|
|
+ if diffStore == nil {
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ rawData := rawdb.ReadDiffLayerRLP(diffStore, blockHash)
|
|
|
|
|
+ if len(rawData) != 0 {
|
|
|
|
|
+ bc.diffLayerRLPCache.Add(blockHash, rawData)
|
|
|
|
|
+ }
|
|
|
|
|
+ return rawData
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// HasBlock checks if a block is fully present in the database or not.
|
|
// HasBlock checks if a block is fully present in the database or not.
|
|
|
func (bc *BlockChain) HasBlock(hash common.Hash, number uint64) bool {
|
|
func (bc *BlockChain) HasBlock(hash common.Hash, number uint64) bool {
|
|
|
if bc.blockCache.Contains(hash) {
|
|
if bc.blockCache.Contains(hash) {
|
|
@@ -1506,10 +1607,19 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
|
|
|
wg.Done()
|
|
wg.Done()
|
|
|
}()
|
|
}()
|
|
|
// Commit all cached state changes into underlying memory database.
|
|
// Commit all cached state changes into underlying memory database.
|
|
|
- root, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
|
|
|
|
|
|
|
+ root, diffLayer, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
return NonStatTy, err
|
|
return NonStatTy, err
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ // Ensure no empty block body
|
|
|
|
|
+ if diffLayer != nil && block.Header().TxHash != types.EmptyRootHash {
|
|
|
|
|
+ // Filling necessary field
|
|
|
|
|
+ diffLayer.Receipts = receipts
|
|
|
|
|
+ diffLayer.BlockHash = block.Hash()
|
|
|
|
|
+ diffLayer.Number = block.NumberU64()
|
|
|
|
|
+ bc.cacheDiffLayer(diffLayer)
|
|
|
|
|
+ }
|
|
|
triedb := bc.stateCache.TrieDB()
|
|
triedb := bc.stateCache.TrieDB()
|
|
|
|
|
|
|
|
// If we're running an archive node, always flush
|
|
// If we're running an archive node, always flush
|
|
@@ -1885,18 +1995,15 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
|
|
|
}
|
|
}
|
|
|
// Enable prefetching to pull in trie node paths while processing transactions
|
|
// Enable prefetching to pull in trie node paths while processing transactions
|
|
|
statedb.StartPrefetcher("chain")
|
|
statedb.StartPrefetcher("chain")
|
|
|
- activeState = statedb
|
|
|
|
|
- statedb.TryPreload(block, signer)
|
|
|
|
|
|
|
|
|
|
//Process block using the parent state as reference point
|
|
//Process block using the parent state as reference point
|
|
|
substart := time.Now()
|
|
substart := time.Now()
|
|
|
- receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig)
|
|
|
|
|
|
|
+ statedb, receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig)
|
|
|
|
|
+ activeState = statedb
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
bc.reportBlock(block, receipts, err)
|
|
bc.reportBlock(block, receipts, err)
|
|
|
return it.index, err
|
|
return it.index, err
|
|
|
}
|
|
}
|
|
|
- bc.CacheReceipts(block.Hash(), receipts)
|
|
|
|
|
- bc.CacheBlock(block.Hash(), block)
|
|
|
|
|
// Update the metrics touched during block processing
|
|
// Update the metrics touched during block processing
|
|
|
accountReadTimer.Update(statedb.AccountReads) // Account reads are complete, we can mark them
|
|
accountReadTimer.Update(statedb.AccountReads) // Account reads are complete, we can mark them
|
|
|
storageReadTimer.Update(statedb.StorageReads) // Storage reads are complete, we can mark them
|
|
storageReadTimer.Update(statedb.StorageReads) // Storage reads are complete, we can mark them
|
|
@@ -1904,18 +2011,20 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
|
|
|
storageUpdateTimer.Update(statedb.StorageUpdates) // Storage updates are complete, we can mark them
|
|
storageUpdateTimer.Update(statedb.StorageUpdates) // Storage updates are complete, we can mark them
|
|
|
snapshotAccountReadTimer.Update(statedb.SnapshotAccountReads) // Account reads are complete, we can mark them
|
|
snapshotAccountReadTimer.Update(statedb.SnapshotAccountReads) // Account reads are complete, we can mark them
|
|
|
snapshotStorageReadTimer.Update(statedb.SnapshotStorageReads) // Storage reads are complete, we can mark them
|
|
snapshotStorageReadTimer.Update(statedb.SnapshotStorageReads) // Storage reads are complete, we can mark them
|
|
|
- trieproc := statedb.SnapshotAccountReads + statedb.AccountReads + statedb.AccountUpdates
|
|
|
|
|
- trieproc += statedb.SnapshotStorageReads + statedb.StorageReads + statedb.StorageUpdates
|
|
|
|
|
|
|
|
|
|
blockExecutionTimer.Update(time.Since(substart))
|
|
blockExecutionTimer.Update(time.Since(substart))
|
|
|
|
|
|
|
|
// Validate the state using the default validator
|
|
// Validate the state using the default validator
|
|
|
substart = time.Now()
|
|
substart = time.Now()
|
|
|
- if err := bc.validator.ValidateState(block, statedb, receipts, usedGas); err != nil {
|
|
|
|
|
- bc.reportBlock(block, receipts, err)
|
|
|
|
|
- log.Error("validate state failed", "error", err)
|
|
|
|
|
- return it.index, err
|
|
|
|
|
|
|
+ if !statedb.IsLightProcessed() {
|
|
|
|
|
+ if err := bc.validator.ValidateState(block, statedb, receipts, usedGas); err != nil {
|
|
|
|
|
+ log.Error("validate state failed", "error", err)
|
|
|
|
|
+ bc.reportBlock(block, receipts, err)
|
|
|
|
|
+ return it.index, err
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
+ bc.cacheReceipts(block.Hash(), receipts)
|
|
|
|
|
+ bc.cacheBlock(block.Hash(), block)
|
|
|
proctime := time.Since(start)
|
|
proctime := time.Since(start)
|
|
|
|
|
|
|
|
// Update the metrics touched during block validation
|
|
// Update the metrics touched during block validation
|
|
@@ -2292,6 +2401,279 @@ func (bc *BlockChain) update() {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+func (bc *BlockChain) trustedDiffLayerLoop() {
|
|
|
|
|
+ recheck := time.Tick(diffLayerFreezerRecheckInterval)
|
|
|
|
|
+ bc.wg.Add(1)
|
|
|
|
|
+ defer bc.wg.Done()
|
|
|
|
|
+ for {
|
|
|
|
|
+ select {
|
|
|
|
|
+ case diff := <-bc.diffQueueBuffer:
|
|
|
|
|
+ bc.diffQueue.Push(diff, -(int64(diff.Number)))
|
|
|
|
|
+ case <-bc.quit:
|
|
|
|
|
+ // Persist all diffLayers when shutdown, it will introduce redundant storage, but it is acceptable.
|
|
|
|
|
+ // If the client been ungracefully shutdown, it will missing all cached diff layers, it is acceptable as well.
|
|
|
|
|
+ var batch ethdb.Batch
|
|
|
|
|
+ for !bc.diffQueue.Empty() {
|
|
|
|
|
+ diff, _ := bc.diffQueue.Pop()
|
|
|
|
|
+ diffLayer := diff.(*types.DiffLayer)
|
|
|
|
|
+ if batch == nil {
|
|
|
|
|
+ batch = bc.db.DiffStore().NewBatch()
|
|
|
|
|
+ }
|
|
|
|
|
+ rawdb.WriteDiffLayer(batch, diffLayer.BlockHash, diffLayer)
|
|
|
|
|
+ if batch.ValueSize() > ethdb.IdealBatchSize {
|
|
|
|
|
+ if err := batch.Write(); err != nil {
|
|
|
|
|
+ log.Error("Failed to write diff layer", "err", err)
|
|
|
|
|
+ return
|
|
|
|
|
+ }
|
|
|
|
|
+ batch.Reset()
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if batch != nil {
|
|
|
|
|
+ // flush data
|
|
|
|
|
+ if err := batch.Write(); err != nil {
|
|
|
|
|
+ log.Error("Failed to write diff layer", "err", err)
|
|
|
|
|
+ return
|
|
|
|
|
+ }
|
|
|
|
|
+ batch.Reset()
|
|
|
|
|
+ }
|
|
|
|
|
+ return
|
|
|
|
|
+ case <-recheck:
|
|
|
|
|
+ currentHeight := bc.CurrentBlock().NumberU64()
|
|
|
|
|
+ var batch ethdb.Batch
|
|
|
|
|
+ for !bc.diffQueue.Empty() {
|
|
|
|
|
+ diff, prio := bc.diffQueue.Pop()
|
|
|
|
|
+ diffLayer := diff.(*types.DiffLayer)
|
|
|
|
|
+
|
|
|
|
|
+ // if the block old enough
|
|
|
|
|
+ if int64(currentHeight)+prio >= int64(bc.triesInMemory) {
|
|
|
|
|
+ canonicalHash := bc.GetCanonicalHash(uint64(-prio))
|
|
|
|
|
+ // on the canonical chain
|
|
|
|
|
+ if canonicalHash == diffLayer.BlockHash {
|
|
|
|
|
+ if batch == nil {
|
|
|
|
|
+ batch = bc.db.DiffStore().NewBatch()
|
|
|
|
|
+ }
|
|
|
|
|
+ rawdb.WriteDiffLayer(batch, diffLayer.BlockHash, diffLayer)
|
|
|
|
|
+ staleHash := bc.GetCanonicalHash(uint64(-prio) - bc.diffLayerFreezerBlockLimit)
|
|
|
|
|
+ rawdb.DeleteDiffLayer(batch, staleHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ } else {
|
|
|
|
|
+ bc.diffQueue.Push(diffLayer, prio)
|
|
|
|
|
+ break
|
|
|
|
|
+ }
|
|
|
|
|
+ if batch != nil && batch.ValueSize() > ethdb.IdealBatchSize {
|
|
|
|
|
+ if err := batch.Write(); err != nil {
|
|
|
|
|
+ panic(fmt.Sprintf("Failed to write diff layer, error %v", err))
|
|
|
|
|
+ }
|
|
|
|
|
+ batch.Reset()
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if batch != nil {
|
|
|
|
|
+ if err := batch.Write(); err != nil {
|
|
|
|
|
+ panic(fmt.Sprintf("Failed to write diff layer, error %v", err))
|
|
|
|
|
+ }
|
|
|
|
|
+ batch.Reset()
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (bc *BlockChain) GetUnTrustedDiffLayer(blockHash common.Hash, pid string) *types.DiffLayer {
|
|
|
|
|
+ bc.diffMux.RLock()
|
|
|
|
|
+ defer bc.diffMux.RUnlock()
|
|
|
|
|
+ if diffs, exist := bc.blockHashToDiffLayers[blockHash]; exist && len(diffs) != 0 {
|
|
|
|
|
+ if len(diffs) == 1 {
|
|
|
|
|
+ // return the only one diff layer
|
|
|
|
|
+ for _, diff := range diffs {
|
|
|
|
|
+ return diff
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ // pick the one from exact same peer if we know where the block comes from
|
|
|
|
|
+ if pid != "" {
|
|
|
|
|
+ if diffHashes, exist := bc.diffPeersToDiffHashes[pid]; exist {
|
|
|
|
|
+ for diff := range diffs {
|
|
|
|
|
+ if _, overlap := diffHashes[diff]; overlap {
|
|
|
|
|
+ return bc.blockHashToDiffLayers[blockHash][diff]
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ // Do not find overlap, do random pick
|
|
|
|
|
+ for _, diff := range diffs {
|
|
|
|
|
+ return diff
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (bc *BlockChain) removeDiffLayers(diffHash common.Hash) {
|
|
|
|
|
+ bc.diffMux.Lock()
|
|
|
|
|
+ defer bc.diffMux.Unlock()
|
|
|
|
|
+
|
|
|
|
|
+ // Untrusted peers
|
|
|
|
|
+ pids := bc.diffHashToPeers[diffHash]
|
|
|
|
|
+ invalidDiffHashes := make(map[common.Hash]struct{})
|
|
|
|
|
+ for pid := range pids {
|
|
|
|
|
+ invaliDiffHashesPeer := bc.diffPeersToDiffHashes[pid]
|
|
|
|
|
+ for invaliDiffHash := range invaliDiffHashesPeer {
|
|
|
|
|
+ invalidDiffHashes[invaliDiffHash] = struct{}{}
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(bc.diffPeersToDiffHashes, pid)
|
|
|
|
|
+ }
|
|
|
|
|
+ for invalidDiffHash := range invalidDiffHashes {
|
|
|
|
|
+ delete(bc.diffHashToPeers, invalidDiffHash)
|
|
|
|
|
+ affectedBlockHash := bc.diffHashToBlockHash[invalidDiffHash]
|
|
|
|
|
+ if diffs, exist := bc.blockHashToDiffLayers[affectedBlockHash]; exist {
|
|
|
|
|
+ delete(diffs, invalidDiffHash)
|
|
|
|
|
+ if len(diffs) == 0 {
|
|
|
|
|
+ delete(bc.blockHashToDiffLayers, affectedBlockHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(bc.diffHashToBlockHash, invalidDiffHash)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (bc *BlockChain) RemoveDiffPeer(pid string) {
|
|
|
|
|
+ bc.diffMux.Lock()
|
|
|
|
|
+ defer bc.diffMux.Unlock()
|
|
|
|
|
+ if invaliDiffHashes := bc.diffPeersToDiffHashes[pid]; invaliDiffHashes != nil {
|
|
|
|
|
+ for invalidDiffHash := range invaliDiffHashes {
|
|
|
|
|
+ lastDiffHash := false
|
|
|
|
|
+ if peers, ok := bc.diffHashToPeers[invalidDiffHash]; ok {
|
|
|
|
|
+ delete(peers, pid)
|
|
|
|
|
+ if len(peers) == 0 {
|
|
|
|
|
+ lastDiffHash = true
|
|
|
|
|
+ delete(bc.diffHashToPeers, invalidDiffHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if lastDiffHash {
|
|
|
|
|
+ affectedBlockHash := bc.diffHashToBlockHash[invalidDiffHash]
|
|
|
|
|
+ if diffs, exist := bc.blockHashToDiffLayers[affectedBlockHash]; exist {
|
|
|
|
|
+ delete(diffs, invalidDiffHash)
|
|
|
|
|
+ if len(diffs) == 0 {
|
|
|
|
|
+ delete(bc.blockHashToDiffLayers, affectedBlockHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(bc.diffHashToBlockHash, invalidDiffHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(bc.diffPeersToDiffHashes, pid)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (bc *BlockChain) untrustedDiffLayerPruneLoop() {
|
|
|
|
|
+ recheck := time.NewTicker(diffLayerPruneRecheckInterval)
|
|
|
|
|
+ bc.wg.Add(1)
|
|
|
|
|
+ defer func() {
|
|
|
|
|
+ bc.wg.Done()
|
|
|
|
|
+ recheck.Stop()
|
|
|
|
|
+ }()
|
|
|
|
|
+ for {
|
|
|
|
|
+ select {
|
|
|
|
|
+ case <-bc.quit:
|
|
|
|
|
+ return
|
|
|
|
|
+ case <-recheck.C:
|
|
|
|
|
+ bc.pruneDiffLayer()
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (bc *BlockChain) pruneDiffLayer() {
|
|
|
|
|
+ currentHeight := bc.CurrentBlock().NumberU64()
|
|
|
|
|
+ bc.diffMux.Lock()
|
|
|
|
|
+ defer bc.diffMux.Unlock()
|
|
|
|
|
+ sortNumbers := make([]uint64, 0, len(bc.diffNumToBlockHashes))
|
|
|
|
|
+ for number := range bc.diffNumToBlockHashes {
|
|
|
|
|
+ sortNumbers = append(sortNumbers, number)
|
|
|
|
|
+ }
|
|
|
|
|
+ sort.Slice(sortNumbers, func(i, j int) bool {
|
|
|
|
|
+ return sortNumbers[i] <= sortNumbers[j]
|
|
|
|
|
+ })
|
|
|
|
|
+ staleBlockHashes := make(map[common.Hash]struct{})
|
|
|
|
|
+ for _, number := range sortNumbers {
|
|
|
|
|
+ if number >= currentHeight-maxDiffForkDist {
|
|
|
|
|
+ break
|
|
|
|
|
+ }
|
|
|
|
|
+ affectedHashes := bc.diffNumToBlockHashes[number]
|
|
|
|
|
+ if affectedHashes != nil {
|
|
|
|
|
+ for affectedHash := range affectedHashes {
|
|
|
|
|
+ staleBlockHashes[affectedHash] = struct{}{}
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(bc.diffNumToBlockHashes, number)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ staleDiffHashes := make(map[common.Hash]struct{})
|
|
|
|
|
+ for blockHash := range staleBlockHashes {
|
|
|
|
|
+ if diffHashes, exist := bc.blockHashToDiffLayers[blockHash]; exist {
|
|
|
|
|
+ for diffHash := range diffHashes {
|
|
|
|
|
+ staleDiffHashes[diffHash] = struct{}{}
|
|
|
|
|
+ delete(bc.diffHashToBlockHash, diffHash)
|
|
|
|
|
+ delete(bc.diffHashToPeers, diffHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(bc.blockHashToDiffLayers, blockHash)
|
|
|
|
|
+ }
|
|
|
|
|
+ for diffHash := range staleDiffHashes {
|
|
|
|
|
+ for p, diffHashes := range bc.diffPeersToDiffHashes {
|
|
|
|
|
+ delete(diffHashes, diffHash)
|
|
|
|
|
+ if len(diffHashes) == 0 {
|
|
|
|
|
+ delete(bc.diffPeersToDiffHashes, p)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// Process received diff layers
|
|
|
|
|
+func (bc *BlockChain) HandleDiffLayer(diffLayer *types.DiffLayer, pid string, fulfilled bool) error {
|
|
|
|
|
+ // Basic check
|
|
|
|
|
+ currentHeight := bc.CurrentBlock().NumberU64()
|
|
|
|
|
+ if diffLayer.Number > currentHeight && diffLayer.Number-currentHeight > maxDiffQueueDist {
|
|
|
|
|
+ log.Error("diff layers too new from current", "pid", pid)
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ if diffLayer.Number < currentHeight && currentHeight-diffLayer.Number > maxDiffForkDist {
|
|
|
|
|
+ log.Error("diff layers too old from current", "pid", pid)
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ bc.diffMux.Lock()
|
|
|
|
|
+ defer bc.diffMux.Unlock()
|
|
|
|
|
+
|
|
|
|
|
+ if !fulfilled && len(bc.diffPeersToDiffHashes[pid]) > maxDiffLimitForBroadcast {
|
|
|
|
|
+ log.Error("too many accumulated diffLayers", "pid", pid)
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if len(bc.diffPeersToDiffHashes[pid]) > maxDiffLimit {
|
|
|
|
|
+ log.Error("too many accumulated diffLayers", "pid", pid)
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ if _, exist := bc.diffPeersToDiffHashes[pid]; exist {
|
|
|
|
|
+ if _, alreadyHas := bc.diffPeersToDiffHashes[pid][diffLayer.DiffHash]; alreadyHas {
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ } else {
|
|
|
|
|
+ bc.diffPeersToDiffHashes[pid] = make(map[common.Hash]struct{})
|
|
|
|
|
+ }
|
|
|
|
|
+ bc.diffPeersToDiffHashes[pid][diffLayer.DiffHash] = struct{}{}
|
|
|
|
|
+ if _, exist := bc.diffNumToBlockHashes[diffLayer.Number]; !exist {
|
|
|
|
|
+ bc.diffNumToBlockHashes[diffLayer.Number] = make(map[common.Hash]struct{})
|
|
|
|
|
+ }
|
|
|
|
|
+ bc.diffNumToBlockHashes[diffLayer.Number][diffLayer.BlockHash] = struct{}{}
|
|
|
|
|
+
|
|
|
|
|
+ if _, exist := bc.diffHashToPeers[diffLayer.DiffHash]; !exist {
|
|
|
|
|
+ bc.diffHashToPeers[diffLayer.DiffHash] = make(map[string]struct{})
|
|
|
|
|
+ }
|
|
|
|
|
+ bc.diffHashToPeers[diffLayer.DiffHash][pid] = struct{}{}
|
|
|
|
|
+
|
|
|
|
|
+ if _, exist := bc.blockHashToDiffLayers[diffLayer.BlockHash]; !exist {
|
|
|
|
|
+ bc.blockHashToDiffLayers[diffLayer.BlockHash] = make(map[common.Hash]*types.DiffLayer)
|
|
|
|
|
+ }
|
|
|
|
|
+ bc.blockHashToDiffLayers[diffLayer.BlockHash][diffLayer.DiffHash] = diffLayer
|
|
|
|
|
+ bc.diffHashToBlockHash[diffLayer.DiffHash] = diffLayer.BlockHash
|
|
|
|
|
+
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// maintainTxIndex is responsible for the construction and deletion of the
|
|
// maintainTxIndex is responsible for the construction and deletion of the
|
|
|
// transaction index.
|
|
// transaction index.
|
|
|
//
|
|
//
|
|
@@ -2541,3 +2923,16 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
|
|
|
func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
|
|
func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
|
|
|
return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
|
|
return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+// Options
|
|
|
|
|
+func EnableLightProcessor(bc *BlockChain) *BlockChain {
|
|
|
|
|
+ bc.processor = NewLightStateProcessor(bc.Config(), bc, bc.engine)
|
|
|
|
|
+ return bc
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func EnablePersistDiff(limit uint64) BlockChainOption {
|
|
|
|
|
+ return func(chain *BlockChain) *BlockChain {
|
|
|
|
|
+ chain.diffLayerFreezerBlockLimit = limit
|
|
|
|
|
+ return chain
|
|
|
|
|
+ }
|
|
|
|
|
+}
|