diff --git a/miner/scroll_worker.go b/miner/scroll_worker.go index 321ce4c367..c5614c5981 100644 --- a/miner/scroll_worker.go +++ b/miner/scroll_worker.go @@ -43,9 +43,6 @@ import ( ) const ( - // resultQueueSize is the size of channel listening to sealing result. - resultQueueSize = 10 - // txChanSize is the size of channel listening to NewTxsEvent. // The number is referenced from the size of tx pool. txChanSize = 4096 @@ -56,15 +53,9 @@ const ( // chainSideChanSize is the size of channel listening to ChainSideEvent. chainSideChanSize = 10 - // miningLogAtDepth is the number of confirmations before logging successful mining. - miningLogAtDepth = 7 - // minRecommitInterval is the minimal time interval to recreate the mining block with // any newly arrived transactions. minRecommitInterval = 1 * time.Second - - // staleThreshold is the maximum depth of the acceptable stale block. - staleThreshold = 7 ) var ( @@ -80,23 +71,12 @@ var ( prepareTimer = metrics.NewRegisteredTimer("miner/prepare", nil) collectL2Timer = metrics.NewRegisteredTimer("miner/collect_l2_txns", nil) l2CommitTimer = metrics.NewRegisteredTimer("miner/commit", nil) - resultTimer = metrics.NewRegisteredTimer("miner/result", nil) commitReasonCCCCounter = metrics.NewRegisteredCounter("miner/commit_reason_ccc", nil) commitReasonDeadlineCounter = metrics.NewRegisteredCounter("miner/commit_reason_deadline", nil) commitGasCounter = metrics.NewRegisteredCounter("miner/commit_gas", nil) ) -// task contains all information for consensus engine sealing and result submitting. -type task struct { - receipts []*types.Receipt - state *state.StateDB - block *types.Block - createdAt time.Time - accRows *types.RowConsumption // accumulated row consumption in the circuit side - nextL1MsgIndex uint64 // next L1 queue index to be processed -} - // newWorkReq represents a request for new sealing work submitting with relative interrupt notifier. type newWorkReq struct { noempty bool @@ -133,8 +113,6 @@ type worker struct { // Channels newWorkCh chan *newWorkReq - taskCh chan *task - resultCh chan *types.Block startCh chan struct{} exitCh chan struct{} @@ -143,17 +121,10 @@ type worker struct { currentPipelineStart time.Time currentPipeline *pipeline.Pipeline - 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 extra []byte - pendingMu sync.RWMutex - pendingTasks map[common.Hash]*task - snapshotMu sync.RWMutex // The lock used to protect the snapshots below snapshotBlock *types.Block snapshotReceipts types.Receipts @@ -178,9 +149,7 @@ type worker struct { prioritizedTx *prioritizedTransaction // Test hooks - newTaskHook func(*task) // Method to call upon receiving a new sealing task. - skipSealHook func(*task) bool // Method to decide whether skipping the sealing. - beforeTxHook func() // Method to call before processing a transaction. + beforeTxHook func() // Method to call before processing a transaction. } func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(*types.Block) bool, init bool) *worker { @@ -192,16 +161,10 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus mux: mux, chain: eth.BlockChain(), isLocalBlock: isLocalBlock, - localUncles: make(map[common.Hash]*types.Block), - remoteUncles: make(map[common.Hash]*types.Block), - unconfirmed: newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth), - 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), - taskCh: make(chan *task), - resultCh: make(chan *types.Block, resultQueueSize), exitCh: make(chan struct{}), startCh: make(chan struct{}, 1), circuitCapacityChecker: circuitcapacitychecker.NewCircuitCapacityChecker(true), @@ -228,11 +191,9 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus worker.config.MaxAccountsNum = math.MaxInt } - worker.wg.Add(4) + worker.wg.Add(2) go worker.mainLoop() go worker.newWorkLoop(recommit) - go worker.resultLoop() - go worker.taskLoop() // Submit first work to initialize pending state. if init { @@ -345,25 +306,13 @@ func (w *worker) newWorkLoop(recommit time.Duration) { atomic.StoreInt32(&w.newTxs, 0) atomic.StoreInt32(&w.newL1Msgs, 0) } - // clearPending cleans the stale pending tasks. - clearPending := func(number uint64) { - w.pendingMu.Lock() - for h, t := range w.pendingTasks { - if t.block.NumberU64()+staleThreshold <= number { - delete(w.pendingTasks, h) - } - } - w.pendingMu.Unlock() - } for { select { case <-w.startCh: - clearPending(w.chain.CurrentBlock().NumberU64()) timestamp = time.Now().Unix() commit(false) - case head := <-w.chainHeadCh: - clearPending(head.Block.NumberU64()) + case <-w.chainHeadCh: timestamp = time.Now().Unix() commit(true) case <-w.exitCh: @@ -426,171 +375,6 @@ func (w *worker) mainLoop() { } } -// taskLoop is a standalone goroutine to fetch sealing task from the generator and -// push them to consensus engine. -func (w *worker) taskLoop() { - defer w.wg.Done() - var ( - stopCh chan struct{} - prev common.Hash - ) - - // interrupt aborts the in-flight sealing task. - interrupt := func() { - if stopCh != nil { - close(stopCh) - stopCh = nil - } - } - for { - select { - case task := <-w.taskCh: - if w.newTaskHook != nil { - w.newTaskHook(task) - } - // Reject duplicate sealing work due to resubmitting. - sealHash := w.engine.SealHash(task.block.Header()) - if sealHash == prev { - continue - } - // Interrupt previous sealing operation - interrupt() - stopCh, prev = make(chan struct{}), sealHash - - if w.skipSealHook != nil && w.skipSealHook(task) { - continue - } - w.pendingMu.Lock() - w.pendingTasks[sealHash] = task - w.pendingMu.Unlock() - - 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) - w.pendingMu.Unlock() - } - case <-w.exitCh: - interrupt() - return - } - } -} - -// resultLoop is a standalone goroutine to handle sealing result submitting -// and flush relative data to the database. -func (w *worker) resultLoop() { - defer w.wg.Done() - for { - select { - case block := <-w.resultCh: - // Short circuit when receiving empty result. - if block == nil { - continue - } - // Short circuit when receiving duplicate result caused by resubmitting. - if w.chain.HasBlock(block.Hash(), block.NumberU64()) { - continue - } - - var ( - sealhash = w.engine.SealHash(block.Header()) - hash = block.Hash() - ) - - w.pendingMu.RLock() - task, exist := w.pendingTasks[sealhash] - w.pendingMu.RUnlock() - - if !exist { - log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", hash) - continue - } - - startTime := time.Now() - - // Different block could share same sealhash, deep copy here to prevent write-write conflict. - var ( - receipts = make([]*types.Receipt, len(task.receipts)) - logs []*types.Log - ) - 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) - - // 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...) - } - // It's possible that we've stored L1 queue index for this block previously, - // in this case do not overwrite it. - if index := rawdb.ReadFirstQueueIndexNotInL2Block(w.eth.ChainDb(), hash); index == nil { - // Store first L1 queue index not processed by this block. - // Note: This accounts for both included and skipped messages. This - // way, if a block only skips messages, we won't reprocess the same - // messages from the next block. - log.Trace( - "Worker WriteFirstQueueIndexNotInL2Block", - "number", block.Number(), - "hash", hash.String(), - "task.nextL1MsgIndex", task.nextL1MsgIndex, - ) - rawdb.WriteFirstQueueIndexNotInL2Block(w.eth.ChainDb(), hash, task.nextL1MsgIndex) - } else { - log.Trace( - "Worker WriteFirstQueueIndexNotInL2Block: not overwriting existing index", - "number", block.Number(), - "hash", hash.String(), - "index", *index, - "task.nextL1MsgIndex", task.nextL1MsgIndex, - ) - } - // Store circuit row consumption. - log.Trace( - "Worker write block row consumption", - "id", w.circuitCapacityChecker.ID, - "number", block.Number(), - "hash", hash.String(), - "accRows", task.accRows, - ) - rawdb.WriteBlockRowConsumption(w.eth.ChainDb(), hash, task.accRows) - // Commit block and state to database. - _, err := w.chain.WriteBlockWithState(block, receipts, logs, task.state, true) - if err != nil { - resultTimer.Update(time.Since(startTime)) - log.Error("Failed writing block to chain", "err", err) - continue - } - log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", hash, - "elapsed", common.PrettyDuration(time.Since(task.createdAt))) - - // 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()) - - resultTimer.Update(time.Since(startTime)) - - case <-w.exitCh: - return - } - } -} - // updateSnapshot updates pending snapshot block and state. // Note this function assumes the current variable is thread safe. func (w *worker) updateSnapshot(current *pipeline.BlockCandidate) { @@ -826,11 +610,15 @@ func (w *worker) handlePipelineResult(res *pipeline.Result) error { return nil } - if res == nil || res.FinalBlock == nil { - w.startNewPipeline(time.Now().Unix()) - return nil + var commitError error + if res != nil && res.FinalBlock != nil { + if commitError = w.commit(res); commitError == nil { + return nil + } + log.Error("Commit failed", "header", res.FinalBlock.Header, "reason", commitError) } - return w.commit(res) + w.startNewPipeline(time.Now().Unix()) + return commitError } // commit runs any post-transaction state modifications, assembles the final block @@ -853,18 +641,85 @@ func (w *worker) commit(res *pipeline.Result) error { return err } - select { - case w.taskCh <- &task{receipts: res.FinalBlock.Receipts, state: res.FinalBlock.State, block: block, createdAt: time.Now(), - accRows: res.Rows, nextL1MsgIndex: res.FinalBlock.NextL1MsgIndex}: - w.unconfirmed.Shift(block.NumberU64() - 1) - log.Info("Commit new mining work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()), - "txs", res.FinalBlock.Txs.Len(), - "gas", block.GasUsed(), "fees", totalFees(block, res.FinalBlock.Receipts), - "elapsed", common.PrettyDuration(time.Since(w.currentPipelineStart))) - case <-w.exitCh: - log.Info("Worker has exited") + sealHash := w.engine.SealHash(block.Header()) + log.Info("Committing new mining work", "number", block.Number(), "sealhash", sealHash, + "txs", res.FinalBlock.Txs.Len(), + "gas", block.GasUsed(), "fees", totalFees(block, res.FinalBlock.Receipts), + "elapsed", common.PrettyDuration(time.Since(w.currentPipelineStart))) + + resultCh, stopCh := make(chan *types.Block), make(chan struct{}) + if err := w.engine.Seal(w.chain, block, resultCh, stopCh); err != nil { + return err } + block = <-resultCh + + // verify the generated block with local consensus engine to make sure everything is as expected + if err = w.engine.VerifyHeader(w.chain, block.Header(), true); err != nil { + return err + } + + blockHash := block.Hash() + + for i, receipt := range res.FinalBlock.Receipts { + // add block location fields + receipt.BlockHash = blockHash + receipt.BlockNumber = block.Number() + receipt.TransactionIndex = uint(i) + + for _, log := range receipt.Logs { + log.BlockHash = blockHash + } + } + + for _, log := range res.FinalBlock.CoalescedLogs { + log.BlockHash = blockHash + } + + // It's possible that we've stored L1 queue index for this block previously, + // in this case do not overwrite it. + if index := rawdb.ReadFirstQueueIndexNotInL2Block(w.eth.ChainDb(), blockHash); index == nil { + // Store first L1 queue index not processed by this block. + // Note: This accounts for both included and skipped messages. This + // way, if a block only skips messages, we won't reprocess the same + // messages from the next block. + log.Trace( + "Worker WriteFirstQueueIndexNotInL2Block", + "number", block.Number(), + "hash", blockHash.String(), + "nextL1MsgIndex", res.FinalBlock.NextL1MsgIndex, + ) + rawdb.WriteFirstQueueIndexNotInL2Block(w.eth.ChainDb(), blockHash, res.FinalBlock.NextL1MsgIndex) + } else { + log.Trace( + "Worker WriteFirstQueueIndexNotInL2Block: not overwriting existing index", + "number", block.Number(), + "hash", blockHash.String(), + "index", *index, + "nextL1MsgIndex", res.FinalBlock.NextL1MsgIndex, + ) + } + // Store circuit row consumption. + log.Trace( + "Worker write block row consumption", + "id", w.circuitCapacityChecker.ID, + "number", block.Number(), + "hash", blockHash.String(), + "accRows", res.Rows, + ) + + rawdb.WriteBlockRowConsumption(w.eth.ChainDb(), blockHash, res.Rows) + // Commit block and state to database. + _, err = w.chain.WriteBlockWithState(block, res.FinalBlock.Receipts, res.FinalBlock.CoalescedLogs, res.FinalBlock.State, true) + if err != nil { + return err + } + + log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealHash, "hash", blockHash) + + // Broadcast the block and announce chain insertion event + w.mux.Post(core.NewMinedBlockEvent{Block: block}) + w.currentPipeline = nil return nil } diff --git a/params/version.go b/params/version.go index 261757fb5e..a9559904c1 100644 --- a/params/version.go +++ b/params/version.go @@ -24,7 +24,7 @@ import ( const ( VersionMajor = 5 // Major version component of the current release VersionMinor = 3 // Minor version component of the current release - VersionPatch = 21 // Patch version component of the current release + VersionPatch = 22 // Patch version component of the current release VersionMeta = "mainnet" // Version metadata to append to the version string )