added throttling by num transactions per account past hour

node/node.go

@@ -255,17 +255,21 @@ func (node *Node) AddPendingTransaction(newTx *types.Transaction) {
 // Note the pending transaction list will then contain the rest of the txs
 func (node *Node) getTransactionsForNewBlock(coinbase common.Address) types.Transactions {
 	node.pendingTxMutex.Lock()
-	selected, unselected, invalid, blockTxsCounts := node.Worker.SelectTransactionsForNewBlock(node.pendingTransactions, core.ShardingSchedule.TxsThrottleConfig(), coinbase)
 
-	node.pendingTransactions = unselected
-	node.reducePendingTransactions()
-	node.recentTxsStats[node.Consensus.ChainReader.CurrentHeader().Number.Uint64()+1] = blockTxsCounts
+	// update recentTxsStats and initiailize for the new block
+	newBlockNum := node.Consensus.ChainReader.CurrentHeader().Number.Uint64() + 1
 	for blockNum := range node.recentTxsStats {
-		blockNumPastHour := (time.Hour / time.Second) / node.BlockPeriod
-		if blockNum < node.Consensus.ChainReader.CurrentHeader().Number.Uint64()-uint64(blockNumPastHour) {
+		blockNumHourAgo := (time.Hour / time.Second) / node.BlockPeriod
+		if blockNum < node.Consensus.ChainReader.CurrentHeader().Number.Uint64()-uint64(blockNumHourAgo) {
 			delete(node.recentTxsStats, blockNum)
 		}
 	}
+	node.recentTxsStats[newBlockNum] = types.BlockTxsCounts{}
+
+	selected, unselected, invalid := node.Worker.SelectTransactionsForNewBlock(newBlockNum, node.pendingTransactions, node.recentTxsStats, core.ShardingSchedule.TxsThrottleConfig(), coinbase)
+
+	node.pendingTransactions = unselected
+	node.reducePendingTransactions()
 
 	utils.Logger().Info().
 		Int("remainPending", len(node.pendingTransactions)).

node/worker/worker.go

@@ -46,7 +46,7 @@ type Worker struct {
 // Returns a tuple where the first value is the txs sender account address,
 // the second is the throttling result enum for the transaction of interest.
 // Throttling happens based on the amount, frequency, etc.
-func (w *Worker) throttleTxs(selected types.Transactions, txsThrottleConfig *shardingconfig.TxsThrottleConfig, txnCnts types.BlockTxsCounts, tx *types.Transaction) (common.Address, shardingconfig.TxThrottleFlag) {
+func (w *Worker) throttleTxs(selected types.Transactions, recentTxsStats types.RecentTxsStats, txsThrottleConfig *shardingconfig.TxsThrottleConfig, tx *types.Transaction) (common.Address, shardingconfig.TxThrottleFlag) {
 	chainID := tx.ChainID()
 	// Depending on the presence of the chain ID, sign with EIP155 or homestead
 	var s types.Signer
@@ -82,7 +82,11 @@ func (w *Worker) throttleTxs(selected types.Transactions, txsThrottleConfig *sha
 	}
 
 	// throttle too large transaction
-	if txnCnts[sender] >= (*txsThrottleConfig).MaxTxsPerAccountInBlockLimit {
+	var numTxsPastHour uint64
+	for _, blockTxsCounts := range recentTxsStats {
+		numTxsPastHour += blockTxsCounts[sender]
+	}
+	if numTxsPastHour >= (*txsThrottleConfig).MaxTxsPerAccountInBlockLimit {
 		utils.Logger().Debug().
 			Uint64("MaxTxsPerAccountInBlockLimit", (*txsThrottleConfig).MaxTxsPerAccountInBlockLimit).
 			Msg("Throttling tx with max txs per account in a single block limit")
@@ -93,14 +97,11 @@ func (w *Worker) throttleTxs(selected types.Transactions, txsThrottleConfig *sha
 }
 
 // SelectTransactionsForNewBlock selects transactions for new block.
-func (w *Worker) SelectTransactionsForNewBlock(txs types.Transactions, txsThrottleConfig *shardingconfig.TxsThrottleConfig, coinbase common.Address) (types.Transactions, types.Transactions, types.Transactions, types.BlockTxsCounts) {
+func (w *Worker) SelectTransactionsForNewBlock(newBlockNum uint64, txs types.Transactions, recentTxsStats types.RecentTxsStats, txsThrottleConfig *shardingconfig.TxsThrottleConfig, coinbase common.Address) (types.Transactions, types.Transactions, types.Transactions) {
 	if w.current.gasPool == nil {
 		w.current.gasPool = new(core.GasPool).AddGas(w.current.header.GasLimit)
 	}
 
-	// used for per account transaction frequency throttling
-	txnCnts := types.BlockTxsCounts{}
-
 	selected := types.Transactions{}
 	unselected := types.Transactions{}
 	invalid := types.Transactions{}
@@ -109,7 +110,7 @@ func (w *Worker) SelectTransactionsForNewBlock(txs types.Transactions, txsThrott
 			invalid = append(invalid, tx)
 		}
 
-		sender, flag := w.throttleTxs(selected, txsThrottleConfig, txnCnts, tx)
+		sender, flag := w.throttleTxs(selected, recentTxsStats, txsThrottleConfig, tx)
 		switch flag {
 		case shardingconfig.Unselect:
 			unselected = append(unselected, tx)
@@ -138,11 +139,11 @@ func (w *Worker) SelectTransactionsForNewBlock(txs types.Transactions, txsThrott
 					Msg("Transaction Throttle flag")
 			} else {
 				selected = append(selected, tx)
-				txnCnts[sender]++
+				recentTxsStats[newBlockNum][sender]++
 			}
 		}
 	}
-	return selected, unselected, invalid, txnCnts
+	return selected, unselected, invalid
 }
 
 func (w *Worker) commitTransaction(tx *types.Transaction, coinbase common.Address) ([]*types.Log, error) {