以太坊机制详解:交易与交易池

type TxPoolConfig struct {
	Locals    []common.Address // Addresses that should be treated by default as local
	NoLocals  bool             // Whether local transaction handling should be disabled
	Journal   string           // Journal of local transactions to survive node restarts
	Rejournal time.Duration    // Time interval to regenerate the local transaction journal

	PriceLimit uint64 // Minimum gas price to enforce for acceptance into the pool
	PriceBump  uint64 // Minimum price bump percentage to replace an already existing transaction (nonce)

	AccountSlots uint64 // Number of executable transaction slots guaranteed per account
	GlobalSlots  uint64 // Maximum number of executable transaction slots for all accounts
	AccountQueue uint64 // Maximum number of non-executable transaction slots permitted per account
	GlobalQueue  uint64 // Maximum number of non-executable transaction slots for all accounts

	Lifetime time.Duration // Maximum amount of time non-executable transaction are queued
}

func numSlots(tx *types.Transaction) int {
	return int((tx.Size() + txSlotSize - 1) / txSlotSize)
}

var DefaultTxPoolConfig = TxPoolConfig{
	Journal:   "transactions.rlp",
	Rejournal: time.Hour,

	PriceLimit: 1,
	PriceBump:  10,

	AccountSlots: 16,
	GlobalSlots:  4096 + 1024, // urgent + floating queue capacity with 4:1 ratio
	AccountQueue: 64,
	GlobalQueue:  1024,

	Lifetime: 3 * time.Hour,
}

const transactionParameters = {
  to: '0x0000000000000000000000000000000000000000', 
  from: ethereum.selectedAddress, 
  value: '0x00', 
  data:
    '0x7f7465737432000000000000000000000000000000000000000000000000000000600057'
};

const txHash = await ethereum.request({
  method: 'eth_sendTransaction',
  params: [transactionParameters],
});

cast send 0x11475691C2CAA465E19F99c445abB31A4a64955C --value 0.001ether --gas-limit 21000 --gas-price 5gwei --priority-gas-price 1.5gwei --private-key $pk --rpc-url https://goerli.infura.io/v3/9aa3d95b3bc440fa88ea12eaa4456161

type TxData interface {
	txType() byte // returns the type ID
	copy() TxData // creates a deep copy and initializes all fields

	chainID() *big.Int
	accessList() AccessList
	data() []byte
	gas() uint64
	gasPrice() *big.Int
	gasTipCap() *big.Int
	gasFeeCap() *big.Int
	value() *big.Int
	nonce() uint64
	to() *common.Address

	rawSignatureValues() (v, r, s *big.Int)
	setSignatureValues(chainID, v, r, s *big.Int)
}

pending map[common.Address]*txList   // All currently processable transactions
queue   map[common.Address]*txList   // Queued but non-processable transactions

func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err error)

hash := tx.Hash()
if pool.all.Get(hash) != nil {
	log.Trace("Discarding already known transaction", "hash", hash)
	knownTxMeter.Mark(1)
	return false, ErrAlreadyKnown
}

isLocal := local || pool.locals.containsTx(tx)

if err := pool.validateTx(tx, isLocal); err != nil {
	log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
	invalidTxMeter.Mark(1)
	return false, err
}

uint64(pool.all.Slots()+numSlots(tx)) > pool.config.GlobalSlots+pool.config.GlobalQueue

