diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 7e0242d8f0..607414bbb6 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -66,6 +66,7 @@ var ( utils.EthashDatasetDirFlag, utils.EthashDatasetsInMemoryFlag, utils.EthashDatasetsOnDiskFlag, + utils.TxPoolNoLocalsFlag, utils.TxPoolPriceLimitFlag, utils.TxPoolPriceBumpFlag, utils.TxPoolAccountSlotsFlag, diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go index 022c8e899e..275aad674b 100644 --- a/cmd/geth/usage.go +++ b/cmd/geth/usage.go @@ -95,6 +95,7 @@ var AppHelpFlagGroups = []flagGroup{ { Name: "TRANSACTION POOL", Flags: []cli.Flag{ + utils.TxPoolNoLocalsFlag, utils.TxPoolPriceLimitFlag, utils.TxPoolPriceBumpFlag, utils.TxPoolAccountSlotsFlag, diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index b5a593ab6e..0159364af7 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -209,6 +209,10 @@ var ( Value: eth.DefaultConfig.EthashDatasetsOnDisk, } // Transaction pool settings + TxPoolNoLocalsFlag = cli.BoolFlag{ + Name: "txpool.nolocals", + Usage: "Disables price exemptions for locally submitted transactions", + } TxPoolPriceLimitFlag = cli.Uint64Flag{ Name: "txpool.pricelimit", Usage: "Minimum gas price limit to enforce for acceptance into the pool", @@ -831,6 +835,9 @@ func setGPO(ctx *cli.Context, cfg *gasprice.Config) { } func setTxPool(ctx *cli.Context, cfg *core.TxPoolConfig) { + if ctx.GlobalIsSet(TxPoolNoLocalsFlag.Name) { + cfg.NoLocals = ctx.GlobalBool(TxPoolNoLocalsFlag.Name) + } if ctx.GlobalIsSet(TxPoolPriceLimitFlag.Name) { cfg.PriceLimit = ctx.GlobalUint64(TxPoolPriceLimitFlag.Name) } diff --git a/core/tx_list.go b/core/tx_list.go index 626d3a3b71..0d87c20bc2 100644 --- a/core/tx_list.go +++ b/core/tx_list.go @@ -420,18 +420,16 @@ func (l *txPricedList) Removed() { heap.Init(l.items) } -// Discard finds all the transactions below the given price threshold, drops them +// Cap finds all the transactions below the given price threshold, drops them // from the priced list and returs them for further removal from the entire pool. -func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions { +func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transactions { drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep for len(*l.items) > 0 { // Discard stale transactions if found during cleanup tx := heap.Pop(l.items).(*types.Transaction) - - hash := tx.Hash() - if _, ok := (*l.all)[hash]; !ok { + if _, ok := (*l.all)[tx.Hash()]; !ok { l.stales-- continue } @@ -440,7 +438,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions break } // Non stale transaction found, discard unless local - if local.contains(hash) { + if local.containsTx(tx) { save = append(save, tx) } else { drop = append(drop, tx) @@ -454,9 +452,9 @@ func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions // 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 *txSet) bool { +func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) bool { // Local transactions cannot be underpriced - if local.contains(tx.Hash()) { + if local.containsTx(tx) { return false } // Discard stale price points if found at the heap start @@ -479,22 +477,20 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *txSet) bool { } // Discard finds a number of most underpriced transactions, removes them from the -// priced list and returs them for further removal from the entire pool. -func (l *txPricedList) Discard(count int, local *txSet) types.Transactions { +// priced list and returns them for further removal from the entire pool. +func (l *txPricedList) Discard(count int, local *accountSet) types.Transactions { drop := make(types.Transactions, 0, count) // Remote underpriced transactions to drop save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep for len(*l.items) > 0 && count > 0 { // Discard stale transactions if found during cleanup tx := heap.Pop(l.items).(*types.Transaction) - - hash := tx.Hash() - if _, ok := (*l.all)[hash]; !ok { + if _, ok := (*l.all)[tx.Hash()]; !ok { l.stales-- continue } // Non stale transaction found, discard unless local - if local.contains(hash) { + if local.containsTx(tx) { save = append(save, tx) } else { drop = append(drop, tx) diff --git a/core/tx_pool.go b/core/tx_pool.go index 2f3cd1e93d..8e2d1b31d8 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -99,6 +99,8 @@ type stateFn func() (*state.StateDB, error) // TxPoolConfig are the configuration parameters of the transaction pool. type TxPoolConfig struct { + NoLocals bool // Whether local transaction handling should be disabled + PriceLimit uint64 // Minimum gas price to enforce for acceptance into the pool PriceBump uint64 // Minimum price bump percentage to replace an already existing transaction (nonce) @@ -155,7 +157,7 @@ type TxPool struct { gasPrice *big.Int eventMux *event.TypeMux events *event.TypeMuxSubscription - locals *txSet + locals *accountSet signer types.Signer mu sync.RWMutex @@ -191,10 +193,10 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, eventMux *e gasLimit: gasLimitFn, gasPrice: new(big.Int).SetUint64(config.PriceLimit), pendingState: nil, - locals: newTxSet(), events: eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}), quit: make(chan struct{}), } + pool.locals = newAccountSet(pool.signer) pool.priced = newTxPricedList(&pool.all) pool.resetState() @@ -237,7 +239,7 @@ func (pool *TxPool) eventLoop() { pool.mu.Unlock() case RemovedTransactionEvent: - pool.AddBatch(ev.Txs) + pool.addTxs(ev.Txs, false) } // Handle stats reporting ticks @@ -371,50 +373,40 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) { return pending, nil } -// SetLocal marks a transaction as local, skipping gas price -// check against local miner minimum in the future -func (pool *TxPool) SetLocal(tx *types.Transaction) { - pool.mu.Lock() - defer pool.mu.Unlock() - pool.locals.add(tx.Hash()) -} - -// validateTx checks whether a transaction is valid according -// to the consensus rules. -func (pool *TxPool) validateTx(tx *types.Transaction) error { - local := pool.locals.contains(tx.Hash()) - // Drop transactions under our own minimal accepted gas price - if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { - return ErrUnderpriced +// 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 { + return ErrOversizedData } - - currentState, err := pool.currentState() - if err != nil { - return err + // Transactions can't be negative. This may never happen using RLP decoded + // transactions but may occur if you create a transaction using the RPC. + if tx.Value().Sign() < 0 { + return ErrNegativeValue } - + // Ensure the transaction doesn't exceed the current block limit gas. + if pool.gasLimit().Cmp(tx.Gas()) < 0 { + return ErrGasLimit + } + // Make sure the transaction is signed properly from, err := types.Sender(pool.signer, tx) if err != nil { return ErrInvalidSender } - // Last but not least check for nonce errors + // Drop non-local transactions under our own minimal accepted gas price + local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network + if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { + return ErrUnderpriced + } + // Ensure the transaction adheres to nonce ordering + currentState, err := pool.currentState() + if err != nil { + return err + } if currentState.GetNonce(from) > tx.Nonce() { return ErrNonceTooLow } - - // Check the transaction doesn't exceed the current - // block limit gas. - if pool.gasLimit().Cmp(tx.Gas()) < 0 { - return ErrGasLimit - } - - // Transactions can't be negative. This may never happen - // using RLP decoded transactions but may occur if you create - // a transaction using the RPC for example. - if tx.Value().Sign() < 0 { - return ErrNegativeValue - } - // Transactor should have enough funds to cover the costs // cost == V + GP * GL if currentState.GetBalance(from).Cmp(tx.Cost()) < 0 { @@ -424,11 +416,6 @@ func (pool *TxPool) validateTx(tx *types.Transaction) error { if tx.Gas().Cmp(intrGas) < 0 { return ErrIntrinsicGas } - - // Heuristic limit, reject transactions over 32KB to prevent DOS attacks - if tx.Size() > 32*1024 { - return ErrOversizedData - } return nil } @@ -436,7 +423,11 @@ func (pool *TxPool) validateTx(tx *types.Transaction) error { // later pending promotion and execution. If the transaction is a replacement for // an already pending or queued one, it overwrites the previous and returns this // so outer code doesn't uselessly call promote. -func (pool *TxPool) add(tx *types.Transaction) (bool, error) { +// +// If a newly added transaction is marked as local, its sending account will be +// whitelisted, preventing any associated transaction from being dropped out of +// the pool due to pricing constraints. +func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { // If the transaction is already known, discard it hash := tx.Hash() if pool.all[hash] != nil { @@ -444,7 +435,7 @@ func (pool *TxPool) add(tx *types.Transaction) (bool, error) { return false, fmt.Errorf("known transaction: %x", hash) } // If the transaction fails basic validation, discard it - if err := pool.validateTx(tx); err != nil { + if err := pool.validateTx(tx, local); err != nil { log.Trace("Discarding invalid transaction", "hash", hash, "err", err) invalidTxCounter.Inc(1) return false, err @@ -486,11 +477,14 @@ func (pool *TxPool) add(tx *types.Transaction) (bool, error) { log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To()) return old != nil, nil } - // New transaction isn't replacing a pending one, push into queue + // New transaction isn't replacing a pending one, push into queue and potentially mark local replace, err := pool.enqueueTx(hash, tx) if err != nil { return false, err } + if local { + pool.locals.add(from) + } log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To()) return replace, nil } @@ -558,13 +552,41 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T go pool.eventMux.Post(TxPreEvent{tx}) } -// Add queues a single transaction in the pool if it is valid. -func (pool *TxPool) Add(tx *types.Transaction) error { +// AddLocal enqueues a single transaction into the pool if it is valid, marking +// the sender as a local one in the mean time, ensuring it goes around the local +// pricing constraints. +func (pool *TxPool) AddLocal(tx *types.Transaction) error { + return pool.addTx(tx, !pool.config.NoLocals) +} + +// AddRemote enqueues a single transaction into the pool if it is valid. If the +// sender is not among the locally tracked ones, full pricing constraints will +// apply. +func (pool *TxPool) AddRemote(tx *types.Transaction) error { + return pool.addTx(tx, false) +} + +// AddLocals enqueues a batch of transactions into the pool if they are valid, +// marking the senders as a local ones in the mean time, ensuring they go around +// the local pricing constraints. +func (pool *TxPool) AddLocals(txs []*types.Transaction) error { + return pool.addTxs(txs, !pool.config.NoLocals) +} + +// AddRemotes enqueues a batch of transactions into the pool if they are valid. +// If the senders are not among the locally tracked ones, full pricing constraints +// will apply. +func (pool *TxPool) AddRemotes(txs []*types.Transaction) error { + return pool.addTxs(txs, false) +} + +// addTx enqueues a single transaction into the pool if it is valid. +func (pool *TxPool) addTx(tx *types.Transaction, local bool) error { pool.mu.Lock() defer pool.mu.Unlock() // Try to inject the transaction and update any state - replace, err := pool.add(tx) + replace, err := pool.add(tx, local) if err != nil { return err } @@ -580,15 +602,15 @@ func (pool *TxPool) Add(tx *types.Transaction) error { return nil } -// AddBatch attempts to queue a batch of transactions. -func (pool *TxPool) AddBatch(txs []*types.Transaction) error { +// addTxs attempts to queue a batch of transactions if they are valid. +func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) error { pool.mu.Lock() defer pool.mu.Unlock() // Add the batch of transaction, tracking the accepted ones dirty := make(map[common.Address]struct{}) for _, tx := range txs { - if replace, err := pool.add(tx); err == nil { + if replace, err := pool.add(tx, local); err == nil { if !replace { from, _ := types.Sender(pool.signer, tx) // already validated dirty[from] = struct{}{} @@ -694,7 +716,6 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A } } // Iterate over all accounts and promote any executable transactions - queued := uint64(0) for _, addr := range accounts { list := pool.queue[addr] if list == nil { @@ -723,15 +744,15 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A pool.promoteTx(addr, hash, tx) } // Drop all transactions over the allowed limit - for _, tx := range list.Cap(int(pool.config.AccountQueue)) { - hash := tx.Hash() - delete(pool.all, hash) - pool.priced.Removed() - queuedRateLimitCounter.Inc(1) - log.Trace("Removed cap-exceeding queued transaction", "hash", hash) + if !pool.locals.contains(addr) { + for _, tx := range list.Cap(int(pool.config.AccountQueue)) { + hash := tx.Hash() + delete(pool.all, hash) + pool.priced.Removed() + queuedRateLimitCounter.Inc(1) + log.Trace("Removed cap-exceeding queued transaction", "hash", hash) + } } - queued += uint64(list.Len()) - // Delete the entire queue entry if it became empty. if list.Empty() { delete(pool.queue, addr) @@ -748,14 +769,8 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A spammers := prque.New() for addr, list := range pool.pending { // Only evict transactions from high rollers - if uint64(list.Len()) > pool.config.AccountSlots { - // Skip local accounts as pools should maintain backlogs for themselves - for _, tx := range list.txs.items { - if !pool.locals.contains(tx.Hash()) { - spammers.Push(addr, float32(list.Len())) - } - break // Checking on transaction for locality is enough - } + if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots { + spammers.Push(addr, float32(list.Len())) } } // Gradually drop transactions from offenders @@ -815,16 +830,22 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A pendingRateLimitCounter.Inc(int64(pendingBeforeCap - pending)) } // If we've queued more transactions than the hard limit, drop oldest ones + queued := uint64(0) + for _, list := range pool.queue { + queued += uint64(list.Len()) + } if queued > pool.config.GlobalQueue { // Sort all accounts with queued transactions by heartbeat addresses := make(addresssByHeartbeat, 0, len(pool.queue)) for addr := range pool.queue { - addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]}) + if !pool.locals.contains(addr) { // don't drop locals + addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]}) + } } sort.Sort(addresses) - // Drop transactions until the total is below the limit - for drop := queued - pool.config.GlobalQueue; drop > 0; { + // Drop transactions until the total is below the limit or only locals remain + for drop := queued - pool.config.GlobalQueue; drop > 0 && len(addresses) > 0; { addr := addresses[len(addresses)-1] list := pool.queue[addr.address] @@ -903,6 +924,11 @@ func (pool *TxPool) expirationLoop() { case <-evict.C: pool.mu.Lock() for addr := range pool.queue { + // 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 { for _, tx := range pool.queue[addr].Flatten() { pool.removeTx(tx.Hash()) @@ -929,48 +955,38 @@ func (a addresssByHeartbeat) Len() int { return len(a) } func (a addresssByHeartbeat) Less(i, j int) bool { return a[i].heartbeat.Before(a[j].heartbeat) } func (a addresssByHeartbeat) Swap(i, j int) { a[i], a[j] = a[j], a[i] } -// txSet represents a set of transaction hashes in which entries -// are automatically dropped after txSetDuration time -type txSet struct { - txMap map[common.Hash]struct{} - txOrd map[uint64]txOrdType - addPtr, delPtr uint64 +// accountSet is simply a set of addresses to check for existance, and a signer +// capable of deriving addresses from transactions. +type accountSet struct { + accounts map[common.Address]struct{} + signer types.Signer } -const txSetDuration = time.Hour * 2 - -// txOrdType represents an entry in the time-ordered list of transaction hashes -type txOrdType struct { - hash common.Hash - time time.Time -} - -// newTxSet creates a new transaction set -func newTxSet() *txSet { - return &txSet{ - txMap: make(map[common.Hash]struct{}), - txOrd: make(map[uint64]txOrdType), +// newAccountSet creates a new address set with an associated signer for sender +// derivations. +func newAccountSet(signer types.Signer) *accountSet { + return &accountSet{ + accounts: make(map[common.Address]struct{}), + signer: signer, } } -// contains returns true if the set contains the given transaction hash -// (not thread safe, should be called from a locked environment) -func (ts *txSet) contains(hash common.Hash) bool { - _, ok := ts.txMap[hash] - return ok +// contains checks if a given address is contained within the set. +func (as *accountSet) contains(addr common.Address) bool { + _, exist := as.accounts[addr] + return exist } -// add adds a transaction hash to the set, then removes entries older than txSetDuration -// (not thread safe, should be called from a locked environment) -func (ts *txSet) add(hash common.Hash) { - ts.txMap[hash] = struct{}{} - now := time.Now() - ts.txOrd[ts.addPtr] = txOrdType{hash: hash, time: now} - ts.addPtr++ - delBefore := now.Add(-txSetDuration) - for ts.delPtr < ts.addPtr && ts.txOrd[ts.delPtr].time.Before(delBefore) { - delete(ts.txMap, ts.txOrd[ts.delPtr].hash) - delete(ts.txOrd, ts.delPtr) - ts.delPtr++ +// containsTx checks if the sender of a given tx is within the set. If the sender +// cannot be derived, this method returns false. +func (as *accountSet) containsTx(tx *types.Transaction) bool { + if addr, err := types.Sender(as.signer, tx); err == nil { + return as.contains(addr) } + return false +} + +// add inserts a new address into the set to track. +func (as *accountSet) add(addr common.Address) { + as.accounts[addr] = struct{}{} } diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go index 4903bc3ca4..03ece3886f 100644 --- a/core/tx_pool_test.go +++ b/core/tx_pool_test.go @@ -47,10 +47,10 @@ func setupTxPool() (*TxPool, *ecdsa.PrivateKey) { statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) key, _ := crypto.GenerateKey() - newPool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) - newPool.resetState() + pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + pool.resetState() - return newPool, key + return pool, key } // validateTxPoolInternals checks various consistency invariants within the pool. @@ -125,17 +125,18 @@ func TestStateChangeDuringPoolReset(t *testing.T) { gasLimitFunc := func() *big.Int { return big.NewInt(1000000000) } - txpool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, mux, stateFunc, gasLimitFunc) - txpool.resetState() + pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, mux, stateFunc, gasLimitFunc) + defer pool.Stop() + pool.resetState() - nonce := txpool.State().GetNonce(address) + nonce := pool.State().GetNonce(address) if nonce != 0 { t.Fatalf("Invalid nonce, want 0, got %d", nonce) } - txpool.AddBatch(types.Transactions{tx0, tx1}) + pool.AddRemotes(types.Transactions{tx0, tx1}) - nonce = txpool.State().GetNonce(address) + nonce = pool.State().GetNonce(address) if nonce != 2 { t.Fatalf("Invalid nonce, want 2, got %d", nonce) } @@ -143,9 +144,9 @@ func TestStateChangeDuringPoolReset(t *testing.T) { // trigger state change in the background trigger = true - txpool.resetState() + pool.resetState() - pendingTx, err := txpool.Pending() + pendingTx, err := pool.Pending() if err != nil { t.Fatalf("Could not fetch pending transactions: %v", err) } @@ -154,7 +155,7 @@ func TestStateChangeDuringPoolReset(t *testing.T) { t.Logf("%0x: %d\n", addr, len(txs)) } - nonce = txpool.State().GetNonce(address) + nonce = pool.State().GetNonce(address) if nonce != 2 { t.Fatalf("Invalid nonce, want 2, got %d", nonce) } @@ -162,42 +163,43 @@ func TestStateChangeDuringPoolReset(t *testing.T) { func TestInvalidTransactions(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() tx := transaction(0, big.NewInt(100), key) from, _ := deriveSender(tx) currentState, _ := pool.currentState() currentState.AddBalance(from, big.NewInt(1)) - if err := pool.Add(tx); err != ErrInsufficientFunds { + if err := pool.AddRemote(tx); err != ErrInsufficientFunds { t.Error("expected", ErrInsufficientFunds) } balance := new(big.Int).Add(tx.Value(), new(big.Int).Mul(tx.Gas(), tx.GasPrice())) currentState.AddBalance(from, balance) - if err := pool.Add(tx); err != ErrIntrinsicGas { + if err := pool.AddRemote(tx); err != ErrIntrinsicGas { t.Error("expected", ErrIntrinsicGas, "got", err) } currentState.SetNonce(from, 1) currentState.AddBalance(from, big.NewInt(0xffffffffffffff)) tx = transaction(0, big.NewInt(100000), key) - if err := pool.Add(tx); err != ErrNonceTooLow { + if err := pool.AddRemote(tx); err != ErrNonceTooLow { t.Error("expected", ErrNonceTooLow) } tx = transaction(1, big.NewInt(100000), key) pool.gasPrice = big.NewInt(1000) - if err := pool.Add(tx); err != ErrUnderpriced { + if err := pool.AddRemote(tx); err != ErrUnderpriced { t.Error("expected", ErrUnderpriced, "got", err) } - - pool.SetLocal(tx) - if err := pool.Add(tx); err != nil { + if err := pool.AddLocal(tx); err != nil { t.Error("expected", nil, "got", err) } } func TestTransactionQueue(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() + tx := transaction(0, big.NewInt(100), key) from, _ := deriveSender(tx) currentState, _ := pool.currentState() @@ -248,6 +250,8 @@ func TestTransactionQueue(t *testing.T) { func TestRemoveTx(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() + addr := crypto.PubkeyToAddress(key.PublicKey) currentState, _ := pool.currentState() currentState.AddBalance(addr, big.NewInt(1)) @@ -277,18 +281,21 @@ func TestRemoveTx(t *testing.T) { func TestNegativeValue(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() tx, _ := types.SignTx(types.NewTransaction(0, common.Address{}, big.NewInt(-1), big.NewInt(100), big.NewInt(1), nil), types.HomesteadSigner{}, key) from, _ := deriveSender(tx) currentState, _ := pool.currentState() currentState.AddBalance(from, big.NewInt(1)) - if err := pool.Add(tx); err != ErrNegativeValue { + if err := pool.AddRemote(tx); err != ErrNegativeValue { t.Error("expected", ErrNegativeValue, "got", err) } } func TestTransactionChainFork(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() + addr := crypto.PubkeyToAddress(key.PublicKey) resetState := func() { db, _ := ethdb.NewMemDatabase() @@ -301,20 +308,22 @@ func TestTransactionChainFork(t *testing.T) { resetState() tx := transaction(0, big.NewInt(100000), key) - if _, err := pool.add(tx); err != nil { + if _, err := pool.add(tx, false); err != nil { t.Error("didn't expect error", err) } pool.RemoveBatch([]*types.Transaction{tx}) // reset the pool's internal state resetState() - if _, err := pool.add(tx); err != nil { + if _, err := pool.add(tx, false); err != nil { t.Error("didn't expect error", err) } } func TestTransactionDoubleNonce(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() + addr := crypto.PubkeyToAddress(key.PublicKey) resetState := func() { db, _ := ethdb.NewMemDatabase() @@ -332,10 +341,10 @@ func TestTransactionDoubleNonce(t *testing.T) { tx3, _ := types.SignTx(types.NewTransaction(0, common.Address{}, big.NewInt(100), big.NewInt(1000000), big.NewInt(1), nil), signer, key) // Add the first two transaction, ensure higher priced stays only - if replace, err := pool.add(tx1); err != nil || replace { + if replace, err := pool.add(tx1, false); err != nil || replace { t.Errorf("first transaction insert failed (%v) or reported replacement (%v)", err, replace) } - if replace, err := pool.add(tx2); err != nil || !replace { + if replace, err := pool.add(tx2, false); err != nil || !replace { t.Errorf("second transaction insert failed (%v) or not reported replacement (%v)", err, replace) } state, _ := pool.currentState() @@ -347,7 +356,7 @@ func TestTransactionDoubleNonce(t *testing.T) { t.Errorf("transaction mismatch: have %x, want %x", tx.Hash(), tx2.Hash()) } // Add the third transaction and ensure it's not saved (smaller price) - pool.add(tx3) + pool.add(tx3, false) pool.promoteExecutables(state, []common.Address{addr}) if pool.pending[addr].Len() != 1 { t.Error("expected 1 pending transactions, got", pool.pending[addr].Len()) @@ -363,11 +372,13 @@ func TestTransactionDoubleNonce(t *testing.T) { func TestMissingNonce(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() + addr := crypto.PubkeyToAddress(key.PublicKey) currentState, _ := pool.currentState() currentState.AddBalance(addr, big.NewInt(100000000000000)) tx := transaction(1, big.NewInt(100000), key) - if _, err := pool.add(tx); err != nil { + if _, err := pool.add(tx, false); err != nil { t.Error("didn't expect error", err) } if len(pool.pending) != 0 { @@ -384,13 +395,15 @@ func TestMissingNonce(t *testing.T) { func TestNonceRecovery(t *testing.T) { const n = 10 pool, key := setupTxPool() + defer pool.Stop() + addr := crypto.PubkeyToAddress(key.PublicKey) currentState, _ := pool.currentState() currentState.SetNonce(addr, n) currentState.AddBalance(addr, big.NewInt(100000000000000)) pool.resetState() tx := transaction(n, big.NewInt(100000), key) - if err := pool.Add(tx); err != nil { + if err := pool.AddRemote(tx); err != nil { t.Error(err) } // simulate some weird re-order of transactions and missing nonce(s) @@ -403,6 +416,8 @@ func TestNonceRecovery(t *testing.T) { func TestRemovedTxEvent(t *testing.T) { pool, key := setupTxPool() + defer pool.Stop() + tx := transaction(0, big.NewInt(1000000), key) from, _ := deriveSender(tx) currentState, _ := pool.currentState() @@ -423,6 +438,8 @@ func TestRemovedTxEvent(t *testing.T) { func TestTransactionDropping(t *testing.T) { // Create a test account and fund it pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() @@ -516,6 +533,8 @@ func TestTransactionDropping(t *testing.T) { func TestTransactionPostponing(t *testing.T) { // Create a test account and fund it pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() @@ -590,6 +609,8 @@ func TestTransactionPostponing(t *testing.T) { func TestTransactionQueueAccountLimiting(t *testing.T) { // Create a test account and fund it pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() @@ -598,7 +619,7 @@ func TestTransactionQueueAccountLimiting(t *testing.T) { // Keep queuing up transactions and make sure all above a limit are dropped for i := uint64(1); i <= DefaultTxPoolConfig.AccountQueue+5; i++ { - if err := pool.Add(transaction(i, big.NewInt(100000), key)); err != nil { + if err := pool.AddRemote(transaction(i, big.NewInt(100000), key)); err != nil { t.Fatalf("tx %d: failed to add transaction: %v", i, err) } if len(pool.pending) != 0 { @@ -621,19 +642,30 @@ func TestTransactionQueueAccountLimiting(t *testing.T) { // Tests that if the transaction count belonging to multiple accounts go above // some threshold, the higher transactions are dropped to prevent DOS attacks. +// +// This logic should not hold for local transactions, unless the local tracking +// mechanism is disabled. func TestTransactionQueueGlobalLimiting(t *testing.T) { - // Reduce the queue limits to shorten test time - defer func(old uint64) { DefaultTxPoolConfig.GlobalQueue = old }(DefaultTxPoolConfig.GlobalQueue) - DefaultTxPoolConfig.GlobalQueue = DefaultTxPoolConfig.AccountQueue * 3 + testTransactionQueueGlobalLimiting(t, false) +} +func TestTransactionQueueGlobalLimitingNoLocals(t *testing.T) { + testTransactionQueueGlobalLimiting(t, true) +} +func testTransactionQueueGlobalLimiting(t *testing.T, nolocals bool) { // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) - pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + config := DefaultTxPoolConfig + config.NoLocals = nolocals + config.GlobalQueue = config.AccountQueue*3 - 1 // reduce the queue limits to shorten test time (-1 to make it non divisible) + + pool := NewTxPool(config, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() - // Create a number of test accounts and fund them + // Create a number of test accounts and fund them (last one will be the local) state, _ := pool.currentState() keys := make([]*ecdsa.PrivateKey, 5) @@ -641,59 +673,132 @@ func TestTransactionQueueGlobalLimiting(t *testing.T) { keys[i], _ = crypto.GenerateKey() state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000)) } + local := keys[len(keys)-1] + // Generate and queue a batch of transactions nonces := make(map[common.Address]uint64) - txs := make(types.Transactions, 0, 3*DefaultTxPoolConfig.GlobalQueue) + txs := make(types.Transactions, 0, 3*config.GlobalQueue) for len(txs) < cap(txs) { - key := keys[rand.Intn(len(keys))] + key := keys[rand.Intn(len(keys)-1)] // skip adding transactions with the local account addr := crypto.PubkeyToAddress(key.PublicKey) txs = append(txs, transaction(nonces[addr]+1, big.NewInt(100000), key)) nonces[addr]++ } // Import the batch and verify that limits have been enforced - pool.AddBatch(txs) + pool.AddRemotes(txs) queued := 0 for addr, list := range pool.queue { - if list.Len() > int(DefaultTxPoolConfig.AccountQueue) { - t.Errorf("addr %x: queued accounts overflown allowance: %d > %d", addr, list.Len(), DefaultTxPoolConfig.AccountQueue) + if list.Len() > int(config.AccountQueue) { + t.Errorf("addr %x: queued accounts overflown allowance: %d > %d", addr, list.Len(), config.AccountQueue) } queued += list.Len() } - if queued > int(DefaultTxPoolConfig.GlobalQueue) { - t.Fatalf("total transactions overflow allowance: %d > %d", queued, DefaultTxPoolConfig.GlobalQueue) + if queued > int(config.GlobalQueue) { + t.Fatalf("total transactions overflow allowance: %d > %d", queued, config.GlobalQueue) + } + // Generate a batch of transactions from the local account and import them + txs = txs[:0] + for i := uint64(0); i < 3*config.GlobalQueue; i++ { + txs = append(txs, transaction(i+1, big.NewInt(100000), local)) + } + pool.AddLocals(txs) + + // If locals are disabled, the previous eviction algorithm should apply here too + if nolocals { + queued := 0 + for addr, list := range pool.queue { + if list.Len() > int(config.AccountQueue) { + t.Errorf("addr %x: queued accounts overflown allowance: %d > %d", addr, list.Len(), config.AccountQueue) + } + queued += list.Len() + } + if queued > int(config.GlobalQueue) { + t.Fatalf("total transactions overflow allowance: %d > %d", queued, config.GlobalQueue) + } + } else { + // Local exemptions are enabled, make sure the local account owned the queue + if len(pool.queue) != 1 { + t.Errorf("multiple accounts in queue: have %v, want %v", len(pool.queue), 1) + } + // Also ensure no local transactions are ever dropped, even if above global limits + if queued := pool.queue[crypto.PubkeyToAddress(local.PublicKey)].Len(); uint64(queued) != 3*config.GlobalQueue { + t.Fatalf("local account queued transaction count mismatch: have %v, want %v", queued, 3*config.GlobalQueue) + } } } // Tests that if an account remains idle for a prolonged amount of time, any // non-executable transactions queued up are dropped to prevent wasting resources // on shuffling them around. -func TestTransactionQueueTimeLimiting(t *testing.T) { - // Reduce the queue limits to shorten test time - defer func(old time.Duration) { DefaultTxPoolConfig.Lifetime = old }(DefaultTxPoolConfig.Lifetime) - defer func(old time.Duration) { evictionInterval = old }(evictionInterval) - DefaultTxPoolConfig.Lifetime = time.Second - evictionInterval = time.Second +// +// This logic should not hold for local transactions, unless the local tracking +// mechanism is disabled. +func TestTransactionQueueTimeLimiting(t *testing.T) { testTransactionQueueTimeLimiting(t, false) } +func TestTransactionQueueTimeLimitingNoLocals(t *testing.T) { testTransactionQueueTimeLimiting(t, true) } - // Create a test account and fund it - pool, key := setupTxPool() - account, _ := deriveSender(transaction(0, big.NewInt(0), key)) +func testTransactionQueueTimeLimiting(t *testing.T, nolocals bool) { + // Reduce the eviction interval to a testable amount + defer func(old time.Duration) { evictionInterval = old }(evictionInterval) + evictionInterval = 250 * time.Millisecond + + // Create the pool to test the non-expiration enforcement + db, _ := ethdb.NewMemDatabase() + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + + config := DefaultTxPoolConfig + config.Lifetime = 250 * time.Millisecond + config.NoLocals = nolocals + + pool := NewTxPool(config, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() + pool.resetState() + + // Create two test accounts to ensure remotes expire but locals do not + local, _ := crypto.GenerateKey() + remote, _ := crypto.GenerateKey() state, _ := pool.currentState() - state.AddBalance(account, big.NewInt(1000000)) + state.AddBalance(crypto.PubkeyToAddress(local.PublicKey), big.NewInt(1000000000)) + state.AddBalance(crypto.PubkeyToAddress(remote.PublicKey), big.NewInt(1000000000)) - // Queue up a batch of transactions - for i := uint64(1); i <= DefaultTxPoolConfig.AccountQueue; i++ { - if err := pool.Add(transaction(i, big.NewInt(100000), key)); err != nil { - t.Fatalf("tx %d: failed to add transaction: %v", i, err) + // Add the two transactions and ensure they both are queued up + if err := pool.AddLocal(pricedTransaction(1, big.NewInt(100000), big.NewInt(1), local)); err != nil { + t.Fatalf("failed to add local transaction: %v", err) + } + if err := pool.AddRemote(pricedTransaction(1, big.NewInt(100000), big.NewInt(1), remote)); err != nil { + t.Fatalf("failed to add remote transaction: %v", err) + } + pending, queued := pool.stats() + if pending != 0 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 0) + } + if queued != 2 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2) + } + if err := validateTxPoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + // Wait a bit for eviction to run and clean up any leftovers, and ensure only the local remains + time.Sleep(2 * config.Lifetime) + + pending, queued = pool.stats() + if pending != 0 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 0) + } + if nolocals { + if queued != 0 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 0) + } + } else { + if queued != 1 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1) } } - // Wait until at least two expiration cycles hit and make sure the transactions are gone - time.Sleep(2 * evictionInterval) - if len(pool.queue) > 0 { - t.Fatalf("old transactions remained after eviction") + if err := validateTxPoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) } } @@ -703,6 +808,8 @@ func TestTransactionQueueTimeLimiting(t *testing.T) { func TestTransactionPendingLimiting(t *testing.T) { // Create a test account and fund it pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() @@ -711,7 +818,7 @@ func TestTransactionPendingLimiting(t *testing.T) { // Keep queuing up transactions and make sure all above a limit are dropped for i := uint64(0); i < DefaultTxPoolConfig.AccountQueue+5; i++ { - if err := pool.Add(transaction(i, big.NewInt(100000), key)); err != nil { + if err := pool.AddRemote(transaction(i, big.NewInt(100000), key)); err != nil { t.Fatalf("tx %d: failed to add transaction: %v", i, err) } if pool.pending[account].Len() != int(i)+1 { @@ -739,7 +846,7 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) { state1.AddBalance(account1, big.NewInt(1000000)) for i := uint64(0); i < DefaultTxPoolConfig.AccountQueue+5; i++ { - if err := pool1.Add(transaction(origin+i, big.NewInt(100000), key1)); err != nil { + if err := pool1.AddRemote(transaction(origin+i, big.NewInt(100000), key1)); err != nil { t.Fatalf("tx %d: failed to add transaction: %v", i, err) } } @@ -753,7 +860,7 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) { for i := uint64(0); i < DefaultTxPoolConfig.AccountQueue+5; i++ { txns = append(txns, transaction(origin+i, big.NewInt(100000), key2)) } - pool2.AddBatch(txns) + pool2.AddRemotes(txns) // Ensure the batch optimization honors the same pool mechanics if len(pool1.pending) != len(pool2.pending) { @@ -777,15 +884,15 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) { // 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) { DefaultTxPoolConfig.GlobalSlots = old }(DefaultTxPoolConfig.GlobalSlots) - DefaultTxPoolConfig.GlobalSlots = DefaultTxPoolConfig.AccountSlots * 10 - // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) - pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + config := DefaultTxPoolConfig + config.GlobalSlots = config.AccountSlots * 10 + + pool := NewTxPool(config, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() // Create a number of test accounts and fund them @@ -802,20 +909,20 @@ func TestTransactionPendingGlobalLimiting(t *testing.T) { txs := types.Transactions{} for _, key := range keys { addr := crypto.PubkeyToAddress(key.PublicKey) - for j := 0; j < int(DefaultTxPoolConfig.GlobalSlots)/len(keys)*2; j++ { + for j := 0; j < int(config.GlobalSlots)/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) + pool.AddRemotes(txs) pending := 0 for _, list := range pool.pending { pending += list.Len() } - if pending > int(DefaultTxPoolConfig.GlobalSlots) { - t.Fatalf("total pending transactions overflow allowance: %d > %d", pending, DefaultTxPoolConfig.GlobalSlots) + if pending > int(config.GlobalSlots) { + t.Fatalf("total pending transactions overflow allowance: %d > %d", pending, config.GlobalSlots) } if err := validateTxPoolInternals(pool); err != nil { t.Fatalf("pool internal state corrupted: %v", err) @@ -824,20 +931,17 @@ func TestTransactionPendingGlobalLimiting(t *testing.T) { // Tests that if transactions start being capped, transasctions are also removed from 'all' func TestTransactionCapClearsFromAll(t *testing.T) { - // Reduce the queue limits to shorten test time - defer func(old uint64) { DefaultTxPoolConfig.AccountSlots = old }(DefaultTxPoolConfig.AccountSlots) - defer func(old uint64) { DefaultTxPoolConfig.AccountQueue = old }(DefaultTxPoolConfig.AccountQueue) - defer func(old uint64) { DefaultTxPoolConfig.GlobalSlots = old }(DefaultTxPoolConfig.GlobalSlots) - - DefaultTxPoolConfig.AccountSlots = 2 - DefaultTxPoolConfig.AccountQueue = 2 - DefaultTxPoolConfig.GlobalSlots = 8 - // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) - pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + config := DefaultTxPoolConfig + config.AccountSlots = 2 + config.AccountQueue = 2 + config.GlobalSlots = 8 + + pool := NewTxPool(config, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() // Create a number of test accounts and fund them @@ -848,11 +952,11 @@ func TestTransactionCapClearsFromAll(t *testing.T) { state.AddBalance(addr, big.NewInt(1000000)) txs := types.Transactions{} - for j := 0; j < int(DefaultTxPoolConfig.GlobalSlots)*2; j++ { + for j := 0; j < int(config.GlobalSlots)*2; j++ { txs = append(txs, transaction(uint64(j), big.NewInt(100000), key)) } // Import the batch and verify that limits have been enforced - pool.AddBatch(txs) + pool.AddRemotes(txs) if err := validateTxPoolInternals(pool); err != nil { t.Fatalf("pool internal state corrupted: %v", err) } @@ -862,15 +966,15 @@ func TestTransactionCapClearsFromAll(t *testing.T) { // 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) { DefaultTxPoolConfig.GlobalSlots = old }(DefaultTxPoolConfig.GlobalSlots) - DefaultTxPoolConfig.GlobalSlots = 0 - // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) - pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + config := DefaultTxPoolConfig + config.GlobalSlots = 0 + + pool := NewTxPool(config, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() // Create a number of test accounts and fund them @@ -887,17 +991,17 @@ func TestTransactionPendingMinimumAllowance(t *testing.T) { txs := types.Transactions{} for _, key := range keys { addr := crypto.PubkeyToAddress(key.PublicKey) - for j := 0; j < int(DefaultTxPoolConfig.AccountSlots)*2; j++ { + for j := 0; j < int(config.AccountSlots)*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) + pool.AddRemotes(txs) for addr, list := range pool.pending { - if list.Len() != int(DefaultTxPoolConfig.AccountSlots) { - t.Errorf("addr %x: total pending transactions mismatch: have %d, want %d", addr, list.Len(), DefaultTxPoolConfig.AccountSlots) + if list.Len() != int(config.AccountSlots) { + t.Errorf("addr %x: total pending transactions mismatch: have %d, want %d", addr, list.Len(), config.AccountSlots) } } if err := validateTxPoolInternals(pool); err != nil { @@ -916,6 +1020,7 @@ func TestTransactionPoolRepricing(t *testing.T) { statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() // Create a number of test accounts and fund them @@ -937,11 +1042,11 @@ func TestTransactionPoolRepricing(t *testing.T) { txs = append(txs, pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1])) txs = append(txs, pricedTransaction(3, big.NewInt(100000), big.NewInt(2), keys[1])) - txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2])) - pool.SetLocal(txs[len(txs)-1]) // prevent this one from ever being dropped + ltx := pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2]) // Import the batch and that both pending and queued transactions match up - pool.AddBatch(txs) + pool.AddRemotes(txs) + pool.AddLocal(ltx) pending, queued := pool.stats() if pending != 4 { @@ -967,10 +1072,10 @@ func TestTransactionPoolRepricing(t *testing.T) { t.Fatalf("pool internal state corrupted: %v", err) } // Check that we can't add the old transactions back - if err := pool.Add(pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[0])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[0])); err != ErrUnderpriced { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced { t.Fatalf("adding underpriced queued transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } if err := validateTxPoolInternals(pool); err != nil { @@ -978,9 +1083,7 @@ func TestTransactionPoolRepricing(t *testing.T) { } // However we can add local underpriced transactions tx := pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[2]) - - pool.SetLocal(tx) // prevent this one from ever being dropped - if err := pool.Add(tx); err != nil { + if err := pool.AddLocal(tx); err != nil { t.Fatalf("failed to add underpriced local transaction: %v", err) } if pending, _ = pool.stats(); pending != 3 { @@ -997,18 +1100,16 @@ func TestTransactionPoolRepricing(t *testing.T) { // // Note, local transactions are never allowed to be dropped. func TestTransactionPoolUnderpricing(t *testing.T) { - // Reduce the queue limits to shorten test time - defer func(old uint64) { DefaultTxPoolConfig.GlobalSlots = old }(DefaultTxPoolConfig.GlobalSlots) - DefaultTxPoolConfig.GlobalSlots = 2 - - defer func(old uint64) { DefaultTxPoolConfig.GlobalQueue = old }(DefaultTxPoolConfig.GlobalQueue) - DefaultTxPoolConfig.GlobalQueue = 2 - // Create the pool to test the pricing enforcement with db, _ := ethdb.NewMemDatabase() statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) - pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + config := DefaultTxPoolConfig + config.GlobalSlots = 2 + config.GlobalQueue = 2 + + pool := NewTxPool(config, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() // Create a number of test accounts and fund them @@ -1027,11 +1128,11 @@ func TestTransactionPoolUnderpricing(t *testing.T) { txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[1])) - txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2])) - pool.SetLocal(txs[len(txs)-1]) // prevent this one from ever being dropped + ltx := pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2]) // Import the batch and that both pending and queued transactions match up - pool.AddBatch(txs) + pool.AddRemotes(txs) + pool.AddLocal(ltx) pending, queued := pool.stats() if pending != 3 { @@ -1044,17 +1145,17 @@ func TestTransactionPoolUnderpricing(t *testing.T) { t.Fatalf("pool internal state corrupted: %v", err) } // Ensure that adding an underpriced transaction on block limit fails - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } // Ensure that adding high priced transactions drops cheap ones, but not own - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(3), keys[1])); err != nil { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(3), keys[1])); err != nil { t.Fatalf("failed to add well priced transaction: %v", err) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(4), keys[1])); err != nil { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100000), big.NewInt(4), keys[1])); err != nil { t.Fatalf("failed to add well priced transaction: %v", err) } - if err := pool.Add(pricedTransaction(3, big.NewInt(100000), big.NewInt(5), keys[1])); err != nil { + if err := pool.AddRemote(pricedTransaction(3, big.NewInt(100000), big.NewInt(5), keys[1])); err != nil { t.Fatalf("failed to add well priced transaction: %v", err) } pending, queued = pool.stats() @@ -1069,9 +1170,7 @@ func TestTransactionPoolUnderpricing(t *testing.T) { } // Ensure that adding local transactions can push out even higher priced ones tx := pricedTransaction(1, big.NewInt(100000), big.NewInt(0), keys[2]) - - pool.SetLocal(tx) // prevent this one from ever being dropped - if err := pool.Add(tx); err != nil { + if err := pool.AddLocal(tx); err != nil { t.Fatalf("failed to add underpriced local transaction: %v", err) } pending, queued = pool.stats() @@ -1094,9 +1193,10 @@ func TestTransactionReplacement(t *testing.T) { statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) pool := NewTxPool(DefaultTxPoolConfig, params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) }) + defer pool.Stop() pool.resetState() - // Create a a test account to add transactions with + // Create a test account to add transactions with key, _ := crypto.GenerateKey() state, _ := pool.currentState() @@ -1106,43 +1206,43 @@ func TestTransactionReplacement(t *testing.T) { price := int64(100) threshold := (price * (100 + int64(DefaultTxPoolConfig.PriceBump))) / 100 - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), key)); err != nil { t.Fatalf("failed to add original cheap pending transaction: %v", err) } - if err := pool.Add(pricedTransaction(0, big.NewInt(100001), big.NewInt(1), key)); err != ErrReplaceUnderpriced { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100001), big.NewInt(1), key)); err != ErrReplaceUnderpriced { t.Fatalf("original cheap pending transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced) } - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(2), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(2), key)); err != nil { t.Fatalf("failed to replace original cheap pending transaction: %v", err) } - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(price), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(price), key)); err != nil { t.Fatalf("failed to add original proper pending transaction: %v", err) } - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(threshold), key)); err != ErrReplaceUnderpriced { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(threshold), key)); err != ErrReplaceUnderpriced { t.Fatalf("original proper pending transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced) } - if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(threshold+1), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(0, big.NewInt(100000), big.NewInt(threshold+1), key)); err != nil { t.Fatalf("failed to replace original proper pending transaction: %v", err) } // Add queued transactions, ensuring the minimum price bump is enforced for replacement (for ultra low prices too) - if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), key)); err != nil { t.Fatalf("failed to add original queued transaction: %v", err) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100001), big.NewInt(1), key)); err != ErrReplaceUnderpriced { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100001), big.NewInt(1), key)); err != ErrReplaceUnderpriced { t.Fatalf("original queued transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(2), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100000), big.NewInt(2), key)); err != nil { t.Fatalf("failed to replace original queued transaction: %v", err) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(price), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100000), big.NewInt(price), key)); err != nil { t.Fatalf("failed to add original queued transaction: %v", err) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100001), big.NewInt(threshold), key)); err != ErrReplaceUnderpriced { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100001), big.NewInt(threshold), key)); err != ErrReplaceUnderpriced { t.Fatalf("original queued transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced) } - if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(threshold+1), key)); err != nil { + if err := pool.AddRemote(pricedTransaction(2, big.NewInt(100000), big.NewInt(threshold+1), key)); err != nil { t.Fatalf("failed to replace original queued transaction: %v", err) } if err := validateTxPoolInternals(pool); err != nil { @@ -1159,6 +1259,8 @@ func BenchmarkPendingDemotion10000(b *testing.B) { benchmarkPendingDemotion(b, 1 func benchmarkPendingDemotion(b *testing.B, size int) { // Add a batch of transactions to a pool one by one pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() state.AddBalance(account, big.NewInt(1000000)) @@ -1183,6 +1285,8 @@ func BenchmarkFuturePromotion10000(b *testing.B) { benchmarkFuturePromotion(b, 1 func benchmarkFuturePromotion(b *testing.B, size int) { // Add a batch of transactions to a pool one by one pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() state.AddBalance(account, big.NewInt(1000000)) @@ -1202,6 +1306,8 @@ func benchmarkFuturePromotion(b *testing.B, size int) { func BenchmarkPoolInsert(b *testing.B) { // Generate a batch of transactions to enqueue into the pool pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() state.AddBalance(account, big.NewInt(1000000)) @@ -1213,7 +1319,7 @@ func BenchmarkPoolInsert(b *testing.B) { // Benchmark importing the transactions into the queue b.ResetTimer() for _, tx := range txs { - pool.Add(tx) + pool.AddRemote(tx) } } @@ -1225,6 +1331,8 @@ func BenchmarkPoolBatchInsert10000(b *testing.B) { benchmarkPoolBatchInsert(b, 1 func benchmarkPoolBatchInsert(b *testing.B, size int) { // Generate a batch of transactions to enqueue into the pool pool, key := setupTxPool() + defer pool.Stop() + account, _ := deriveSender(transaction(0, big.NewInt(0), key)) state, _ := pool.currentState() state.AddBalance(account, big.NewInt(1000000)) @@ -1239,6 +1347,6 @@ func benchmarkPoolBatchInsert(b *testing.B, size int) { // Benchmark importing the transactions into the queue b.ResetTimer() for _, batch := range batches { - pool.AddBatch(batch) + pool.AddRemotes(batch) } } diff --git a/eth/api_backend.go b/eth/api_backend.go index 166b5084d3..7ef7c030dd 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -116,29 +116,18 @@ func (b *EthApiBackend) GetEVM(ctx context.Context, msg core.Message, state *sta } func (b *EthApiBackend) SendTx(ctx context.Context, signedTx *types.Transaction) error { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - - b.eth.txPool.SetLocal(signedTx) - return b.eth.txPool.Add(signedTx) + return b.eth.txPool.AddLocal(signedTx) } func (b *EthApiBackend) RemoveTx(txHash common.Hash) { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - b.eth.txPool.Remove(txHash) } func (b *EthApiBackend) GetPoolTransactions() (types.Transactions, error) { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - pending, err := b.eth.txPool.Pending() if err != nil { return nil, err } - var txs types.Transactions for _, batch := range pending { txs = append(txs, batch...) @@ -147,30 +136,18 @@ func (b *EthApiBackend) GetPoolTransactions() (types.Transactions, error) { } func (b *EthApiBackend) GetPoolTransaction(hash common.Hash) *types.Transaction { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - return b.eth.txPool.Get(hash) } func (b *EthApiBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - return b.eth.txPool.State().GetNonce(addr), nil } func (b *EthApiBackend) Stats() (pending int, queued int) { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - return b.eth.txPool.Stats() } func (b *EthApiBackend) TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) { - b.eth.txMu.Lock() - defer b.eth.txMu.Unlock() - return b.eth.TxPool().Content() } diff --git a/eth/backend.go b/eth/backend.go index 75e0e737b1..78478e86e9 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -63,7 +63,6 @@ type Ethereum struct { stopDbUpgrade func() // stop chain db sequential key upgrade // Handlers txPool *core.TxPool - txMu sync.Mutex blockchain *core.BlockChain protocolManager *ProtocolManager lesServer LesServer diff --git a/eth/handler.go b/eth/handler.go index b2422d71c4..6c64493401 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -658,7 +658,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } p.MarkTransaction(tx.Hash()) } - pm.txpool.AddBatch(txs) + pm.txpool.AddRemotes(txs) default: return errResp(ErrInvalidMsgCode, "%v", msg.Code) diff --git a/eth/helper_test.go b/eth/helper_test.go index 0260b9d77c..546478a3ee 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -94,9 +94,9 @@ type testTxPool struct { lock sync.RWMutex // Protects the transaction pool } -// AddBatch appends a batch of transactions to the pool, and notifies any +// AddRemotes appends a batch of transactions to the pool, and notifies any // listeners if the addition channel is non nil -func (p *testTxPool) AddBatch(txs []*types.Transaction) error { +func (p *testTxPool) AddRemotes(txs []*types.Transaction) error { p.lock.Lock() defer p.lock.Unlock() diff --git a/eth/protocol.go b/eth/protocol.go index 4bc8bee729..376e4663e3 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -94,8 +94,8 @@ var errorToString = map[int]string{ } type txPool interface { - // AddBatch should add the given transactions to the pool. - AddBatch([]*types.Transaction) error + // AddRemotes should add the given transactions to the pool. + AddRemotes([]*types.Transaction) error // Pending should return pending transactions. // The slice should be modifiable by the caller. diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 2056ee0a8e..d3a44ae91f 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -130,7 +130,7 @@ func testSendTransactions(t *testing.T, protocol int) { for nonce := range alltxs { alltxs[nonce] = newTestTransaction(testAccount, uint64(nonce), txsize) } - pm.txpool.AddBatch(alltxs) + pm.txpool.AddRemotes(alltxs) // Connect several peers. They should all receive the pending transactions. var wg sync.WaitGroup diff --git a/les/handler.go b/les/handler.go index 39045ecbea..f50abaaa3e 100644 --- a/les/handler.go +++ b/les/handler.go @@ -87,8 +87,8 @@ type BlockChain interface { } type txPool interface { - // AddTransactions should add the given transactions to the pool. - AddBatch([]*types.Transaction) error + // AddRemotes should add the given transactions to the pool. + AddRemotes([]*types.Transaction) error } type ProtocolManager struct { @@ -801,7 +801,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrRequestRejected, "") } - if err := pm.txpool.AddBatch(txs); err != nil { + if err := pm.txpool.AddRemotes(txs); err != nil { return errResp(ErrUnexpectedResponse, "msg: %v", err) }