aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--core/tx_pool.go282
-rw-r--r--core/tx_pool_test.go133
-rw-r--r--eth/api_backend.go2
-rw-r--r--internal/ethapi/api.go60
-rw-r--r--internal/ethapi/backend.go2
-rw-r--r--miner/worker.go10
6 files changed, 282 insertions, 207 deletions
diff --git a/core/tx_pool.go b/core/tx_pool.go
index 596356377..ec3d5c16b 100644
--- a/core/tx_pool.go
+++ b/core/tx_pool.go
@@ -51,6 +51,11 @@ const (
type stateFn func() (*state.StateDB, error)
+// TxList is a "list" of transactions belonging to an account, sorted by account
+// nonce. To allow gaps and avoid constant copying, the list is represented as a
+// hash map.
+type TxList map[uint64]*types.Transaction
+
// TxPool contains all currently known transactions. Transactions
// enter the pool when they are received from the network or submitted
// locally. They exit the pool when they are included in the blockchain.
@@ -68,8 +73,10 @@ type TxPool struct {
events event.Subscription
localTx *txSet
mu sync.RWMutex
- pending map[common.Hash]*types.Transaction // processable transactions
- queue map[common.Address]map[common.Hash]*types.Transaction
+
+ pending map[common.Address]TxList // All currently processable transactions
+ queue map[common.Address]TxList // Queued but non-processable transactions
+ all map[common.Hash]*types.Transaction // All transactions to allow lookups
wg sync.WaitGroup // for shutdown sync
@@ -79,8 +86,9 @@ type TxPool struct {
func NewTxPool(config *ChainConfig, eventMux *event.TypeMux, currentStateFn stateFn, gasLimitFn func() *big.Int) *TxPool {
pool := &TxPool{
config: config,
- pending: make(map[common.Hash]*types.Transaction),
- queue: make(map[common.Address]map[common.Hash]*types.Transaction),
+ pending: make(map[common.Address]TxList),
+ queue: make(map[common.Address]TxList),
+ all: make(map[common.Hash]*types.Transaction),
eventMux: eventMux,
currentState: currentStateFn,
gasLimit: gasLimitFn,
@@ -143,12 +151,12 @@ func (pool *TxPool) resetState() {
// Loop over the pending transactions and base the nonce of the new
// pending transaction set.
- for _, tx := range pool.pending {
- if addr, err := tx.From(); err == nil {
- // Set the nonce. Transaction nonce can never be lower
- // than the state nonce; validatePool took care of that.
- if pool.pendingState.GetNonce(addr) <= tx.Nonce() {
- pool.pendingState.SetNonce(addr, tx.Nonce()+1)
+ for addr, txs := range pool.pending {
+ // Set the nonce. Transaction nonce can never be lower
+ // than the state nonce; validatePool took care of that.
+ for nonce, _ := range txs {
+ if pool.pendingState.GetNonce(addr) <= nonce {
+ pool.pendingState.SetNonce(addr, nonce+1)
}
}
}
@@ -174,7 +182,9 @@ func (pool *TxPool) Stats() (pending int, queued int) {
pool.mu.RLock()
defer pool.mu.RUnlock()
- pending = len(pool.pending)
+ for _, txs := range pool.pending {
+ pending += len(txs)
+ }
for _, txs := range pool.queue {
queued += len(txs)
}
@@ -183,30 +193,27 @@ func (pool *TxPool) Stats() (pending int, queued int) {
// Content retrieves the data content of the transaction pool, returning all the
// pending as well as queued transactions, grouped by account and nonce.
-func (pool *TxPool) Content() (map[common.Address]map[uint64][]*types.Transaction, map[common.Address]map[uint64][]*types.Transaction) {
+func (pool *TxPool) Content() (map[common.Address]TxList, map[common.Address]TxList) {
pool.mu.RLock()
defer pool.mu.RUnlock()
// Retrieve all the pending transactions and sort by account and by nonce
- pending := make(map[common.Address]map[uint64][]*types.Transaction)
- for _, tx := range pool.pending {
- account, _ := tx.From()
-
- owned, ok := pending[account]
- if !ok {
- owned = make(map[uint64][]*types.Transaction)
- pending[account] = owned
+ pending := make(map[common.Address]TxList)
+ for addr, txs := range pool.pending {
+ copy := make(TxList)
+ for nonce, tx := range txs {
+ copy[nonce] = tx
}
- owned[tx.Nonce()] = append(owned[tx.Nonce()], tx)
+ pending[addr] = copy
}
// Retrieve all the queued transactions and sort by account and by nonce
- queued := make(map[common.Address]map[uint64][]*types.Transaction)
- for account, txs := range pool.queue {
- owned := make(map[uint64][]*types.Transaction)
- for _, tx := range txs {
- owned[tx.Nonce()] = append(owned[tx.Nonce()], tx)
+ queued := make(map[common.Address]TxList)
+ for addr, txs := range pool.queue {
+ copy := make(TxList)
+ for nonce, tx := range txs {
+ copy[nonce] = tx
}
- queued[account] = owned
+ queued[addr] = copy
}
return pending, queued
}
@@ -280,7 +287,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction) error {
func (self *TxPool) add(tx *types.Transaction) error {
hash := tx.Hash()
- if self.pending[hash] != nil {
+ if self.all[hash] != nil {
return fmt.Errorf("Known transaction (%x)", hash[:4])
}
err := self.validateTx(tx)
@@ -306,33 +313,63 @@ func (self *TxPool) add(tx *types.Transaction) error {
return nil
}
-// queueTx will queue an unknown transaction
+// queueTx will queue an unknown transaction.
func (self *TxPool) queueTx(hash common.Hash, tx *types.Transaction) {
- from, _ := tx.From() // already validated
- if self.queue[from] == nil {
- self.queue[from] = make(map[common.Hash]*types.Transaction)
+ addr, _ := tx.From() // already validated
+ if self.queue[addr] == nil {
+ self.queue[addr] = make(TxList)
}
- self.queue[from][hash] = tx
+ // If the nonce is already used, discard the lower priced transaction
+ nonce := tx.Nonce()
+
+ if old, ok := self.queue[addr][nonce]; ok {
+ if old.GasPrice().Cmp(tx.GasPrice()) >= 0 {
+ return // Old was better, discard this
+ }
+ delete(self.all, old.Hash()) // New is better, drop and overwrite old one
+ }
+ self.queue[addr][nonce] = tx
+ self.all[hash] = tx
}
-// addTx will add a transaction to the pending (processable queue) list of transactions
-func (pool *TxPool) addTx(hash common.Hash, addr common.Address, tx *types.Transaction) {
- // init delayed since tx pool could have been started before any state sync
+// addTx will moves a transaction from the non-executable queue to the pending
+// (processable) list of transactions.
+func (pool *TxPool) addTx(addr common.Address, tx *types.Transaction) {
+ // Init delayed since tx pool could have been started before any state sync
if pool.pendingState == nil {
pool.resetState()
}
+ // If the nonce is already used, discard the lower priced transaction
+ hash, nonce := tx.Hash(), tx.Nonce()
- if _, ok := pool.pending[hash]; !ok {
- pool.pending[hash] = tx
+ if old, ok := pool.pending[addr][nonce]; ok {
+ oldHash := old.Hash()
- // Increment the nonce on the pending state. This can only happen if
- // the nonce is +1 to the previous one.
- pool.pendingState.SetNonce(addr, tx.Nonce()+1)
- // Notify the subscribers. This event is posted in a goroutine
- // because it's possible that somewhere during the post "Remove transaction"
- // gets called which will then wait for the global tx pool lock and deadlock.
- go pool.eventMux.Post(TxPreEvent{tx})
+ switch {
+ case oldHash == hash: // Nothing changed, noop
+ return
+ case old.GasPrice().Cmp(tx.GasPrice()) >= 0: // Old was better, discard this
+ delete(pool.all, hash)
+ return
+ default: // New is better, discard old
+ delete(pool.all, oldHash)
+ }
+ }
+ // The transaction is being kept, insert it into the tx pool
+ if _, ok := pool.pending[addr]; !ok {
+ pool.pending[addr] = make(TxList)
}
+ pool.pending[addr][nonce] = tx
+ pool.all[hash] = tx
+
+ // Increment the nonce on the pending state. This can only happen if
+ // the nonce is +1 to the previous one.
+ pool.pendingState.SetNonce(addr, nonce+1)
+
+ // Notify the subscribers. This event is posted in a goroutine
+ // because it's possible that somewhere during the post "Remove transaction"
+ // gets called which will then wait for the global tx pool lock and deadlock.
+ go pool.eventMux.Post(TxPreEvent{tx})
}
// Add queues a single transaction in the pool if it is valid.
@@ -371,58 +408,39 @@ func (tp *TxPool) GetTransaction(hash common.Hash) *types.Transaction {
tp.mu.RLock()
defer tp.mu.RUnlock()
- // check the txs first
- if tx, ok := tp.pending[hash]; ok {
- return tx
- }
- // check queue
- for _, txs := range tp.queue {
- if tx, ok := txs[hash]; ok {
- return tx
- }
- }
- return nil
+ return tp.all[hash]
}
// GetTransactions returns all currently processable transactions.
// The returned slice may be modified by the caller.
-func (self *TxPool) GetTransactions() (txs types.Transactions) {
+func (self *TxPool) GetTransactions() types.Transactions {
self.mu.Lock()
defer self.mu.Unlock()
// check queue first
self.checkQueue()
+
// invalidate any txs
self.validatePool()
- txs = make(types.Transactions, len(self.pending))
- i := 0
- for _, tx := range self.pending {
- txs[i] = tx
- i++
+ count := 0
+ for _, txs := range self.pending {
+ count += len(txs)
}
- return txs
-}
-
-// GetQueuedTransactions returns all non-processable transactions.
-func (self *TxPool) GetQueuedTransactions() types.Transactions {
- self.mu.RLock()
- defer self.mu.RUnlock()
-
- var ret types.Transactions
- for _, txs := range self.queue {
+ pending := make(types.Transactions, 0, count)
+ for _, txs := range self.pending {
for _, tx := range txs {
- ret = append(ret, tx)
+ pending = append(pending, tx)
}
}
- sort.Sort(types.TxByNonce(ret))
- return ret
+ return pending
}
// RemoveTransactions removes all given transactions from the pool.
func (self *TxPool) RemoveTransactions(txs types.Transactions) {
self.mu.Lock()
defer self.mu.Unlock()
+
for _, tx := range txs {
self.removeTx(tx.Hash())
}
@@ -432,29 +450,35 @@ func (self *TxPool) RemoveTransactions(txs types.Transactions) {
func (pool *TxPool) RemoveTx(hash common.Hash) {
pool.mu.Lock()
defer pool.mu.Unlock()
+
pool.removeTx(hash)
}
func (pool *TxPool) removeTx(hash common.Hash) {
- // delete from pending pool
- delete(pool.pending, hash)
- // delete from queue
- for address, txs := range pool.queue {
- if _, ok := txs[hash]; ok {
- if len(txs) == 1 {
- // if only one tx, remove entire address entry.
- delete(pool.queue, address)
- } else {
- delete(txs, hash)
- }
- break
- }
+ // Fetch the transaction we wish to delete
+ tx, ok := pool.all[hash]
+ if !ok {
+ return
+ }
+ addr, _ := tx.From()
+
+ // Remove it from all internal lists
+ delete(pool.all, hash)
+
+ delete(pool.pending[addr], tx.Nonce())
+ if len(pool.pending[addr]) == 0 {
+ delete(pool.pending, addr)
+ }
+ delete(pool.queue[addr], tx.Nonce())
+ if len(pool.queue[addr]) == 0 {
+ delete(pool.queue, addr)
}
}
-// checkQueue moves transactions that have become processable to main pool.
+// checkQueue moves transactions that have become processable from the pool's
+// queue to the set of pending transactions.
func (pool *TxPool) checkQueue() {
- // init delayed since tx pool could have been started before any state sync
+ // Init delayed since tx pool could have been started before any state sync
if pool.pendingState == nil {
pool.resetState()
}
@@ -473,17 +497,19 @@ func (pool *TxPool) checkQueue() {
trueNonce = currentState.GetNonce(address) // nonce known by the last state
)
promote = promote[:0]
- for hash, tx := range txs {
+ for nonce, tx := range txs {
// Drop processed or out of fund transactions
- if tx.Nonce() < trueNonce || balance.Cmp(tx.Cost()) < 0 {
+ if nonce < trueNonce || balance.Cmp(tx.Cost()) < 0 {
if glog.V(logger.Core) {
glog.Infof("removed tx (%v) from pool queue: low tx nonce or out of funds\n", tx)
}
- delete(txs, hash)
+ delete(txs, nonce)
+ delete(pool.all, tx.Hash())
+
continue
}
// Collect the remaining transactions for the next pass.
- promote = append(promote, txQueueEntry{hash, address, tx})
+ promote = append(promote, txQueueEntry{address, tx})
}
// Find the next consecutive nonce range starting at the current account nonce,
// pushing the guessed nonce forward if we add consecutive transactions.
@@ -493,17 +519,18 @@ func (pool *TxPool) checkQueue() {
if entry.Nonce() > guessedNonce {
if len(promote)-i > maxQueued {
if glog.V(logger.Debug) {
- glog.Infof("Queued tx limit exceeded for %s. Tx %s removed\n", common.PP(address[:]), common.PP(entry.hash[:]))
+ glog.Infof("Queued tx limit exceeded for %s. Tx %s removed\n", common.PP(address[:]), common.PP(entry.Hash().Bytes()))
}
for _, drop := range promote[i+maxQueued:] {
- delete(txs, drop.hash)
+ delete(txs, drop.Nonce())
+ delete(pool.all, drop.Hash())
}
}
break
}
// Otherwise promote the transaction and move the guess nonce if needed
- pool.addTx(entry.hash, address, entry.Transaction)
- delete(txs, entry.hash)
+ pool.addTx(address, entry.Transaction)
+ delete(txs, entry.Nonce())
if entry.Nonce() == guessedNonce {
guessedNonce++
@@ -532,40 +559,48 @@ func (pool *TxPool) validatePool() {
// Clean up the pending pool, accumulating invalid nonces
gaps := make(map[common.Address]uint64)
- for hash, tx := range pool.pending {
- sender, _ := tx.From() // err already checked
-
- // Perform light nonce and balance validation
- balance := balanceCache[sender]
- if balance == nil {
- balance = state.GetBalance(sender)
- balanceCache[sender] = balance
- }
- if past := state.GetNonce(sender) > tx.Nonce(); past || balance.Cmp(tx.Cost()) < 0 {
- // Remove an already past it invalidated transaction
- if glog.V(logger.Core) {
- glog.Infof("removed tx (%v) from pool: low tx nonce or out of funds\n", tx)
+ for addr, txs := range pool.pending {
+ for nonce, tx := range txs {
+ // Perform light nonce and balance validation
+ balance := balanceCache[addr]
+ if balance == nil {
+ balance = state.GetBalance(addr)
+ balanceCache[addr] = balance
}
- delete(pool.pending, hash)
+ if past := state.GetNonce(addr) > nonce; past || balance.Cmp(tx.Cost()) < 0 {
+ // Remove an already past it invalidated transaction
+ if glog.V(logger.Core) {
+ glog.Infof("removed tx (%v) from pool: low tx nonce or out of funds\n", tx)
+ }
+ delete(pool.pending[addr], nonce)
+ if len(pool.pending[addr]) == 0 {
+ delete(pool.pending, addr)
+ }
+ delete(pool.all, tx.Hash())
- // Track the smallest invalid nonce to postpone subsequent transactions
- if !past {
- if prev, ok := gaps[sender]; !ok || tx.Nonce() < prev {
- gaps[sender] = tx.Nonce()
+ // Track the smallest invalid nonce to postpone subsequent transactions
+ if !past {
+ if prev, ok := gaps[addr]; !ok || nonce < prev {
+ gaps[addr] = nonce
+ }
}
}
}
}
// Move all transactions after a gap back to the future queue
if len(gaps) > 0 {
- for hash, tx := range pool.pending {
- sender, _ := tx.From()
- if gap, ok := gaps[sender]; ok && tx.Nonce() >= gap {
- if glog.V(logger.Core) {
- glog.Infof("postponed tx (%v) due to introduced gap\n", tx)
+ for addr, txs := range pool.pending {
+ for nonce, tx := range txs {
+ if gap, ok := gaps[addr]; ok && nonce >= gap {
+ if glog.V(logger.Core) {
+ glog.Infof("postponed tx (%v) due to introduced gap\n", tx)
+ }
+ delete(pool.pending[addr], nonce)
+ if len(pool.pending[addr]) == 0 {
+ delete(pool.pending, addr)
+ }
+ pool.queueTx(tx.Hash(), tx)
}
- pool.queueTx(hash, tx)
- delete(pool.pending, hash)
}
}
}
@@ -574,7 +609,6 @@ func (pool *TxPool) validatePool() {
type txQueue []txQueueEntry
type txQueueEntry struct {
- hash common.Hash
addr common.Address
*types.Transaction
}
diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go
index ed9bea75f..8aa69233d 100644
--- a/core/tx_pool_test.go
+++ b/core/tx_pool_test.go
@@ -103,7 +103,7 @@ func TestTransactionQueue(t *testing.T) {
currentState.SetNonce(from, 2)
pool.queueTx(tx.Hash(), tx)
pool.checkQueue()
- if _, ok := pool.pending[tx.Hash()]; ok {
+ if _, ok := pool.pending[from][tx.Nonce()]; ok {
t.Error("expected transaction to be in tx pool")
}
@@ -139,7 +139,7 @@ func TestRemoveTx(t *testing.T) {
currentState, _ := pool.currentState()
currentState.AddBalance(from, big.NewInt(1))
pool.queueTx(tx.Hash(), tx)
- pool.addTx(tx.Hash(), from, tx)
+ pool.addTx(from, tx)
if len(pool.queue) != 1 {
t.Error("expected queue to be 1, got", len(pool.queue))
}
@@ -210,18 +210,38 @@ func TestTransactionDoubleNonce(t *testing.T) {
}
resetState()
- tx := transaction(0, big.NewInt(100000), key)
- tx2 := transaction(0, big.NewInt(1000000), key)
- if err := pool.add(tx); err != nil {
+ tx1, _ := types.NewTransaction(0, common.Address{}, big.NewInt(100), big.NewInt(100000), big.NewInt(1), nil).SignECDSA(key)
+ tx2, _ := types.NewTransaction(0, common.Address{}, big.NewInt(100), big.NewInt(1000000), big.NewInt(2), nil).SignECDSA(key)
+ tx3, _ := types.NewTransaction(0, common.Address{}, big.NewInt(100), big.NewInt(1000000), big.NewInt(1), nil).SignECDSA(key)
+
+ // Add the first two transaction, ensure higher priced stays only
+ if err := pool.add(tx1); err != nil {
t.Error("didn't expect error", err)
}
if err := pool.add(tx2); err != nil {
t.Error("didn't expect error", err)
}
-
pool.checkQueue()
- if len(pool.pending) != 2 {
- t.Error("expected 2 pending txs. Got", len(pool.pending))
+ if len(pool.pending[addr]) != 1 {
+ t.Error("expected 1 pending transactions, got", len(pool.pending))
+ }
+ if tx := pool.pending[addr][0]; tx.Hash() != tx2.Hash() {
+ t.Errorf("transaction mismatch: have %x, want %x", tx.Hash(), tx2.Hash())
+ }
+ // Add the thid transaction and ensure it's not saved (smaller price)
+ if err := pool.add(tx3); err != nil {
+ t.Error("didn't expect error", err)
+ }
+ pool.checkQueue()
+ if len(pool.pending[addr]) != 1 {
+ t.Error("expected 1 pending transactions, got", len(pool.pending))
+ }
+ if tx := pool.pending[addr][0]; tx.Hash() != tx2.Hash() {
+ t.Errorf("transaction mismatch: have %x, want %x", tx.Hash(), tx2.Hash())
+ }
+ // Ensure the total transaction count is correct
+ if len(pool.all) != 1 {
+ t.Error("expected 1 total transactions, got", len(pool.all))
}
}
@@ -234,12 +254,15 @@ func TestMissingNonce(t *testing.T) {
if err := pool.add(tx); err != nil {
t.Error("didn't expect error", err)
}
- if len(pool.pending) != 0 {
- t.Error("expected 0 pending transactions, got", len(pool.pending))
+ if len(pool.pending[addr]) != 0 {
+ t.Error("expected 0 pending transactions, got", len(pool.pending[addr]))
}
if len(pool.queue[addr]) != 1 {
t.Error("expected 1 queued transaction, got", len(pool.queue[addr]))
}
+ if len(pool.all) != 1 {
+ t.Error("expected 1 total transactions, got", len(pool.all))
+ }
}
func TestNonceRecovery(t *testing.T) {
@@ -270,8 +293,11 @@ func TestRemovedTxEvent(t *testing.T) {
currentState.AddBalance(from, big.NewInt(1000000000000))
pool.eventMux.Post(RemovedTransactionEvent{types.Transactions{tx}})
pool.eventMux.Post(ChainHeadEvent{nil})
- if len(pool.pending) != 1 {
- t.Error("expected 1 pending tx, got", len(pool.pending))
+ if len(pool.pending[from]) != 1 {
+ t.Error("expected 1 pending tx, got", len(pool.pending[from]))
+ }
+ if len(pool.all) != 1 {
+ t.Error("expected 1 total transactions, got", len(pool.all))
}
}
@@ -292,41 +318,50 @@ func TestTransactionDropping(t *testing.T) {
tx10 = transaction(10, big.NewInt(100), key)
tx11 = transaction(11, big.NewInt(200), key)
)
- pool.addTx(tx0.Hash(), account, tx0)
- pool.addTx(tx1.Hash(), account, tx1)
+ pool.addTx(account, tx0)
+ pool.addTx(account, tx1)
pool.queueTx(tx10.Hash(), tx10)
pool.queueTx(tx11.Hash(), tx11)
// Check that pre and post validations leave the pool as is
- if len(pool.pending) != 2 {
- t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending), 2)
+ if len(pool.pending[account]) != 2 {
+ t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending[account]), 2)
}
if len(pool.queue[account]) != 2 {
- t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue), 2)
+ t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue[account]), 2)
+ }
+ if len(pool.all) != 4 {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 4)
}
pool.resetState()
- if len(pool.pending) != 2 {
- t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending), 2)
+ if len(pool.pending[account]) != 2 {
+ t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending[account]), 2)
}
if len(pool.queue[account]) != 2 {
- t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue), 2)
+ t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue[account]), 2)
+ }
+ if len(pool.all) != 4 {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 4)
}
// Reduce the balance of the account, and check that invalidated transactions are dropped
state.AddBalance(account, big.NewInt(-750))
pool.resetState()
- if _, ok := pool.pending[tx0.Hash()]; !ok {
+ if _, ok := pool.pending[account][tx0.Nonce()]; !ok {
t.Errorf("funded pending transaction missing: %v", tx0)
}
- if _, ok := pool.pending[tx1.Hash()]; ok {
+ if _, ok := pool.pending[account][tx1.Nonce()]; ok {
t.Errorf("out-of-fund pending transaction present: %v", tx1)
}
- if _, ok := pool.queue[account][tx10.Hash()]; !ok {
+ if _, ok := pool.queue[account][tx10.Nonce()]; !ok {
t.Errorf("funded queued transaction missing: %v", tx10)
}
- if _, ok := pool.queue[account][tx11.Hash()]; ok {
+ if _, ok := pool.queue[account][tx11.Nonce()]; ok {
t.Errorf("out-of-fund queued transaction present: %v", tx11)
}
+ if len(pool.all) != 2 {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 2)
+ }
}
// Tests that if a transaction is dropped from the current pending pool (e.g. out
@@ -349,50 +384,59 @@ func TestTransactionPostponing(t *testing.T) {
} else {
tx = transaction(uint64(i), big.NewInt(500), key)
}
- pool.addTx(tx.Hash(), account, tx)
+ pool.addTx(account, tx)
txns = append(txns, tx)
}
// Check that pre and post validations leave the pool as is
- if len(pool.pending) != len(txns) {
- t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending), len(txns))
+ if len(pool.pending[account]) != len(txns) {
+ t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending[account]), len(txns))
}
if len(pool.queue[account]) != 0 {
- t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue), 0)
+ t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue[account]), 0)
+ }
+ if len(pool.all) != len(txns) {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), len(txns))
}
pool.resetState()
- if len(pool.pending) != len(txns) {
- t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending), len(txns))
+ if len(pool.pending[account]) != len(txns) {
+ t.Errorf("pending transaction mismatch: have %d, want %d", len(pool.pending[account]), len(txns))
}
if len(pool.queue[account]) != 0 {
- t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue), 0)
+ t.Errorf("queued transaction mismatch: have %d, want %d", len(pool.queue[account]), 0)
+ }
+ if len(pool.all) != len(txns) {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), len(txns))
}
// Reduce the balance of the account, and check that transactions are reorganised
state.AddBalance(account, big.NewInt(-750))
pool.resetState()
- if _, ok := pool.pending[txns[0].Hash()]; !ok {
+ if _, ok := pool.pending[account][txns[0].Nonce()]; !ok {
t.Errorf("tx %d: valid and funded transaction missing from pending pool: %v", 0, txns[0])
}
- if _, ok := pool.queue[account][txns[0].Hash()]; ok {
+ if _, ok := pool.queue[account][txns[0].Nonce()]; ok {
t.Errorf("tx %d: valid and funded transaction present in future queue: %v", 0, txns[0])
}
for i, tx := range txns[1:] {
if i%2 == 1 {
- if _, ok := pool.pending[tx.Hash()]; ok {
+ if _, ok := pool.pending[account][tx.Nonce()]; ok {
t.Errorf("tx %d: valid but future transaction present in pending pool: %v", i+1, tx)
}
- if _, ok := pool.queue[account][tx.Hash()]; !ok {
+ if _, ok := pool.queue[account][tx.Nonce()]; !ok {
t.Errorf("tx %d: valid but future transaction missing from future queue: %v", i+1, tx)
}
} else {
- if _, ok := pool.pending[tx.Hash()]; ok {
+ if _, ok := pool.pending[account][tx.Nonce()]; ok {
t.Errorf("tx %d: out-of-fund transaction present in pending pool: %v", i+1, tx)
}
- if _, ok := pool.queue[account][tx.Hash()]; ok {
+ if _, ok := pool.queue[account][tx.Nonce()]; ok {
t.Errorf("tx %d: out-of-fund transaction present in future queue: %v", i+1, tx)
}
}
}
+ if len(pool.all) != len(txns)/2 {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), len(txns)/2)
+ }
}
// Tests that if the transaction count belonging to a single account goes above
@@ -423,6 +467,9 @@ func TestTransactionQueueLimiting(t *testing.T) {
}
}
}
+ if len(pool.all) != maxQueued {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), maxQueued)
+ }
}
// Tests that even if the transaction count belonging to a single account goes
@@ -441,13 +488,16 @@ func TestTransactionPendingLimiting(t *testing.T) {
if err := pool.Add(transaction(i, big.NewInt(100000), key)); err != nil {
t.Fatalf("tx %d: failed to add transaction: %v", i, err)
}
- if len(pool.pending) != int(i)+1 {
- t.Errorf("tx %d: pending pool size mismatch: have %d, want %d", i, len(pool.pending), i+1)
+ if len(pool.pending[account]) != int(i)+1 {
+ t.Errorf("tx %d: pending pool size mismatch: have %d, want %d", i, len(pool.pending[account]), i+1)
}
if len(pool.queue[account]) != 0 {
t.Errorf("tx %d: queue size mismatch: have %d, want %d", i, len(pool.queue[account]), 0)
}
}
+ if len(pool.all) != maxQueued+5 {
+ t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), maxQueued+5)
+ }
}
// Tests that the transaction limits are enforced the same way irrelevant whether
@@ -486,6 +536,9 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) {
if len(pool1.queue[account1]) != len(pool2.queue[account2]) {
t.Errorf("queued transaction count mismatch: one-by-one algo: %d, batch algo: %d", len(pool1.queue[account1]), len(pool2.queue[account2]))
}
+ if len(pool1.all) != len(pool2.all) {
+ t.Errorf("total transaction count mismatch: one-by-one algo %d, batch algo %d", len(pool1.all), len(pool2.all))
+ }
}
// Benchmarks the speed of validating the contents of the pending queue of the
@@ -503,7 +556,7 @@ func benchmarkValidatePool(b *testing.B, size int) {
for i := 0; i < size; i++ {
tx := transaction(uint64(i), big.NewInt(100000), key)
- pool.addTx(tx.Hash(), account, tx)
+ pool.addTx(account, tx)
}
// Benchmark the speed of pool validation
b.ResetTimer()
diff --git a/eth/api_backend.go b/eth/api_backend.go
index efcdb3361..e19254374 100644
--- a/eth/api_backend.go
+++ b/eth/api_backend.go
@@ -149,7 +149,7 @@ func (b *EthApiBackend) Stats() (pending int, queued int) {
return b.eth.txPool.Stats()
}
-func (b *EthApiBackend) TxPoolContent() (map[common.Address]map[uint64][]*types.Transaction, map[common.Address]map[uint64][]*types.Transaction) {
+func (b *EthApiBackend) TxPoolContent() (map[common.Address]core.TxList, map[common.Address]core.TxList) {
b.eth.txMu.Lock()
defer b.eth.txMu.Unlock()
diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go
index 184b5831f..0fba55ae8 100644
--- a/internal/ethapi/api.go
+++ b/internal/ethapi/api.go
@@ -100,32 +100,26 @@ func NewPublicTxPoolAPI(b Backend) *PublicTxPoolAPI {
}
// Content returns the transactions contained within the transaction pool.
-func (s *PublicTxPoolAPI) Content() map[string]map[string]map[string][]*RPCTransaction {
- content := map[string]map[string]map[string][]*RPCTransaction{
- "pending": make(map[string]map[string][]*RPCTransaction),
- "queued": make(map[string]map[string][]*RPCTransaction),
+func (s *PublicTxPoolAPI) Content() map[string]map[string]map[string]*RPCTransaction {
+ content := map[string]map[string]map[string]*RPCTransaction{
+ "pending": make(map[string]map[string]*RPCTransaction),
+ "queued": make(map[string]map[string]*RPCTransaction),
}
pending, queue := s.b.TxPoolContent()
// Flatten the pending transactions
- for account, batches := range pending {
- dump := make(map[string][]*RPCTransaction)
- for nonce, txs := range batches {
- nonce := fmt.Sprintf("%d", nonce)
- for _, tx := range txs {
- dump[nonce] = append(dump[nonce], newRPCPendingTransaction(tx))
- }
+ for account, txs := range pending {
+ dump := make(map[string]*RPCTransaction)
+ for nonce, tx := range txs {
+ dump[fmt.Sprintf("%d", nonce)] = newRPCPendingTransaction(tx)
}
content["pending"][account.Hex()] = dump
}
// Flatten the queued transactions
- for account, batches := range queue {
- dump := make(map[string][]*RPCTransaction)
- for nonce, txs := range batches {
- nonce := fmt.Sprintf("%d", nonce)
- for _, tx := range txs {
- dump[nonce] = append(dump[nonce], newRPCPendingTransaction(tx))
- }
+ for account, txs := range queue {
+ dump := make(map[string]*RPCTransaction)
+ for nonce, tx := range txs {
+ dump[fmt.Sprintf("%d", nonce)] = newRPCPendingTransaction(tx)
}
content["queued"][account.Hex()] = dump
}
@@ -143,10 +137,10 @@ func (s *PublicTxPoolAPI) Status() map[string]*rpc.HexNumber {
// Inspect retrieves the content of the transaction pool and flattens it into an
// easily inspectable list.
-func (s *PublicTxPoolAPI) Inspect() map[string]map[string]map[string][]string {
- content := map[string]map[string]map[string][]string{
- "pending": make(map[string]map[string][]string),
- "queued": make(map[string]map[string][]string),
+func (s *PublicTxPoolAPI) Inspect() map[string]map[string]map[string]string {
+ content := map[string]map[string]map[string]string{
+ "pending": make(map[string]map[string]string),
+ "queued": make(map[string]map[string]string),
}
pending, queue := s.b.TxPoolContent()
@@ -158,24 +152,18 @@ func (s *PublicTxPoolAPI) Inspect() map[string]map[string]map[string][]string {
return fmt.Sprintf("contract creation: %v wei + %v × %v gas", tx.Value(), tx.Gas(), tx.GasPrice())
}
// Flatten the pending transactions
- for account, batches := range pending {
- dump := make(map[string][]string)
- for nonce, txs := range batches {
- nonce := fmt.Sprintf("%d", nonce)
- for _, tx := range txs {
- dump[nonce] = append(dump[nonce], format(tx))
- }
+ for account, txs := range pending {
+ dump := make(map[string]string)
+ for nonce, tx := range txs {
+ dump[fmt.Sprintf("%d", nonce)] = format(tx)
}
content["pending"][account.Hex()] = dump
}
// Flatten the queued transactions
- for account, batches := range queue {
- dump := make(map[string][]string)
- for nonce, txs := range batches {
- nonce := fmt.Sprintf("%d", nonce)
- for _, tx := range txs {
- dump[nonce] = append(dump[nonce], format(tx))
- }
+ for account, txs := range queue {
+ dump := make(map[string]string)
+ for nonce, tx := range txs {
+ dump[fmt.Sprintf("%d", nonce)] = format(tx)
}
content["queued"][account.Hex()] = dump
}
diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go
index 791a06925..673ab0546 100644
--- a/internal/ethapi/backend.go
+++ b/internal/ethapi/backend.go
@@ -58,7 +58,7 @@ type Backend interface {
GetPoolTransaction(txHash common.Hash) *types.Transaction
GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error)
Stats() (pending int, queued int)
- TxPoolContent() (map[common.Address]map[uint64][]*types.Transaction, map[common.Address]map[uint64][]*types.Transaction)
+ TxPoolContent() (map[common.Address]core.TxList, map[common.Address]core.TxList)
}
type State interface {
diff --git a/miner/worker.go b/miner/worker.go
index 59406bf4e..f243fe799 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -68,12 +68,12 @@ type Work struct {
ancestors *set.Set // ancestor set (used for checking uncle parent validity)
family *set.Set // family set (used for checking uncle invalidity)
uncles *set.Set // uncle set
- remove *set.Set // tx which will be removed
tcount int // tx count in cycle
ignoredTransactors *set.Set
lowGasTransactors *set.Set
ownedAccounts *set.Set
lowGasTxs types.Transactions
+ failedTxs types.Transactions
localMinedBlocks *uint64RingBuffer // the most recent block numbers that were mined locally (used to check block inclusion)
Block *types.Block // the new block
@@ -383,7 +383,6 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
accounts := self.eth.AccountManager().Accounts()
// Keep track of transactions which return errors so they can be removed
- work.remove = set.New()
work.tcount = 0
work.ignoredTransactors = set.New()
work.lowGasTransactors = set.New()
@@ -533,7 +532,9 @@ func (self *worker) commitNewWork() {
*/
work.commitTransactions(self.mux, transactions, self.gasPrice, self.chain)
+
self.eth.TxPool().RemoveTransactions(work.lowGasTxs)
+ self.eth.TxPool().RemoveTransactions(work.failedTxs)
// compute uncles for the new block.
var (
@@ -639,11 +640,10 @@ func (env *Work) commitTransactions(mux *event.TypeMux, transactions types.Trans
// ignore the transactor so no nonce errors will be thrown for this account
// next time the worker is run, they'll be picked up again.
env.ignoredTransactors.Add(from)
-
glog.V(logger.Detail).Infof("Gas limit reached for (%x) in this block. Continue to try smaller txs\n", from[:4])
- case err != nil:
- env.remove.Add(tx.Hash())
+ case err != nil:
+ env.failedTxs = append(env.failedTxs, tx)
if glog.V(logger.Detail) {
glog.Infof("TX (%x) failed, will be removed: %v\n", tx.Hash().Bytes()[:4], err)
}
generated by cgit (git 2.34.1) at 2024-11-12 22:25:29 +0800