if !isLocal && pool.priced.Underpriced(tx) {
	log.Trace("Discarding underpriced transaction", "hash", hash, "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
	underpricedTxMeter.Mark(1)
	return false, ErrUnderpriced
}

if pool.changesSinceReorg > int(pool.config.GlobalSlots/4) {
	throttleTxMeter.Mark(1)
	return false, ErrTxPoolOverflow
}

drop, success := pool.priced.Discard(pool.all.Slots()-int(pool.config.GlobalSlots+pool.config.GlobalQueue)+numSlots(tx), isLocal)

// Special case, we still can't make the room for the new remote one.
if !isLocal && !success {
	log.Trace("Discarding overflown transaction", "hash", hash)
	overflowedTxMeter.Mark(1)
	return false, ErrTxPoolOverflow
}
// Bump the counter of rejections-since-reorg
pool.changesSinceReorg += len(drop)
// Kick out the underpriced remote transactions.
for _, tx := range drop {
	log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
	underpricedTxMeter.Mark(1)
	pool.removeTx(tx.Hash(), false)
}

from, _ := types.Sender(pool.signer, tx) // already validated
if list := pool.pending[from]; list != nil && list.Overlaps(tx)

inserted, old := list.Add(tx, pool.config.PriceBump)
if !inserted {
	pendingDiscardMeter.Mark(1)
	return false, ErrReplaceUnderpriced
}

func (l *txList) Add(tx *types.Transaction, priceBump uint64) (bool, *types.Transaction) {
	// 获取交易列表内 Nonce 相同的旧交易
	old := l.txs.Get(tx.Nonce())
	if old != nil {
		// 要求新交易的 GasFeeCap 和 GasTipCap 大于旧交易
		if old.GasFeeCapCmp(tx) >= 0 || old.GasTipCapCmp(tx) >= 0 {
			return false, nil
		}
		// thresholdFeeCap = oldFC  * (100 + priceBump) / 100
		a := big.NewInt(100 + int64(priceBump))
		aFeeCap := new(big.Int).Mul(a, old.GasFeeCap())
		aTip := a.Mul(a, old.GasTipCap())

		// thresholdTip    = oldTip * (100 + priceBump) / 100
		b := big.NewInt(100)
		thresholdFeeCap := aFeeCap.Div(aFeeCap, b)
		thresholdTip := aTip.Div(aTip, b)

		// 要求新交易的 GasFeeCap 和 GasTipCapCmp 增加幅度大于 PriceBump
		if tx.GasFeeCapIntCmp(thresholdFeeCap) < 0 || tx.GasTipCapIntCmp(thresholdTip) < 0 {
			return false, nil
		}
	}
	// 使用新交易覆盖旧交易
	l.txs.Put(tx)

	// 以下内容为刷新参数
	if cost := tx.Cost(); l.costcap.Cmp(cost) < 0 {
		l.costcap = cost
	}
	if gas := tx.Gas(); l.gascap < gas {
		l.gascap = gas
	}
	return true, old
}

if old != nil {
	pool.all.Remove(old.Hash())
	pool.priced.Removed(1)
	pendingReplaceMeter.Mark(1)
}
pool.all.Add(tx, isLocal)
pool.priced.Put(tx, isLocal)
pool.journalTx(from, tx)
pool.queueTxEvent(tx)
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())

case tx := <-pool.queueTxEventCh:
	addr, _ := types.Sender(pool.signer, tx)
	if _, ok := queuedEvents[addr]; !ok {
		queuedEvents[addr] = newTxSortedMap()
	}
	queuedEvents[addr].Put(tx)

queuedEvents  = make(map[common.Address]*txSortedMap)

if len(events) > 0 {
	var txs []*types.Transaction
	for _, set := range events {
		txs = append(txs, set.Flatten()...)
	}
	pool.txFeed.Send(NewTxsEvent{txs})
}

pool.beats[from] = time.Now()
return old != nil, nil

replaced, err = pool.enqueueTx(hash, tx, isLocal, true)
if err != nil {
	return false, err
}

