Upgrade txpool module to increase tps (ethereum#43)

core/tx_pool.go

@@ -129,6 +129,19 @@ var (
  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.
 type TxStatus uint
 
@@ -267,6 +280,8 @@ type TxPool struct {
  wg sync.WaitGroup // tracks loop, scheduleReorgLoop
  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.
 }
 
@@ -298,6 +313,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
  reorgShutdownCh: make(chan struct{}),
  initDoneCh: make(chan struct{}),
  gasPrice: new(big.Int).SetUint64(config.PriceLimit),
+ spammers: prque.New(nil),
  }
  pool.locals = newAccountSet(pool.signer)
  for _, addr := range config.Locals {
@@ -1166,13 +1182,14 @@ func (pool *TxPool) runReorg(done chan struct{}, reset *txpoolResetRequest, dirt
  }
  }
  // Reset needs promote for all addresses
- promoteAddrs = make([]common.Address, 0, len(pool.queue))
+ promoteAddrs = addrsPool.Get().([]common.Address)
  for addr := range pool.queue {
  promoteAddrs = append(promoteAddrs, addr)
  }
  }
  // Check for pending transactions for every account that sent new ones
  promoted := pool.promoteExecutables(promoteAddrs)
+ defer addrsPool.Put(promoteAddrs[:0])
 
  // 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
@@ -1387,18 +1404,19 @@ func (pool *TxPool) truncatePending() {
 
  pendingBeforeCap := pending
  // Assemble a spam order to penalize large transactors first
- spammers := prque.New(nil)
+ pool.spammers.Reset()
  for addr, list := range pool.pending {
  // Only evict transactions from high rollers
  if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots {
- spammers.Push(addr, int64(list.Len()))
+ pool.spammers.Push(addr, int64(list.Len()))
  }
  }
  // Gradually drop transactions from offenders
- offenders := []common.Address{}
- for pending > pool.config.GlobalSlots && !spammers.Empty() {
+ offenders := addrsPool.Get().([]common.Address)
+ defer addrsPool.Put(offenders[:0])
+ for pending > pool.config.GlobalSlots && !pool.spammers.Empty() {
  // Retrieve the next offender if not local address
- offender, _ := spammers.Pop()
+ offender, _ := pool.spammers.Pop()
  offenders = append(offenders, offender.(common.Address))
 
  // Equalize balances until all the same or below threshold
@@ -1471,7 +1489,8 @@ func (pool *TxPool) truncateQueue() {
  }
 
  // Sort all accounts with queued transactions by heartbeat
- addresses := make(addressesByHeartbeat, 0, len(pool.queue))
+ addresses := addrBeatPool.Get().(addressesByHeartbeat)
+ defer addrBeatPool.Put(addresses[:0])
  for addr := range pool.queue {
  if !pool.locals.contains(addr) { // don't drop locals
  addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]})
@@ -1620,7 +1639,10 @@ func (as *accountSet) containsTx(tx *types.Transaction) bool {
 // add inserts a new address into the set to track.
 func (as *accountSet) add(addr common.Address) {
  as.accounts[addr] = struct{}{}
- as.cache = nil
+ if as.cache != nil {
+ addrsPool.Put((*as.cache)[:0])
+ as.cache = nil
+ }
 }
 
 // addTx adds the sender of tx into the set.
@@ -1634,7 +1656,7 @@ func (as *accountSet) addTx(tx *types.Transaction) {
 // reuse. The returned slice should not be changed!
 func (as *accountSet) flatten() []common.Address {
  if as.cache == nil {
- accounts := make([]common.Address, 0, len(as.accounts))
+ accounts := addrsPool.Get().([]common.Address)
  for account := range as.accounts {
  accounts = append(accounts, account)
  }
@@ -1648,7 +1670,10 @@ func (as *accountSet) merge(other *accountSet) {
  for addr := range other.accounts {
  as.accounts[addr] = struct{}{}
  }
- as.cache = nil
+ if as.cache != nil {
+ addrsPool.Put((*as.cache)[:0])
+ as.cache = nil
+ }
 }
 
 // txLookup is used internally by TxPool to track transactions while allowing