兄弟连区Go语言+区块链技术培训以太坊源码分析(26)core-txpool交易池源码分析

1/25兄弟连区Go语言+区块链技术培训以太坊源码分析(26)core-txpool交易池源码分析txpool主要用来存放当前提交的等待写入区块的交易，有远端和本地的。txpool里面的交易分为两种，1.提交但是还不能执行的，放在queue里面等待能够执行(比如说nonce太高)。2.等待执行的，放在pending里面等待执行。从txpool的测试案例来看，txpool主要功能有下面几点。1.交易验证的功能，包括余额不足，Gas不足，Nonce太低,value值是合法的，不能为负数。2.能够缓存Nonce比当前本地账号状态高的交易。存放在queue字段。如果是能够执行的交易存放在pending字段3.相同用户的相同Nonce的交易只会保留一个GasPrice最大的那个。其他的插入不成功。4.如果账号没有钱了，那么queue和pending中对应账号的交易会被删除。5.如果账号的余额小于一些交易的额度，那么对应的交易会被删除，同时有效的交易会从pending移动到queue里面。防止被广播。6.txPool支持一些限制PriceLimit(remove的最低GasPrice限制)，PriceBump(替换相同Nonce的交易的价格的百分比)AccountSlots(每个账户的pending的槽位的最小值)GlobalSlots(全局pending队列的最大值)AccountQueue(每个账户的queueing的槽位的最小值)GlobalQueue(全局queueing的最大值)Lifetime(在queue队列的最长等待时间)7.有限的资源情况下按照GasPrice的优先级进行替换。8.本地的交易会使用journal的功能存放在磁盘上，重启之后会重新导入。远程的交易不会。数据结构//TxPoolcontainsallcurrentlyknowntransactions.Transactions//enterthepoolwhentheyarereceivedfromthenetworkorsubmitted//locally.Theyexitthepoolwhentheyareincludedintheblockchain.2/25//TxPool包含了当前知的交易，当前网络接收到交易，或者本地提交的交易会加入到TxPool。//当他们已经被添加到区块链的时候被移除。//Thepoolseparatesprocessabletransactions(whichcanbeappliedtothe//currentstate)andfuturetransactions.Transactionsmovebetweenthose//twostatesovertimeastheyarereceivedandprocessed.//TxPool分为可执行的交易(可以应用到当前的状态)和未来的交易。交易在这两种状态之间转换，typeTxPoolstruct{configTxPoolConfigchainconfig*params.ChainConfigchainblockChaingasPrice*big.Int//最低的GasPrice限制txFeedevent.Feed//通过txFeed来订阅TxPool的消息scopeevent.SubscriptionScopechainHeadChchanChainHeadEvent//订阅了区块头的消息，当有了新的区块头生成的时候会在这里收到通知chainHeadSubevent.Subscription//区块头消息的订阅器。signertypes.Signer//封装了事务签名处理。musync.RWMutexcurrentState*state.StateDB//CurrentstateintheblockchainheadpendingState*state.ManagedState//PendingstatetrackingvirtualnoncescurrentMaxGas*big.Int//Currentgaslimitfortransactioncaps目前交易上限的GasLimitlocals*accountSet//Setoflocaltransactiontoexepmtfromevicionrules本地交易免除驱逐规则journal*txJournal//Journaloflocaltransactiontobackuptodisk本地交易会写入磁盘pendingmap[common.Address]*txList//Allcurrentlyprocessabletransactions所有当前可以处理的交易queuemap[common.Address]*txList//Queuedbutnon-processabletransactions当前还不能处理的交易beatsmap[common.Address]time.Time//Lastheartbeatfromeachknownaccount每一个已知账号的最后一次心跳信息的时间allmap[common.Hash]*types.Transaction//Alltransactionstoallowlookups可以查找到所有交易3/25priced*txPricedList//Alltransactionssortedbyprice按照价格排序的交易wgsync.WaitGroup//forshutdownsynchomesteadbool//家园版本}构建//NewTxPoolcreatesanewtransactionpooltogather,sortandfilterinbound//trnsactionsfromthenetwork.funcNewTxPool(configTxPoolConfig,chainconfig*params.ChainConfig,chainblockChain)*TxPool{//Sanitizetheinputtoensurenovulnerablegaspricesaresetconfig=(&config).sanitize()//Createthetransactionpoolwithitsinitialsettingspool:=&TxPool{config:config,chainconfig:chainconfig,chain:chain,signer:types.NewEIP155Signer(chainconfig.ChainId),pending:make(map[common.Address]*txList),queue:make(map[common.Address]*txList),beats:make(map[common.Address]time.Time),all:make(map[common.Hash]*types.Transaction),chainHeadCh:make(chanChainHeadEvent,chainHeadChanSize),gasPrice:new(big.Int).SetUint64(config.PriceLimit),}pool.locals=newAccountSet(pool.signer)pool.priced=newTxPricedList(&pool.all)pool.reset(nil,chain.CurrentBlock().Header())//Iflocaltransactionsandjournalingisenabled,loadfromdisk//如果本地交易被允许,而且配置的Journal目录不为空,那么从指定的目录加载日志.4/25//然后rotate交易日志.因为老的交易可能已经失效了,所以调用add方法之后再把被接收的交易写入日志.//if!config.NoLocals&&config.Journal!={pool.journal=newTxJournal(config.Journal)iferr:=pool.journal.load(pool.AddLocal);err!=nil{log.Warn(Failedtoloadtransactionjournal,err,err)}iferr:=pool.journal.rotate(pool.local());err!=nil{log.Warn(Failedtorotatetransactionjournal,err,err)}}//Subscribeeventsfromblockchain从区块链订阅事件。pool.chainHeadSub=pool.chain.SubscribeChainHeadEvent(pool.chainHeadCh)//Starttheeventloopandreturnpool.wg.Add(1)gopool.loop()returnpool}reset方法检索区块链的当前状态并且确保事务池的内容关于当前的区块链状态是有效的。主要功能包括：1.因为更换了区块头，所以原有的区块中有一些交易因为区块头的更换而作废，这部分交易需要重新加入到txPool里面等待插入新的区块2.生成新的currentState和pendingState3.因为状态的改变。将pending中的部分交易移到queue里面4.因为状态的改变，将queue里面的交易移入到pending里面。reset代码//resetretrievesthecurrentstateoftheblockchainandensuresthecontent//ofthetransactionpoolisvalidwithregardtothechainstate.func(pool*TxPool)reset(oldHead,newHead*types.Header){//Ifwe'rereorginganoldstate,reinjectalldroppedtransactions5/25varreinjecttypes.TransactionsifoldHead!=nil&&oldHead.Hash()!=newHead.ParentHash{//Ifthereorgistoodeep,avoiddoingit(willhappenduringfastsync)oldNum:=oldHead.Number.Uint64()newNum:=newHead.Number.Uint64()ifdepth:=uint64(math.Abs(float64(oldNum)-float64(newNum)));depth64{//如果老的头和新的头差距太远,那么取消重建log.Warn(Skippingdeeptransactionreorg,depth,depth)}else{//Reorgseemsshallowenoughtopullinalltransactionsintomemoryvardiscarded,includedtypes.Transactionsvar(rem=pool.chain.GetBlock(oldHead.Hash(),oldHead.Number.Uint64())add=pool.chain.GetBlock(newHead.Hash(),newHead.Number.Uint64()))//如果老的高度大于新的.那么需要把多的全部删除.forrem.NumberU64()add.NumberU64(){discarded=append(discarded,rem.Transactions()...)ifrem=pool.chain.GetBlock(rem.ParentHash(),rem.NumberU64()-1);rem==nil{log.Error(Unrootedoldchainseen