func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction, local bool, addAll bool) (bool, error) {
	from, _ := types.Sender(pool.signer, tx) // already validated
	// 如果 queue 队列内没有此地址记录，则创建一个新的映射
	// queue 的定义为 map[common.Address]*txList
	if pool.queue[from] == nil {
		pool.queue[from] = newTxList(false)
	}
	// 使用 (l *txList) Add 直接加入交易
	// 详情请参考上文给出的代码解析
	inserted, old := pool.queue[from].Add(tx, pool.config.PriceBump)

	// 处理插入失败的情况
	if !inserted {
		queuedDiscardMeter.Mark(1)
		return false, ErrReplaceUnderpriced
	}

	if old != nil {
		// 发生替换交易情况，刷新 pool 参数
		// 类似上文提到的 替换交易 的后续处理
		pool.all.Remove(old.Hash())
		pool.priced.Removed(1)
		queuedReplaceMeter.Mark(1)
	} else {
		// 没有发生替换交易的情况，增加计数器
		// 计数器用于评估等功能
		queuedGauge.Inc(1)
	}
	// If the transaction isn't in lookup set but it's expected to be there,
	// show the error log.
	if pool.all.Get(hash) == nil && !addAll {
		log.Error("Missing transaction in lookup set, please report the issue", "hash", hash)
	}
	// 刷新 pool 中的 all 和 priced 变量
	if addAll {
		pool.all.Add(tx, local)
		pool.priced.Put(tx, local)
	}
	// 刷新交易账户的生命周期
	if _, exist := pool.beats[from]; !exist {
		pool.beats[from] = time.Now()
	}
	return old != nil, nil
}

if local && !pool.locals.contains(from) {
	log.Info("Setting new local account", "address", from)
	pool.locals.add(from)
	pool.priced.Removed(pool.all.RemoteToLocals(pool.locals)) // Migrate the remotes if it's marked as local first time.
}
if isLocal {
	localGauge.Inc(1)
}
pool.journalTx(from, tx)

log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
return replaced, nil

func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) ([]error, *accountSet) {
	dirty := newAccountSet(pool.signer)
	errs := make([]error, len(txs))
	for i, tx := range txs {
		replaced, err := pool.add(tx, local)
		errs[i] = err
		if err == nil && !replaced {
			dirty.addTx(tx)
		}
	}
	validTxMeter.Mark(int64(len(dirty.accounts)))
	return errs, dirty
}

func (pool *TxPool) addTxs(txs []*types.Transaction, local, sync bool) []error

var (
	errs = make([]error, len(txs))
	news = make([]*types.Transaction, 0, len(txs))
)
for i, tx := range txs {
	// 验证交易是否已经存在在交易池内
	if pool.all.Get(tx.Hash()) != nil {
		errs[i] = ErrAlreadyKnown
		knownTxMeter.Mark(1)
		continue
	}
	// 验证交易签名是否正确
	_, err := types.Sender(pool.signer, tx)
	if err != nil {
		errs[i] = ErrInvalidSender
		invalidTxMeter.Mark(1)
		continue
	}
	// 将交易添加到 news 列表内
	news = append(news, tx)
}
if len(news) == 0 {
	return errs
}

pool.mu.Lock()
newErrs, dirtyAddrs := pool.addTxsLocked(news, local)
pool.mu.Unlock()

var nilSlot = 0
for _, err := range newErrs {
	for errs[nilSlot] != nil {
		nilSlot++
	}
	errs[nilSlot] = err
	nilSlot++
}

done := pool.requestPromoteExecutables(dirtyAddrs)
if sync {
	<-done
}
return errs

func (pool *TxPool) requestPromoteExecutables(set *accountSet) chan struct{} {
	select {
	case pool.reqPromoteCh <- set:
		return <-pool.reorgDoneCh
	case <-pool.reorgShutdownCh:
		return pool.reorgShutdownCh
	}
}

go pool.scheduleReorgLoop()

reqResetCh      chan *txpoolResetRequest
reqPromoteCh    chan *accountSet
queueTxEventCh  chan *types.Transaction
reorgDoneCh     chan chan struct{}
reorgShutdownCh chan struct{}  // requests shutdown of scheduleReorgLoop

case req := <-pool.reqResetCh:
	// Reset request: update head if request is already pending.
	if reset == nil {
		reset = req
	} else {
		reset.newHead = req.newHead
	}
	launchNextRun = true
	pool.reorgDoneCh <- nextDone

