From 5c3fa5b91bff90419d135fd9ece94ce6ab37838d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 9 Jan 2020 12:53:48 +0200 Subject: [PATCH] core: minor tx slotting polishes, add slot tracking metric --- core/tx_list.go | 10 +-- core/tx_pool.go | 64 ++++++++++--------- core/tx_pool_test.go | 149 ++++++++++++++++++++++--------------------- 3 files changed, 114 insertions(+), 109 deletions(-) diff --git a/core/tx_list.go b/core/tx_list.go index c26d4881bc..6b22cbbebe 100644 --- a/core/tx_list.go +++ b/core/tx_list.go @@ -494,11 +494,11 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) boo // Discard finds a number of most underpriced transactions, removes them from the // priced list and returns them for further removal from the entire pool. -func (l *txPricedList) Discard(numSlots int, local *accountSet) types.Transactions { - drop := make(types.Transactions, 0, numSlots) // Remote underpriced transactions to drop - save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep +func (l *txPricedList) Discard(slots int, local *accountSet) types.Transactions { + drop := make(types.Transactions, 0, slots) // Remote underpriced transactions to drop + save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep - for len(*l.items) > 0 && numSlots > 0 { + for len(*l.items) > 0 && slots > 0 { // Discard stale transactions if found during cleanup tx := heap.Pop(l.items).(*types.Transaction) if l.all.Get(tx.Hash()) == nil { @@ -510,7 +510,7 @@ func (l *txPricedList) Discard(numSlots int, local *accountSet) types.Transactio save = append(save, tx) } else { drop = append(drop, tx) - numSlots -= NumSlots(tx) + slots -= numSlots(tx) } } for _, tx := range save { diff --git a/core/tx_pool.go b/core/tx_pool.go index 8d4bc97279..e95496b13b 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -38,6 +38,18 @@ import ( const ( // chainHeadChanSize is the size of channel listening to ChainHeadEvent. chainHeadChanSize = 10 + + // txSlotSize is used to calculate how many data slots a single transaction + // takes up based on its size. The slots are used as DoS protection, ensuring + // that validating a new transaction remains a constant operation (in reality + // O(maxslots), where max slots are 4 currently). + txSlotSize = 32 * 1024 + + // txMaxSize is the maximum size a single transaction can have. This field has + // non-trivial consequences: larger transactions are significantly harder and + // more expensive to propagate; larger transactions also take more resources + // to validate whether they fit into the pool or not. + txMaxSize = 4 * txSlotSize // 128KB, don't bump without chunking support ) var ( @@ -105,6 +117,7 @@ var ( pendingGauge = metrics.NewRegisteredGauge("txpool/pending", nil) queuedGauge = metrics.NewRegisteredGauge("txpool/queued", nil) localGauge = metrics.NewRegisteredGauge("txpool/local", nil) + slotsGauge = metrics.NewRegisteredGauge("txpool/slots", nil) ) // TxStatus is the current status of a transaction as seen by the pool. @@ -223,7 +236,6 @@ type TxPool struct { currentState *state.StateDB // Current state in the blockchain head pendingNonces *txNoncer // Pending state tracking virtual nonces currentMaxGas uint64 // Current gas limit for transaction caps - maxTxSize uint64 // Maximal size of allowed transaction in bytes locals *accountSet // Set of local transaction to exempt from eviction rules journal *txJournal // Journal of local transaction to back up to disk @@ -271,9 +283,6 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block reorgDoneCh: make(chan chan struct{}), reorgShutdownCh: make(chan struct{}), gasPrice: new(big.Int).SetUint64(config.PriceLimit), - - // Heuristic limit, reject transactions over 32KB to prevent DOS attacks - maxTxSize: 32 * 1024, } pool.locals = newAccountSet(pool.signer) for _, addr := range config.Locals { @@ -515,7 +524,7 @@ func (pool *TxPool) local() map[common.Address]types.Transactions { // rules and adheres to some heuristic limits of the local node (price and size). func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { // Reject transactions over defined size to prevent DOS attacks - if uint64(tx.Size()) > pool.maxTxSize { + if uint64(tx.Size()) > txMaxSize { return ErrOversizedData } // Transactions can't be negative. This may never happen using RLP decoded @@ -587,7 +596,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e return false, ErrUnderpriced } // New transaction is better than our worse ones, make room for it - drop := pool.priced.Discard(pool.all.SlotsUsed()-int(pool.config.GlobalSlots+pool.config.GlobalQueue)+NumSlots(tx), pool.locals) + drop := pool.priced.Discard(pool.all.Slots()-int(pool.config.GlobalSlots+pool.config.GlobalQueue)+numSlots(tx), pool.locals) for _, tx := range drop { log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice()) underpricedTxMeter.Mark(1) @@ -1146,9 +1155,6 @@ func (pool *TxPool) reset(oldHead, newHead *types.Header) { pool.pendingNonces = newTxNoncer(statedb) pool.currentMaxGas = newHead.GasLimit - // Heuristic limit, reject transactions over 4 slots to prevent DOS attacks - pool.maxTxSize = 4 * slotSize - // Inject any transactions discarded due to reorgs log.Debug("Reinjecting stale transactions", "count", len(reinject)) senderCacher.recover(pool.signer, reinject) @@ -1500,16 +1506,15 @@ 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 - numUsedSlots int - lock sync.RWMutex + all map[common.Hash]*types.Transaction + slots int + lock sync.RWMutex } // newTxLookup returns a new txLookup structure. func newTxLookup() *txLookup { return &txLookup{ - all: make(map[common.Hash]*types.Transaction), - numUsedSlots: 0, + all: make(map[common.Hash]*types.Transaction), } } @@ -1541,24 +1546,12 @@ func (t *txLookup) Count() int { return len(t.all) } -// SlotsUsed returns the current number of slots used in the lookup. -func (t *txLookup) SlotsUsed() int { +// Slots returns the current number of slots used in the lookup. +func (t *txLookup) Slots() int { t.lock.RLock() defer t.lock.RUnlock() - return t.numUsedSlots -} - -const ( - // Transactions are kept in slots. - // Each slot is defined to be 32KB. - // This slots mechanism provides protection against - // resources over consumption. - slotSize = 32 * 1024 -) - -func NumSlots(tx *types.Transaction) int { - return int(math.Ceil(float64(tx.Size()) / slotSize)) + return t.slots } // Add adds a transaction to the lookup. @@ -1566,7 +1559,9 @@ func (t *txLookup) Add(tx *types.Transaction) { t.lock.Lock() defer t.lock.Unlock() - t.numUsedSlots += NumSlots(tx) + t.slots += numSlots(tx) + slotsGauge.Update(int64(t.slots)) + t.all[tx.Hash()] = tx } @@ -1575,6 +1570,13 @@ func (t *txLookup) Remove(hash common.Hash) { t.lock.Lock() defer t.lock.Unlock() - t.numUsedSlots -= NumSlots(t.all[hash]) + t.slots -= numSlots(t.all[hash]) + slotsGauge.Update(int64(t.slots)) + delete(t.all, hash) } + +// numSlots calculates the number of slots needed for a single transaction. +func numSlots(tx *types.Transaction) int { + return int((tx.Size() + txSlotSize - 1) / txSlotSize) +} diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go index 3d217c0122..4db3e6deef 100644 --- a/core/tx_pool_test.go +++ b/core/tx_pool_test.go @@ -77,16 +77,17 @@ 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) +func pricedDataTransaction(nonce uint64, gaslimit uint64, gasprice *big.Int, key *ecdsa.PrivateKey, bytes uint64) *types.Transaction { + data := make([]byte, bytes) rand.Read(data) + tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(0), 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)} + blockchain := &testBlockChain{statedb, 10000000, new(event.Feed)} key, _ := crypto.GenerateKey() pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) @@ -472,7 +473,7 @@ func TestTransactionDropping(t *testing.T) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000)) // Add some pending and some queued transactions @@ -681,7 +682,7 @@ func TestTransactionGapFilling(t *testing.T) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000000)) // Keep track of transaction events to ensure all executables get announced @@ -735,7 +736,7 @@ func TestTransactionQueueAccountLimiting(t *testing.T) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000000)) // Keep queuing up transactions and make sure all above a limit are dropped @@ -761,59 +762,6 @@ func TestTransactionQueueAccountLimiting(t *testing.T) { } } -// Test the limit on transaction size is enforced correctly. -// This test verifies every transaction having allowed size -// is added to the pool, and longer transactions are rejected. -func TestAllowedTxSize(t *testing.T) { - t.Parallel() - - // Create a test account and fund it - pool, key := setupTxPool() - defer pool.Stop() - - // Increase block gas limit to something very big - pool.currentMaxGas = 2 << 30 - - account, _ := deriveSender(transaction(0, 0, key)) - pool.currentState.AddBalance(account, big.NewInt(int64(pool.currentMaxGas)<<30)) - - // Compute maximal data size for transactions (lower bound). - // It is assumed the fields in the transaction (except of the data) are: - // nonce : at most 32 bytes - // gasPrice : at most 32 bytes - // gasLimit : at most 32 bytes - // To address : 20 bytes - // value : at most 32 bytes - // signature : 65 bytes - // All those fields are summed up to at most 213 bytes. - txSizeWithoutData := uint64(213) - maxDataSize := pool.maxTxSize - txSizeWithoutData - - // Provide a legitimate gas price - gasPrice := big.NewInt(1) - - // Try adding a transaction with maximal allowed size - currTx := pricedDataTransaction(0, pool.currentMaxGas, gasPrice, key, maxDataSize) - if err := pool.addRemoteSync(currTx); err != nil { - t.Fatalf("failed to add transaction of size close to maximal: %d, %v", int(currTx.Size()), err) - } - - // Try adding a transaction with random allowed size - if err := pool.addRemoteSync(pricedDataTransaction(1, pool.currentMaxGas, gasPrice, key, uint64(rand.Intn(int(maxDataSize))))); err != nil { - t.Fatalf("failed to add transaction of random allowed size: %v", err) - } - - // Try adding a transaction of minimal not allowed size - if pool.addRemoteSync(pricedDataTransaction(2, pool.currentMaxGas, gasPrice, key, pool.maxTxSize)) == nil { - t.Fatalf("expected rejection on slightly oversize transaction") - } - - // Try adding a transaction of random not allowed size - if pool.addRemoteSync(pricedDataTransaction(3, pool.currentMaxGas, gasPrice, key, maxDataSize+1+uint64(rand.Intn(int(10*pool.maxTxSize))))) == nil { - t.Fatalf("expected rejection on oversize transaction") - } -} - // Tests that if the transaction count belonging to multiple accounts go above // some threshold, the higher transactions are dropped to prevent DOS attacks. // @@ -983,7 +931,7 @@ func TestTransactionPendingLimiting(t *testing.T) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000000)) // Keep track of transaction events to ensure all executables get announced @@ -1062,6 +1010,62 @@ func TestTransactionPendingGlobalLimiting(t *testing.T) { } } +// Test the limit on transaction size is enforced correctly. +// This test verifies every transaction having allowed size +// is added to the pool, and longer transactions are rejected. +func TestTransactionAllowedTxSize(t *testing.T) { + t.Parallel() + + // Create a test account and fund it + pool, key := setupTxPool() + defer pool.Stop() + + account := crypto.PubkeyToAddress(key.PublicKey) + pool.currentState.AddBalance(account, big.NewInt(1000000000)) + + // Compute maximal data size for transactions (lower bound). + // + // It is assumed the fields in the transaction (except of the data) are: + // - nonce <= 32 bytes + // - gasPrice <= 32 bytes + // - gasLimit <= 32 bytes + // - recipient == 20 bytes + // - value <= 32 bytes + // - signature == 65 bytes + // All those fields are summed up to at most 213 bytes. + baseSize := uint64(213) + dataSize := txMaxSize - baseSize + + // Try adding a transaction with maximal allowed size + tx := pricedDataTransaction(0, pool.currentMaxGas, big.NewInt(1), key, dataSize) + if err := pool.addRemoteSync(tx); err != nil { + t.Fatalf("failed to add transaction of size %d, close to maximal: %v", int(tx.Size()), err) + } + // Try adding a transaction with random allowed size + if err := pool.addRemoteSync(pricedDataTransaction(1, pool.currentMaxGas, big.NewInt(1), key, uint64(rand.Intn(int(dataSize))))); err != nil { + t.Fatalf("failed to add transaction of random allowed size: %v", err) + } + // Try adding a transaction of minimal not allowed size + if err := pool.addRemoteSync(pricedDataTransaction(2, pool.currentMaxGas, big.NewInt(1), key, txMaxSize)); err == nil { + t.Fatalf("expected rejection on slightly oversize transaction") + } + // Try adding a transaction of random not allowed size + if err := pool.addRemoteSync(pricedDataTransaction(2, pool.currentMaxGas, big.NewInt(1), key, dataSize+1+uint64(rand.Intn(int(10*txMaxSize))))); err == nil { + t.Fatalf("expected rejection on oversize transaction") + } + // Run some sanity checks on the pool internals + pending, queued := pool.Stats() + if pending != 2 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2) + } + if queued != 0 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 0) + } + if err := validateTxPoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } +} + // Tests that if transactions start being capped, transactions are also removed from 'all' func TestTransactionCapClearsFromAll(t *testing.T) { t.Parallel() @@ -1813,21 +1817,20 @@ func TestTransactionStatusCheck(t *testing.T) { } // Test the transaction slots consumption is computed correctly -func TestNumSlots(t *testing.T) { +func TestTransactionSlotCount(t *testing.T) { t.Parallel() key, _ := crypto.GenerateKey() - tinyTx := pricedDataTransaction(0, 0, big.NewInt(0), key, 0) - if NumSlots(tinyTx) != 1 { - t.Fatalf("Small transactions are expected to consume a single slot.") + // Check that an empty transaction consumes a single slot + smallTx := pricedDataTransaction(0, 0, big.NewInt(0), key, 0) + if slots := numSlots(smallTx); slots != 1 { + t.Fatalf("small transactions slot count mismatch: have %d want %d", slots, 1) } - - numAdditionalSlots := rand.Intn(10) - dataLen := uint64(slotSize * numAdditionalSlots) - severalSlotsTx := pricedDataTransaction(0, 0, big.NewInt(0), key, dataLen) - if actual := NumSlots(severalSlotsTx); actual != 1+numAdditionalSlots { - t.Fatalf("Unexpected slots consumptions: expected %d, actual %d", 1+numAdditionalSlots, actual) + // Check that a large transaction consumes the correct number of slots + bigTx := pricedDataTransaction(0, 0, big.NewInt(0), key, uint64(10*txSlotSize)) + if slots := numSlots(bigTx); slots != 11 { + t.Fatalf("big transactions slot count mismatch: have %d want %d", slots, 11) } } @@ -1842,7 +1845,7 @@ func benchmarkPendingDemotion(b *testing.B, size int) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000000)) for i := 0; i < size; i++ { @@ -1867,7 +1870,7 @@ func benchmarkFuturePromotion(b *testing.B, size int) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000000)) for i := 0; i < size; i++ { @@ -1891,7 +1894,7 @@ func benchmarkPoolBatchInsert(b *testing.B, size int) { pool, key := setupTxPool() defer pool.Stop() - account, _ := deriveSender(transaction(0, 0, key)) + account := crypto.PubkeyToAddress(key.PublicKey) pool.currentState.AddBalance(account, big.NewInt(1000000)) batches := make([]types.Transactions, b.N)