From 6a12df288f2e91a9b961640be89ac95947957300 Mon Sep 17 00:00:00 2001 From: Shahar Papini Date: Thu, 23 May 2019 11:03:46 +0300 Subject: [PATCH] core: Limit total size of transaction pool --- core/tx_list.go | 44 ++++++++++++++++++++++++++++ core/tx_pool.go | 47 +++++++++++++++++++++++++++--- core/tx_pool_test.go | 69 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 156 insertions(+), 4 deletions(-) diff --git a/core/tx_list.go b/core/tx_list.go index 75bfdaedac..133f57881d 100644 --- a/core/tx_list.go +++ b/core/tx_list.go @@ -466,6 +466,50 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact return drop } +// FreeSize drops transactions that sum up to at least a certain total size. +func (l *txPricedList) FreeSize(sizeToFree common.StorageSize, threshold *big.Int, locals *accountSet) (types.Transactions, bool) { + drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop + save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep + totalSize := common.StorageSize(0) + + for len(*l.items) > 0 { + // Discard stale transactions if found during cleanup + tx := heap.Pop(l.items).(*types.Transaction) + if l.all.Get(tx.Hash()) == nil { + l.stales-- + continue + } + // Stop the discards if we've reached the threshold + if tx.GasPrice().Cmp(threshold) >= 0 { + save = append(save, tx) + break + } + // Keep local + if locals.containsTx(tx) { + save = append(save, tx) + continue + } + // Non stale transaction found + drop = append(drop, tx) + totalSize += tx.Size() + // Stop the discards if we've reached the required size + if totalSize >= sizeToFree { + break + } + } + for _, tx := range save { + heap.Push(l.items, tx) + } + // If not enough space, bring back removed transactions + if totalSize < sizeToFree { + for _, tx := range drop { + heap.Push(l.items, tx) + } + return make(types.Transactions, 0), false + } + return drop, true +} + // Underpriced checks whether a transaction is cheaper than (or as cheap as) the // lowest priced transaction currently being tracked. func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) bool { diff --git a/core/tx_pool.go b/core/tx_pool.go index 0c422dd99d..df70294aab 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -48,6 +48,10 @@ var ( // one present in the local chain. ErrNonceTooLow = errors.New("nonce too low") + // ErrOverflown is returned if a transaction's size will overflow the allowed memory + // usage. + ErrOverflown = errors.New("transaction overflows memory") + // ErrUnderpriced is returned if a transaction's gas price is below the minimum // configured for the transaction pool. ErrUnderpriced = errors.New("transaction underpriced") @@ -99,6 +103,7 @@ var ( // General tx metrics validMeter = metrics.NewRegisteredMeter("txpool/valid", nil) invalidTxMeter = metrics.NewRegisteredMeter("txpool/invalid", nil) + overflownTxMeter = metrics.NewRegisteredMeter("txpool/overflown", nil) underpricedTxMeter = metrics.NewRegisteredMeter("txpool/underpriced", nil) pendingCounter = metrics.NewRegisteredCounter("txpool/pending", nil) @@ -141,6 +146,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 + MemoryLimit common.StorageSize // Limit on the total size of transactions in the pool. + Lifetime time.Duration // Maximum amount of time non-executable transaction are queued } @@ -158,6 +165,9 @@ var DefaultTxPoolConfig = TxPoolConfig{ AccountQueue: 64, GlobalQueue: 1024, + // Default 160MB total tx size limit + MemoryLimit: common.StorageSize(160 * 1024 * 1024), + Lifetime: 3 * time.Hour, } @@ -507,8 +517,8 @@ func (pool *TxPool) local() map[common.Address]types.Transactions { // validateTx checks whether a transaction is valid according to the consensus // rules and adheres to some heuristic limits of the local node (price and size). func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { - // Heuristic limit, reject transactions over 32KB to prevent DOS attacks - if tx.Size() > 32*1024 { + // Heuristic limit, reject transactions over 120KB to prevent DOS attacks + if tx.Size() > 120*1024 { return ErrOversizedData } // Transactions can't be negative. This may never happen using RLP decoded @@ -588,6 +598,32 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e pool.removeTx(tx.Hash(), false) } } + // If there isn't enough free size in pool, discard underpriced transactions + if pool.all.memory+tx.Size() > pool.config.MemoryLimit { + sizeToFree := pool.all.memory + tx.Size() - pool.config.MemoryLimit + gasPrice := tx.GasPrice() + if local { + // TODO: infinity. + gasPrice = big.NewInt(1000000000) + } + drop, success := pool.priced.FreeSize(sizeToFree, gasPrice, pool.locals) + if !local && !success { + log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice()) + overflownTxMeter.Mark(1) + + // Bring back freed transactions + for _, tx := range drop { + pool.priced.Put(tx) + } + return false, ErrOverflown + } + // New transaction is better than our worse ones, make room for it + for _, tx := range drop { + log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice()) + overflownTxMeter.Mark(1) + pool.removeTx(tx.Hash(), false) + } + } // Try to replace an existing transaction in the pending pool from, _ := types.Sender(pool.signer, tx) // already validated @@ -1461,8 +1497,9 @@ func (as *accountSet) merge(other *accountSet) { // peeking into the pool in TxPool.Get without having to acquire the widely scoped // TxPool.mu mutex. type txLookup struct { - all map[common.Hash]*types.Transaction - lock sync.RWMutex + all map[common.Hash]*types.Transaction + memory common.StorageSize + lock sync.RWMutex } // newTxLookup returns a new txLookup structure. @@ -1505,6 +1542,7 @@ func (t *txLookup) Add(tx *types.Transaction) { t.lock.Lock() defer t.lock.Unlock() + t.memory += tx.Size() t.all[tx.Hash()] = tx } @@ -1513,5 +1551,6 @@ func (t *txLookup) Remove(hash common.Hash) { t.lock.Lock() defer t.lock.Unlock() + t.memory -= t.all[hash].Size() delete(t.all, hash) } diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go index 388668ed8b..d8156815c2 100644 --- a/core/tx_pool_test.go +++ b/core/tx_pool_test.go @@ -77,6 +77,13 @@ func pricedTransaction(nonce uint64, gaslimit uint64, gasprice *big.Int, key *ec return tx } +func pricedDataTransaction(nonce uint64, gaslimit uint64, gasprice *big.Int, key *ecdsa.PrivateKey, numDataBytes uint64) *types.Transaction { + data := make([]byte, numDataBytes) + rand.Read(data) + tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(100), gaslimit, gasprice, data), types.HomesteadSigner{}, key) + return tx +} + func setupTxPool() (*TxPool, *ecdsa.PrivateKey) { statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase())) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} @@ -113,6 +120,15 @@ func validateTxPoolInternals(pool *TxPool) error { return fmt.Errorf("pending nonce mismatch: have %v, want %v", nonce, last+1) } } + // Ensure the total size of the transactions match the running counter + memory := common.StorageSize(0) + for _, tx := range pool.all.all { + memory += tx.Size() + } + if pool.all.memory != memory { + return fmt.Errorf("tracked total transaction size %d != calculated total size %d", + uint64(pool.all.memory), uint64(memory)) + } return nil } @@ -1002,6 +1018,59 @@ func TestTransactionPendingGlobalLimiting(t *testing.T) { } } +// Tests that if the total transaction size in the pool goes above some +// hard threshold, the higher transactions are dropped to prevent DOS attacks. +func TestTransactionTotalSizeLimiting(t *testing.T) { + t.Parallel() + + // Create the pool to test the limit enforcement with + statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase())) + blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} + + config := testTxPoolConfig + config.MemoryLimit = common.StorageSize(10 * 1000) + + pool := NewTxPool(config, params.TestChainConfig, blockchain) + defer pool.Stop() + + // Create a number of test accounts and fund them + keys := make([]*ecdsa.PrivateKey, 5) + for i := 0; i < len(keys); i++ { + keys[i], _ = crypto.GenerateKey() + pool.currentState.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(config.GlobalSlots)/len(keys)*2; j++ { + datalen := uint64(rand.Intn(1000)) + price := big.NewInt(int64(rand.Intn(10))) + tx := pricedDataTransaction(nonces[addr], 100000, price, key, datalen) + txs = append(txs, tx) + nonces[addr]++ + } + } + // Import the batch and verify that limits have been enforced + pool.AddRemotes(txs) + + if pool.all.memory > config.MemoryLimit { + t.Fatalf("Pool size too large: %d. Expected at most %d", uint64(pool.all.memory), uint64(config.MemoryLimit)) + } + if err := validateTxPoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + + for _, tx := range pool.all.all { + // Remaining transactions should all have high price. The highest in our range is 9. + if tx.GasPrice().Int64() < 9 { + t.Fatalf("Remaining transaction with low price") + } + } +} + // Tests that if transactions start being capped, transactions are also removed from 'all' func TestTransactionCapClearsFromAll(t *testing.T) { t.Parallel()