case req := <-pool.reqPromoteCh:
	// Promote request: update address set if request is already pending.
	if dirtyAccounts == nil {
		dirtyAccounts = req
	} else {
		dirtyAccounts.merge(req)
	}
	launchNextRun = true
	pool.reorgDoneCh <- nextDone

func (pool *TxPool) requestReset(oldHead *types.Header, newHead *types.Header) chan struct{} {
	select {
	case pool.reqResetCh <- &txpoolResetRequest{oldHead, newHead}:
		return <-pool.reorgDoneCh
	case <-pool.reorgShutdownCh:
		return pool.reorgShutdownCh
	}
}

func (pool *TxPool) requestPromoteExecutables(set *accountSet) chan struct{} {
	select {
	case pool.reqPromoteCh <- set:
		return <-pool.reorgDoneCh
	case <-pool.reorgShutdownCh:
		return pool.reorgShutdownCh
	}
}

done := pool.requestPromoteExecutables(dirtyAddrs)
if sync {
	<-done
}

case <-pool.reorgShutdownCh:
	// Wait for current run to finish.
	if curDone != nil {
		<-curDone
	}
	close(nextDone)
	return

case <-pool.chainHeadSub.Err():
	close(pool.reorgShutdownCh)
	return

func (pool *TxPool) runReorg(done chan struct{}, reset *txpoolResetRequest, dirtyAccounts *accountSet, events map[common.Address]*txSortedMap)

var promoteAddrs []common.Address
if dirtyAccounts != nil && reset == nil {
	promoteAddrs = dirtyAccounts.flatten()
}
pool.mu.Lock()
promoted := pool.promoteExecutables(promoteAddrs)

func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Transaction {
	var promoted []*types.Transaction

	// 循环 accounts 并寻找可提权交易
	for _, addr := range accounts {
		list := pool.queue[addr]
		if list == nil {
			continue // Just in case someone calls with a non existing account
		}
		// 丢弃所有低于当前状态账户 nonce 的交易
		forwards := list.Forward(pool.currentState.GetNonce(addr))
		for _, tx := range forwards {
			hash := tx.Hash()
			pool.all.Remove(hash)
		}
		log.Trace("Removed old queued transactions", "count", len(forwards))
		// 丢弃所有账户余额无法满足 gas 消耗的交易
		drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
		for _, tx := range drops {
			hash := tx.Hash()
			pool.all.Remove(hash)
		}
		log.Trace("Removed unpayable queued transactions", "count", len(drops))
		queuedNofundsMeter.Mark(int64(len(drops)))

		// 使用 pool.pendingNonces.get(addr) 获得 pending 队列中的最大 nonce
		// 使用 list.Ready 在 quenue 队列中获得上述 nonce 后的交易
		readies := list.Ready(pool.pendingNonces.get(addr))
		for _, tx := range readies {
			hash := tx.Hash()
			// 使用 pool.promoteTx 函数对交易进行提权
			// 即将交易自 quenue 转移到 pending
			if pool.promoteTx(addr, hash, tx) {
				promoted = append(promoted, tx)
			}
		}
		log.Trace("Promoted queued transactions", "count", len(promoted))
		queuedGauge.Dec(int64(len(readies)))

		// 判断用户的 quenue 队列中交易数量是否符合需求
		// 如不符合则删除部分交易
		var caps types.Transactions
		if !pool.locals.contains(addr) {
			caps = list.Cap(int(pool.config.AccountQueue))
			for _, tx := range caps {
				hash := tx.Hash()
				pool.all.Remove(hash)
				log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
			}
			queuedRateLimitMeter.Mark(int64(len(caps)))
		}
		// Mark all the items dropped as removed
		pool.priced.Removed(len(forwards) + len(drops) + len(caps))
		queuedGauge.Dec(int64(len(forwards) + len(drops) + len(caps)))
		if pool.locals.contains(addr) {
			localGauge.Dec(int64(len(forwards) + len(drops) + len(caps)))
		}
		// Delete the entire queue entry if it became empty.
		if list.Empty() {
			delete(pool.queue, addr)
			delete(pool.beats, addr)
		}
	}
	return promoted
}

