diff --git a/core/tx_pool.go b/core/tx_pool.go
index 168f2671ee1dbf98e6a80bb649fce45c150ab232..23ea6762ea0963704e4554e3f7bcec16e9ba3b7b 100644
--- a/core/tx_pool.go
+++ b/core/tx_pool.go
@@ -232,12 +232,11 @@ type TxPool struct {
locals *accountSet // Set of local transaction to exempt from eviction rules
journal *txJournal // Journal of local transaction to back up to disk
- pending map[common.Address]*txList // All currently processable transactions
- queue map[common.Address]*txList // Queued but non-processable transactions
- beats map[common.Address]time.Time // Last heartbeat from each known account
- queuedTs map[common.Hash]time.Time // Timestamp for when queued transactions were added
- all *txLookup // All transactions to allow lookups
- priced *txPricedList // All transactions sorted by price
+ pending map[common.Address]*txList // All currently processable transactions
+ queue map[common.Address]*txList // Queued but non-processable transactions
+ beats map[common.Address]time.Time // Last heartbeat from each known account
+ all *txLookup // All transactions to allow lookups
+ priced *txPricedList // All transactions sorted by price
chainHeadCh chan ChainHeadEvent
chainHeadSub event.Subscription
@@ -268,7 +267,6 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
pending: make(map[common.Address]*txList),
queue: make(map[common.Address]*txList),
beats: make(map[common.Address]time.Time),
- queuedTs: make(map[common.Hash]time.Time),
all: newTxLookup(),
chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize),
reqResetCh: make(chan *txpoolResetRequest),
@@ -365,12 +363,11 @@ func (pool *TxPool) loop() {
}
// Any non-locals old enough should be removed
if time.Since(pool.beats[addr]) > pool.config.Lifetime {
- for _, tx := range pool.queue[addr].Flatten() {
- if time.Since(pool.queuedTs[tx.Hash()]) > pool.config.Lifetime {
- queuedEvictionMeter.Mark(1)
- pool.removeTx(tx.Hash(), true)
- }
+ list := pool.queue[addr].Flatten()
+ for _, tx := range list {
+ pool.removeTx(tx.Hash(), true)
}
+ queuedEvictionMeter.Mark(int64(len(list)))
}
}
pool.mu.Unlock()
@@ -622,9 +619,11 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
pool.all.Add(tx)
pool.priced.Put(tx)
pool.journalTx(from, tx)
- pool.beats[from] = time.Now()
pool.queueTxEvent(tx)
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
+
+ // Successful promotion, bump the heartbeat
+ pool.beats[from] = time.Now()
return old != nil, nil
}
// New transaction isn't replacing a pending one, push into queue
@@ -665,20 +664,20 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
}
// Discard any previous transaction and mark this
if old != nil {
- old_hash := old.Hash()
- pool.all.Remove(old_hash)
+ pool.all.Remove(old.Hash())
pool.priced.Removed(1)
- delete(pool.queuedTs, old_hash)
queuedReplaceMeter.Mark(1)
} else {
// Nothing was replaced, bump the queued counter
queuedGauge.Inc(1)
- pool.queuedTs[hash] = time.Now()
}
if pool.all.Get(hash) == nil {
pool.all.Add(tx)
pool.priced.Put(tx)
- pool.queuedTs[hash] = time.Now()
+ }
+ // If we never record the heartbeat, do it right now.
+ if _, exist := pool.beats[from]; !exist {
+ pool.beats[from] = time.Now()
}
return old != nil, nil
}
@@ -711,7 +710,6 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
// An older transaction was better, discard this
pool.all.Remove(hash)
pool.priced.Removed(1)
- delete(pool.queuedTs, hash)
pendingDiscardMeter.Mark(1)
return false
}
@@ -730,10 +728,10 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
pool.priced.Put(tx)
}
// Set the potentially new pending nonce and notify any subsystems of the new tx
- pool.beats[addr] = time.Now()
- delete(pool.queuedTs, hash)
pool.pendingNonces.set(addr, tx.Nonce()+1)
+ // Successful promotion, bump the heartbeat
+ pool.beats[addr] = time.Now()
return true
}
@@ -923,10 +921,10 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
if removed, _ := future.Remove(tx); removed {
// Reduce the queued counter
queuedGauge.Dec(1)
- delete(pool.queuedTs, hash)
}
if future.Empty() {
delete(pool.queue, addr)
+ delete(pool.beats, addr)
}
}
}
@@ -1202,7 +1200,6 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
for _, tx := range forwards {
hash := tx.Hash()
pool.all.Remove(hash)
- delete(pool.queuedTs, hash)
}
log.Trace("Removed old queued transactions", "count", len(forwards))
// Drop all transactions that are too costly (low balance or out of gas)
@@ -1210,7 +1207,6 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
for _, tx := range drops {
hash := tx.Hash()
pool.all.Remove(hash)
- delete(pool.queuedTs, hash)
}
log.Trace("Removed unpayable queued transactions", "count", len(drops))
queuedNofundsMeter.Mark(int64(len(drops)))
@@ -1233,7 +1229,6 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
for _, tx := range caps {
hash := tx.Hash()
pool.all.Remove(hash)
- delete(pool.queuedTs, hash)
log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
}
queuedRateLimitMeter.Mark(int64(len(caps)))
@@ -1247,6 +1242,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
// Delete the entire queue entry if it became empty.
if list.Empty() {
delete(pool.queue, addr)
+ delete(pool.beats, addr)
}
}
return promoted
@@ -1431,7 +1427,6 @@ func (pool *TxPool) demoteUnexecutables() {
// Delete the entire pending entry if it became empty.
if list.Empty() {
delete(pool.pending, addr)
- delete(pool.beats, addr)
}
}
}
diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go
index 3f48ce6ce08b2eb8a939a44990fa910bd57bd8a7..f87d6fbea9e3685b579b50ef8bbf05a72305c7a1 100644
--- a/core/tx_pool_test.go
+++ b/core/tx_pool_test.go
@@ -109,9 +109,6 @@ func validateTxPoolInternals(pool *TxPool) error {
if priced := pool.priced.items.Len() - pool.priced.stales; priced != pending+queued {
return fmt.Errorf("total priced transaction count %d != %d pending + %d queued", priced, pending, queued)
}
- if queued != len(pool.queuedTs) {
- return fmt.Errorf("total queued transaction count %d != %d queuedTs length", queued, len(pool.queuedTs))
- }
// Ensure the next nonce to assign is the correct one
for addr, txs := range pool.pending {
@@ -948,7 +945,6 @@ func testTransactionQueueTimeLimiting(t *testing.T, nolocals bool) {
// remove current transactions and increase nonce to prepare for a reset and cleanup
statedb.SetNonce(crypto.PubkeyToAddress(remote.PublicKey), 2)
statedb.SetNonce(crypto.PubkeyToAddress(local.PublicKey), 2)
-
<-pool.requestReset(nil, nil)
// make sure queue, pending are cleared
@@ -963,93 +959,49 @@ func testTransactionQueueTimeLimiting(t *testing.T, nolocals bool) {
t.Fatalf("pool internal state corrupted: %v", err)
}
- if err := pool.AddLocal(pricedTransaction(2, 100000, big.NewInt(1), local)); err != nil {
- t.Fatalf("failed to add remote transaction: %v", err)
- }
+ // Queue gapped transactions
if err := pool.AddLocal(pricedTransaction(4, 100000, big.NewInt(1), local)); err != nil {
t.Fatalf("failed to add remote transaction: %v", err)
}
- if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(1), remote)); err != nil {
- t.Fatalf("failed to add remote transaction: %v", err)
- }
if err := pool.addRemoteSync(pricedTransaction(4, 100000, big.NewInt(1), remote)); err != nil {
t.Fatalf("failed to add remote transaction: %v", err)
}
+ time.Sleep(5 * evictionInterval) // A half lifetime pass
- // wait a short amount of time to add an additional future queued item to test proper eviction when
- // pending is removed
- time.Sleep(2 * evictionInterval)
- if err := pool.addRemoteSync(pricedTransaction(5, 100000, big.NewInt(1), remote)); err != nil {
+ // Queue executable transactions, the life cycle should be restarted.
+ if err := pool.AddLocal(pricedTransaction(2, 100000, big.NewInt(1), local)); err != nil {
t.Fatalf("failed to add remote transaction: %v", err)
}
+ if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(1), remote)); err != nil {
+ t.Fatalf("failed to add remote transaction: %v", err)
+ }
+ time.Sleep(6 * evictionInterval)
- // Make sure future queue and pending have transactions
+ // All gapped transactions shouldn't be kicked out
pending, queued = pool.Stats()
if pending != 2 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
}
- if queued != 3 {
+ if queued != 2 {
t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 3)
}
if err := validateTxPoolInternals(pool); err != nil {
t.Fatalf("pool internal state corrupted: %v", err)
}
- // Trigger a reset to make sure queued items are not evicted
- statedb.SetNonce(crypto.PubkeyToAddress(remote.PublicKey), 3)
- statedb.SetNonce(crypto.PubkeyToAddress(local.PublicKey), 3)
- <-pool.requestReset(nil, nil)
-
- // Wait for eviction to run
- time.Sleep(evictionInterval * 2)
-
- // a pool reset, empty pending list, or demotion of pending transactions should maintain
- // queued transactions for non locals and locals alike if the lifetime duration has not passed yet
- pending, queued = pool.Stats()
- if pending != 0 {
- t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 0)
- }
- if queued != 3 {
- 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 for the lifetime to run for all transactions except the one that was added later
- time.Sleep(evictionInterval * 7)
- pending, queued = pool.Stats()
- if pending != 0 {
- t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 0)
- }
- if nolocals {
- if queued != 1 {
- t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1)
- }
- } else {
- 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)
- }
-
- // lifetime should pass for the final transaction
- time.Sleep(evictionInterval * 2)
-
+ // The whole life time pass after last promotion, kick out stale transactions
+ time.Sleep(2 * config.Lifetime)
pending, queued = pool.Stats()
- if pending != 0 {
- t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 0)
+ if pending != 2 {
+ t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
}
if nolocals {
if queued != 0 {
- t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2)
+ 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, 0)
+ t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1)
}
}
if err := validateTxPoolInternals(pool); err != nil {