diff --git a/eth/api_backend.go b/eth/api_backend.go
index 927dcdef10..c9ad4311e7 100644
--- a/eth/api_backend.go
+++ b/eth/api_backend.go
@@ -68,9 +68,6 @@ func (b *EthAPIBackend) HeaderByNumber(ctx context.Context, number rpc.BlockNumb
// Pending block is only known by the miner
if number == rpc.PendingBlockNumber {
block := b.eth.miner.PendingBlock()
- if block == nil {
- return nil, errors.New("pending block is not available")
- }
return block.Header(), nil
}
// Otherwise resolve and return the block
@@ -125,9 +122,6 @@ func (b *EthAPIBackend) BlockByNumber(ctx context.Context, number rpc.BlockNumbe
// Pending block is only known by the miner
if number == rpc.PendingBlockNumber {
block := b.eth.miner.PendingBlock()
- if block == nil {
- return nil, errors.New("pending block is not available")
- }
return block, nil
}
// Otherwise resolve and return the block
@@ -202,9 +196,6 @@ func (b *EthAPIBackend) StateAndHeaderByNumber(ctx context.Context, number rpc.B
// Pending state is only known by the miner
if number == rpc.PendingBlockNumber {
block, state := b.eth.miner.Pending()
- if block == nil || state == nil {
- return nil, nil, errors.New("pending state is not available")
- }
return state, block.Header(), nil
}
// Otherwise resolve the block number and return its state
diff --git a/eth/api_debug.go b/eth/api_debug.go
index 7cf0239174..929e0460f7 100644
--- a/eth/api_debug.go
+++ b/eth/api_debug.go
@@ -56,9 +56,6 @@ func (api *DebugAPI) DumpBlock(blockNr rpc.BlockNumber) (state.Dump, error) {
// both the pending block as well as the pending state from
// the miner and operate on those
_, stateDb := api.eth.miner.Pending()
- if stateDb == nil {
- return state.Dump{}, errors.New("pending state is not available")
- }
return stateDb.RawDump(opts), nil
}
var header *types.Header
@@ -144,9 +141,6 @@ func (api *DebugAPI) AccountRange(blockNrOrHash rpc.BlockNumberOrHash, start hex
// both the pending block as well as the pending state from
// the miner and operate on those
_, stateDb = api.eth.miner.Pending()
- if stateDb == nil {
- return state.IteratorDump{}, errors.New("pending state is not available")
- }
} else {
var header *types.Header
if number == rpc.LatestBlockNumber {
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index d65bbfacfc..7543c06033 100644
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@@ -334,7 +334,7 @@ func (f *Filter) checkMatches(ctx context.Context, header *types.Header) ([]*typ
// pendingLogs returns the logs matching the filter criteria within the pending block.
func (f *Filter) pendingLogs() []*types.Log {
block, receipts := f.sys.backend.PendingBlockAndReceipts()
- if block == nil || receipts == nil {
+ if block == nil {
return nil
}
if bloomFilter(block.Bloom(), f.addresses, f.topics) {
diff --git a/miner/miner.go b/miner/miner.go
index b7273948f5..b1d1f7c4cb 100644
--- a/miner/miner.go
+++ b/miner/miner.go
@@ -131,22 +131,16 @@ func (miner *Miner) update() {
shouldStart = true
log.Info("Mining aborted due to sync")
}
- miner.worker.syncing.Store(true)
-
case downloader.FailedEvent:
canStart = true
if shouldStart {
miner.worker.start()
}
- miner.worker.syncing.Store(false)
-
case downloader.DoneEvent:
canStart = true
if shouldStart {
miner.worker.start()
}
- miner.worker.syncing.Store(false)
-
// Stop reacting to downloader events
events.Unsubscribe()
}
@@ -202,14 +196,12 @@ func (miner *Miner) SetRecommitInterval(interval time.Duration) {
miner.worker.setRecommitInterval(interval)
}
-// Pending returns the currently pending block and associated state. The returned
-// values can be nil in case the pending block is not initialized
+// Pending returns the currently pending block and associated state.
func (miner *Miner) Pending() (*types.Block, *state.StateDB) {
return miner.worker.pending()
}
-// PendingBlock returns the currently pending block. The returned block can be
-// nil in case the pending block is not initialized.
+// PendingBlock returns the currently pending block.
//
// Note, to access both the pending block and the pending state
// simultaneously, please use Pending(), as the pending state can
@@ -219,7 +211,6 @@ func (miner *Miner) PendingBlock() *types.Block {
}
// PendingBlockAndReceipts returns the currently pending block and corresponding receipts.
-// The returned values can be nil in case the pending block is not initialized.
func (miner *Miner) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
return miner.worker.pendingBlockAndReceipts()
}
@@ -234,6 +225,23 @@ func (miner *Miner) SetGasCeil(ceil uint64) {
miner.worker.setGasCeil(ceil)
}
+// EnablePreseal turns on the preseal mining feature. It's enabled by default.
+// Note this function shouldn't be exposed to API, it's unnecessary for users
+// (miners) to actually know the underlying detail. It's only for outside project
+// which uses this library.
+func (miner *Miner) EnablePreseal() {
+ miner.worker.enablePreseal()
+}
+
+// DisablePreseal turns off the preseal mining feature. It's necessary for some
+// fake consensus engine which can seal blocks instantaneously.
+// Note this function shouldn't be exposed to API, it's unnecessary for users
+// (miners) to actually know the underlying detail. It's only for outside project
+// which uses this library.
+func (miner *Miner) DisablePreseal() {
+ miner.worker.disablePreseal()
+}
+
// SubscribePendingLogs starts delivering logs from pending transactions
// to the given channel.
func (miner *Miner) SubscribePendingLogs(ch chan<- []*types.Log) event.Subscription {
diff --git a/miner/unconfirmed.go b/miner/unconfirmed.go
new file mode 100644
index 0000000000..0489f1ea4a
--- /dev/null
+++ b/miner/unconfirmed.go
@@ -0,0 +1,136 @@
+// Copyright 2016 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 .
+
+package miner
+
+import (
+ "container/ring"
+ "sync"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/log"
+)
+
+// chainRetriever is used by the unconfirmed block set to verify whether a previously
+// mined block is part of the canonical chain or not.
+type chainRetriever interface {
+ // GetHeaderByNumber retrieves the canonical header associated with a block number.
+ GetHeaderByNumber(number uint64) *types.Header
+
+ // GetBlockByNumber retrieves the canonical block associated with a block number.
+ GetBlockByNumber(number uint64) *types.Block
+}
+
+// unconfirmedBlock is a small collection of metadata about a locally mined block
+// that is placed into a unconfirmed set for canonical chain inclusion tracking.
+type unconfirmedBlock struct {
+ index uint64
+ hash common.Hash
+}
+
+// unconfirmedBlocks implements a data structure to maintain locally mined blocks
+// have not yet reached enough maturity to guarantee chain inclusion. It is
+// used by the miner to provide logs to the user when a previously mined block
+// has a high enough guarantee to not be reorged out of the canonical chain.
+type unconfirmedBlocks struct {
+ chain chainRetriever // Blockchain to verify canonical status through
+ depth uint // Depth after which to discard previous blocks
+ blocks *ring.Ring // Block infos to allow canonical chain cross checks
+ lock sync.Mutex // Protects the fields from concurrent access
+}
+
+// newUnconfirmedBlocks returns new data structure to track currently unconfirmed blocks.
+func newUnconfirmedBlocks(chain chainRetriever, depth uint) *unconfirmedBlocks {
+ return &unconfirmedBlocks{
+ chain: chain,
+ depth: depth,
+ }
+}
+
+// Insert adds a new block to the set of unconfirmed ones.
+func (set *unconfirmedBlocks) Insert(index uint64, hash common.Hash) {
+ // If a new block was mined locally, shift out any old enough blocks
+ set.Shift(index)
+
+ // Create the new item as its own ring
+ item := ring.New(1)
+ item.Value = &unconfirmedBlock{
+ index: index,
+ hash: hash,
+ }
+ // Set as the initial ring or append to the end
+ set.lock.Lock()
+ defer set.lock.Unlock()
+
+ if set.blocks == nil {
+ set.blocks = item
+ } else {
+ set.blocks.Move(-1).Link(item)
+ }
+ // Display a log for the user to notify of a new mined block unconfirmed
+ log.Info("🔨 mined potential block", "number", index, "hash", hash)
+}
+
+// Shift drops all unconfirmed blocks from the set which exceed the unconfirmed sets depth
+// allowance, checking them against the canonical chain for inclusion or staleness
+// report.
+func (set *unconfirmedBlocks) Shift(height uint64) {
+ set.lock.Lock()
+ defer set.lock.Unlock()
+
+ for set.blocks != nil {
+ // Retrieve the next unconfirmed block and abort if too fresh
+ next := set.blocks.Value.(*unconfirmedBlock)
+ if next.index+uint64(set.depth) > height {
+ break
+ }
+ // Block seems to exceed depth allowance, check for canonical status
+ header := set.chain.GetHeaderByNumber(next.index)
+ switch {
+ case header == nil:
+ log.Warn("Failed to retrieve header of mined block", "number", next.index, "hash", next.hash)
+ case header.Hash() == next.hash:
+ log.Info("🔗 block reached canonical chain", "number", next.index, "hash", next.hash)
+ default:
+ // Block is not canonical, check whether we have an uncle or a lost block
+ included := false
+ for number := next.index; !included && number < next.index+uint64(set.depth) && number <= height; number++ {
+ if block := set.chain.GetBlockByNumber(number); block != nil {
+ for _, uncle := range block.Uncles() {
+ if uncle.Hash() == next.hash {
+ included = true
+ break
+ }
+ }
+ }
+ }
+ if included {
+ log.Info("â‘‚ block became an uncle", "number", next.index, "hash", next.hash)
+ } else {
+ log.Info("😱 block lost", "number", next.index, "hash", next.hash)
+ }
+ }
+ // Drop the block out of the ring
+ if set.blocks.Value == set.blocks.Next().Value {
+ set.blocks = nil
+ } else {
+ set.blocks = set.blocks.Move(-1)
+ set.blocks.Unlink(1)
+ set.blocks = set.blocks.Move(1)
+ }
+ }
+}
diff --git a/miner/unconfirmed_test.go b/miner/unconfirmed_test.go
new file mode 100644
index 0000000000..60958f658a
--- /dev/null
+++ b/miner/unconfirmed_test.go
@@ -0,0 +1,87 @@
+// Copyright 2016 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 .
+
+package miner
+
+import (
+ "testing"
+
+ "github.com/ethereum/go-ethereum/core/types"
+)
+
+// noopChainRetriever is an implementation of headerRetriever that always
+// returns nil for any requested headers.
+type noopChainRetriever struct{}
+
+func (r *noopChainRetriever) GetHeaderByNumber(number uint64) *types.Header {
+ return nil
+}
+func (r *noopChainRetriever) GetBlockByNumber(number uint64) *types.Block {
+ return nil
+}
+
+// Tests that inserting blocks into the unconfirmed set accumulates them until
+// the desired depth is reached, after which they begin to be dropped.
+func TestUnconfirmedInsertBounds(t *testing.T) {
+ limit := uint(10)
+
+ pool := newUnconfirmedBlocks(new(noopChainRetriever), limit)
+ for depth := uint64(0); depth < 2*uint64(limit); depth++ {
+ // Insert multiple blocks for the same level just to stress it
+ for i := 0; i < int(depth); i++ {
+ pool.Insert(depth, [32]byte{byte(depth), byte(i)})
+ }
+ // Validate that no blocks below the depth allowance are left in
+ pool.blocks.Do(func(block interface{}) {
+ if block := block.(*unconfirmedBlock); block.index+uint64(limit) <= depth {
+ t.Errorf("depth %d: block %x not dropped", depth, block.hash)
+ }
+ })
+ }
+}
+
+// Tests that shifting blocks out of the unconfirmed set works both for normal
+// cases as well as for corner cases such as empty sets, empty shifts or full
+// shifts.
+func TestUnconfirmedShifts(t *testing.T) {
+ // Create a pool with a few blocks on various depths
+ limit, start := uint(10), uint64(25)
+
+ pool := newUnconfirmedBlocks(new(noopChainRetriever), limit)
+ for depth := start; depth < start+uint64(limit); depth++ {
+ pool.Insert(depth, [32]byte{byte(depth)})
+ }
+ // Try to shift below the limit and ensure no blocks are dropped
+ pool.Shift(start + uint64(limit) - 1)
+ if n := pool.blocks.Len(); n != int(limit) {
+ t.Errorf("unconfirmed count mismatch: have %d, want %d", n, limit)
+ }
+ // Try to shift half the blocks out and verify remainder
+ pool.Shift(start + uint64(limit) - 1 + uint64(limit/2))
+ if n := pool.blocks.Len(); n != int(limit)/2 {
+ t.Errorf("unconfirmed count mismatch: have %d, want %d", n, limit/2)
+ }
+ // Try to shift all the remaining blocks out and verify emptiness
+ pool.Shift(start + 2*uint64(limit))
+ if n := pool.blocks.Len(); n != 0 {
+ t.Errorf("unconfirmed count mismatch: have %d, want %d", n, 0)
+ }
+ // Try to shift out from the empty set and make sure it doesn't break
+ pool.Shift(start + 3*uint64(limit))
+ if n := pool.blocks.Len(); n != 0 {
+ t.Errorf("unconfirmed count mismatch: have %d, want %d", n, 0)
+ }
+}
diff --git a/miner/worker.go b/miner/worker.go
index d61f8dcada..936a9e74a5 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -24,6 +24,7 @@ import (
"sync/atomic"
"time"
+ mapset "github.com/deckarep/golang-set/v2"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/consensus/misc"
@@ -47,9 +48,15 @@ const (
// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
chainHeadChanSize = 10
+ // chainSideChanSize is the size of channel listening to ChainSideEvent.
+ chainSideChanSize = 10
+
// resubmitAdjustChanSize is the size of resubmitting interval adjustment channel.
resubmitAdjustChanSize = 10
+ // sealingLogAtDepth is the number of confirmations before logging successful sealing.
+ sealingLogAtDepth = 7
+
// minRecommitInterval is the minimal time interval to recreate the sealing block with
// any newly arrived transactions.
minRecommitInterval = 1 * time.Second
@@ -79,36 +86,57 @@ var (
// environment is the worker's current environment and holds all
// information of the sealing block generation.
type environment struct {
- signer types.Signer
- state *state.StateDB // apply state changes here
- tcount int // tx count in cycle
- gasPool *core.GasPool // available gas used to pack transactions
- coinbase common.Address
+ signer types.Signer
+
+ state *state.StateDB // apply state changes here
+ ancestors mapset.Set[common.Hash] // ancestor set (used for checking uncle parent validity)
+ family mapset.Set[common.Hash] // family set (used for checking uncle invalidity)
+ tcount int // tx count in cycle
+ gasPool *core.GasPool // available gas used to pack transactions
+ coinbase common.Address
header *types.Header
txs []*types.Transaction
receipts []*types.Receipt
+ uncles map[common.Hash]*types.Header
}
// copy creates a deep copy of environment.
func (env *environment) copy() *environment {
cpy := &environment{
- signer: env.signer,
- state: env.state.Copy(),
- tcount: env.tcount,
- coinbase: env.coinbase,
- header: types.CopyHeader(env.header),
- receipts: copyReceipts(env.receipts),
+ signer: env.signer,
+ state: env.state.Copy(),
+ ancestors: env.ancestors.Clone(),
+ family: env.family.Clone(),
+ tcount: env.tcount,
+ coinbase: env.coinbase,
+ header: types.CopyHeader(env.header),
+ receipts: copyReceipts(env.receipts),
}
if env.gasPool != nil {
gasPool := *env.gasPool
cpy.gasPool = &gasPool
}
+ // The content of txs and uncles are immutable, unnecessary
+ // to do the expensive deep copy for them.
cpy.txs = make([]*types.Transaction, len(env.txs))
copy(cpy.txs, env.txs)
+ cpy.uncles = make(map[common.Hash]*types.Header)
+ for hash, uncle := range env.uncles {
+ cpy.uncles[hash] = uncle
+ }
return cpy
}
+// unclelist returns the contained uncles as the list format.
+func (env *environment) unclelist() []*types.Header {
+ var uncles []*types.Header
+ for _, uncle := range env.uncles {
+ uncles = append(uncles, uncle)
+ }
+ return uncles
+}
+
// discard terminates the background prefetcher go-routine. It should
// always be called for all created environment instances otherwise
// the go-routine leak can happen.
@@ -137,6 +165,7 @@ const (
// newWorkReq represents a request for new sealing work submitting with relative interrupt notifier.
type newWorkReq struct {
interrupt *atomic.Int32
+ noempty bool
timestamp int64
}
@@ -177,6 +206,8 @@ type worker struct {
txsSub event.Subscription
chainHeadCh chan core.ChainHeadEvent
chainHeadSub event.Subscription
+ chainSideCh chan core.ChainSideEvent
+ chainSideSub event.Subscription
// Channels
newWorkCh chan *newWorkReq
@@ -190,7 +221,10 @@ type worker struct {
wg sync.WaitGroup
- current *environment // An environment for current running cycle.
+ current *environment // An environment for current running cycle.
+ localUncles map[common.Hash]*types.Block // A set of side blocks generated locally as the possible uncle blocks.
+ remoteUncles map[common.Hash]*types.Block // A set of side blocks as the possible uncle blocks.
+ unconfirmed *unconfirmedBlocks // A set of locally mined blocks pending canonicalness confirmations.
mu sync.RWMutex // The lock used to protect the coinbase and extra fields
coinbase common.Address
@@ -207,7 +241,13 @@ type worker struct {
// atomic status counters
running atomic.Bool // The indicator whether the consensus engine is running or not.
newTxs atomic.Int32 // New arrival transaction count since last sealing work submitting.
- syncing atomic.Bool // The indicator whether the node is still syncing.
+
+ // noempty is the flag used to control whether the feature of pre-seal empty
+ // block is enabled. The default value is false(pre-seal is enabled by default).
+ // But in some special scenario the consensus engine will seal blocks instantaneously,
+ // in this case this feature will add all empty blocks into canonical chain
+ // non-stop and no real transaction will be included.
+ noempty atomic.Bool
// newpayloadTimeout is the maximum timeout allowance for creating payload.
// The default value is 2 seconds but node operator can set it to arbitrary
@@ -239,11 +279,15 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
chain: eth.BlockChain(),
mux: mux,
isLocalBlock: isLocalBlock,
+ localUncles: make(map[common.Hash]*types.Block),
+ remoteUncles: make(map[common.Hash]*types.Block),
+ unconfirmed: newUnconfirmedBlocks(eth.BlockChain(), sealingLogAtDepth),
coinbase: config.Etherbase,
extra: config.ExtraData,
pendingTasks: make(map[common.Hash]*task),
txsCh: make(chan core.NewTxsEvent, txChanSize),
chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize),
+ chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize),
newWorkCh: make(chan *newWorkReq),
getWorkCh: make(chan *getWorkReq),
taskCh: make(chan *task),
@@ -257,6 +301,7 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
worker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh)
// Subscribe events for blockchain
worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
+ worker.chainSideSub = eth.BlockChain().SubscribeChainSideEvent(worker.chainSideCh)
// Sanitize recommit interval if the user-specified one is too short.
recommit := worker.config.Recommit
@@ -325,9 +370,19 @@ func (w *worker) setRecommitInterval(interval time.Duration) {
}
}
-// pending returns the pending state and corresponding block. The returned
-// values can be nil in case the pending block is not initialized.
+// disablePreseal disables pre-sealing feature
+func (w *worker) disablePreseal() {
+ w.noempty.Store(true)
+}
+
+// enablePreseal enables pre-sealing feature
+func (w *worker) enablePreseal() {
+ w.noempty.Store(false)
+}
+
+// pending returns the pending state and corresponding block.
func (w *worker) pending() (*types.Block, *state.StateDB) {
+ // return a snapshot to avoid contention on currentMu mutex
w.snapshotMu.RLock()
defer w.snapshotMu.RUnlock()
if w.snapshotState == nil {
@@ -336,17 +391,17 @@ func (w *worker) pending() (*types.Block, *state.StateDB) {
return w.snapshotBlock, w.snapshotState.Copy()
}
-// pendingBlock returns pending block. The returned block can be nil in case the
-// pending block is not initialized.
+// pendingBlock returns pending block.
func (w *worker) pendingBlock() *types.Block {
+ // return a snapshot to avoid contention on currentMu mutex
w.snapshotMu.RLock()
defer w.snapshotMu.RUnlock()
return w.snapshotBlock
}
// pendingBlockAndReceipts returns pending block and corresponding receipts.
-// The returned values can be nil in case the pending block is not initialized.
func (w *worker) pendingBlockAndReceipts() (*types.Block, types.Receipts) {
+ // return a snapshot to avoid contention on currentMu mutex
w.snapshotMu.RLock()
defer w.snapshotMu.RUnlock()
return w.snapshotBlock, w.snapshotReceipts
@@ -412,13 +467,13 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
<-timer.C // discard the initial tick
// commit aborts in-flight transaction execution with given signal and resubmits a new one.
- commit := func(s int32) {
+ commit := func(noempty bool, s int32) {
if interrupt != nil {
interrupt.Store(s)
}
interrupt = new(atomic.Int32)
select {
- case w.newWorkCh <- &newWorkReq{interrupt: interrupt, timestamp: timestamp}:
+ case w.newWorkCh <- &newWorkReq{interrupt: interrupt, noempty: noempty, timestamp: timestamp}:
case <-w.exitCh:
return
}
@@ -441,12 +496,12 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
case <-w.startCh:
clearPending(w.chain.CurrentBlock().Number.Uint64())
timestamp = time.Now().Unix()
- commit(commitInterruptNewHead)
+ commit(false, commitInterruptNewHead)
case head := <-w.chainHeadCh:
clearPending(head.Block.NumberU64())
timestamp = time.Now().Unix()
- commit(commitInterruptNewHead)
+ commit(false, commitInterruptNewHead)
case <-timer.C:
// If sealing is running resubmit a new work cycle periodically to pull in
@@ -457,7 +512,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
timer.Reset(recommit)
continue
}
- commit(commitInterruptResubmit)
+ commit(true, commitInterruptResubmit)
}
case interval := <-w.resubmitIntervalCh:
@@ -503,16 +558,20 @@ func (w *worker) mainLoop() {
defer w.wg.Done()
defer w.txsSub.Unsubscribe()
defer w.chainHeadSub.Unsubscribe()
+ defer w.chainSideSub.Unsubscribe()
defer func() {
if w.current != nil {
w.current.discard()
}
}()
+ cleanTicker := time.NewTicker(time.Second * 10)
+ defer cleanTicker.Stop()
+
for {
select {
case req := <-w.newWorkCh:
- w.commitWork(req.interrupt, req.timestamp)
+ w.commitWork(req.interrupt, req.noempty, req.timestamp)
case req := <-w.getWorkCh:
block, fees, err := w.generateWork(req.params)
@@ -521,6 +580,42 @@ func (w *worker) mainLoop() {
block: block,
fees: fees,
}
+ case ev := <-w.chainSideCh:
+ // Short circuit for duplicate side blocks
+ if _, exist := w.localUncles[ev.Block.Hash()]; exist {
+ continue
+ }
+ if _, exist := w.remoteUncles[ev.Block.Hash()]; exist {
+ continue
+ }
+ // Add side block to possible uncle block set depending on the author.
+ if w.isLocalBlock != nil && w.isLocalBlock(ev.Block.Header()) {
+ w.localUncles[ev.Block.Hash()] = ev.Block
+ } else {
+ w.remoteUncles[ev.Block.Hash()] = ev.Block
+ }
+ // If our sealing block contains less than 2 uncle blocks,
+ // add the new uncle block if valid and regenerate a new
+ // sealing block for higher profit.
+ if w.isRunning() && w.current != nil && len(w.current.uncles) < 2 {
+ start := time.Now()
+ if err := w.commitUncle(w.current, ev.Block.Header()); err == nil {
+ w.commit(w.current.copy(), nil, true, start)
+ }
+ }
+
+ case <-cleanTicker.C:
+ chainHead := w.chain.CurrentBlock()
+ for hash, uncle := range w.localUncles {
+ if uncle.NumberU64()+staleThreshold <= chainHead.Number.Uint64() {
+ delete(w.localUncles, hash)
+ }
+ }
+ for hash, uncle := range w.remoteUncles {
+ if uncle.NumberU64()+staleThreshold <= chainHead.Number.Uint64() {
+ delete(w.remoteUncles, hash)
+ }
+ }
case ev := <-w.txsCh:
// Apply transactions to the pending state if we're not sealing
@@ -552,7 +647,7 @@ func (w *worker) mainLoop() {
// submit sealing work here since all empty submission will be rejected
// by clique. Of course the advance sealing(empty submission) is disabled.
if w.chainConfig.Clique != nil && w.chainConfig.Clique.Period == 0 {
- w.commitWork(nil, time.Now().Unix())
+ w.commitWork(nil, true, time.Now().Unix())
}
}
w.newTxs.Add(int32(len(ev.Txs)))
@@ -564,6 +659,8 @@ func (w *worker) mainLoop() {
return
case <-w.chainHeadSub.Err():
return
+ case <-w.chainSideSub.Err():
+ return
}
}
}
@@ -683,6 +780,9 @@ func (w *worker) resultLoop() {
// Broadcast the block and announce chain insertion event
w.mux.Post(core.NewMinedBlockEvent{Block: block})
+ // Insert the block into the set of pending ones to resultLoop for confirmations
+ w.unconfirmed.Insert(block.NumberU64(), block.Hash())
+
case <-w.exitCh:
return
}
@@ -701,16 +801,49 @@ func (w *worker) makeEnv(parent *types.Header, header *types.Header, coinbase co
// Note the passed coinbase may be different with header.Coinbase.
env := &environment{
- signer: types.MakeSigner(w.chainConfig, header.Number, header.Time),
- state: state,
- coinbase: coinbase,
- header: header,
+ signer: types.MakeSigner(w.chainConfig, header.Number, header.Time),
+ state: state,
+ coinbase: coinbase,
+ ancestors: mapset.NewSet[common.Hash](),
+ family: mapset.NewSet[common.Hash](),
+ header: header,
+ uncles: make(map[common.Hash]*types.Header),
+ }
+ // when 08 is processed ancestors contain 07 (quick block)
+ for _, ancestor := range w.chain.GetBlocksFromHash(parent.Hash(), 7) {
+ for _, uncle := range ancestor.Uncles() {
+ env.family.Add(uncle.Hash())
+ }
+ env.family.Add(ancestor.Hash())
+ env.ancestors.Add(ancestor.Hash())
}
// Keep track of transactions which return errors so they can be removed
env.tcount = 0
return env, nil
}
+// commitUncle adds the given block to uncle block set, returns error if failed to add.
+func (w *worker) commitUncle(env *environment, uncle *types.Header) error {
+ if w.isTTDReached(env.header) {
+ return errors.New("ignore uncle for beacon block")
+ }
+ hash := uncle.Hash()
+ if _, exist := env.uncles[hash]; exist {
+ return errors.New("uncle not unique")
+ }
+ if env.header.ParentHash == uncle.ParentHash {
+ return errors.New("uncle is sibling")
+ }
+ if !env.ancestors.Contains(uncle.ParentHash) {
+ return errors.New("uncle's parent unknown")
+ }
+ if env.family.Contains(hash) {
+ return errors.New("uncle already included")
+ }
+ env.uncles[hash] = uncle
+ return nil
+}
+
// updateSnapshot updates pending snapshot block, receipts and state.
func (w *worker) updateSnapshot(env *environment) {
w.snapshotMu.Lock()
@@ -719,7 +852,7 @@ func (w *worker) updateSnapshot(env *environment) {
w.snapshotBlock = types.NewBlock(
env.header,
env.txs,
- nil,
+ env.unclelist(),
env.receipts,
trie.NewStackTrie(nil),
)
@@ -829,6 +962,7 @@ type generateParams struct {
coinbase common.Address // The fee recipient address for including transaction
random common.Hash // The randomness generated by beacon chain, empty before the merge
withdrawals types.Withdrawals // List of withdrawals to include in block.
+ noUncle bool // Flag whether the uncle block inclusion is allowed
noTxs bool // Flag whether an empty block without any transaction is expected
}
@@ -894,6 +1028,24 @@ func (w *worker) prepareWork(genParams *generateParams) (*environment, error) {
log.Error("Failed to create sealing context", "err", err)
return nil, err
}
+ // Accumulate the uncles for the sealing work only if it's allowed.
+ if !genParams.noUncle {
+ commitUncles := func(blocks map[common.Hash]*types.Block) {
+ for hash, uncle := range blocks {
+ if len(env.uncles) == 2 {
+ break
+ }
+ if err := w.commitUncle(env, uncle.Header()); err != nil {
+ log.Trace("Possible uncle rejected", "hash", hash, "reason", err)
+ } else {
+ log.Debug("Committing new uncle to block", "hash", hash)
+ }
+ }
+ }
+ // Prefer to locally generated uncle
+ commitUncles(w.localUncles)
+ commitUncles(w.remoteUncles)
+ }
return env, nil
}
@@ -946,7 +1098,7 @@ func (w *worker) generateWork(params *generateParams) (*types.Block, *big.Int, e
log.Warn("Block building is interrupted", "allowance", common.PrettyDuration(w.newpayloadTimeout))
}
}
- block, err := w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, nil, work.receipts, params.withdrawals)
+ block, err := w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts, params.withdrawals)
if err != nil {
return nil, nil, err
}
@@ -955,11 +1107,7 @@ func (w *worker) generateWork(params *generateParams) (*types.Block, *big.Int, e
// commitWork generates several new sealing tasks based on the parent block
// and submit them to the sealer.
-func (w *worker) commitWork(interrupt *atomic.Int32, timestamp int64) {
- // Abort committing if node is still syncing
- if w.syncing.Load() {
- return
- }
+func (w *worker) commitWork(interrupt *atomic.Int32, noempty bool, timestamp int64) {
start := time.Now()
// Set the coinbase if the worker is running or it's required
@@ -978,6 +1126,11 @@ func (w *worker) commitWork(interrupt *atomic.Int32, timestamp int64) {
if err != nil {
return
}
+ // Create an empty block based on temporary copied state for
+ // sealing in advance without waiting block execution finished.
+ if !noempty && !w.noempty.Load() {
+ w.commit(work.copy(), nil, false, start)
+ }
// Fill pending transactions from the txpool into the block.
err = w.fillTransactions(interrupt, work)
switch {
@@ -1031,7 +1184,7 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti
// https://github.com/ethereum/go-ethereum/issues/24299
env := env.copy()
// Withdrawals are set to nil here, because this is only called in PoW.
- block, err := w.engine.FinalizeAndAssemble(w.chain, env.header, env.state, env.txs, nil, env.receipts, nil)
+ block, err := w.engine.FinalizeAndAssemble(w.chain, env.header, env.state, env.txs, env.unclelist(), env.receipts, nil)
if err != nil {
return err
}
@@ -1039,10 +1192,13 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti
if !w.isTTDReached(block.Header()) {
select {
case w.taskCh <- &task{receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}:
+ w.unconfirmed.Shift(block.NumberU64() - 1)
+
fees := totalFees(block, env.receipts)
feesInEther := new(big.Float).Quo(new(big.Float).SetInt(fees), big.NewFloat(params.Ether))
log.Info("Commit new sealing work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()),
- "txs", env.tcount, "gas", block.GasUsed(), "fees", feesInEther,
+ "uncles", len(env.uncles), "txs", env.tcount,
+ "gas", block.GasUsed(), "fees", feesInEther,
"elapsed", common.PrettyDuration(time.Since(start)))
case <-w.exitCh:
@@ -1068,6 +1224,7 @@ func (w *worker) getSealingBlock(parent common.Hash, timestamp uint64, coinbase
coinbase: coinbase,
random: random,
withdrawals: withdrawals,
+ noUncle: true,
noTxs: noTxs,
},
result: make(chan *newPayloadResult, 1),
@@ -1101,6 +1258,14 @@ func copyReceipts(receipts []*types.Receipt) []*types.Receipt {
return result
}
+// postSideBlock fires a side chain event, only use it for testing.
+func (w *worker) postSideBlock(event core.ChainSideEvent) {
+ select {
+ case w.chainSideCh <- event:
+ case <-w.exitCh:
+ }
+}
+
// totalFees computes total consumed miner fees in Wei. Block transactions and receipts have to have the same order.
func totalFees(block *types.Block, receipts []*types.Receipt) *big.Int {
feesWei := new(big.Int)
diff --git a/miner/worker_test.go b/miner/worker_test.go
index fb15d365a7..683d019d2d 100644
--- a/miner/worker_test.go
+++ b/miner/worker_test.go
@@ -17,6 +17,8 @@
package miner
import (
+ "crypto/rand"
+ "errors"
"math/big"
"sync/atomic"
"testing"
@@ -29,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/consensus/ethash"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
+ "github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm"
@@ -106,10 +109,11 @@ func init() {
// testWorkerBackend implements worker.Backend interfaces and wraps all information needed during the testing.
type testWorkerBackend struct {
- db ethdb.Database
- txPool *txpool.TxPool
- chain *core.BlockChain
- genesis *core.Genesis
+ db ethdb.Database
+ txPool *txpool.TxPool
+ chain *core.BlockChain
+ genesis *core.Genesis
+ uncleBlock *types.Block
}
func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine, db ethdb.Database, n int) *testWorkerBackend {
@@ -132,16 +136,58 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine
if err != nil {
t.Fatalf("core.NewBlockChain failed: %v", err)
}
+ txpool := txpool.NewTxPool(testTxPoolConfig, chainConfig, chain)
+
+ // Generate a small n-block chain and an uncle block for it
+ var uncle *types.Block
+ if n > 0 {
+ genDb, blocks, _ := core.GenerateChainWithGenesis(gspec, engine, n, func(i int, gen *core.BlockGen) {
+ gen.SetCoinbase(testBankAddress)
+ })
+ if _, err := chain.InsertChain(blocks); err != nil {
+ t.Fatalf("failed to insert origin chain: %v", err)
+ }
+ parent := chain.GetBlockByHash(chain.CurrentBlock().ParentHash)
+ blocks, _ = core.GenerateChain(chainConfig, parent, engine, genDb, 1, func(i int, gen *core.BlockGen) {
+ gen.SetCoinbase(testUserAddress)
+ })
+ uncle = blocks[0]
+ } else {
+ _, blocks, _ := core.GenerateChainWithGenesis(gspec, engine, 1, func(i int, gen *core.BlockGen) {
+ gen.SetCoinbase(testUserAddress)
+ })
+ uncle = blocks[0]
+ }
return &testWorkerBackend{
- db: db,
- chain: chain,
- txPool: txpool.NewTxPool(testTxPoolConfig, chainConfig, chain),
- genesis: gspec,
+ db: db,
+ chain: chain,
+ txPool: txpool,
+ genesis: gspec,
+ uncleBlock: uncle,
}
}
func (b *testWorkerBackend) BlockChain() *core.BlockChain { return b.chain }
func (b *testWorkerBackend) TxPool() *txpool.TxPool { return b.txPool }
+func (b *testWorkerBackend) StateAtBlock(block *types.Block, reexec uint64, base *state.StateDB, checkLive bool, preferDisk bool) (statedb *state.StateDB, err error) {
+ return nil, errors.New("not supported")
+}
+
+func (b *testWorkerBackend) newRandomUncle() *types.Block {
+ var parent *types.Block
+ cur := b.chain.CurrentBlock()
+ if cur.Number.Uint64() == 0 {
+ parent = b.chain.Genesis()
+ } else {
+ parent = b.chain.GetBlockByHash(b.chain.CurrentBlock().ParentHash)
+ }
+ blocks, _ := core.GenerateChain(b.chain.Config(), parent, b.chain.Engine(), b.db, 1, func(i int, gen *core.BlockGen) {
+ var addr = make([]byte, common.AddressLength)
+ rand.Read(addr)
+ gen.SetCoinbase(common.BytesToAddress(addr))
+ })
+ return blocks[0]
+}
func (b *testWorkerBackend) newRandomTx(creation bool) *types.Transaction {
var tx *types.Transaction
@@ -162,15 +208,25 @@ func newTestWorker(t *testing.T, chainConfig *params.ChainConfig, engine consens
return w, backend
}
-func TestGenerateAndImportBlock(t *testing.T) {
- var (
- db = rawdb.NewMemoryDatabase()
- config = *params.AllCliqueProtocolChanges
- )
- config.Clique = ¶ms.CliqueConfig{Period: 1, Epoch: 30000}
- engine := clique.New(config.Clique, db)
+func TestGenerateBlockAndImportClique(t *testing.T) {
+ testGenerateBlockAndImport(t, true)
+}
- w, b := newTestWorker(t, &config, engine, db, 0)
+func testGenerateBlockAndImport(t *testing.T, isClique bool) {
+ var (
+ engine consensus.Engine
+ chainConfig params.ChainConfig
+ db = rawdb.NewMemoryDatabase()
+ )
+ if isClique {
+ chainConfig = *params.AllCliqueProtocolChanges
+ chainConfig.Clique = ¶ms.CliqueConfig{Period: 1, Epoch: 30000}
+ engine = clique.New(chainConfig.Clique, db)
+ } else {
+ chainConfig = *params.AllEthashProtocolChanges
+ engine = ethash.NewFaker()
+ }
+ w, b := newTestWorker(t, &chainConfig, engine, db, 0)
defer w.close()
// This test chain imports the mined blocks.
@@ -192,6 +248,8 @@ func TestGenerateAndImportBlock(t *testing.T) {
for i := 0; i < 5; i++ {
b.txPool.AddLocal(b.newRandomTx(true))
b.txPool.AddLocal(b.newRandomTx(false))
+ w.postSideBlock(core.ChainSideEvent{Block: b.newRandomUncle()})
+ w.postSideBlock(core.ChainSideEvent{Block: b.newRandomUncle()})
select {
case ev := <-sub.Chan():
@@ -218,10 +276,17 @@ func testEmptyWork(t *testing.T, chainConfig *params.ChainConfig, engine consens
w, _ := newTestWorker(t, chainConfig, engine, rawdb.NewMemoryDatabase(), 0)
defer w.close()
- taskCh := make(chan struct{}, 2)
- checkEqual := func(t *testing.T, task *task) {
- // The work should contain 1 tx
- receiptLen, balance := 1, big.NewInt(1000)
+ var (
+ taskIndex int
+ taskCh = make(chan struct{}, 2)
+ )
+ checkEqual := func(t *testing.T, task *task, index int) {
+ // The first empty work without any txs included
+ receiptLen, balance := 0, big.NewInt(0)
+ if index == 1 {
+ // The second full work with 1 tx included
+ receiptLen, balance = 1, big.NewInt(1000)
+ }
if len(task.receipts) != receiptLen {
t.Fatalf("receipt number mismatch: have %d, want %d", len(task.receipts), receiptLen)
}
@@ -231,7 +296,8 @@ func testEmptyWork(t *testing.T, chainConfig *params.ChainConfig, engine consens
}
w.newTaskHook = func(task *task) {
if task.block.NumberU64() == 1 {
- checkEqual(t, task)
+ checkEqual(t, task, taskIndex)
+ taskIndex += 1
taskCh <- struct{}{}
}
}
@@ -240,9 +306,122 @@ func testEmptyWork(t *testing.T, chainConfig *params.ChainConfig, engine consens
time.Sleep(100 * time.Millisecond)
}
w.start() // Start mining!
+ for i := 0; i < 2; i += 1 {
+ select {
+ case <-taskCh:
+ case <-time.NewTimer(3 * time.Second).C:
+ t.Error("new task timeout")
+ }
+ }
+}
+
+func TestStreamUncleBlock(t *testing.T) {
+ ethash := ethash.NewFaker()
+ defer ethash.Close()
+
+ w, b := newTestWorker(t, ethashChainConfig, ethash, rawdb.NewMemoryDatabase(), 1)
+ defer w.close()
+
+ var taskCh = make(chan struct{}, 3)
+
+ taskIndex := 0
+ w.newTaskHook = func(task *task) {
+ if task.block.NumberU64() == 2 {
+ // The first task is an empty task, the second
+ // one has 1 pending tx, the third one has 1 tx
+ // and 1 uncle.
+ if taskIndex == 2 {
+ have := task.block.Header().UncleHash
+ want := types.CalcUncleHash([]*types.Header{b.uncleBlock.Header()})
+ if have != want {
+ t.Errorf("uncle hash mismatch: have %s, want %s", have.Hex(), want.Hex())
+ }
+ }
+ taskCh <- struct{}{}
+ taskIndex += 1
+ }
+ }
+ w.skipSealHook = func(task *task) bool {
+ return true
+ }
+ w.fullTaskHook = func() {
+ time.Sleep(100 * time.Millisecond)
+ }
+ w.start()
+
+ for i := 0; i < 2; i += 1 {
+ select {
+ case <-taskCh:
+ case <-time.NewTimer(time.Second).C:
+ t.Error("new task timeout")
+ }
+ }
+
+ w.postSideBlock(core.ChainSideEvent{Block: b.uncleBlock})
+
select {
case <-taskCh:
- case <-time.NewTimer(3 * time.Second).C:
+ case <-time.NewTimer(time.Second).C:
+ t.Error("new task timeout")
+ }
+}
+
+func TestRegenerateMiningBlockEthash(t *testing.T) {
+ testRegenerateMiningBlock(t, ethashChainConfig, ethash.NewFaker())
+}
+
+func TestRegenerateMiningBlockClique(t *testing.T) {
+ testRegenerateMiningBlock(t, cliqueChainConfig, clique.New(cliqueChainConfig.Clique, rawdb.NewMemoryDatabase()))
+}
+
+func testRegenerateMiningBlock(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) {
+ defer engine.Close()
+
+ w, b := newTestWorker(t, chainConfig, engine, rawdb.NewMemoryDatabase(), 0)
+ defer w.close()
+
+ var taskCh = make(chan struct{}, 3)
+
+ taskIndex := 0
+ w.newTaskHook = func(task *task) {
+ if task.block.NumberU64() == 1 {
+ // The first task is an empty task, the second
+ // one has 1 pending tx, the third one has 2 txs
+ if taskIndex == 2 {
+ receiptLen, balance := 2, big.NewInt(2000)
+ if len(task.receipts) != receiptLen {
+ t.Errorf("receipt number mismatch: have %d, want %d", len(task.receipts), receiptLen)
+ }
+ if task.state.GetBalance(testUserAddress).Cmp(balance) != 0 {
+ t.Errorf("account balance mismatch: have %d, want %d", task.state.GetBalance(testUserAddress), balance)
+ }
+ }
+ taskCh <- struct{}{}
+ taskIndex += 1
+ }
+ }
+ w.skipSealHook = func(task *task) bool {
+ return true
+ }
+ w.fullTaskHook = func() {
+ time.Sleep(100 * time.Millisecond)
+ }
+
+ w.start()
+ // Ignore the first two works
+ for i := 0; i < 2; i += 1 {
+ select {
+ case <-taskCh:
+ case <-time.NewTimer(time.Second).C:
+ t.Error("new task timeout")
+ }
+ }
+ b.txPool.AddLocals(newTxs)
+ time.Sleep(time.Second)
+
+ select {
+ case <-taskCh:
+ case <-time.NewTimer(time.Second).C:
t.Error("new task timeout")
}
}
@@ -363,6 +542,7 @@ func testGetSealingWork(t *testing.T, chainConfig *params.ChainConfig, engine co
defer w.close()
w.setExtra([]byte{0x01, 0x02})
+ w.postSideBlock(core.ChainSideEvent{Block: b.uncleBlock})
w.skipSealHook = func(task *task) bool {
return true
@@ -377,6 +557,9 @@ func testGetSealingWork(t *testing.T, chainConfig *params.ChainConfig, engine co
// is even smaller than parent block's. It's OK.
t.Logf("Invalid timestamp, want %d, get %d", timestamp, block.Time())
}
+ if len(block.Uncles()) != 0 {
+ t.Error("Unexpected uncle block")
+ }
_, isClique := engine.(*clique.Clique)
if !isClique {
if len(block.Extra()) != 2 {