pool.truncatePending()
pool.truncateQueue()

dropBetweenReorgHistogram.Update(int64(pool.changesSinceReorg))
pool.changesSinceReorg = 0 // Reset change counter
pool.mu.Unlock()

func (w *worker) fillTransactions(interrupt *int32, env *environment) error {
	// 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]types.Transactions), pending
	for _, account := range w.eth.TxPool().Locals() {
		if txs := remoteTxs[account]; len(txs) > 0 {
			delete(remoteTxs, account)
			localTxs[account] = txs
		}
	}
	if len(localTxs) > 0 {
		txs := types.NewTransactionsByPriceAndNonce(env.signer, localTxs, env.header.BaseFee)
		if err := w.commitTransactions(env, txs, interrupt); err != nil {
			return err
		}
	}
	if len(remoteTxs) > 0 {
		txs := types.NewTransactionsByPriceAndNonce(env.signer, remoteTxs, env.header.BaseFee)
		if err := w.commitTransactions(env, txs, interrupt); err != nil {
			return err
		}
	}
	return nil
}

type TransactionsByPriceAndNonce struct {
	txs     map[common.Address]Transactions // Per account nonce-sorted list of transactions
	heads   TxByPriceAndTime                // Next transaction for each unique account (price heap)
	signer  Signer                          // Signer for the set of transactions
	baseFee *big.Int                        // Current base fee
}

type TxWithMinerFee struct {
	tx       *Transaction
	minerFee *big.Int
}

func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByPriceAndNonce, interrupt *int32) error

gasLimit := env.header.GasLimit
if env.gasPool == nil {
	env.gasPool = new(core.GasPool).AddGas(gasLimit)
}

if env.gasPool.Gas() < params.TxGas {
	log.Trace("Not enough gas for further transactions", "have", env.gasPool, "want", params.TxGas)
	break
}

tx := txs.Peek()
if tx == nil {
	break
}

if tx.Protected() && !w.chainConfig.IsEIP155(env.header.Number) {
	log.Trace("Ignoring reply protected transaction", "hash", tx.Hash(), "eip155", w.chainConfig.EIP155Block)

	txs.Pop()
	continue
}

env.state.Prepare(tx.Hash(), env.tcount)

logs, err := w.commitTransaction(env, tx)

if contractCreation {
	ret, _, st.gas, vmerr = st.evm.Create(sender, st.data, st.gas, st.value)
} else {
	// Increment the nonce for the next transaction
	st.state.SetNonce(msg.From(), st.state.GetNonce(sender.Address())+1)
	ret, st.gas, vmerr = st.evm.Call(sender, st.to(), st.data, st.gas, st.value)
}

case errors.Is(err, nil):
	// Everything ok, collect the logs and shift in the next transaction from the same account
	coalescedLogs = append(coalescedLogs, logs...)
	env.tcount++
	txs.Shift()

if !w.isRunning() && len(coalescedLogs) > 0 {
	cpy := make([]*types.Log, len(coalescedLogs))
	for i, l := range coalescedLogs {
		cpy[i] = new(types.Log)
		*cpy[i] = *l
	}
	w.pendingLogsFeed.Send(cpy)
}
return nil

uwsc wss://mainnet.infura.io/ws/v3/YOUR_API_KEY

{"jsonrpc":"2.0", "id": 1, "method": "eth_subscribe", "params": ["newPendingTransactions"]}

