mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-26 06:36:43 +00:00
miner, consensus, core: remove tracing in mining workflow (#1422)
This PR removes open telemetry based tracing from the whole mining workflow. The tracing extended into consensus and core blockchain functions from miner is also removed in this PR. We weren't checking the tracing from a long time and the way it was implemented increased the code complexity a lot.
This commit is contained in:
Manav Darji
GitHub
parent
984e074a9b
commit
9da0432ec2
9 changed files with 253 additions and 656 deletions
|
|
@ -17,7 +17,6 @@
|
|||
package beacon
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/big"
|
||||
|
|
@ -398,9 +397,9 @@ func (beacon *Beacon) FinalizeAndAssemble(chain consensus.ChainHeaderReader, hea
|
|||
//
|
||||
// Note, the method returns immediately and will send the result async. More
|
||||
// than one result may also be returned depending on the consensus algorithm.
|
||||
func (beacon *Beacon) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
func (beacon *Beacon) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
if !beacon.IsPoSHeader(block.Header()) {
|
||||
return beacon.ethone.Seal(ctx, chain, block, results, stop)
|
||||
return beacon.ethone.Seal(chain, block, results, stop)
|
||||
}
|
||||
// The seal verification is done by the external consensus engine,
|
||||
// return directly without pushing any block back. In another word
|
||||
|
|
|
|||
|
|
@ -17,13 +17,10 @@ import (
|
|||
|
||||
lru "github.com/hashicorp/golang-lru"
|
||||
"github.com/holiman/uint256"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"golang.org/x/crypto/sha3"
|
||||
|
||||
"github.com/ethereum/go-ethereum/accounts"
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/common/tracing"
|
||||
balance_tracing "github.com/ethereum/go-ethereum/core/tracing"
|
||||
|
||||
"github.com/ethereum/go-ethereum/consensus"
|
||||
|
|
@ -820,30 +817,28 @@ func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header) e
|
|||
// Finalize implements consensus.Engine, ensuring no uncles are set, nor block
|
||||
// rewards given.
|
||||
func (c *Bor) Finalize(chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, body *types.Body) {
|
||||
headerNumber := header.Number.Uint64()
|
||||
if body.Withdrawals != nil || header.WithdrawalsHash != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var (
|
||||
stateSyncData []*types.StateSyncData
|
||||
err error
|
||||
)
|
||||
|
||||
headerNumber := header.Number.Uint64()
|
||||
|
||||
if body.Withdrawals != nil || header.WithdrawalsHash != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if IsSprintStart(headerNumber, c.config.CalculateSprint(headerNumber)) {
|
||||
start := time.Now()
|
||||
ctx := context.Background()
|
||||
cx := statefull.ChainContext{Chain: chain, Bor: c}
|
||||
// check and commit span
|
||||
if err := c.checkAndCommitSpan(ctx, state, header, cx); err != nil {
|
||||
if err := c.checkAndCommitSpan(state, header, cx); err != nil {
|
||||
log.Error("Error while committing span", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
if c.HeimdallClient != nil {
|
||||
// commit states
|
||||
stateSyncData, err = c.CommitStates(ctx, state, header, cx)
|
||||
stateSyncData, err = c.CommitStates(state, header, cx)
|
||||
if err != nil {
|
||||
log.Error("Error while committing states", "error", err)
|
||||
return
|
||||
|
|
@ -908,38 +903,28 @@ func (c *Bor) changeContractCodeIfNeeded(headerNumber uint64, state *state.State
|
|||
// FinalizeAndAssemble implements consensus.Engine, ensuring no uncles are set,
|
||||
// nor block rewards given, and returns the final block.
|
||||
func (c *Bor) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *types.Header, state *state.StateDB, body *types.Body, receipts []*types.Receipt) (*types.Block, error) {
|
||||
finalizeCtx, finalizeSpan := tracing.StartSpan(context.Background(), "bor.FinalizeAndAssemble")
|
||||
defer tracing.EndSpan(finalizeSpan)
|
||||
|
||||
headerNumber := header.Number.Uint64()
|
||||
|
||||
if body.Withdrawals != nil || header.WithdrawalsHash != nil {
|
||||
return nil, consensus.ErrUnexpectedWithdrawals
|
||||
}
|
||||
|
||||
stateSyncData := []*types.StateSyncData{}
|
||||
|
||||
var err error
|
||||
var (
|
||||
stateSyncData []*types.StateSyncData
|
||||
err error
|
||||
)
|
||||
|
||||
if IsSprintStart(headerNumber, c.config.CalculateSprint(headerNumber)) {
|
||||
cx := statefull.ChainContext{Chain: chain, Bor: c}
|
||||
|
||||
tracing.Exec(finalizeCtx, "", "bor.checkAndCommitSpan", func(ctx context.Context, span trace.Span) {
|
||||
// check and commit span
|
||||
err = c.checkAndCommitSpan(finalizeCtx, state, header, cx)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
// check and commit span
|
||||
if err = c.checkAndCommitSpan(state, header, cx); err != nil {
|
||||
log.Error("Error while committing span", "error", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if c.HeimdallClient != nil {
|
||||
tracing.Exec(finalizeCtx, "", "bor.checkAndCommitSpan", func(ctx context.Context, span trace.Span) {
|
||||
// commit states
|
||||
stateSyncData, err = c.CommitStates(finalizeCtx, state, header, cx)
|
||||
})
|
||||
|
||||
// commit states
|
||||
stateSyncData, err = c.CommitStates(state, header, cx)
|
||||
if err != nil {
|
||||
log.Error("Error while committing states", "error", err)
|
||||
return nil, err
|
||||
|
|
@ -947,19 +932,13 @@ func (c *Bor) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *typ
|
|||
}
|
||||
}
|
||||
|
||||
tracing.Exec(finalizeCtx, "", "bor.changeContractCodeIfNeeded", func(ctx context.Context, span trace.Span) {
|
||||
err = c.changeContractCodeIfNeeded(headerNumber, state)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
if err = c.changeContractCodeIfNeeded(headerNumber, state); err != nil {
|
||||
log.Error("Error changing contract code", "error", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// No block rewards in PoA, so the state remains as it is
|
||||
tracing.Exec(finalizeCtx, "", "bor.IntermediateRoot", func(ctx context.Context, span trace.Span) {
|
||||
header.Root = state.IntermediateRoot(chain.Config().IsEIP158(header.Number))
|
||||
})
|
||||
header.Root = state.IntermediateRoot(chain.Config().IsEIP158(header.Number))
|
||||
|
||||
// Uncles are dropped
|
||||
header.UncleHash = types.CalcUncleHash(nil)
|
||||
|
|
@ -971,14 +950,6 @@ func (c *Bor) FinalizeAndAssemble(chain consensus.ChainHeaderReader, header *typ
|
|||
bc := chain.(core.BorStateSyncer)
|
||||
bc.SetStateSync(stateSyncData)
|
||||
|
||||
tracing.SetAttributes(
|
||||
finalizeSpan,
|
||||
attribute.Int("number", int(header.Number.Int64())),
|
||||
attribute.String("hash", header.Hash().String()),
|
||||
attribute.Int("number of txs", len(body.Transactions)),
|
||||
attribute.Int("gas used", int(block.GasUsed())),
|
||||
)
|
||||
|
||||
// return the final block for sealing
|
||||
return block, nil
|
||||
}
|
||||
|
|
@ -994,18 +965,7 @@ func (c *Bor) Authorize(currentSigner common.Address, signFn SignerFn) {
|
|||
|
||||
// Seal implements consensus.Engine, attempting to create a sealed block using
|
||||
// the local signing credentials.
|
||||
func (c *Bor) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
_, sealSpan := tracing.StartSpan(ctx, "bor.Seal")
|
||||
|
||||
var endSpan bool = true
|
||||
|
||||
defer func() {
|
||||
// Only end span in case of early-returns/errors
|
||||
if endSpan {
|
||||
tracing.EndSpan(sealSpan)
|
||||
}
|
||||
}()
|
||||
|
||||
func (c *Bor) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
header := block.Header()
|
||||
// Sealing the genesis block is not supported
|
||||
number := header.Number.Uint64()
|
||||
|
|
@ -1051,7 +1011,7 @@ func (c *Bor) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block
|
|||
// Wait until sealing is terminated or delay timeout.
|
||||
log.Info("Waiting for slot to sign and propagate", "number", number, "hash", header.Hash, "delay-in-sec", uint(delay), "delay", common.PrettyDuration(delay))
|
||||
|
||||
go func(sealSpan trace.Span) {
|
||||
go func() {
|
||||
select {
|
||||
case <-stop:
|
||||
log.Debug("Discarding sealing operation for block", "number", number)
|
||||
|
|
@ -1074,27 +1034,13 @@ func (c *Bor) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block
|
|||
"delay", delay,
|
||||
"headerDifficulty", header.Difficulty,
|
||||
)
|
||||
|
||||
tracing.SetAttributes(
|
||||
sealSpan,
|
||||
attribute.Int("number", int(number)),
|
||||
attribute.String("hash", header.Hash().String()),
|
||||
attribute.Int("delay", int(delay.Milliseconds())),
|
||||
attribute.Int("wiggle", int(wiggle.Milliseconds())),
|
||||
attribute.Bool("out-of-turn", wiggle > 0),
|
||||
)
|
||||
|
||||
tracing.EndSpan(sealSpan)
|
||||
}
|
||||
select {
|
||||
case results <- block.WithSeal(header):
|
||||
default:
|
||||
log.Warn("Sealing result was not read by miner", "number", number, "sealhash", SealHash(header, c.config))
|
||||
}
|
||||
}(sealSpan)
|
||||
|
||||
// Set the endSpan flag to false, as the go routine will handle it
|
||||
endSpan = false
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -1150,11 +1096,11 @@ func (c *Bor) Close() error {
|
|||
}
|
||||
|
||||
func (c *Bor) checkAndCommitSpan(
|
||||
ctx context.Context,
|
||||
state *state.StateDB,
|
||||
header *types.Header,
|
||||
chain core.ChainContext,
|
||||
) error {
|
||||
var ctx = context.Background()
|
||||
headerNumber := header.Number.Uint64()
|
||||
|
||||
span, err := c.spanner.GetCurrentSpan(ctx, header.ParentHash)
|
||||
|
|
@ -1228,7 +1174,6 @@ func (c *Bor) FetchAndCommitSpan(
|
|||
|
||||
// CommitStates commit states
|
||||
func (c *Bor) CommitStates(
|
||||
ctx context.Context,
|
||||
state *state.StateDB,
|
||||
header *types.Header,
|
||||
chain statefull.ChainContext,
|
||||
|
|
@ -1269,7 +1214,7 @@ func (c *Bor) CommitStates(
|
|||
"fromID", from,
|
||||
"to", to.Format(time.RFC3339))
|
||||
|
||||
eventRecords, err := c.HeimdallClient.StateSyncEvents(ctx, from, to.Unix())
|
||||
eventRecords, err := c.HeimdallClient.StateSyncEvents(context.Background(), from, to.Unix())
|
||||
if err != nil {
|
||||
log.Error("Error occurred when fetching state sync events", "fromID", from, "to", to.Unix(), "err", err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,7 +19,6 @@ package clique
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
|
|
@ -652,7 +651,7 @@ func (c *Clique) Authorize(signer common.Address, signFn SignerFn) {
|
|||
|
||||
// Seal implements consensus.Engine, attempting to create a sealed block using
|
||||
// the local signing credentials.
|
||||
func (c *Clique) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
func (c *Clique) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
header := block.Header()
|
||||
|
||||
// Sealing the genesis block is not supported
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@
|
|||
package consensus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/big"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
|
|
@ -103,7 +102,7 @@ type Engine interface {
|
|||
//
|
||||
// Note, the method returns immediately and will send the result async. More
|
||||
// than one result may also be returned depending on the consensus algorithm.
|
||||
Seal(ctx context.Context, chain ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error
|
||||
Seal(chain ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error
|
||||
|
||||
// SealHash returns the hash of a block prior to it being sealed.
|
||||
SealHash(header *types.Header) common.Hash
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@
|
|||
package ethash
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/consensus"
|
||||
|
|
@ -81,6 +80,6 @@ func (ethash *Ethash) APIs(chain consensus.ChainHeaderReader) []rpc.API {
|
|||
// Seal generates a new sealing request for the given input block and pushes
|
||||
// the result into the given channel. For the ethash engine, this method will
|
||||
// just panic as sealing is not supported anymore.
|
||||
func (ethash *Ethash) Seal(ctx context.Context, chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
func (ethash *Ethash) Seal(chain consensus.ChainHeaderReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error {
|
||||
panic("ethash (pow) sealing not supported any more")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1922,18 +1922,18 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
|
|||
|
||||
// WriteBlockAndSetHead writes the given block and all associated state to the database,
|
||||
// and applies the block as the new chain head.
|
||||
func (bc *BlockChain) WriteBlockAndSetHead(ctx context.Context, block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
|
||||
func (bc *BlockChain) WriteBlockAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
|
||||
if !bc.chainmu.TryLock() {
|
||||
return NonStatTy, errChainStopped
|
||||
}
|
||||
defer bc.chainmu.Unlock()
|
||||
|
||||
return bc.writeBlockAndSetHead(ctx, block, receipts, logs, state, emitHeadEvent)
|
||||
return bc.writeBlockAndSetHead(block, receipts, logs, state, emitHeadEvent)
|
||||
}
|
||||
|
||||
// writeBlockAndSetHead is the internal implementation of WriteBlockAndSetHead.
|
||||
// This function expects the chain mutex to be held.
|
||||
func (bc *BlockChain) writeBlockAndSetHead(ctx context.Context, block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
|
||||
func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
|
||||
stateSyncLogs, err := bc.writeBlockWithState(block, receipts, logs, state)
|
||||
if err != nil {
|
||||
return NonStatTy, err
|
||||
|
|
@ -2401,7 +2401,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
|
|||
// Don't set the head, only insert the block
|
||||
_, err = bc.writeBlockWithState(block, receipts, logs, statedb)
|
||||
} else {
|
||||
status, err = bc.writeBlockAndSetHead(context.Background(), block, receipts, logs, statedb, false)
|
||||
status, err = bc.writeBlockAndSetHead(block, receipts, logs, statedb, false)
|
||||
}
|
||||
|
||||
followupInterrupt.Store(true)
|
||||
|
|
@ -2574,7 +2574,7 @@ func (bc *BlockChain) processBlock(block *types.Block, statedb *state.StateDB, s
|
|||
// Don't set the head, only insert the block
|
||||
_, err = bc.writeBlockWithState(block, receipts, logs, statedb)
|
||||
} else {
|
||||
status, err = bc.writeBlockAndSetHead(context.Background(), block, receipts, logs, statedb, false)
|
||||
status, err = bc.writeBlockAndSetHead(block, receipts, logs, statedb, false)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
|
|||
|
|
@ -3,8 +3,6 @@ package miner
|
|||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
|
@ -12,9 +10,8 @@ import (
|
|||
// nolint:typecheck
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
cmath "github.com/ethereum/go-ethereum/common/math"
|
||||
"github.com/ethereum/go-ethereum/common/tracing"
|
||||
"github.com/ethereum/go-ethereum/consensus"
|
||||
"github.com/ethereum/go-ethereum/consensus/misc/eip4844"
|
||||
"github.com/ethereum/go-ethereum/core"
|
||||
"github.com/ethereum/go-ethereum/core/blockstm"
|
||||
"github.com/ethereum/go-ethereum/core/state"
|
||||
|
|
@ -27,10 +24,6 @@ import (
|
|||
"github.com/ethereum/go-ethereum/rlp"
|
||||
"github.com/holiman/uint256"
|
||||
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
|
||||
lru "github.com/hashicorp/golang-lru"
|
||||
)
|
||||
|
||||
|
|
@ -102,12 +95,10 @@ func newWorkerWithDelay(config *Config, chainConfig *params.ChainConfig, engine
|
|||
|
||||
worker.newpayloadTimeout = newpayloadTimeout
|
||||
|
||||
ctx := tracing.WithTracer(context.Background(), otel.GetTracerProvider().Tracer("MinerWorker"))
|
||||
|
||||
worker.wg.Add(4)
|
||||
|
||||
go worker.mainLoopWithDelay(ctx, delay, opcodeDelay)
|
||||
go worker.newWorkLoop(ctx, recommit)
|
||||
go worker.mainLoopWithDelay(delay, opcodeDelay)
|
||||
go worker.newWorkLoop(recommit)
|
||||
go worker.resultLoop()
|
||||
go worker.taskLoop()
|
||||
|
||||
|
|
@ -121,7 +112,7 @@ func newWorkerWithDelay(config *Config, chainConfig *params.ChainConfig, engine
|
|||
|
||||
// mainLoopWithDelay is mainLoop() with extra params to induce artficial delays for tests such as commit-interrupt.
|
||||
// nolint:gocognit
|
||||
func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay uint) {
|
||||
func (w *worker) mainLoopWithDelay(delay uint, opcodeDelay uint) {
|
||||
defer w.wg.Done()
|
||||
defer w.txsSub.Unsubscribe()
|
||||
defer w.chainHeadSub.Unsubscribe()
|
||||
|
|
@ -137,15 +128,15 @@ func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay
|
|||
if w.chainConfig.ChainID.Cmp(params.BorMainnetChainConfig.ChainID) == 0 || w.chainConfig.ChainID.Cmp(params.MumbaiChainConfig.ChainID) == 0 {
|
||||
if w.eth.PeerCount() > 0 {
|
||||
//nolint:contextcheck
|
||||
w.commitWorkWithDelay(req.ctx, req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay)
|
||||
w.commitWorkWithDelay(req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay)
|
||||
}
|
||||
} else {
|
||||
//nolint:contextcheck
|
||||
w.commitWorkWithDelay(req.ctx, req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay)
|
||||
w.commitWorkWithDelay(req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay)
|
||||
}
|
||||
|
||||
case req := <-w.getWorkCh:
|
||||
req.result <- w.generateWork(ctx, req.params)
|
||||
req.result <- w.generateWork(req.params)
|
||||
|
||||
case ev := <-w.txsCh:
|
||||
// Apply transactions to the pending state if we're not sealing
|
||||
|
|
@ -187,7 +178,7 @@ func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay
|
|||
// 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(ctx, nil, true, time.Now().Unix())
|
||||
w.commitWork(nil, true, time.Now().Unix())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -205,7 +196,7 @@ func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay
|
|||
}
|
||||
|
||||
// commitWorkWithDelay is commitWork() with extra params to induce artficial delays for tests such as commit-interrupt.
|
||||
func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int32, noempty bool, timestamp int64, delay uint, opcodeDelay uint) {
|
||||
func (w *worker) commitWorkWithDelay(interrupt *atomic.Int32, noempty bool, timestamp int64, delay uint, opcodeDelay uint) {
|
||||
// Abort committing if node is still syncing
|
||||
if w.syncing.Load() {
|
||||
return
|
||||
|
|
@ -217,23 +208,20 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int3
|
|||
err error
|
||||
)
|
||||
|
||||
tracing.Exec(ctx, "", "worker.prepareWork", func(ctx context.Context, span trace.Span) {
|
||||
// Set the coinbase if the worker is running or it's required
|
||||
var coinbase common.Address
|
||||
if w.IsRunning() {
|
||||
coinbase = w.etherbase()
|
||||
if coinbase == (common.Address{}) {
|
||||
log.Error("Refusing to mine without etherbase")
|
||||
return
|
||||
}
|
||||
// Set the coinbase if the worker is running or it's required
|
||||
var coinbase common.Address
|
||||
if w.IsRunning() {
|
||||
coinbase = w.etherbase()
|
||||
if coinbase == (common.Address{}) {
|
||||
log.Error("Refusing to mine without etherbase")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
work, err = w.prepareWork(&generateParams{
|
||||
timestamp: uint64(timestamp),
|
||||
coinbase: coinbase,
|
||||
})
|
||||
work, err = w.prepareWork(&generateParams{
|
||||
timestamp: uint64(timestamp),
|
||||
coinbase: coinbase,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -248,7 +236,7 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int3
|
|||
|
||||
if !noempty && w.interruptCommitFlag {
|
||||
block := w.chain.GetBlockByHash(w.chain.CurrentBlock().Hash())
|
||||
interruptCtx, stopFn = getInterruptTimer(ctx, work, block)
|
||||
interruptCtx, stopFn = getInterruptTimer(work, block)
|
||||
// nolint : staticcheck
|
||||
interruptCtx = vm.PutCache(interruptCtx, w.interruptedTxCache)
|
||||
// nolint : staticcheck
|
||||
|
|
@ -257,21 +245,13 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int3
|
|||
interruptCtx = context.WithValue(interruptCtx, vm.InterruptCtxOpcodeDelayKey, opcodeDelay)
|
||||
}
|
||||
|
||||
ctx, span := tracing.StartSpan(ctx, "commitWork")
|
||||
defer tracing.EndSpan(span)
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("number", int(work.header.Number.Uint64())),
|
||||
)
|
||||
|
||||
// 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(ctx, work.copy(), nil, false, start)
|
||||
_ = w.commit(work.copy(), nil, false, start)
|
||||
}
|
||||
// Fill pending transactions from the txpool into the block.
|
||||
err = w.fillTransactionsWithDelay(ctx, interrupt, work, interruptCtx)
|
||||
err = w.fillTransactionsWithDelay(interrupt, work, interruptCtx)
|
||||
|
||||
switch {
|
||||
case err == nil:
|
||||
|
|
@ -302,7 +282,7 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int3
|
|||
return
|
||||
}
|
||||
// Submit the generated block for consensus sealing.
|
||||
_ = w.commit(ctx, work.copy(), w.fullTaskHook, true, start)
|
||||
_ = w.commit(work.copy(), w.fullTaskHook, true, start)
|
||||
|
||||
// Swap out the old work with the new one, terminating any leftover
|
||||
// prefetcher processes in the mean time and starting a new one.
|
||||
|
|
@ -315,182 +295,78 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int3
|
|||
|
||||
// fillTransactionsWithDelay is fillTransactions() with extra params to induce artficial delays for tests such as commit-interrupt.
|
||||
// nolint:gocognit
|
||||
func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *atomic.Int32, env *environment, interruptCtx context.Context) error {
|
||||
ctx, span := tracing.StartSpan(ctx, "fillTransactions")
|
||||
defer tracing.EndSpan(span)
|
||||
func (w *worker) fillTransactionsWithDelay(interrupt *atomic.Int32, env *environment, interruptCtx context.Context) error {
|
||||
w.mu.RLock()
|
||||
tip := w.tip
|
||||
w.mu.RUnlock()
|
||||
|
||||
// Retrieve the pending transactions pre-filtered by the 1559/4844 dynamic fees
|
||||
filter := txpool.PendingFilter{
|
||||
MinTip: uint256.MustFromBig(tip.ToBig()),
|
||||
}
|
||||
|
||||
if env.header.BaseFee != nil {
|
||||
filter.BaseFee = uint256.MustFromBig(env.header.BaseFee)
|
||||
}
|
||||
|
||||
if env.header.ExcessBlobGas != nil {
|
||||
filter.BlobFee = uint256.MustFromBig(eip4844.CalcBlobFee(*env.header.ExcessBlobGas))
|
||||
}
|
||||
|
||||
var (
|
||||
localPlainTxs, remotePlainTxs, localBlobTxs, remoteBlobTxs map[common.Address][]*txpool.LazyTransaction
|
||||
)
|
||||
|
||||
filter.OnlyPlainTxs, filter.OnlyBlobTxs = true, false
|
||||
pendingPlainTxs := w.eth.TxPool().Pending(filter)
|
||||
|
||||
filter.OnlyPlainTxs, filter.OnlyBlobTxs = false, true
|
||||
pendingBlobTxs := w.eth.TxPool().Pending(filter)
|
||||
|
||||
// Split the pending transactions into locals and remotes.
|
||||
localPlainTxs, remotePlainTxs = make(map[common.Address][]*txpool.LazyTransaction), pendingPlainTxs
|
||||
localBlobTxs, remoteBlobTxs = make(map[common.Address][]*txpool.LazyTransaction), pendingBlobTxs
|
||||
|
||||
for _, account := range w.eth.TxPool().Locals() {
|
||||
if txs := remotePlainTxs[account]; len(txs) > 0 {
|
||||
delete(remotePlainTxs, account)
|
||||
localPlainTxs[account] = txs
|
||||
}
|
||||
if txs := remoteBlobTxs[account]; len(txs) > 0 {
|
||||
delete(remoteBlobTxs, account)
|
||||
localBlobTxs[account] = txs
|
||||
}
|
||||
}
|
||||
|
||||
// Split the pending transactions into locals and remotes
|
||||
// Fill the block with all available pending transactions.
|
||||
pending := w.eth.TxPool().Pending(txpool.PendingFilter{})
|
||||
localTxs, remoteTxs := make(map[common.Address][]*txpool.LazyTransaction), pending
|
||||
if len(localPlainTxs) > 0 || len(localBlobTxs) > 0 {
|
||||
var plainTxs, blobTxs *transactionsByPriceAndNonce
|
||||
|
||||
var (
|
||||
localTxsCount int
|
||||
remoteTxsCount int
|
||||
)
|
||||
plainTxs = newTransactionsByPriceAndNonce(env.signer, localPlainTxs, env.header.BaseFee)
|
||||
blobTxs = newTransactionsByPriceAndNonce(env.signer, localBlobTxs, env.header.BaseFee)
|
||||
|
||||
// TODO: move to config or RPC
|
||||
const profiling = false
|
||||
|
||||
if profiling {
|
||||
doneCh := make(chan struct{})
|
||||
|
||||
defer func() {
|
||||
close(doneCh)
|
||||
}()
|
||||
|
||||
go func(number uint64) {
|
||||
closeFn := func() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
// Check if we've not crossed limit
|
||||
if attempt := atomic.AddInt32(w.profileCount, 1); attempt >= 10 {
|
||||
log.Info("Completed profiling", "attempt", attempt)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Starting profiling in fill transactions", "number", number)
|
||||
|
||||
dir, err := os.MkdirTemp("", fmt.Sprintf("bor-traces-%s-", time.Now().UTC().Format("2006-01-02-150405Z")))
|
||||
if err != nil {
|
||||
log.Error("Error in profiling", "path", dir, "number", number, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
// grab the cpu profile
|
||||
closeFnInternal, err := startProfiler("cpu", dir, number)
|
||||
if err != nil {
|
||||
log.Error("Error in profiling", "path", dir, "number", number, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
closeFn = func() error {
|
||||
err := closeFnInternal()
|
||||
|
||||
log.Info("Completed profiling", "path", dir, "number", number, "error", err)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
case <-doneCh:
|
||||
err := closeFn()
|
||||
|
||||
if err != nil {
|
||||
log.Info("closing fillTransactions", "number", number, "error", err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
}(env.header.Number.Uint64())
|
||||
}
|
||||
|
||||
tracing.Exec(ctx, "", "worker.SplittingTransactions", func(ctx context.Context, span trace.Span) {
|
||||
prePendingTime := time.Now()
|
||||
|
||||
pending := w.eth.TxPool().Pending(txpool.PendingFilter{})
|
||||
remoteTxs = pending
|
||||
|
||||
postPendingTime := time.Now()
|
||||
|
||||
for _, account := range w.eth.TxPool().Locals() {
|
||||
if txs := remoteTxs[account]; len(txs) > 0 {
|
||||
delete(remoteTxs, account)
|
||||
|
||||
localTxs[account] = txs
|
||||
}
|
||||
}
|
||||
|
||||
postLocalsTime := time.Now()
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of local txs", localTxsCount),
|
||||
attribute.Int("len of remote txs", remoteTxsCount),
|
||||
attribute.String("time taken by Pending()", fmt.Sprintf("%v", postPendingTime.Sub(prePendingTime))),
|
||||
attribute.String("time taken by Locals()", fmt.Sprintf("%v", postLocalsTime.Sub(postPendingTime))),
|
||||
)
|
||||
})
|
||||
|
||||
var (
|
||||
localEnvTCount int
|
||||
remoteEnvTCount int
|
||||
err error
|
||||
)
|
||||
|
||||
if len(localTxs) > 0 {
|
||||
var txs *transactionsByPriceAndNonce
|
||||
|
||||
tracing.Exec(ctx, "", "worker.LocalTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) {
|
||||
var baseFee *uint256.Int
|
||||
if env.header.BaseFee != nil {
|
||||
baseFee = cmath.FromBig(env.header.BaseFee)
|
||||
}
|
||||
|
||||
txs = newTransactionsByPriceAndNonce(env.signer, localTxs, baseFee.ToBig())
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of tx local Heads", txs.GetTxs()),
|
||||
)
|
||||
})
|
||||
|
||||
tracing.Exec(ctx, "", "worker.LocalCommitTransactions", func(ctx context.Context, span trace.Span) {
|
||||
err = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
if err := w.commitTransactionsWithDelay(env, plainTxs, blobTxs, interrupt, new(uint256.Int), interruptCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
localEnvTCount = env.tcount
|
||||
}
|
||||
|
||||
if len(remoteTxs) > 0 {
|
||||
var txs *transactionsByPriceAndNonce
|
||||
if len(remotePlainTxs) > 0 || len(remoteBlobTxs) > 0 {
|
||||
var plainTxs, blobTxs *transactionsByPriceAndNonce
|
||||
|
||||
tracing.Exec(ctx, "", "worker.RemoteTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) {
|
||||
var baseFee *uint256.Int
|
||||
if env.header.BaseFee != nil {
|
||||
baseFee = cmath.FromBig(env.header.BaseFee)
|
||||
}
|
||||
plainTxs = newTransactionsByPriceAndNonce(env.signer, remotePlainTxs, env.header.BaseFee)
|
||||
blobTxs = newTransactionsByPriceAndNonce(env.signer, remoteBlobTxs, env.header.BaseFee)
|
||||
|
||||
txs = newTransactionsByPriceAndNonce(env.signer, remoteTxs, baseFee.ToBig())
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of tx remote Heads", txs.GetTxs()),
|
||||
)
|
||||
})
|
||||
|
||||
tracing.Exec(ctx, "", "worker.RemoteCommitTransactions", func(ctx context.Context, span trace.Span) {
|
||||
err = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
if err := w.commitTransactionsWithDelay(env, plainTxs, blobTxs, interrupt, new(uint256.Int), interruptCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
remoteEnvTCount = env.tcount
|
||||
}
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of final local txs ", localEnvTCount),
|
||||
attribute.Int("len of final remote txs", remoteEnvTCount),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// commitTransactionsWithDelay is commitTransactions() with extra params to induce artficial delays for tests such as commit-interrupt.
|
||||
// nolint:gocognit, unparam
|
||||
func (w *worker) commitTransactionsWithDelay(env *environment, txs *transactionsByPriceAndNonce, interrupt *atomic.Int32, interruptCtx context.Context) error {
|
||||
func (w *worker) commitTransactionsWithDelay(env *environment, plainTxs, blobTxs *transactionsByPriceAndNonce, interrupt *atomic.Int32, minTip *uint256.Int, interruptCtx context.Context) error {
|
||||
gasLimit := env.header.GasLimit
|
||||
if env.gasPool == nil {
|
||||
env.gasPool = new(core.GasPool).AddGas(gasLimit)
|
||||
|
|
@ -498,35 +374,25 @@ func (w *worker) commitTransactionsWithDelay(env *environment, txs *transactions
|
|||
|
||||
var coalescedLogs []*types.Log
|
||||
|
||||
var depsMVReadList [][]blockstm.ReadDescriptor
|
||||
|
||||
var depsMVFullWriteList [][]blockstm.WriteDescriptor
|
||||
|
||||
var mvReadMapList []map[blockstm.Key]blockstm.ReadDescriptor
|
||||
|
||||
var deps map[int]map[int]bool
|
||||
|
||||
chDeps := make(chan blockstm.TxDep)
|
||||
|
||||
var count int
|
||||
|
||||
var depsWg sync.WaitGroup
|
||||
var once sync.Once
|
||||
|
||||
EnableMVHashMap := w.chainConfig.IsCancun(env.header.Number)
|
||||
|
||||
// create and add empty mvHashMap in statedb
|
||||
if EnableMVHashMap {
|
||||
depsMVReadList = [][]blockstm.ReadDescriptor{}
|
||||
|
||||
depsMVFullWriteList = [][]blockstm.WriteDescriptor{}
|
||||
|
||||
mvReadMapList = []map[blockstm.Key]blockstm.ReadDescriptor{}
|
||||
|
||||
if EnableMVHashMap && w.IsRunning() {
|
||||
deps = map[int]map[int]bool{}
|
||||
|
||||
chDeps = make(chan blockstm.TxDep)
|
||||
|
||||
count = 0
|
||||
// Make sure we safely close the channel in case of interrupt
|
||||
defer once.Do(func() {
|
||||
close(chDeps)
|
||||
})
|
||||
|
||||
depsWg.Add(1)
|
||||
|
||||
|
|
@ -543,8 +409,15 @@ func (w *worker) commitTransactionsWithDelay(env *environment, txs *transactions
|
|||
|
||||
mainloop:
|
||||
for {
|
||||
// Check interruption signal and abort building if it's fired.
|
||||
if interrupt != nil {
|
||||
if signal := interrupt.Load(); signal != commitInterruptNone {
|
||||
return signalToErr(signal)
|
||||
}
|
||||
}
|
||||
|
||||
if interruptCtx != nil {
|
||||
if EnableMVHashMap {
|
||||
if EnableMVHashMap && w.IsRunning() {
|
||||
env.state.AddEmptyMVHashMap()
|
||||
}
|
||||
|
||||
|
|
@ -558,19 +431,39 @@ mainloop:
|
|||
}
|
||||
}
|
||||
|
||||
// Check interruption signal and abort building if it's fired.
|
||||
if interrupt != nil {
|
||||
if signal := interrupt.Load(); signal != commitInterruptNone {
|
||||
return signalToErr(signal)
|
||||
}
|
||||
}
|
||||
// If we don't have enough gas for any further transactions then we're done.
|
||||
if env.gasPool.Gas() < params.TxGas {
|
||||
log.Trace("Not enough gas for further transactions", "have", env.gasPool, "want", params.TxGas)
|
||||
break
|
||||
}
|
||||
// If we don't have enough blob space for any further blob transactions,
|
||||
// skip that list altogether
|
||||
if !blobTxs.Empty() && env.blobs*params.BlobTxBlobGasPerBlob >= params.MaxBlobGasPerBlock {
|
||||
log.Trace("Not enough blob space for further blob transactions")
|
||||
blobTxs.Clear()
|
||||
// Fall though to pick up any plain txs
|
||||
}
|
||||
// Retrieve the next transaction and abort if all done.
|
||||
ltx, _ := txs.Peek()
|
||||
|
||||
var (
|
||||
ltx *txpool.LazyTransaction
|
||||
txs *transactionsByPriceAndNonce
|
||||
)
|
||||
pltx, ptip := plainTxs.Peek()
|
||||
bltx, btip := blobTxs.Peek()
|
||||
|
||||
switch {
|
||||
case pltx == nil:
|
||||
txs, ltx = blobTxs, bltx
|
||||
case bltx == nil:
|
||||
txs, ltx = plainTxs, pltx
|
||||
default:
|
||||
if ptip.Lt(btip) {
|
||||
txs, ltx = blobTxs, bltx
|
||||
} else {
|
||||
txs, ltx = plainTxs, pltx
|
||||
}
|
||||
}
|
||||
if ltx == nil {
|
||||
break
|
||||
}
|
||||
|
|
@ -586,6 +479,11 @@ mainloop:
|
|||
txs.Pop()
|
||||
continue
|
||||
}
|
||||
// If we don't receive enough tip for the next transaction, skip the account
|
||||
if ptip.Cmp(minTip) < 0 {
|
||||
log.Trace("Not enough tip for transaction", "hash", ltx.Hash, "tip", ptip, "needed", minTip)
|
||||
break // If the next-best is too low, surely no better will be available
|
||||
}
|
||||
// Transaction seems to fit, pull it up from the pool
|
||||
tx := ltx.Resolve()
|
||||
if tx == nil {
|
||||
|
|
@ -594,7 +492,7 @@ mainloop:
|
|||
continue
|
||||
}
|
||||
// Error may be ignored here. The error has already been checked
|
||||
// during transaction acceptance is the transaction pool.
|
||||
// during transaction acceptance in the transaction pool.
|
||||
from, _ := types.Sender(env.signer, tx)
|
||||
|
||||
// not prioritising conditional transaction, yet.
|
||||
|
|
@ -652,19 +550,21 @@ mainloop:
|
|||
coalescedLogs = append(coalescedLogs, logs...)
|
||||
env.tcount++
|
||||
|
||||
if EnableMVHashMap {
|
||||
depsMVReadList = append(depsMVReadList, env.state.MVReadList())
|
||||
depsMVFullWriteList = append(depsMVFullWriteList, env.state.MVFullWriteList())
|
||||
mvReadMapList = append(mvReadMapList, env.state.MVReadMap())
|
||||
if EnableMVHashMap && w.IsRunning() {
|
||||
env.depsMVFullWriteList = append(env.depsMVFullWriteList, env.state.MVFullWriteList())
|
||||
env.mvReadMapList = append(env.mvReadMapList, env.state.MVReadMap())
|
||||
|
||||
if env.tcount > len(env.depsMVFullWriteList) {
|
||||
log.Warn("blockstm - env.tcount > len(env.depsMVFullWriteList)", "env.tcount", env.tcount, "len(depsMVFullWriteList)", len(env.depsMVFullWriteList))
|
||||
}
|
||||
|
||||
temp := blockstm.TxDep{
|
||||
Index: env.tcount - 1,
|
||||
ReadList: depsMVReadList[count],
|
||||
FullWriteList: depsMVFullWriteList,
|
||||
ReadList: env.state.MVReadList(),
|
||||
FullWriteList: env.depsMVFullWriteList,
|
||||
}
|
||||
|
||||
chDeps <- temp
|
||||
count++
|
||||
}
|
||||
|
||||
txs.Shift()
|
||||
|
|
@ -675,7 +575,7 @@ mainloop:
|
|||
txs.Pop()
|
||||
}
|
||||
|
||||
if EnableMVHashMap {
|
||||
if EnableMVHashMap && w.IsRunning() {
|
||||
env.state.ClearReadMap()
|
||||
env.state.ClearWriteMap()
|
||||
}
|
||||
|
|
@ -683,7 +583,9 @@ mainloop:
|
|||
|
||||
// nolint:nestif
|
||||
if EnableMVHashMap && w.IsRunning() {
|
||||
close(chDeps)
|
||||
once.Do(func() {
|
||||
close(chDeps)
|
||||
})
|
||||
depsWg.Wait()
|
||||
|
||||
var blockExtraData types.BlockExtraData
|
||||
|
|
@ -691,8 +593,8 @@ mainloop:
|
|||
tempVanity := env.header.Extra[:types.ExtraVanityLength]
|
||||
tempSeal := env.header.Extra[len(env.header.Extra)-types.ExtraSealLength:]
|
||||
|
||||
if len(mvReadMapList) > 0 {
|
||||
tempDeps := make([][]uint64, len(mvReadMapList))
|
||||
if len(env.mvReadMapList) > 0 {
|
||||
tempDeps := make([][]uint64, len(env.mvReadMapList))
|
||||
|
||||
for j := range deps[0] {
|
||||
tempDeps[0] = append(tempDeps[0], uint64(j))
|
||||
|
|
@ -700,14 +602,15 @@ mainloop:
|
|||
|
||||
delayFlag := true
|
||||
|
||||
for i := 1; i <= len(mvReadMapList)-1; i++ {
|
||||
reads := mvReadMapList[i-1]
|
||||
for i := 1; i <= len(env.mvReadMapList)-1; i++ {
|
||||
reads := env.mvReadMapList[i-1]
|
||||
|
||||
_, ok1 := reads[blockstm.NewSubpathKey(env.coinbase, state.BalancePath)]
|
||||
_, ok2 := reads[blockstm.NewSubpathKey(common.HexToAddress(w.chainConfig.Bor.CalculateBurntContract(env.header.Number.Uint64())), state.BalancePath)]
|
||||
|
||||
if ok1 || ok2 {
|
||||
delayFlag = false
|
||||
break
|
||||
}
|
||||
|
||||
for j := range deps[i] {
|
||||
|
|
|
|||
400
miner/worker.go
400
miner/worker.go
|
|
@ -17,15 +17,10 @@
|
|||
package miner
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/big"
|
||||
"os"
|
||||
"runtime"
|
||||
"runtime/pprof"
|
||||
ptrace "runtime/trace"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
|
@ -33,8 +28,6 @@ import (
|
|||
lru "github.com/hashicorp/golang-lru"
|
||||
"github.com/holiman/uint256"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/common/tracing"
|
||||
|
|
@ -155,8 +148,6 @@ func (env *environment) discard() {
|
|||
|
||||
// task contains all information for consensus engine sealing and result submitting.
|
||||
type task struct {
|
||||
//nolint:containedctx
|
||||
ctx context.Context
|
||||
receipts []*types.Receipt
|
||||
state *state.StateDB
|
||||
block *types.Block
|
||||
|
|
@ -172,8 +163,6 @@ const (
|
|||
|
||||
// newWorkReq represents a request for new sealing work submitting with relative interrupt notifier.
|
||||
type newWorkReq struct {
|
||||
//nolint:containedctx
|
||||
ctx context.Context
|
||||
interrupt *atomic.Int32
|
||||
noempty bool
|
||||
timestamp int64
|
||||
|
|
@ -352,12 +341,10 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
|
|||
|
||||
worker.newpayloadTimeout = newpayloadTimeout
|
||||
|
||||
ctx := tracing.WithTracer(context.Background(), otel.GetTracerProvider().Tracer("MinerWorker"))
|
||||
|
||||
worker.wg.Add(4)
|
||||
|
||||
go worker.mainLoop(ctx)
|
||||
go worker.newWorkLoop(ctx, recommit)
|
||||
go worker.mainLoop()
|
||||
go worker.newWorkLoop(recommit)
|
||||
go worker.resultLoop()
|
||||
go worker.taskLoop()
|
||||
|
||||
|
|
@ -483,7 +470,7 @@ func recalcRecommit(minRecommit, prev time.Duration, target float64, inc bool) t
|
|||
// newWorkLoop is a standalone goroutine to submit new sealing work upon received events.
|
||||
//
|
||||
//nolint:gocognit
|
||||
func (w *worker) newWorkLoop(ctx context.Context, recommit time.Duration) {
|
||||
func (w *worker) newWorkLoop(recommit time.Duration) {
|
||||
defer w.wg.Done()
|
||||
|
||||
var (
|
||||
|
|
@ -498,15 +485,13 @@ func (w *worker) newWorkLoop(ctx context.Context, recommit time.Duration) {
|
|||
|
||||
// commit aborts in-flight transaction execution with given signal and resubmits a new one.
|
||||
commit := func(noempty bool, s int32) {
|
||||
ctx, span := tracing.Trace(ctx, "worker.newWorkLoop.commit")
|
||||
tracing.EndSpan(span)
|
||||
if interrupt != nil {
|
||||
interrupt.Store(s)
|
||||
}
|
||||
|
||||
interrupt = new(atomic.Int32)
|
||||
select {
|
||||
case w.newWorkCh <- &newWorkReq{interrupt: interrupt, timestamp: timestamp, ctx: ctx, noempty: noempty}:
|
||||
case w.newWorkCh <- &newWorkReq{interrupt: interrupt, timestamp: timestamp, noempty: noempty}:
|
||||
case <-w.exitCh:
|
||||
return
|
||||
}
|
||||
|
|
@ -515,9 +500,6 @@ func (w *worker) newWorkLoop(ctx context.Context, recommit time.Duration) {
|
|||
}
|
||||
// clearPending cleans the stale pending tasks.
|
||||
clearPending := func(number uint64) {
|
||||
_, span := tracing.Trace(ctx, "worker.newWorkLoop.clearPending")
|
||||
tracing.EndSpan(span)
|
||||
|
||||
w.pendingMu.Lock()
|
||||
for h, t := range w.pendingTasks {
|
||||
if t.block.NumberU64()+staleThreshold <= number {
|
||||
|
|
@ -594,7 +576,7 @@ func (w *worker) newWorkLoop(ctx context.Context, recommit time.Duration) {
|
|||
// the received event. It can support two modes: automatically generate task and
|
||||
// submit it or return task according to given parameters for various proposes.
|
||||
// nolint: gocognit, contextcheck
|
||||
func (w *worker) mainLoop(ctx context.Context) {
|
||||
func (w *worker) mainLoop() {
|
||||
defer w.wg.Done()
|
||||
defer w.txsSub.Unsubscribe()
|
||||
defer w.chainHeadSub.Unsubscribe()
|
||||
|
|
@ -610,15 +592,15 @@ func (w *worker) mainLoop(ctx context.Context) {
|
|||
if w.chainConfig.ChainID.Cmp(params.BorMainnetChainConfig.ChainID) == 0 || w.chainConfig.ChainID.Cmp(params.MumbaiChainConfig.ChainID) == 0 || w.chainConfig.ChainID.Cmp(params.AmoyChainConfig.ChainID) == 0 {
|
||||
if w.eth.PeerCount() > 0 {
|
||||
//nolint:contextcheck
|
||||
w.commitWork(req.ctx, req.interrupt, req.noempty, req.timestamp)
|
||||
w.commitWork(req.interrupt, req.noempty, req.timestamp)
|
||||
}
|
||||
} else {
|
||||
//nolint:contextcheck
|
||||
w.commitWork(req.ctx, req.interrupt, req.noempty, req.timestamp)
|
||||
w.commitWork(req.interrupt, req.noempty, req.timestamp)
|
||||
}
|
||||
|
||||
case req := <-w.getWorkCh:
|
||||
req.result <- w.generateWork(ctx, req.params)
|
||||
req.result <- w.generateWork(req.params)
|
||||
|
||||
case ev := <-w.txsCh:
|
||||
// Apply transactions to the pending state if we're not sealing
|
||||
|
|
@ -663,7 +645,7 @@ func (w *worker) mainLoop(ctx context.Context) {
|
|||
// 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(ctx, nil, true, time.Now().Unix())
|
||||
w.commitWork(nil, true, time.Now().Unix())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -722,7 +704,7 @@ func (w *worker) taskLoop() {
|
|||
w.pendingTasks[sealHash] = task
|
||||
w.pendingMu.Unlock()
|
||||
|
||||
if err := w.engine.Seal(task.ctx, w.chain, task.block, w.resultCh, stopCh); err != nil {
|
||||
if err := w.engine.Seal(w.chain, task.block, w.resultCh, stopCh); err != nil {
|
||||
log.Warn("Block sealing failed", "err", err)
|
||||
w.pendingMu.Lock()
|
||||
delete(w.pendingTasks, sealHash)
|
||||
|
|
@ -783,45 +765,31 @@ func (w *worker) resultLoop() {
|
|||
err error
|
||||
)
|
||||
|
||||
tracing.Exec(task.ctx, "", "resultLoop", func(ctx context.Context, span trace.Span) {
|
||||
for i, taskReceipt := range task.receipts {
|
||||
receipt := new(types.Receipt)
|
||||
receipts[i] = receipt
|
||||
*receipt = *taskReceipt
|
||||
for i, taskReceipt := range task.receipts {
|
||||
receipt := new(types.Receipt)
|
||||
receipts[i] = receipt
|
||||
*receipt = *taskReceipt
|
||||
|
||||
// add block location fields
|
||||
receipt.BlockHash = hash
|
||||
receipt.BlockNumber = block.Number()
|
||||
receipt.TransactionIndex = uint(i)
|
||||
// add block location fields
|
||||
receipt.BlockHash = hash
|
||||
receipt.BlockNumber = block.Number()
|
||||
receipt.TransactionIndex = uint(i)
|
||||
|
||||
// Update the block hash in all logs since it is now available and not when the
|
||||
// receipt/log of individual transactions were created.
|
||||
receipt.Logs = make([]*types.Log, len(taskReceipt.Logs))
|
||||
// Update the block hash in all logs since it is now available and not when the
|
||||
// receipt/log of individual transactions were created.
|
||||
receipt.Logs = make([]*types.Log, len(taskReceipt.Logs))
|
||||
|
||||
for i, taskLog := range taskReceipt.Logs {
|
||||
log := new(types.Log)
|
||||
receipt.Logs[i] = log
|
||||
*log = *taskLog
|
||||
log.BlockHash = hash
|
||||
}
|
||||
|
||||
logs = append(logs, receipt.Logs...)
|
||||
for i, taskLog := range taskReceipt.Logs {
|
||||
log := new(types.Log)
|
||||
receipt.Logs[i] = log
|
||||
*log = *taskLog
|
||||
log.BlockHash = hash
|
||||
}
|
||||
// Commit block and state to database.
|
||||
tracing.Exec(ctx, "", "resultLoop.WriteBlockAndSetHead", func(ctx context.Context, span trace.Span) {
|
||||
_, err = w.chain.WriteBlockAndSetHead(ctx, block, receipts, logs, task.state, true)
|
||||
})
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.String("hash", hash.String()),
|
||||
attribute.Int("number", int(block.Number().Uint64())),
|
||||
attribute.Int("txns", block.Transactions().Len()),
|
||||
attribute.Int("gas used", int(block.GasUsed())),
|
||||
attribute.Int("elapsed", int(time.Since(task.createdAt).Milliseconds())),
|
||||
attribute.Bool("error", err != nil),
|
||||
)
|
||||
})
|
||||
logs = append(logs, receipt.Logs...)
|
||||
}
|
||||
// Commit block and state to database.
|
||||
_, err = w.chain.WriteBlockAndSetHead(block, receipts, logs, task.state, true)
|
||||
|
||||
if err != nil {
|
||||
log.Error("Failed writing block to chain", "err", err)
|
||||
|
|
@ -1301,80 +1269,13 @@ func (w *worker) prepareWork(genParams *generateParams) (*environment, error) {
|
|||
return env, nil
|
||||
}
|
||||
|
||||
func startProfiler(profile string, filepath string, number uint64) (func() error, error) {
|
||||
var (
|
||||
buf bytes.Buffer
|
||||
err error
|
||||
)
|
||||
|
||||
closeFn := func() {}
|
||||
|
||||
switch profile {
|
||||
case "cpu":
|
||||
err = pprof.StartCPUProfile(&buf)
|
||||
|
||||
if err == nil {
|
||||
closeFn = func() {
|
||||
pprof.StopCPUProfile()
|
||||
}
|
||||
}
|
||||
case "trace":
|
||||
err = ptrace.Start(&buf)
|
||||
|
||||
if err == nil {
|
||||
closeFn = func() {
|
||||
ptrace.Stop()
|
||||
}
|
||||
}
|
||||
case "heap":
|
||||
runtime.GC()
|
||||
|
||||
err = pprof.WriteHeapProfile(&buf)
|
||||
default:
|
||||
log.Info("Incorrect profile name")
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return func() error {
|
||||
closeFn()
|
||||
return nil
|
||||
}, err
|
||||
}
|
||||
|
||||
closeFnNew := func() error {
|
||||
var err error
|
||||
|
||||
closeFn()
|
||||
|
||||
if buf.Len() == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
f, err := os.Create(filepath + "/" + profile + "-" + fmt.Sprint(number) + ".prof")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
defer f.Close()
|
||||
|
||||
_, err = f.Write(buf.Bytes())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return closeFnNew, nil
|
||||
}
|
||||
|
||||
// fillTransactions retrieves the pending transactions from the txpool and fills them
|
||||
// into the given sealing block. The transaction selection and ordering strategy can
|
||||
// be customized with the plugin in the future.
|
||||
|
||||
//
|
||||
//nolint:gocognit
|
||||
func (w *worker) fillTransactions(ctx context.Context, interrupt *atomic.Int32, env *environment, interruptCtx context.Context) error {
|
||||
ctx, span := tracing.StartSpan(ctx, "fillTransactions")
|
||||
defer tracing.EndSpan(span)
|
||||
|
||||
func (w *worker) fillTransactions(interrupt *atomic.Int32, env *environment, interruptCtx context.Context) error {
|
||||
w.mu.RLock()
|
||||
tip := w.tip
|
||||
w.mu.RUnlock()
|
||||
|
|
@ -1392,179 +1293,59 @@ func (w *worker) fillTransactions(ctx context.Context, interrupt *atomic.Int32,
|
|||
filter.BlobFee = uint256.MustFromBig(eip4844.CalcBlobFee(*env.header.ExcessBlobGas))
|
||||
}
|
||||
|
||||
var (
|
||||
localPlainTxsCount int
|
||||
remotePlainTxsCount int
|
||||
)
|
||||
|
||||
// TODO: move to config or RPC
|
||||
const profiling = false
|
||||
|
||||
if profiling {
|
||||
doneCh := make(chan struct{})
|
||||
|
||||
defer func() {
|
||||
close(doneCh)
|
||||
}()
|
||||
|
||||
go func(number uint64) {
|
||||
closeFn := func() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
// Check if we've not crossed limit
|
||||
if attempt := atomic.AddInt32(w.profileCount, 1); attempt >= 10 {
|
||||
log.Info("Completed profiling", "attempt", attempt)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Starting profiling in fill transactions", "number", number)
|
||||
|
||||
dir, err := os.MkdirTemp("", fmt.Sprintf("bor-traces-%s-", time.Now().UTC().Format("2006-01-02-150405Z")))
|
||||
if err != nil {
|
||||
log.Error("Error in profiling", "path", dir, "number", number, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
// grab the cpu profile
|
||||
closeFnInternal, err := startProfiler("cpu", dir, number)
|
||||
if err != nil {
|
||||
log.Error("Error in profiling", "path", dir, "number", number, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
closeFn = func() error {
|
||||
err := closeFnInternal()
|
||||
|
||||
log.Info("Completed profiling", "path", dir, "number", number, "error", err)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
case <-doneCh:
|
||||
err := closeFn()
|
||||
|
||||
if err != nil {
|
||||
log.Info("closing fillTransactions", "number", number, "error", err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
}(env.header.Number.Uint64())
|
||||
}
|
||||
|
||||
var (
|
||||
localPlainTxs, remotePlainTxs, localBlobTxs, remoteBlobTxs map[common.Address][]*txpool.LazyTransaction
|
||||
)
|
||||
|
||||
tracing.Exec(ctx, "", "worker.SplittingTransactions", func(ctx context.Context, span trace.Span) {
|
||||
prePendingTime := time.Now()
|
||||
filter.OnlyPlainTxs, filter.OnlyBlobTxs = true, false
|
||||
pendingPlainTxs := w.eth.TxPool().Pending(filter)
|
||||
|
||||
filter.OnlyPlainTxs, filter.OnlyBlobTxs = true, false
|
||||
pendingPlainTxs := w.eth.TxPool().Pending(filter)
|
||||
filter.OnlyPlainTxs, filter.OnlyBlobTxs = false, true
|
||||
pendingBlobTxs := w.eth.TxPool().Pending(filter)
|
||||
|
||||
filter.OnlyPlainTxs, filter.OnlyBlobTxs = false, true
|
||||
pendingBlobTxs := w.eth.TxPool().Pending(filter)
|
||||
// Split the pending transactions into locals and remotes.
|
||||
localPlainTxs, remotePlainTxs = make(map[common.Address][]*txpool.LazyTransaction), pendingPlainTxs
|
||||
localBlobTxs, remoteBlobTxs = make(map[common.Address][]*txpool.LazyTransaction), pendingBlobTxs
|
||||
|
||||
// Split the pending transactions into locals and remotes.
|
||||
localPlainTxs, remotePlainTxs = make(map[common.Address][]*txpool.LazyTransaction), pendingPlainTxs
|
||||
localBlobTxs, remoteBlobTxs = make(map[common.Address][]*txpool.LazyTransaction), pendingBlobTxs
|
||||
|
||||
postPendingTime := time.Now()
|
||||
|
||||
for _, account := range w.eth.TxPool().Locals() {
|
||||
if txs := remotePlainTxs[account]; len(txs) > 0 {
|
||||
delete(remotePlainTxs, account)
|
||||
localPlainTxs[account] = txs
|
||||
}
|
||||
if txs := remoteBlobTxs[account]; len(txs) > 0 {
|
||||
delete(remoteBlobTxs, account)
|
||||
localBlobTxs[account] = txs
|
||||
}
|
||||
for _, account := range w.eth.TxPool().Locals() {
|
||||
if txs := remotePlainTxs[account]; len(txs) > 0 {
|
||||
delete(remotePlainTxs, account)
|
||||
localPlainTxs[account] = txs
|
||||
}
|
||||
|
||||
postLocalsTime := time.Now()
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of local txs", localPlainTxsCount),
|
||||
attribute.Int("len of remote txs", remotePlainTxsCount),
|
||||
attribute.String("time taken by Pending()", fmt.Sprintf("%v", postPendingTime.Sub(prePendingTime))),
|
||||
attribute.String("time taken by Locals()", fmt.Sprintf("%v", postLocalsTime.Sub(postPendingTime))),
|
||||
)
|
||||
})
|
||||
|
||||
var (
|
||||
localEnvTCount int
|
||||
remoteEnvTCount int
|
||||
err error
|
||||
)
|
||||
if txs := remoteBlobTxs[account]; len(txs) > 0 {
|
||||
delete(remoteBlobTxs, account)
|
||||
localBlobTxs[account] = txs
|
||||
}
|
||||
}
|
||||
|
||||
// Fill the block with all available pending transactions.
|
||||
if len(localPlainTxs) > 0 || len(localBlobTxs) > 0 {
|
||||
var plainTxs, blobTxs *transactionsByPriceAndNonce
|
||||
|
||||
tracing.Exec(ctx, "", "worker.LocalTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) {
|
||||
plainTxs = newTransactionsByPriceAndNonce(env.signer, localPlainTxs, env.header.BaseFee)
|
||||
blobTxs = newTransactionsByPriceAndNonce(env.signer, localBlobTxs, env.header.BaseFee)
|
||||
plainTxs = newTransactionsByPriceAndNonce(env.signer, localPlainTxs, env.header.BaseFee)
|
||||
blobTxs = newTransactionsByPriceAndNonce(env.signer, localBlobTxs, env.header.BaseFee)
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of tx local Heads", plainTxs.GetTxs()),
|
||||
)
|
||||
})
|
||||
|
||||
tracing.Exec(ctx, "", "worker.LocalCommitTransactions", func(ctx context.Context, span trace.Span) {
|
||||
err = w.commitTransactions(env, plainTxs, blobTxs, interrupt, new(uint256.Int), interruptCtx)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
if err := w.commitTransactions(env, plainTxs, blobTxs, interrupt, new(uint256.Int), interruptCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
localEnvTCount = env.tcount
|
||||
}
|
||||
|
||||
if len(remotePlainTxs) > 0 || len(remoteBlobTxs) > 0 {
|
||||
var plainTxs, blobTxs *transactionsByPriceAndNonce
|
||||
|
||||
tracing.Exec(ctx, "", "worker.RemoteTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) {
|
||||
plainTxs = newTransactionsByPriceAndNonce(env.signer, remotePlainTxs, env.header.BaseFee)
|
||||
blobTxs = newTransactionsByPriceAndNonce(env.signer, remoteBlobTxs, env.header.BaseFee)
|
||||
plainTxs = newTransactionsByPriceAndNonce(env.signer, remotePlainTxs, env.header.BaseFee)
|
||||
blobTxs = newTransactionsByPriceAndNonce(env.signer, remoteBlobTxs, env.header.BaseFee)
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of tx remote Heads", plainTxs.GetTxs()),
|
||||
)
|
||||
})
|
||||
|
||||
tracing.Exec(ctx, "", "worker.RemoteCommitTransactions", func(ctx context.Context, span trace.Span) {
|
||||
err = w.commitTransactions(env, plainTxs, blobTxs, interrupt, new(uint256.Int), interruptCtx)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
if err := w.commitTransactions(env, plainTxs, blobTxs, interrupt, new(uint256.Int), interruptCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
remoteEnvTCount = env.tcount
|
||||
}
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("len of final local txs ", localEnvTCount),
|
||||
attribute.Int("len of final remote txs", remoteEnvTCount),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// generateWork generates a sealing block based on the given parameters.
|
||||
func (w *worker) generateWork(ctx context.Context, params *generateParams) *newPayloadResult {
|
||||
func (w *worker) generateWork(params *generateParams) *newPayloadResult {
|
||||
work, err := w.prepareWork(params)
|
||||
if err != nil {
|
||||
return &newPayloadResult{err: err}
|
||||
|
|
@ -1582,7 +1363,7 @@ func (w *worker) generateWork(ctx context.Context, params *generateParams) *newP
|
|||
})
|
||||
defer timer.Stop()
|
||||
|
||||
err := w.fillTransactions(ctx, interrupt, work, interruptCtx)
|
||||
err := w.fillTransactions(interrupt, work, interruptCtx)
|
||||
if errors.Is(err, errBlockInterruptedByTimeout) {
|
||||
log.Warn("Block building is interrupted", "allowance", common.PrettyDuration(w.newpayloadTimeout))
|
||||
}
|
||||
|
|
@ -1605,7 +1386,7 @@ func (w *worker) generateWork(ctx context.Context, params *generateParams) *newP
|
|||
|
||||
// commitWork generates several new sealing tasks based on the parent block
|
||||
// and submit them to the sealer.
|
||||
func (w *worker) commitWork(ctx context.Context, interrupt *atomic.Int32, noempty bool, timestamp int64) {
|
||||
func (w *worker) commitWork(interrupt *atomic.Int32, noempty bool, timestamp int64) {
|
||||
// Abort committing if node is still syncing
|
||||
if w.syncing.Load() {
|
||||
return
|
||||
|
|
@ -1617,23 +1398,20 @@ func (w *worker) commitWork(ctx context.Context, interrupt *atomic.Int32, noempt
|
|||
err error
|
||||
)
|
||||
|
||||
tracing.Exec(ctx, "", "worker.prepareWork", func(ctx context.Context, span trace.Span) {
|
||||
// Set the coinbase if the worker is running or it's required
|
||||
var coinbase common.Address
|
||||
if w.IsRunning() {
|
||||
coinbase = w.etherbase()
|
||||
if coinbase == (common.Address{}) {
|
||||
log.Error("Refusing to mine without etherbase")
|
||||
return
|
||||
}
|
||||
// Set the coinbase if the worker is running or it's required
|
||||
var coinbase common.Address
|
||||
if w.IsRunning() {
|
||||
coinbase = w.etherbase()
|
||||
if coinbase == (common.Address{}) {
|
||||
log.Error("Refusing to mine without etherbase")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
work, err = w.prepareWork(&generateParams{
|
||||
timestamp: uint64(timestamp),
|
||||
coinbase: coinbase,
|
||||
})
|
||||
work, err = w.prepareWork(&generateParams{
|
||||
timestamp: uint64(timestamp),
|
||||
coinbase: coinbase,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -1648,26 +1426,18 @@ func (w *worker) commitWork(ctx context.Context, interrupt *atomic.Int32, noempt
|
|||
|
||||
if !noempty && w.interruptCommitFlag {
|
||||
block := w.chain.GetBlockByHash(w.chain.CurrentBlock().Hash())
|
||||
interruptCtx, stopFn = getInterruptTimer(ctx, work, block)
|
||||
interruptCtx, stopFn = getInterruptTimer(work, block)
|
||||
// nolint : staticcheck
|
||||
interruptCtx = vm.PutCache(interruptCtx, w.interruptedTxCache)
|
||||
}
|
||||
|
||||
ctx, span := tracing.StartSpan(ctx, "commitWork")
|
||||
defer tracing.EndSpan(span)
|
||||
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("number", int(work.header.Number.Uint64())),
|
||||
)
|
||||
|
||||
// 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(ctx, work.copy(), nil, false, start)
|
||||
_ = w.commit(work.copy(), nil, false, start)
|
||||
}
|
||||
// Fill pending transactions from the txpool into the block.
|
||||
err = w.fillTransactions(ctx, interrupt, work, interruptCtx)
|
||||
err = w.fillTransactions(interrupt, work, interruptCtx)
|
||||
|
||||
switch {
|
||||
case err == nil:
|
||||
|
|
@ -1698,7 +1468,7 @@ func (w *worker) commitWork(ctx context.Context, interrupt *atomic.Int32, noempt
|
|||
return
|
||||
}
|
||||
// Submit the generated block for consensus sealing.
|
||||
_ = w.commit(ctx, work.copy(), w.fullTaskHook, true, start)
|
||||
_ = w.commit(work.copy(), w.fullTaskHook, true, start)
|
||||
|
||||
// Swap out the old work with the new one, terminating any leftover
|
||||
// prefetcher processes in the mean time and starting a new one.
|
||||
|
|
@ -1709,7 +1479,7 @@ func (w *worker) commitWork(ctx context.Context, interrupt *atomic.Int32, noempt
|
|||
w.current = work
|
||||
}
|
||||
|
||||
func getInterruptTimer(ctx context.Context, work *environment, current *types.Block) (context.Context, func()) {
|
||||
func getInterruptTimer(work *environment, current *types.Block) (context.Context, func()) {
|
||||
delay := time.Until(time.Unix(int64(work.header.Time), 0))
|
||||
|
||||
interruptCtx, cancel := context.WithTimeout(context.Background(), delay)
|
||||
|
|
@ -1717,13 +1487,10 @@ func getInterruptTimer(ctx context.Context, work *environment, current *types.Bl
|
|||
blockNumber := current.NumberU64() + 1
|
||||
|
||||
go func() {
|
||||
select {
|
||||
case <-interruptCtx.Done():
|
||||
if interruptCtx.Err() != context.Canceled {
|
||||
log.Info("Commit Interrupt. Pre-committing the current block", "block", blockNumber)
|
||||
cancel()
|
||||
}
|
||||
case <-ctx.Done(): // nothing to do
|
||||
<-interruptCtx.Done()
|
||||
if interruptCtx.Err() != context.Canceled {
|
||||
log.Info("Commit Interrupt. Pre-committing the current block", "block", blockNumber)
|
||||
cancel()
|
||||
}
|
||||
}()
|
||||
|
||||
|
|
@ -1734,11 +1501,8 @@ func getInterruptTimer(ctx context.Context, work *environment, current *types.Bl
|
|||
// and commits new work if consensus engine is running.
|
||||
// Note the assumption is held that the mutation is allowed to the passed env, do
|
||||
// the deep copy first.
|
||||
func (w *worker) commit(ctx context.Context, env *environment, interval func(), update bool, start time.Time) error {
|
||||
func (w *worker) commit(env *environment, interval func(), update bool, start time.Time) error {
|
||||
if w.IsRunning() {
|
||||
ctx, span := tracing.StartSpan(ctx, "commit")
|
||||
defer tracing.EndSpan(span)
|
||||
|
||||
if interval != nil {
|
||||
interval()
|
||||
}
|
||||
|
|
@ -1749,21 +1513,13 @@ func (w *worker) commit(ctx context.Context, env *environment, interval func(),
|
|||
block, err := w.engine.FinalizeAndAssemble(w.chain, env.header, env.state, &types.Body{
|
||||
Transactions: env.txs,
|
||||
}, env.receipts)
|
||||
tracing.SetAttributes(
|
||||
span,
|
||||
attribute.Int("number", int(env.header.Number.Uint64())),
|
||||
attribute.String("hash", env.header.Hash().String()),
|
||||
attribute.String("sealhash", w.engine.SealHash(env.header).String()),
|
||||
attribute.Int("len of env.txs", len(env.txs)),
|
||||
attribute.Bool("error", err != nil),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
select {
|
||||
case w.taskCh <- &task{ctx: ctx, receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}:
|
||||
case w.taskCh <- &task{receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}:
|
||||
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()),
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@
|
|||
package bor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ecdsa"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
|
|
@ -219,8 +218,6 @@ func buildNextBlock(t *testing.T, _bor consensus.Engine, chain *core.BlockChain,
|
|||
b.addTxWithChain(chain, state, tx, addr)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Finalize and seal the block
|
||||
block, err := _bor.FinalizeAndAssemble(chain, b.header, state, &types.Body{
|
||||
Transactions: b.txs,
|
||||
|
|
@ -242,7 +239,7 @@ func buildNextBlock(t *testing.T, _bor consensus.Engine, chain *core.BlockChain,
|
|||
|
||||
res := make(chan *types.Block, 1)
|
||||
|
||||
err = _bor.Seal(ctx, chain, block, res, nil)
|
||||
err = _bor.Seal(chain, block, res, nil)
|
||||
if err != nil {
|
||||
// an error case - sign manually
|
||||
sign(t, header, signer, borConfig)
|
||||
|
|
|
|||
Loading…
Reference in a new issue