diff --git a/core/tx_pool.go b/core/tx_pool.go
index 4c9410d8ed4c0d79e94d1aeaf34e71a7364cf2d2..2c8a5c396c53a8118324434f1ff266cb0dac1f48 100644
--- a/core/tx_pool.go
+++ b/core/tx_pool.go
@@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
+ "gopkg.in/karalabe/cookiejar.v2/collections/prque"
)
var (
@@ -46,10 +47,12 @@ var (
)
var (
- maxQueuedPerAccount = uint64(64) // Max limit of queued transactions per address
- maxQueuedInTotal = uint64(8192) // Max limit of queued transactions from all accounts
- maxQueuedLifetime = 3 * time.Hour // Max amount of time transactions from idle accounts are queued
- evictionInterval = time.Minute // Time interval to check for evictable transactions
+ minPendingPerAccount = uint64(16) // Min number of guaranteed transaction slots per address
+ maxPendingTotal = uint64(4096) // Max limit of pending transactions from all accounts (soft)
+ maxQueuedPerAccount = uint64(64) // Max limit of queued transactions per address
+ maxQueuedInTotal = uint64(1024) // Max limit of queued transactions from all accounts
+ maxQueuedLifetime = 3 * time.Hour // Max amount of time transactions from idle accounts are queued
+ evictionInterval = time.Minute // Time interval to check for evictable transactions
)
type stateFn func() (*state.StateDB, error)
@@ -481,7 +484,6 @@ func (pool *TxPool) promoteExecutables() {
}
// Iterate over all accounts and promote any executable transactions
queued := uint64(0)
-
for addr, list := range pool.queue {
// Drop all transactions that are deemed too old (low nonce)
for _, tx := range list.Forward(state.GetNonce(addr)) {
@@ -519,6 +521,59 @@ func (pool *TxPool) promoteExecutables() {
delete(pool.queue, addr)
}
}
+ // If the pending limit is overflown, start equalizing allowances
+ pending := uint64(0)
+ for _, list := range pool.pending {
+ pending += uint64(list.Len())
+ }
+ if pending > maxPendingTotal {
+ // Assemble a spam order to penalize large transactors first
+ spammers := prque.New()
+ for addr, list := range pool.pending {
+ // Only evict transactions from high rollers
+ if uint64(list.Len()) > minPendingPerAccount {
+ // Skip local accounts as pools should maintain backlogs for themselves
+ for _, tx := range list.txs.items {
+ if !pool.localTx.contains(tx.Hash()) {
+ spammers.Push(addr, float32(list.Len()))
+ }
+ break // Checking on transaction for locality is enough
+ }
+ }
+ }
+ // Gradually drop transactions from offenders
+ offenders := []common.Address{}
+ for pending > maxPendingTotal && !spammers.Empty() {
+ // Retrieve the next offender if not local address
+ offender, _ := spammers.Pop()
+ offenders = append(offenders, offender.(common.Address))
+
+ // Equalize balances until all the same or below threshold
+ if len(offenders) > 1 {
+ // Calculate the equalization threshold for all current offenders
+ threshold := pool.pending[offender.(common.Address)].Len()
+
+ // Iteratively reduce all offenders until below limit or threshold reached
+ for pending > maxPendingTotal && pool.pending[offenders[len(offenders)-2]].Len() > threshold {
+ for i := 0; i < len(offenders)-1; i++ {
+ list := pool.pending[offenders[i]]
+ list.Cap(list.Len() - 1)
+ pending--
+ }
+ }
+ }
+ }
+ // If still above threshold, reduce to limit or min allowance
+ if pending > maxPendingTotal && len(offenders) > 0 {
+ for pending > maxPendingTotal && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > minPendingPerAccount {
+ for _, addr := range offenders {
+ list := pool.pending[addr]
+ list.Cap(list.Len() - 1)
+ pending--
+ }
+ }
+ }
+ }
// If we've queued more transactions than the hard limit, drop oldest ones
if queued > maxQueuedInTotal {
// Sort all accounts with queued transactions by heartbeat
diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go
index 4bc5aed38ddce0586b6c51229c1acf1e4a3b34c3..dbe6fa635005f163898e032e20b7347acb817163 100644
--- a/core/tx_pool_test.go
+++ b/core/tx_pool_test.go
@@ -618,6 +618,96 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) {
}
}
+// Tests that if the transaction count belonging to multiple accounts go above
+// some hard threshold, the higher transactions are dropped to prevent DOS
+// attacks.
+func TestTransactionPendingGlobalLimiting(t *testing.T) {
+ // Reduce the queue limits to shorten test time
+ defer func(old uint64) { maxPendingTotal = old }(maxPendingTotal)
+ maxPendingTotal = minPendingPerAccount * 10
+
+ // Create the pool to test the limit enforcement with
+ db, _ := ethdb.NewMemDatabase()
+ statedb, _ := state.New(common.Hash{}, db)
+
+ pool := NewTxPool(testChainConfig(), new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
+ pool.resetState()
+
+ // Create a number of test accounts and fund them
+ state, _ := pool.currentState()
+
+ keys := make([]*ecdsa.PrivateKey, 5)
+ for i := 0; i < len(keys); i++ {
+ keys[i], _ = crypto.GenerateKey()
+ state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
+ }
+ // Generate and queue a batch of transactions
+ nonces := make(map[common.Address]uint64)
+
+ txs := types.Transactions{}
+ for _, key := range keys {
+ addr := crypto.PubkeyToAddress(key.PublicKey)
+ for j := 0; j < int(maxPendingTotal)/len(keys)*2; j++ {
+ txs = append(txs, transaction(nonces[addr], big.NewInt(100000), key))
+ nonces[addr]++
+ }
+ }
+ // Import the batch and verify that limits have been enforced
+ pool.AddBatch(txs)
+
+ pending := 0
+ for _, list := range pool.pending {
+ pending += list.Len()
+ }
+ if pending > int(maxPendingTotal) {
+ t.Fatalf("total pending transactions overflow allowance: %d > %d", pending, maxPendingTotal)
+ }
+}
+
+// Tests that if the transaction count belonging to multiple accounts go above
+// some hard threshold, if they are under the minimum guaranteed slot count then
+// the transactions are still kept.
+func TestTransactionPendingMinimumAllowance(t *testing.T) {
+ // Reduce the queue limits to shorten test time
+ defer func(old uint64) { maxPendingTotal = old }(maxPendingTotal)
+ maxPendingTotal = 0
+
+ // Create the pool to test the limit enforcement with
+ db, _ := ethdb.NewMemDatabase()
+ statedb, _ := state.New(common.Hash{}, db)
+
+ pool := NewTxPool(testChainConfig(), new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
+ pool.resetState()
+
+ // Create a number of test accounts and fund them
+ state, _ := pool.currentState()
+
+ keys := make([]*ecdsa.PrivateKey, 5)
+ for i := 0; i < len(keys); i++ {
+ keys[i], _ = crypto.GenerateKey()
+ state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
+ }
+ // Generate and queue a batch of transactions
+ nonces := make(map[common.Address]uint64)
+
+ txs := types.Transactions{}
+ for _, key := range keys {
+ addr := crypto.PubkeyToAddress(key.PublicKey)
+ for j := 0; j < int(minPendingPerAccount)*2; j++ {
+ txs = append(txs, transaction(nonces[addr], big.NewInt(100000), key))
+ nonces[addr]++
+ }
+ }
+ // Import the batch and verify that limits have been enforced
+ pool.AddBatch(txs)
+
+ for addr, list := range pool.pending {
+ if list.Len() != int(minPendingPerAccount) {
+ t.Errorf("addr %x: total pending transactions mismatch: have %d, want %d", addr, list.Len(), minPendingPerAccount)
+ }
+ }
+}
+
// Benchmarks the speed of validating the contents of the pending queue of the
// transaction pool.
func BenchmarkPendingDemotion100(b *testing.B) { benchmarkPendingDemotion(b, 100) }