diff --git a/miner/test_backend.go b/miner/test_backend.go index 5c5c217424..0b8a995c72 100644 --- a/miner/test_backend.go +++ b/miner/test_backend.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "os" + "sync" "sync/atomic" "time" @@ -15,12 +16,15 @@ import ( "github.com/ethereum/go-ethereum/common/tracing" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/blockstm" + "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" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" + "github.com/ethereum/go-ethereum/rlp" "github.com/holiman/uint256" "go.opentelemetry.io/otel" @@ -38,19 +42,20 @@ func newWorkerWithDelay(config *Config, chainConfig *params.ChainConfig, engine chainConfig: chainConfig, engine: engine, eth: eth, - mux: mux, chain: eth.BlockChain(), + mux: mux, isLocalBlock: isLocalBlock, + 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), resultCh: make(chan *types.Block, resultQueueSize), - exitCh: make(chan struct{}), startCh: make(chan struct{}, 1), + exitCh: make(chan struct{}), resubmitIntervalCh: make(chan time.Duration), resubmitAdjustCh: make(chan *intervalAdjust, resubmitAdjustChanSize), interruptCommitFlag: config.CommitInterruptFlag, @@ -61,7 +66,6 @@ func newWorkerWithDelay(config *Config, chainConfig *params.ChainConfig, engine 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) interruptedTxCache, err := lru.New(vm.InterruptedTxCacheSize) if err != nil { @@ -83,6 +87,21 @@ func newWorkerWithDelay(config *Config, chainConfig *params.ChainConfig, engine recommit = minRecommitInterval } + worker.recommit = recommit + + // Sanitize the timeout config for creating payload. + newpayloadTimeout := worker.config.NewPayloadTimeout + if newpayloadTimeout == 0 { + log.Warn("Sanitizing new payload timeout to default", "provided", newpayloadTimeout, "updated", DefaultConfig.NewPayloadTimeout) + newpayloadTimeout = DefaultConfig.NewPayloadTimeout + } + + if newpayloadTimeout < time.Millisecond*100 { + log.Warn("Low payload timeout may cause high amount of non-full blocks", "provided", newpayloadTimeout, "default", DefaultConfig.NewPayloadTimeout) + } + + worker.newpayloadTimeout = newpayloadTimeout + ctx := tracing.WithTracer(context.Background(), otel.GetTracerProvider().Tracer("MinerWorker")) worker.wg.Add(4) @@ -106,35 +125,31 @@ func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay 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: - i := req.interrupt.Load() - //nolint:contextcheck - w.commitWorkWithDelay(req.ctx, &i, req.noempty, req.timestamp, delay, 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) + } + } else { + //nolint:contextcheck + w.commitWorkWithDelay(req.ctx, req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay) + } case req := <-w.getWorkCh: - //nolint:contextcheck - block, _, err := w.generateWork(req.ctx, req.params) - if err != nil { - req.result <- nil - } else { - payload := newPayloadResult{ - err: nil, - block: block, - fees: block.BaseFee(), - } - req.result <- &payload + block, fees, err := w.generateWork(req.ctx, req.params) + req.result <- &newPayloadResult{ + err: err, + block: block, + fees: fees, } case ev := <-w.txsCh: @@ -143,6 +158,7 @@ func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay // Note all transactions received may not be continuous with transactions // already included in the current sealing block. These transactions will // be automatically eliminated. + // nolint : nestif if !w.IsRunning() && w.current != nil { // If block is already full, abort if gp := w.current.gasPool; gp != nil && gp.Gas() < params.TxGas { @@ -186,14 +202,16 @@ func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay return case <-w.chainHeadSub.Err(): return - case <-w.chainSideSub.Err(): - return } } } // commitWorkWithDelay is commitWork() with extra params to induce artficial delays for tests such as commit-interrupt. -func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *int32, noempty bool, timestamp int64, delay uint, opcodeDelay uint) { +func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int32, noempty bool, timestamp int64, delay uint, opcodeDelay uint) { + // Abort committing if node is still syncing + if w.syncing.Load() { + return + } start := time.Now() var ( @@ -205,12 +223,11 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *int32, noem // Set the coinbase if the worker is running or it's required var coinbase common.Address if w.IsRunning() { - if w.coinbase == (common.Address{}) { + coinbase = w.etherbase() + if coinbase == (common.Address{}) { log.Error("Refusing to mine without etherbase") return } - - coinbase = w.coinbase // Use the preset address as the fee recipient } work, err = w.prepareWork(&generateParams{ @@ -223,7 +240,7 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *int32, noem return } - //nolint:contextcheck + // nolint:contextcheck var interruptCtx = context.Background() stopFn := func() {} @@ -253,19 +270,41 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *int32, noem // Create an empty block based on temporary copied state for // sealing in advance without waiting block execution finished. if !noempty && !w.noempty.Load() { - err = w.commit(ctx, work.copy(), nil, false, start) - if err != nil { - return - } + _ = w.commit(ctx, work.copy(), nil, false, start) } + // Fill pending transactions from the txpool into the block. + err = w.fillTransactionsWithDelay(ctx, interrupt, work, interruptCtx) - // Fill pending transactions from the txpool - w.fillTransactionsWithDelay(ctx, interrupt, work, interruptCtx) + switch { + case err == nil: + // The entire block is filled, decrease resubmit interval in case + // of current interval is larger than the user-specified one. + w.resubmitAdjustCh <- &intervalAdjust{inc: false} - err = w.commit(ctx, work.copy(), w.fullTaskHook, true, start) - if err != nil { + case errors.Is(err, errBlockInterruptedByRecommit): + // Notify resubmit loop to increase resubmitting interval if the + // interruption is due to frequent commits. + gaslimit := work.header.GasLimit + + ratio := float64(gaslimit-work.gasPool.Gas()) / float64(gaslimit) + if ratio < 0.1 { + ratio = 0.1 + } + w.resubmitAdjustCh <- &intervalAdjust{ + ratio: ratio, + inc: true, + } + + case errors.Is(err, errBlockInterruptedByNewHead): + // If the block building is interrupted by newhead event, discard it + // totally. Committing the interrupted block introduces unnecessary + // delay, and possibly causes miner to mine on the previous head, + // which could result in higher uncle rate. + work.discard() return } + // Submit the generated block for consensus sealing. + _ = w.commit(ctx, 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. @@ -278,21 +317,20 @@ func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *int32, noem // 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 *int32, env *environment, interruptCtx context.Context) { +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) // Split the pending transactions into locals and remotes // Fill the block with all available pending transactions. + pending := w.eth.TxPool().Pending(true) + localTxs, remoteTxs := make(map[common.Address][]*txpool.LazyTransaction), pending var ( localTxsCount int remoteTxsCount int ) - pending := w.eth.TxPool().Pending(true) - localTxs, remoteTxs := make(map[common.Address][]*txpool.LazyTransaction), pending - // TODO: move to config or RPC const profiling = false @@ -384,10 +422,10 @@ func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *int32 var ( localEnvTCount int remoteEnvTCount int - committed bool + err error ) - if localTxsCount > 0 { + if len(localTxs) > 0 { var txs *transactionsByPriceAndNonce tracing.Exec(ctx, "", "worker.LocalTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) { @@ -396,7 +434,7 @@ func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *int32 baseFee = cmath.FromBig(env.header.BaseFee) } - txs := newTransactionsByPriceAndNonce(env.signer, localTxs, baseFee.ToBig()) + txs = newTransactionsByPriceAndNonce(env.signer, localTxs, baseFee.ToBig()) tracing.SetAttributes( span, @@ -405,17 +443,17 @@ func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *int32 }) tracing.Exec(ctx, "", "worker.LocalCommitTransactions", func(ctx context.Context, span trace.Span) { - committed = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx) + err = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx) }) - if committed { - return + if err != nil { + return err } localEnvTCount = env.tcount } - if remoteTxsCount > 0 { + if len(remoteTxs) > 0 { var txs *transactionsByPriceAndNonce tracing.Exec(ctx, "", "worker.RemoteTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) { @@ -433,11 +471,11 @@ func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *int32 }) tracing.Exec(ctx, "", "worker.RemoteCommitTransactions", func(ctx context.Context, span trace.Span) { - committed = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx) + err = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx) }) - if committed { - return + if err != nil { + return err } remoteEnvTCount = env.tcount @@ -448,11 +486,13 @@ func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *int32 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 *int32, interruptCtx context.Context) bool { +func (w *worker) commitTransactionsWithDelay(env *environment, txs *transactionsByPriceAndNonce, interrupt *atomic.Int32, interruptCtx context.Context) error { gasLimit := env.header.GasLimit if env.gasPool == nil { env.gasPool = new(core.GasPool).AddGas(gasLimit) @@ -460,7 +500,49 @@ 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 + + EnableMVHashMap := false + + // create and add empty mvHashMap in statedb + if EnableMVHashMap { + depsMVReadList = [][]blockstm.ReadDescriptor{} + + depsMVFullWriteList = [][]blockstm.WriteDescriptor{} + + mvReadMapList = []map[blockstm.Key]blockstm.ReadDescriptor{} + + deps = map[int]map[int]bool{} + + chDeps = make(chan blockstm.TxDep) + + count = 0 + + depsWg.Add(1) + + go func(chDeps chan blockstm.TxDep) { + for t := range chDeps { + deps = blockstm.UpdateDeps(deps, t) + } + + depsWg.Done() + }(chDeps) + } + initialGasLimit := env.gasPool.Gas() + initialTxs := txs.GetTxs() var breakCause string @@ -479,6 +561,10 @@ func (w *worker) commitTransactionsWithDelay(env *environment, txs *transactions mainloop: for { if interruptCtx != nil { + if EnableMVHashMap { + env.state.AddEmptyMVHashMap() + } + // case of interrupting by timeout select { case <-interruptCtx.Done(): @@ -489,29 +575,12 @@ mainloop: } } - // In the following three cases, we will interrupt the execution of the transaction. - // (1) new head block event arrival, the interrupt signal is 1 - // (2) worker start or restart, the interrupt signal is 1 - // (3) worker recreate the sealing block with any newly arrived transactions, the interrupt signal is 2. - // For the first two cases, the semi-finished work will be discarded. - // For the third case, the semi-finished work will be submitted to the consensus engine. - if interrupt != nil && atomic.LoadInt32(interrupt) != commitInterruptNone { - // Notify resubmit loop to increase resubmitting interval due to too frequent commits. - if atomic.LoadInt32(interrupt) == commitInterruptResubmit { - ratio := float64(gasLimit-env.gasPool.Gas()) / float64(gasLimit) - if ratio < 0.1 { - // nolint:goconst - ratio = 0.1 - } - w.resubmitAdjustCh <- &intervalAdjust{ - ratio: ratio, - inc: true, - } + // Check interruption signal and abort building if it's fired. + if interrupt != nil { + if signal := interrupt.Load(); signal != commitInterruptNone { + breakCause = "interrupt" + return signalToErr(signal) } - // nolint:goconst - breakCause = "interrupt" - - return atomic.LoadInt32(interrupt) == commitInterruptNewHead } // If we don't have enough gas for any further transactions then we're done. if env.gasPool.Gas() < params.TxGas { @@ -536,6 +605,31 @@ mainloop: // during transaction acceptance is the transaction pool. from, _ := types.Sender(env.signer, tx.Tx) + // not prioritising conditional transaction, yet. + //nolint:nestif + if options := tx.Tx.GetOptions(); options != nil { + if err := env.header.ValidateBlockNumberOptions4337(options.BlockNumberMin, options.BlockNumberMax); err != nil { + log.Trace("Dropping conditional transaction", "from", from, "hash", tx.Tx.Hash(), "reason", err) + txs.Pop() + + continue + } + + if err := env.header.ValidateTimestampOptions4337(options.TimestampMin, options.TimestampMax); err != nil { + log.Trace("Dropping conditional transaction", "from", from, "hash", tx.Tx.Hash(), "reason", err) + txs.Pop() + + continue + } + + if err := env.state.ValidateKnownAccounts(options.KnownAccounts); err != nil { + log.Trace("Dropping conditional transaction", "from", from, "hash", tx.Tx.Hash(), "reason", err) + txs.Pop() + + continue + } + } + // Check whether the tx is replay protected. If we're not in the EIP155 hf // phase, start ignoring the sender until we do. if tx.Tx.Protected() && !w.chainConfig.IsEIP155(env.header.Number) { @@ -573,6 +667,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()) + + temp := blockstm.TxDep{ + Index: env.tcount - 1, + ReadList: depsMVReadList[count], + FullWriteList: depsMVFullWriteList, + } + + chDeps <- temp + count++ + } + txs.Shift() log.OnDebug(func(lg log.Logging) { @@ -585,6 +694,72 @@ mainloop: log.Debug("Transaction failed, account skipped", "hash", tx.Tx.Hash(), "err", err) txs.Pop() } + + if EnableMVHashMap { + env.state.ClearReadMap() + env.state.ClearWriteMap() + } + } + + // nolint:nestif + if EnableMVHashMap && w.IsRunning() { + close(chDeps) + depsWg.Wait() + + var blockExtraData types.BlockExtraData + + 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)) + + for j := range deps[0] { + tempDeps[0] = append(tempDeps[0], uint64(j)) + } + + delayFlag := true + + for i := 1; i <= len(mvReadMapList)-1; i++ { + reads := 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 + } + + for j := range deps[i] { + tempDeps[i] = append(tempDeps[i], uint64(j)) + } + } + + if err := rlp.DecodeBytes(env.header.Extra[types.ExtraVanityLength:len(env.header.Extra)-types.ExtraSealLength], &blockExtraData); err != nil { + log.Error("error while decoding block extra data", "err", err) + return err + } + + if delayFlag { + blockExtraData.TxDependency = tempDeps + } else { + blockExtraData.TxDependency = nil + } + } else { + blockExtraData.TxDependency = nil + } + + blockExtraDataBytes, err := rlp.EncodeToBytes(blockExtraData) + if err != nil { + log.Error("error while encoding block extra data: %v", err) + return err + } + + env.header.Extra = []byte{} + + env.header.Extra = append(tempVanity, blockExtraDataBytes...) + + env.header.Extra = append(env.header.Extra, tempSeal...) } if !w.IsRunning() && len(coalescedLogs) > 0 { @@ -602,11 +777,6 @@ mainloop: w.pendingLogsFeed.Send(cpy) } - // Notify resubmit loop to decrease resubmitting interval if current interval is larger - // than the user-specified one. - if interrupt != nil { - w.resubmitAdjustCh <- &intervalAdjust{inc: false} - } - return false + return nil } diff --git a/miner/worker_test.go b/miner/worker_test.go index e137bf7667..58a5ff2948 100644 --- a/miner/worker_test.go +++ b/miner/worker_test.go @@ -175,8 +175,9 @@ var ( newTxs []*types.Transaction testConfig = &Config{ - Recommit: time.Second, - GasCeil: params.GenesisGasLimit, + Recommit: time.Second, + GasCeil: params.GenesisGasLimit, + CommitInterruptFlag: true, } )