Skip to content

Commit

Permalink
disable tx acceleration
Browse files Browse the repository at this point in the history
  • Loading branch information
@libangzhu
libangzhu committed Nov 3, 2023
1 parent b34fdf8 commit 96011b5
Show file tree
Hide file tree
Showing 2 changed files with 44 additions and 27 deletions.
58 changes: 37 additions & 21 deletions system/mempool/base.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import (

var mlog = log.New("module", "mempool.base")

//Mempool mempool 基础类
// Mempool mempool 基础类
type Mempool struct {
proxyMtx sync.RWMutex
in chan *queue.Message
Expand Down Expand Up @@ -54,14 +54,14 @@ func (mem *Mempool) getAPI() client.QueueProtocolAPI {
return mem.api
}

//GetSync 判断是否mempool 同步
// GetSync 判断是否mempool 同步
func (mem *Mempool) getSync() bool {
mem.proxyMtx.RLock()
defer mem.proxyMtx.RUnlock()
return mem.sync
}

//NewMempool 新建mempool 实例
// NewMempool 新建mempool 实例
func NewMempool(cfg *types.Mempool) *Mempool {
pool := &Mempool{}
if cfg.MaxTxNumPerAccount == 0 {
Expand All @@ -85,7 +85,7 @@ func NewMempool(cfg *types.Mempool) *Mempool {
return pool
}

//Close 关闭mempool
// Close 关闭mempool
func (mem *Mempool) Close() {
if mem.isClose() {
return
Expand All @@ -101,7 +101,7 @@ func (mem *Mempool) Close() {
mlog.Info("mempool module closed")
}

//SetQueueClient 初始化mempool模块
// SetQueueClient 初始化mempool模块
func (mem *Mempool) SetQueueClient(cli queue.Client) {
mem.client = cli
mem.client.Sub("mempool")
Expand Down Expand Up @@ -136,7 +136,7 @@ func (mem *Mempool) SetMinFee(fee int64) {
mem.proxyMtx.Unlock()
}

//SetQueueCache 设置排队策略
// SetQueueCache 设置排队策略
func (mem *Mempool) SetQueueCache(qcache QueueCache) {
mem.cache.SetQueueCache(qcache)
}
Expand All @@ -159,33 +159,40 @@ func (mem *Mempool) filterTxList(count int64, dupMap map[string]bool, isAll bool
blockTime := mem.header.GetBlockTime()
types.AssertConfig(mem.client)
cfg := mem.client.GetConfig()
//由于mempool可能存在过期交易，先遍历所有，满足目标交易数再退出，否则存在无法获取到实际交易情况
var temptxs []*types.Transaction
mem.cache.Walk(0, func(tx *Item) bool {
if len(dupMap) > 0 {
if _, ok := dupMap[string(tx.Value.Hash())]; ok {
return true
}
}
if isExpired(cfg, tx, height, blockTime) && !isAll {
return true
}
txs = append(txs, tx.Value)
temptxs = append(temptxs, tx.Value)
//达到设定的交易数，退出循环, count为0获取所有
if count > 0 && len(txs) == int(count) {
if count > 0 && len(temptxs) == int(count) {
return false
}
return true
})

if mem.client.GetConfig().IsFork(mem.header.GetHeight(), "ForkCheckEthTxSort") {
//对txs 进行排序
txs = mem.sortEthSignTyTx(txs)
temptxs = mem.sortEthSignTyTx(temptxs)
}

if len(dupMap) > 0 {
for _, tx := range temptxs {
if _, ok := dupMap[string(tx.Hash())]; ok {
continue
}
txs = append(txs, tx)
}
} else {
txs = temptxs
}

return txs
}

//对eth signtype 的交易，同地址下nonce 按照从小到达的顺序排序
//确保nonce 按照递增顺序发给blockchain
// 对eth signtype 的交易，同地址下nonce 按照从小到达的顺序排序
// 确保nonce 按照递增顺序发给blockchain
func (mem *Mempool) sortEthSignTyTx(txs []*types.Transaction) []*types.Transaction {
//平行链架构下，主链节点无法获取到平行链evm的nonce
var merge []*types.Transaction
Expand All @@ -212,14 +219,15 @@ func (mem *Mempool) sortEthSignTyTx(txs []*types.Transaction) []*types.Transacti
})
//check exts[0].Nonce 是否等于current nonce, merge
if len(etxs) != 0 && mem.getCurrentNonce(from) == etxs[0].GetNonce() {

merge = append(merge, etxs[0])
for i, etx := range etxs {
if i == 0 {
continue
}
//要求nonce 具有连续性
if etx.GetNonce() == etxs[i-1].GetNonce()+1 {
merge = append(merge, etxs[i])
if etx.GetNonce() == etxs[0].GetNonce()+int64(i) {
merge = append(merge, etx)
continue
}
break
Expand Down Expand Up @@ -272,7 +280,7 @@ func (mem *Mempool) PushTx(tx *types.Transaction) error {
return err
}

// setHeader设置mempool.header
// setHeader设置mempool.header
func (mem *Mempool) setHeader(h *types.Header) {
atomic.StoreInt64(&mem.currHeight, h.Height)
mem.proxyMtx.Lock()
Expand All @@ -287,7 +295,7 @@ func (mem *Mempool) GetHeader() *types.Header {
return mem.header
}

//IsClose 判断是否mempool 关闭
// IsClose 判断是否mempool 关闭
func (mem *Mempool) isClose() bool {
return atomic.LoadInt32(&mem.isclose) == 1
}
Expand Down Expand Up @@ -482,6 +490,14 @@ func (mem *Mempool) delBlock(block *types.Block) {
if !mem.checkExpireValid(tx) {
continue
}

//检查mempool内是否有相同的tx.nonce
err = mem.evmTxNonceCheck(tx.Tx())
if err != nil {
mlog.Error("delBlock mem", "block height:", block.Height, "evmTxNonceCheck push tx err", err)
continue
}

err = mem.PushTx(tx)
if err != nil {
mlog.Error("mem", "push tx err", err)
Expand Down
13 changes: 7 additions & 6 deletions system/mempool/check.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,6 @@ package mempool
import (
"bytes"
"errors"
"fmt"
"math/big"
"sort"
"sync/atomic"
"time"
Expand Down Expand Up @@ -150,7 +148,7 @@ func (mem *Mempool) checkLevelFee(tx *types.TransactionCache) error {
return nil
}

//checkTxRemote 检查账户余额是否足够，并加入到Mempool，成功则传入goodChan，若加入Mempool失败则传入badChan
// checkTxRemote 检查账户余额是否足够，并加入到Mempool，成功则传入goodChan，若加入Mempool失败则传入badChan
func (mem *Mempool) checkTxRemote(msg *queue.Message) *queue.Message {
tx := msg.GetData().(types.TxGroup)
lastheader := mem.GetHeader()
Expand Down Expand Up @@ -221,7 +219,7 @@ func (mem *Mempool) checkTxRemote(msg *queue.Message) *queue.Message {
return msg
}

//evmTxNonceCheck 检查eth noce 是否有重复值，如果有的话，需要比较txFee大小，用于替换较小的fee的那笔交易
// evmTxNonceCheck 检查eth noce 是否有重复值，如果有的话，需要比较txFee大小，用于替换较小的fee的那笔交易
func (mem *Mempool) evmTxNonceCheck(tx *types.Transaction) error {
if !types.IsEthSignID(tx.Tx().GetSignature().GetTy()) {
return nil
Expand All @@ -243,8 +241,11 @@ func (mem *Mempool) evmTxNonceCheck(tx *types.Transaction) error {
if bytes.Equal(stx.Tx.Hash(), tx.Hash()) {
continue
}

if txs[i].GetTx().GetNonce() == tx.GetNonce() {
mlog.Info("evmTxNonceCheck", "from:", tx.From(), "pre tx hash", common.ToHex(txs[i].GetTx().Hash()), "detect transaction acceleration action:", "reject")
return errors.New("disable transaction acceleration")
}
/*if txs[i].GetTx().GetNonce() == tx.GetNonce() {
bnfee := big.NewInt(txs[i].GetTx().Fee)
//相同的nonce，fee 必须提升至1.1 倍 才能有效替换之前的交易
bnfee = bnfee.Mul(bnfee, big.NewInt(110))
Expand All @@ -261,7 +262,7 @@ func (mem *Mempool) evmTxNonceCheck(tx *types.Transaction) error {
})
mlog.Info("evmTxNonceCheck", "remote txhash:", common.ToHex(txs[i].GetTx().Hash()), "replace txHash:", common.ToHex(tx.Hash()))
return nil
}
}*/
}

}
Expand Down

0 comments on commit 96011b5

Please sign in to comment.