func (w *worker) generateWork(params *generateParams) (*types.Block, error) {
	work, err := w.prepareWork(params)
	if err != nil {
		return nil, err
	}
	defer work.discard()

	if !params.noTxs {
		interrupt := new(int32)
		timer := time.AfterFunc(w.newpayloadTimeout, func() {
			atomic.StoreInt32(interrupt, commitInterruptTimeout)
		})
		defer timer.Stop()

		err := w.fillTransactions(interrupt, work)
		if errors.Is(err, errBlockInterruptedByTimeout) {
			log.Warn("Block building is interrupted", "allowance", common.PrettyDuration(w.newpayloadTimeout))
		}
	}
	return w.engine.FinalizeAndAssemble(w.chain, work.header, work.state, work.txs, work.unclelist(), work.receipts)
}

case ev := <-pool.chainHeadCh:
	if ev.Block != nil {
		pool.requestReset(head.Header(), ev.Block.Header())
		head = ev.Block
	}

case req := <-pool.reqResetCh:
	// Reset request: update head if request is already pending.
	if reset == nil {
		reset = req
	} else {
		reset.newHead = req.newHead
	}
	launchNextRun = true
	pool.reorgDoneCh <- nextDone

if reset != nil {
	// Reset from the old head to the new, rescheduling any reorged transactions
	pool.reset(reset.oldHead, reset.newHead)

	// Nonces were reset, discard any events that became stale
	for addr := range events {
		events[addr].Forward(pool.pendingNonces.get(addr))
		if events[addr].Len() == 0 {
			delete(events, addr)
		}
	}
	// Reset needs promote for all addresses
	promoteAddrs = make([]common.Address, 0, len(pool.queue))
	for addr := range pool.queue {
		promoteAddrs = append(promoteAddrs, addr)
	}
}

if oldHead != nil && oldHead.Hash() != newHead.ParentHash {
	// 获得区块编号
	oldNum := oldHead.Number.Uint64()
	newNum := newHead.Number.Uint64()
	// 当新旧区块差异过大时，不进行处理
	// 此种区块一般发生在节点同步数据时
	if depth := uint64(math.Abs(float64(oldNum) - float64(newNum))); depth > 64 {
		log.Debug("Skipping deep transaction reorg", "depth", depth)
	} else {
		// 声明需要丢弃和包含的交易
		var discarded, included types.Transactions
		var (
			// 获取区块完整数据
			rem = pool.chain.GetBlock(oldHead.Hash(), oldHead.Number.Uint64())
			add = pool.chain.GetBlock(newHead.Hash(), newHead.Number.Uint64())
		)
		if rem == nil {
			// 此情况属于特殊情况，即旧区块无法检索
			if newNum >= oldNum {
				log.Warn("Transaction pool reset with missing oldhead",
					"old", oldHead.Hash(), "oldnum", oldNum, "new", newHead.Hash(), "newnum", newNum)
				return
			}
			log.Debug("Skipping transaction reset caused by setHead",
				"old", oldHead.Hash(), "oldnum", oldNum, "new", newHead.Hash(), "newnum", newNum)
		} else {
			for rem.NumberU64() > add.NumberU64() {
				discarded = append(discarded, rem.Transactions()...)
				if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil {
					log.Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash())
					return
				}
			}
			for add.NumberU64() > rem.NumberU64() {
				included = append(included, add.Transactions()...)
				if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil {
					log.Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash())
					return
				}
			}
			for rem.Hash() != add.Hash() {
				discarded = append(discarded, rem.Transactions()...)
				if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil {
					log.Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash())
					return
				}
				included = append(included, add.Transactions()...)
				if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil {
					log.Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash())
					return
				}
			}
			reinject = types.TxDifference(discarded, included)
		}
	}
}

rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil

if newHead == nil {
	newHead = pool.chain.CurrentBlock().Header() // Special case during testing
}

statedb, err := pool.chain.StateAt(newHead.Root)
if err != nil {
	log.Error("Failed to reset txpool state", "err", err)
	return
}
pool.currentState = statedb
pool.pendingNonces = newTxNoncer(statedb)
pool.currentMaxGas = newHead.GasLimit

