core/txpool: refactor processing of pending transactions

core/txpool: make heads into smaller struct

core/txpool miner: handle incoming txs in worker

core/txpool: implement PendingHashes

core/txpool: unexport types, use interface + minor tweaks

core/txpool, miner: post-rebase fixup

core/txpool, miner: add back handling of locals

core/txpool, miner: move new types to pending.go

core/txpool: testing

core/txpool: make tests pass

core/txpool, eth: fix up more tests

core/txpool: fix pendingset empty-check

core/txpool: add back sorting by time

Author: holiman

core/txpool/blobpool/blobpool.go

@@ -1446,11 +1446,15 @@ func (p *BlobPool) drop() {
 //
 // The transactions can also be pre-filtered by the dynamic fee components to
 // reduce allocations and load on downstream subsystems.
-func (p *BlobPool) Pending(filter txpool.PendingFilter) map[common.Address][]*txpool.LazyTransaction {
+func (p *BlobPool) Pending(filter txpool.PendingFilter) txpool.Pending {
 	// If only plain transactions are requested, this pool is unsuitable as it
 	// contains none, don't even bother.
 	if filter.OnlyPlainTxs {
-		return nil
+		return txpool.EmptyPending
+	}
+	if filter.OnlyLocals {
+		// There is no notion of local accounts in the blob pool.
+		return txpool.EmptyPending
 	}
 	// Track the amount of time waiting to retrieve the list of pending blob txs
 	// from the pool and the amount of time actually spent on assembling the data.
@@ -1466,19 +1470,31 @@ func (p *BlobPool) Pending(filter txpool.PendingFilter) map[common.Address][]*tx
 		pendtimeHist.Update(time.Since(execStart).Nanoseconds())
 	}()
 
-	pending := make(map[common.Address][]*txpool.LazyTransaction, len(p.index))
+	var (
+		baseFee = new(uint256.Int)
+		heads   = make(txpool.TipList, 0, len(p.index))
+		tails   = make(map[common.Address][]*txpool.LazyTransaction, len(p.index))
+	)
+	if filter.BaseFee != nil {
+		baseFee = filter.BaseFee
+	}
 	for addr, txs := range p.index {
-		lazies := make([]*txpool.LazyTransaction, 0, len(txs))
-		for _, tx := range txs {
-			// If transaction filtering was requested, discard badly priced ones
-			if filter.MinTip != nil && filter.BaseFee != nil {
+		var (
+			tail  []*txpool.LazyTransaction
+			first = true
+		)
+		for i, tx := range txs {
+			var tip *uint256.Int
+			if filter.BaseFee != nil {
 				if tx.execFeeCap.Lt(filter.BaseFee) {
 					break // basefee too low, cannot be included, discard rest of txs from the account
 				}
-				tip := new(uint256.Int).Sub(tx.execFeeCap, filter.BaseFee)
-				if tip.Gt(tx.execTipCap) {
-					tip = tx.execTipCap
-				}
+			}
+			tip = new(uint256.Int).Sub(tx.execFeeCap, baseFee)
+			if tip.Gt(tx.execTipCap) {
+				tip = tx.execTipCap
+			}
+			if filter.MinTip != nil {
 				if tip.Lt(filter.MinTip) {
 					break // allowed or remaining tip too low, cannot be included, discard rest of txs from the account
 				}
@@ -1488,22 +1504,49 @@ func (p *BlobPool) Pending(filter txpool.PendingFilter) map[common.Address][]*tx
 					break // blobfee too low, cannot be included, discard rest of txs from the account
 				}
 			}
-			// Transaction was accepted according to the filter, append to the pending list
-			lazies = append(lazies, &txpool.LazyTransaction{
-				Pool:      p,
-				Hash:      tx.hash,
-				Time:      execStart, // TODO(karalabe): Maybe save these and use that?
-				GasFeeCap: tx.execFeeCap,
-				GasTipCap: tx.execTipCap,
-				Gas:       tx.execGas,
-				BlobGas:   tx.blobGas,
-			})
+			// Transaction was accepted according to the filter, append to the pending set
+			lazyTx := &txpool.LazyTransaction{
+				Pool:    p,
+				Hash:    tx.hash,
+				Time:    execStart,
+				Fees:    *tip,
+				Gas:     tx.execGas,
+				BlobGas: tx.blobGas,
+			}
+			if first {
+				first = false
+				tail = make([]*txpool.LazyTransaction, 0, len(txs)-i)
+				heads = append(heads, &txpool.TxTips{
+					From: addr,
+					Tips: lazyTx.Fees,
+					Time: lazyTx.Time.UnixNano(),
+				})
+			}
+			tail = append(tail, lazyTx)
+		}
+		if len(tail) > 0 {
+			tails[addr] = tail
 		}
-		if len(lazies) > 0 {
-			pending[addr] = lazies
+	}
+	return txpool.NewPendingSet(heads, tails)
+}
+
+// PendingHashes retrieves the hashes of all currently processable transactions.
+// The returned list is grouped by origin account and sorted by nonce
+func (p *BlobPool) PendingHashes(filter txpool.PendingFilter) []common.Hash {
+	if filter.OnlyPlainTxs || filter.OnlyLocals {
+		return nil
+	}
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	var hashes = make([]common.Hash, 0, len(p.index))
+	for _, txs := range p.index {
+		for _, tx := range txs {
+			hashes = append(hashes, tx.hash)
 		}
 	}
-	return pending
+	return hashes
 }
 
 // updateStorageMetrics retrieves a bunch of stats from the data store and pushes

core/txpool/blobpool/blobpool_test.go

@@ -1356,15 +1356,22 @@ func benchmarkPoolPending(b *testing.B, datacap uint64) {
 	// Benchmark assembling the pending
 	b.ResetTimer()
 	b.ReportAllocs()
-
+	var p txpool.Pending
 	for i := 0; i < b.N; i++ {
-		p := pool.Pending(txpool.PendingFilter{
+		p = pool.Pending(txpool.PendingFilter{
 			MinTip:  uint256.NewInt(1),
 			BaseFee: chain.basefee,
 			BlobFee: chain.blobfee,
 		})
-		if len(p) != int(capacity) {
-			b.Fatalf("have %d want %d", len(p), capacity)
-		}
+	}
+	b.StopTimer()
+
+	count := 0
+	for ltx, _ := p.Peek(); ltx != nil; ltx, _ = p.Peek() {
+		p.Shift()
+		count++
+	}
+	if count != int(capacity) {
+		b.Fatalf("have %d want %d", count, capacity)
 	}
 }

core/txpool/legacypool/legacypool.go

@@ -522,57 +522,99 @@ func (pool *LegacyPool) ContentFrom(addr common.Address) ([]*types.Transaction,
 //
 // The transactions can also be pre-filtered by the dynamic fee components to
 // reduce allocations and load on downstream subsystems.
-func (pool *LegacyPool) Pending(filter txpool.PendingFilter) map[common.Address][]*txpool.LazyTransaction {
+func (pool *LegacyPool) Pending(filter txpool.PendingFilter) txpool.Pending {
 	// If only blob transactions are requested, this pool is unsuitable as it
 	// contains none, don't even bother.
 	if filter.OnlyBlobTxs {
-		return nil
+		return txpool.EmptyPending
 	}
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
-	// Convert the new uint256.Int types to the old big.Int ones used by the legacy pool
 	var (
-		minTipBig  *big.Int
-		baseFeeBig *big.Int
+		baseFee = new(uint256.Int)
+		minTip  = new(uint256.Int)
+		heads   = make(txpool.TipList, 0, len(pool.pending))
+		tails   = make(map[common.Address][]*txpool.LazyTransaction, len(pool.pending))
 	)
 	if filter.MinTip != nil {
-		minTipBig = filter.MinTip.ToBig()
-	}
-	if filter.BaseFee != nil {
-		baseFeeBig = filter.BaseFee.ToBig()
+		minTip = filter.MinTip
 	}
-	pending := make(map[common.Address][]*txpool.LazyTransaction, len(pool.pending))
+	baseFeeBig := baseFee.ToBig()
 	for addr, list := range pool.pending {
-		txs := list.Flatten()
-
-		// If the miner requests tip enforcement, cap the lists now
-		if minTipBig != nil && !pool.locals.contains(addr) {
-			for i, tx := range txs {
-				if tx.EffectiveGasTipIntCmp(minTipBig, baseFeeBig) < 0 {
-					txs = txs[:i]
-					break
-				}
-			}
+		if filter.NoLocals && pool.locals.contains(addr) {
+			continue
 		}
-		if len(txs) > 0 {
-			lazies := make([]*txpool.LazyTransaction, len(txs))
-			for i := 0; i < len(txs); i++ {
-				lazies[i] = &txpool.LazyTransaction{
-					Pool:      pool,
-					Hash:      txs[i].Hash(),
-					Tx:        txs[i],
-					Time:      txs[i].Time(),
-					GasFeeCap: uint256.MustFromBig(txs[i].GasFeeCap()),
-					GasTipCap: uint256.MustFromBig(txs[i].GasTipCap()),
-					Gas:       txs[i].Gas(),
-					BlobGas:   txs[i].BlobGas(),
-				}
+		if filter.OnlyLocals && !pool.locals.contains(addr) {
+			continue
+		}
+		var (
+			tail  []*txpool.LazyTransaction
+			first = true
+			txs   = list.Flatten()
+		)
+		for i, tx := range txs {
+			if tx.GasFeeCapIntCmp(baseFeeBig) < 0 {
+				break // basefee too low, cannot be included, discard rest of txs from the account
+			}
+			gasTipCap := uint256.MustFromBig(tx.GasTipCap())
+			gasFeeCap := uint256.MustFromBig(tx.GasFeeCap())
+			tip := new(uint256.Int).Sub(gasFeeCap, baseFee)
+			if tip.Gt(gasTipCap) {
+				tip = gasTipCap
 			}
-			pending[addr] = lazies
+			if tip.Lt(minTip) {
+				break // allowed or remaining tip too low, cannot be included, discard rest of txs from the account
+			}
+			lazyTx := &txpool.LazyTransaction{
+				Pool:    pool,
+				Hash:    txs[i].Hash(),
+				Tx:      txs[i],
+				Time:    txs[i].Time(),
+				Fees:    *tip,
+				Gas:     txs[i].Gas(),
+				BlobGas: txs[i].BlobGas(),
+			}
+			if first {
+				first = false
+				tail = make([]*txpool.LazyTransaction, 0, len(txs)-i)
+				heads = append(heads, &txpool.TxTips{
+					From: addr,
+					Tips: lazyTx.Fees,
+					Time: lazyTx.Time.UnixNano(),
+				})
+			}
+			tail = append(tail, lazyTx)
+		}
+		if len(tail) > 0 {
+			tails[addr] = tail
+		}
+	}
+	return txpool.NewPendingSet(heads, tails)
+}
+
+// PendingHashes retrieves the hashes of all currently processable transactions.
+// The returned list is grouped by origin account and sorted by nonce
+func (pool *LegacyPool) PendingHashes(filter txpool.PendingFilter) []common.Hash {
+	if filter.OnlyBlobTxs {
+		return nil
+	}
+	pool.mu.Lock()
+	defer pool.mu.Unlock()
+
+	var hashes = make([]common.Hash, 0, len(pool.pending))
+	for addr, list := range pool.pending {
+		if filter.NoLocals && pool.locals.contains(addr) {
+			continue
+		}
+		if filter.OnlyLocals && !pool.locals.contains(addr) {
+			continue
+		}
+		for _, tx := range list.Flatten() {
+			hashes = append(hashes, tx.Hash())
 		}
 	}
-	return pending
+	return hashes
 }
 
 // Locals retrieves the accounts currently considered local by the pool.