mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 06:06:44 +00:00
fix(txpool): rollback false optimization (#694)
* feat: add transactions len metrics of block processer * feat: rollback false optimize * feat: update * feat: bump version
This commit is contained in:
parent
0bccb7bc50
commit
4c2b567116
2 changed files with 11 additions and 36 deletions
|
|
@ -131,19 +131,6 @@ var (
|
||||||
reheapTimer = metrics.NewRegisteredTimer("txpool/reheap", nil)
|
reheapTimer = metrics.NewRegisteredTimer("txpool/reheap", nil)
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
|
||||||
addrsPool = sync.Pool{
|
|
||||||
New: func() interface{} {
|
|
||||||
return make([]common.Address, 0, 8)
|
|
||||||
},
|
|
||||||
}
|
|
||||||
addrBeatPool = sync.Pool{
|
|
||||||
New: func() interface{} {
|
|
||||||
return make(addressesByHeartbeat, 0, 8)
|
|
||||||
},
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
// TxStatus is the current status of a transaction as seen by the pool.
|
// TxStatus is the current status of a transaction as seen by the pool.
|
||||||
type TxStatus uint
|
type TxStatus uint
|
||||||
|
|
||||||
|
|
@ -283,8 +270,6 @@ type TxPool struct {
|
||||||
wg sync.WaitGroup // tracks loop, scheduleReorgLoop
|
wg sync.WaitGroup // tracks loop, scheduleReorgLoop
|
||||||
initDoneCh chan struct{} // is closed once the pool is initialized (for tests)
|
initDoneCh chan struct{} // is closed once the pool is initialized (for tests)
|
||||||
|
|
||||||
spammers *prque.Prque
|
|
||||||
|
|
||||||
changesSinceReorg int // A counter for how many drops we've performed in-between reorg.
|
changesSinceReorg int // A counter for how many drops we've performed in-between reorg.
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -316,7 +301,6 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
|
||||||
reorgShutdownCh: make(chan struct{}),
|
reorgShutdownCh: make(chan struct{}),
|
||||||
initDoneCh: make(chan struct{}),
|
initDoneCh: make(chan struct{}),
|
||||||
gasPrice: new(big.Int).SetUint64(config.PriceLimit),
|
gasPrice: new(big.Int).SetUint64(config.PriceLimit),
|
||||||
spammers: prque.New(nil),
|
|
||||||
}
|
}
|
||||||
pool.locals = newAccountSet(pool.signer)
|
pool.locals = newAccountSet(pool.signer)
|
||||||
for _, addr := range config.Locals {
|
for _, addr := range config.Locals {
|
||||||
|
|
@ -1218,14 +1202,13 @@ func (pool *TxPool) runReorg(done chan struct{}, reset *txpoolResetRequest, dirt
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Reset needs promote for all addresses
|
// Reset needs promote for all addresses
|
||||||
promoteAddrs = addrsPool.Get().([]common.Address)
|
promoteAddrs = make([]common.Address, 0, len(pool.queue))
|
||||||
for addr := range pool.queue {
|
for addr := range pool.queue {
|
||||||
promoteAddrs = append(promoteAddrs, addr)
|
promoteAddrs = append(promoteAddrs, addr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Check for pending transactions for every account that sent new ones
|
// Check for pending transactions for every account that sent new ones
|
||||||
promoted := pool.promoteExecutables(promoteAddrs)
|
promoted := pool.promoteExecutables(promoteAddrs)
|
||||||
defer addrsPool.Put(promoteAddrs[:0])
|
|
||||||
|
|
||||||
// If a new block appeared, validate the pool of pending transactions. This will
|
// If a new block appeared, validate the pool of pending transactions. This will
|
||||||
// remove any transaction that has been included in the block or was invalidated
|
// remove any transaction that has been included in the block or was invalidated
|
||||||
|
|
@ -1465,19 +1448,18 @@ func (pool *TxPool) truncatePending() {
|
||||||
|
|
||||||
pendingBeforeCap := pending
|
pendingBeforeCap := pending
|
||||||
// Assemble a spam order to penalize large transactors first
|
// Assemble a spam order to penalize large transactors first
|
||||||
pool.spammers.Reset()
|
spammers := prque.New(nil)
|
||||||
for addr, list := range pool.pending {
|
for addr, list := range pool.pending {
|
||||||
// Only evict transactions from high rollers
|
// Only evict transactions from high rollers
|
||||||
if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots {
|
if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots {
|
||||||
pool.spammers.Push(addr, int64(list.Len()))
|
spammers.Push(addr, int64(list.Len()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Gradually drop transactions from offenders
|
// Gradually drop transactions from offenders
|
||||||
offenders := addrsPool.Get().([]common.Address)
|
offenders := []common.Address{}
|
||||||
defer addrsPool.Put(offenders[:0])
|
for pending > pool.config.GlobalSlots && !spammers.Empty() {
|
||||||
for pending > pool.config.GlobalSlots && !pool.spammers.Empty() {
|
|
||||||
// Retrieve the next offender if not local address
|
// Retrieve the next offender if not local address
|
||||||
offender, _ := pool.spammers.Pop()
|
offender, _ := spammers.Pop()
|
||||||
offenders = append(offenders, offender.(common.Address))
|
offenders = append(offenders, offender.(common.Address))
|
||||||
|
|
||||||
// Equalize balances until all the same or below threshold
|
// Equalize balances until all the same or below threshold
|
||||||
|
|
@ -1550,8 +1532,7 @@ func (pool *TxPool) truncateQueue() {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Sort all accounts with queued transactions by heartbeat
|
// Sort all accounts with queued transactions by heartbeat
|
||||||
addresses := addrBeatPool.Get().(addressesByHeartbeat)
|
addresses := make(addressesByHeartbeat, 0, len(pool.queue))
|
||||||
defer addrBeatPool.Put(addresses[:0])
|
|
||||||
for addr := range pool.queue {
|
for addr := range pool.queue {
|
||||||
if !pool.locals.contains(addr) { // don't drop locals
|
if !pool.locals.contains(addr) { // don't drop locals
|
||||||
addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]})
|
addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]})
|
||||||
|
|
@ -1701,10 +1682,7 @@ func (as *accountSet) containsTx(tx *types.Transaction) bool {
|
||||||
// add inserts a new address into the set to track.
|
// add inserts a new address into the set to track.
|
||||||
func (as *accountSet) add(addr common.Address) {
|
func (as *accountSet) add(addr common.Address) {
|
||||||
as.accounts[addr] = struct{}{}
|
as.accounts[addr] = struct{}{}
|
||||||
if as.cache != nil {
|
as.cache = nil
|
||||||
addrsPool.Put((*as.cache)[:0])
|
|
||||||
as.cache = nil
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// addTx adds the sender of tx into the set.
|
// addTx adds the sender of tx into the set.
|
||||||
|
|
@ -1718,7 +1696,7 @@ func (as *accountSet) addTx(tx *types.Transaction) {
|
||||||
// reuse. The returned slice should not be changed!
|
// reuse. The returned slice should not be changed!
|
||||||
func (as *accountSet) flatten() []common.Address {
|
func (as *accountSet) flatten() []common.Address {
|
||||||
if as.cache == nil {
|
if as.cache == nil {
|
||||||
accounts := addrsPool.Get().([]common.Address)
|
accounts := make([]common.Address, 0, len(as.accounts))
|
||||||
for account := range as.accounts {
|
for account := range as.accounts {
|
||||||
accounts = append(accounts, account)
|
accounts = append(accounts, account)
|
||||||
}
|
}
|
||||||
|
|
@ -1732,10 +1710,7 @@ func (as *accountSet) merge(other *accountSet) {
|
||||||
for addr := range other.accounts {
|
for addr := range other.accounts {
|
||||||
as.accounts[addr] = struct{}{}
|
as.accounts[addr] = struct{}{}
|
||||||
}
|
}
|
||||||
if as.cache != nil {
|
as.cache = nil
|
||||||
addrsPool.Put((*as.cache)[:0])
|
|
||||||
as.cache = nil
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// txLookup is used internally by TxPool to track transactions while allowing
|
// txLookup is used internally by TxPool to track transactions while allowing
|
||||||
|
|
|
||||||
|
|
@ -24,7 +24,7 @@ import (
|
||||||
const (
|
const (
|
||||||
VersionMajor = 5 // Major version component of the current release
|
VersionMajor = 5 // Major version component of the current release
|
||||||
VersionMinor = 1 // Minor version component of the current release
|
VersionMinor = 1 // Minor version component of the current release
|
||||||
VersionPatch = 32 // Patch version component of the current release
|
VersionPatch = 33 // Patch version component of the current release
|
||||||
VersionMeta = "mainnet" // Version metadata to append to the version string
|
VersionMeta = "mainnet" // Version metadata to append to the version string
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue