Merge branch 'develop' into omerfirmak/mpt

Author: omerfirmak

cmd/geth/main.go

@@ -90,6 +90,7 @@ var (
 		utils.TxPoolGlobalSlotsFlag,
 		utils.TxPoolAccountQueueFlag,
 		utils.TxPoolGlobalQueueFlag,
+		utils.TxPoolAccountPendingLimitFlag,
 		utils.TxPoolLifetimeFlag,
 		utils.SyncModeFlag,
 		utils.ExitWhenSyncedFlag,

cmd/geth/usage.go

@@ -109,6 +109,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.TxPoolGlobalSlotsFlag,
 			utils.TxPoolAccountQueueFlag,
 			utils.TxPoolGlobalQueueFlag,
+			utils.TxPoolAccountPendingLimitFlag,
 			utils.TxPoolLifetimeFlag,
 		},
 	},

cmd/utils/flags.go

@@ -406,6 +406,11 @@ var (
 		Usage: "Maximum number of non-executable transaction slots for all accounts",
 		Value: ethconfig.Defaults.TxPool.GlobalQueue,
 	}
+	TxPoolAccountPendingLimitFlag = cli.Uint64Flag{
+		Name:  "txpool.accountpendinglimit",
+		Usage: "Maximum number of executable transactions allowed per account",
+		Value: ethconfig.Defaults.TxPool.AccountPendingLimit,
+	}
 	TxPoolLifetimeFlag = cli.DurationFlag{
 		Name:  "txpool.lifetime",
 		Usage: "Maximum amount of time non-executable transaction are queued",
@@ -1523,6 +1528,9 @@ func setTxPool(ctx *cli.Context, cfg *core.TxPoolConfig) {
 	if ctx.GlobalIsSet(TxPoolGlobalQueueFlag.Name) {
 		cfg.GlobalQueue = ctx.GlobalUint64(TxPoolGlobalQueueFlag.Name)
 	}
+	if ctx.GlobalIsSet(TxPoolAccountPendingLimitFlag.Name) {
+		cfg.AccountPendingLimit = ctx.GlobalUint64(TxPoolAccountPendingLimitFlag.Name)
+	}
 	if ctx.GlobalIsSet(TxPoolLifetimeFlag.Name) {
 		cfg.Lifetime = ctx.GlobalDuration(TxPoolLifetimeFlag.Name)
 	}

core/tx_list.go

@@ -439,7 +439,11 @@ func (l *txList) Ready(start uint64) types.Transactions {
 
 // Len returns the length of the transaction list.
 func (l *txList) Len() int {
-	return l.txs.Len()
+	if l == nil {
+		return 0
+	} else {
+		return l.txs.Len()
+	}
 }
 
 // Empty returns whether the list of transactions is empty or not.

core/tx_pool.go

@@ -104,6 +104,7 @@ var (
 	pendingReplaceMeter   = metrics.NewRegisteredMeter("txpool/pending/replace", nil)
 	pendingRateLimitMeter = metrics.NewRegisteredMeter("txpool/pending/ratelimit", nil) // Dropped due to rate limiting
 	pendingNofundsMeter   = metrics.NewRegisteredMeter("txpool/pending/nofunds", nil)   // Dropped due to out-of-funds
+	pendingEvictionMeter  = metrics.NewRegisteredMeter("txpool/pending/eviction", nil)  // Dropped due to lifetime
 
 	// Metrics for the queued pool
 	queuedDiscardMeter   = metrics.NewRegisteredMeter("txpool/queued/discard", nil)
@@ -178,6 +179,8 @@ type TxPoolConfig struct {
 	AccountQueue uint64 // Maximum number of non-executable transaction slots permitted per account
 	GlobalQueue  uint64 // Maximum number of non-executable transaction slots for all accounts
 
+	AccountPendingLimit uint64 // Number of executable transactions allowed per account
+
 	Lifetime time.Duration // Maximum amount of time non-executable transaction are queued
 }
 
@@ -195,6 +198,8 @@ var DefaultTxPoolConfig = TxPoolConfig{
 	AccountQueue: 64,
 	GlobalQueue:  1024,
 
+	AccountPendingLimit: 1024,
+
 	Lifetime: 3 * time.Hour,
 }
 
@@ -230,6 +235,10 @@ func (config *TxPoolConfig) sanitize() TxPoolConfig {
 		log.Warn("Sanitizing invalid txpool global queue", "provided", conf.GlobalQueue, "updated", DefaultTxPoolConfig.GlobalQueue)
 		conf.GlobalQueue = DefaultTxPoolConfig.GlobalQueue
 	}
+	if conf.AccountPendingLimit < 1 {
+		log.Warn("Sanitizing invalid txpool account pending limit", "provided", conf.AccountPendingLimit, "updated", DefaultTxPoolConfig.AccountPendingLimit)
+		conf.AccountPendingLimit = DefaultTxPoolConfig.AccountPendingLimit
+	}
 	if conf.Lifetime < 1 {
 		log.Warn("Sanitizing invalid txpool lifetime", "provided", conf.Lifetime, "updated", DefaultTxPoolConfig.Lifetime)
 		conf.Lifetime = DefaultTxPoolConfig.Lifetime
@@ -422,6 +431,7 @@ func (pool *TxPool) loop() {
 		// Handle inactive account transaction eviction
 		case <-evict.C:
 			pool.mu.Lock()
+			// Evict queued transactions
 			for addr := range pool.queue {
 				// Skip local transactions from the eviction mechanism
 				if pool.locals.contains(addr) {
@@ -431,12 +441,28 @@ func (pool *TxPool) loop() {
 				if time.Since(pool.beats[addr]) > pool.config.Lifetime {
 					list := pool.queue[addr].Flatten()
 					for _, tx := range list {
-						log.Trace("Evicting transaction due to timeout", "account", addr.Hex(), "hash", tx.Hash().Hex(), "lifetime sec", time.Since(pool.beats[addr]).Seconds(), "lifetime limit sec", pool.config.Lifetime.Seconds())
+						log.Trace("Evicting queued transaction due to timeout", "account", addr.Hex(), "hash", tx.Hash().Hex(), "lifetime sec", time.Since(pool.beats[addr]).Seconds(), "lifetime limit sec", pool.config.Lifetime.Seconds())
 						pool.removeTx(tx.Hash(), true)
 					}
 					queuedEvictionMeter.Mark(int64(len(list)))
 				}
 			}
+			// Evict pending transactions
+			for addr := range pool.pending {
+				// Skip local transactions from the eviction mechanism
+				if pool.locals.contains(addr) {
+					continue
+				}
+				// Any non-locals old enough should be removed
+				if time.Since(pool.beats[addr]) > pool.config.Lifetime {
+					list := pool.pending[addr].Flatten()
+					for _, tx := range list {
+						log.Trace("Evicting pending transaction due to timeout", "account", addr.Hex(), "hash", tx.Hash().Hex(), "lifetime sec", time.Since(pool.beats[addr]).Seconds(), "lifetime limit sec", pool.config.Lifetime.Seconds())
+						pool.removeTx(tx.Hash(), true)
+					}
+					pendingEvictionMeter.Mark(int64(len(list)))
+				}
+			}
 			pool.mu.Unlock()
 
 		// Handle local transaction journal rotation
@@ -957,6 +983,15 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	}
 	list := pool.pending[addr]
 
+	// Account pending list is full
+	if uint64(list.Len()) >= pool.config.AccountPendingLimit {
+		pool.all.Remove(hash)
+		pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
+		pool.priced.Removed(1)
+		pendingDiscardMeter.Mark(1)
+		return false
+	}
+
 	inserted, old := list.Add(tx, pool.currentState, pool.config.PriceBump, pool.chainconfig, pool.currentHead)
 	if !inserted {
 		// An older transaction was better, discard this
@@ -1153,6 +1188,12 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 
 	addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
 
+	defer func(addr common.Address) {
+		if pool.queue[addr].Empty() && pool.pending[addr].Empty() {
+			delete(pool.beats, addr)
+		}
+	}(addr)
+
 	// Remove it from the list of known transactions
 	pool.all.Remove(hash)
 	pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
@@ -1189,7 +1230,6 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 		}
 		if future.Empty() {
 			delete(pool.queue, addr)
-			delete(pool.beats, addr)
 		}
 	}
 }
@@ -1544,6 +1584,8 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 		// Delete the entire queue entry if it became empty.
 		if list.Empty() {
 			delete(pool.queue, addr)
+		}
+		if pool.queue[addr].Empty() && pool.pending[addr].Empty() {
 			delete(pool.beats, addr)
 		}
 	}
@@ -1574,6 +1616,29 @@ func (pool *TxPool) executableTxFilter(costLimit *big.Int) func(tx *types.Transa
 // pending limit. The algorithm tries to reduce transaction counts by an approximately
 // equal number for all for accounts with many pending transactions.
 func (pool *TxPool) truncatePending() {
+	// Truncate pending lists to max length
+	for addr, list := range pool.pending {
+		if list.Len() > int(pool.config.AccountPendingLimit) {
+			caps := list.Cap(int(pool.config.AccountPendingLimit))
+			for _, tx := range caps {
+				// Drop the transaction from the global pools too
+				hash := tx.Hash()
+				pool.all.Remove(hash)
+				pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
+
+				// Update the account nonce to the dropped transaction
+				// note: this will set pending nonce to the min nonce from the discarded txs
+				pool.pendingNonces.setIfLower(addr, tx.Nonce())
+				log.Trace("Removed pending transaction to comply with hard limit", "hash", hash.Hex())
+			}
+			pool.priced.Removed(len(caps))
+			pendingGauge.Dec(int64(len(caps)))
+			if pool.locals.contains(addr) {
+				localGauge.Dec(int64(len(caps)))
+			}
+		}
+	}
+
 	pending := uint64(0)
 	for _, list := range pool.pending {
 		pending += uint64(list.Len())