// Inject any transactions discarded due to reorgs
log.Debug("Reinjecting stale transactions", "count", len(reinject))
senderCacher.recover(pool.signer, reinject)
pool.addTxsLocked(reinject, false)

// Update all fork indicator by next pending block number.
next := new(big.Int).Add(newHead.Number, big.NewInt(1))
pool.istanbul = pool.chainconfig.IsIstanbul(next)
pool.eip2718 = pool.chainconfig.IsBerlin(next)
pool.eip1559 = pool.chainconfig.IsLondon(next)

if reset != nil {
	pool.demoteUnexecutables()
	if reset.newHead != nil && pool.chainconfig.IsLondon(new(big.Int).Add(reset.newHead.Number, big.NewInt(1))) {
		pendingBaseFee := misc.CalcBaseFee(pool.chainconfig, reset.newHead)
		pool.priced.SetBaseFee(pendingBaseFee)
	}
	// Update all accounts to the latest known pending nonce
	nonces := make(map[common.Address]uint64, len(pool.pending))
	for addr, list := range pool.pending {
		highestPending := list.LastElement()
		nonces[addr] = highestPending.Nonce() + 1
	}
	pool.pendingNonces.setAll(nonces)
}

func (pool *TxPool) demoteUnexecutables() {
	// 迭代交易池内的 pending 队列
	for addr, list := range pool.pending {
		nonce := pool.currentState.GetNonce(addr)

		// 删除交易中 nonce 较低的交易
		olds := list.Forward(nonce)
		for _, tx := range olds {
			hash := tx.Hash()
			pool.all.Remove(hash)
			log.Trace("Removed old pending transaction", "hash", hash)
		}
		// 丢弃到所有账户无法支付 gas 费用的交易
		// drops 是所有无法支付 gas 费用的交易
		// invalids 是指低于 drops 中最低 nonce 交易的列表
		drops, invalids := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
		for _, tx := range drops {
			hash := tx.Hash()
			log.Trace("Removed unpayable pending transaction", "hash", hash)
			pool.all.Remove(hash)
		}
		pendingNofundsMeter.Mark(int64(len(drops)))
		// invalids 内包含的交易可能符合要求，也有可能不符合要求
		// 所以交易应该进入 queued 队列
		for _, tx := range invalids {
			hash := tx.Hash()
			log.Trace("Demoting pending transaction", "hash", hash)

			// Internal shuffle shouldn't touch the lookup set.
			pool.enqueueTx(hash, tx, false, false)
		}
		// 更新计数器
		pendingGauge.Dec(int64(len(olds) + len(drops) + len(invalids)))
		if pool.locals.contains(addr) {
			localGauge.Dec(int64(len(olds) + len(drops) + len(invalids)))
		}
		// 交易列表长度大于 0 ，但交易列表内没有找到 nonce 为最新 nonce 的交易
		// 这意味着交易列表内的所有交易都是可能过时的
		if list.Len() > 0 && list.txs.Get(nonce) == nil {
			// 利用 Cap 返回交易列表内所有交易
			// Cap 的功能是限制列表内的交易数，此处限制为 0
			gapped := list.Cap(0)
			// 执行交易降级
			for _, tx := range gapped {
				hash := tx.Hash()
				log.Error("Demoting invalidated transaction", "hash", hash)

				// Internal shuffle shouldn't touch the lookup set.
				pool.enqueueTx(hash, tx, false, false)
			}
			// 重置计数器
			pendingGauge.Dec(int64(len(gapped)))
			blockReorgInvalidatedTx.Mark(int64(len(gapped)))
		}
		// 如果发现 list(账户地址对应的交易列表)为空，则直接在交易池内删除此账户
		if list.Empty() {
			delete(pool.pending, addr)
		}
	}
}

nonces := make(map[common.Address]uint64, len(pool.pending))
for addr, list := range pool.pending {
	highestPending := list.LastElement()
	nonces[addr] = highestPending.Nonce() + 1
}
pool.pendingNonces.setAll(nonces)