core: fix heartbeat in txpool

rjl493456442 · karalabe · commit 5413df1dfa54 · 2020-07-24T11:12:59.000+03:00
core: address comment
diff --git 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,15 +1200,13 @@ 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)
 		drops, _ := list.Filter(pool.currentState.GetBalance(addr).Uint64(), pool.currentMaxGas)
 		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
@@ -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 {
