forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core/txpool: refactor processing of pending transactions

Browse source 
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
This commit is contained in:
Martin Holst Swende 2024-02-17 20:10:34 +01:00
parent aadcb88675
commit 5ccec6f7f8
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
15 changed files with 610 additions and 521 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
core/txpool/blobpool/blobpool.go
View file

@ -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)
}
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{
// Transaction was accepted according to the filter, append to the pending set
lazyTx := &txpool.LazyTransaction{
Pool: p,
Hash: tx.hash,
Time: execStart, // TODO(karalabe): Maybe save these and use that?
GasFeeCap: tx.execFeeCap,
GasTipCap: tx.execTipCap,
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(),
})
}
if len(lazies) > 0 {
pending[addr] = lazies
tail = append(tail, lazyTx)
}
if len(tail) > 0 {
tails[addr] = tail
}
}
return pending
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 hashes
}
// updateStorageMetrics retrieves a bunch of stats from the data store and pushes

17
core/txpool/blobpool/blobpool_test.go
View file

@ -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)
}
}

122
core/txpool/legacypool/legacypool.go
View file

@ -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 txpool.EmptyPending
}
pool.mu.Lock()
defer pool.mu.Unlock()
var (
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 {
minTip = filter.MinTip
}
baseFeeBig := baseFee.ToBig()
for addr, list := range pool.pending {
if filter.NoLocals && pool.locals.contains(addr) {
continue
}
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
}
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()
// Convert the new uint256.Int types to the old big.Int ones used by the legacy pool
var (
minTipBig *big.Int
baseFeeBig *big.Int
)
if filter.MinTip != nil {
minTipBig = filter.MinTip.ToBig()
}
if filter.BaseFee != nil {
baseFeeBig = filter.BaseFee.ToBig()
}
pending := make(map[common.Address][]*txpool.LazyTransaction, len(pool.pending))
var hashes = make([]common.Hash, 0, len(pool.pending))
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 filter.OnlyLocals && !pool.locals.contains(addr) {
continue
}
for _, tx := range list.Flatten() {
hashes = append(hashes, tx.Hash())
}
}
}
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(),
}
}
pending[addr] = lazies
}
}
return pending
return hashes
}
// Locals retrieves the accounts currently considered local by the pool.

154
core/txpool/pending.go Normal file
View file

@ -0,0 +1,154 @@
// Copyright 2024 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package txpool
import (
"container/heap"
"github.com/ethereum/go-ethereum/common"
"github.com/holiman/uint256"
)
type Pending interface {
// Shift replaces the current best head with the next one from the same account.
Shift()
// Peek returns the next transaction by price.
Peek() (*LazyTransaction, *uint256.Int)
// Pop removes the best transaction, *not* replacing it with the next one from
// the same account. This should be used when a transaction cannot be executed
// and hence all subsequent ones should be discarded from the same account.
Pop()
// Empty returns true if the set is empty.
Empty() bool
// Clears the set
Clear()
}
// PendingFilter is a collection of filter rules to allow retrieving a subset
// of transactions for announcement or mining.
//
// Note, the entries here are not arbitrary useful filters, rather each one has
// a very specific call site in mind and each one can be evaluated very cheaply
// by the pool implementations. Only add new ones that satisfy those constraints.
type PendingFilter struct {
MinTip *uint256.Int // Minimum miner tip required to include a transaction
BaseFee *uint256.Int // Minimum 1559 basefee needed to include a transaction
BlobFee *uint256.Int // Minimum 4844 blobfee needed to include a blob transaction
OnlyPlainTxs bool // Return only plain EVM transactions (peer-join announces, block space filling)
OnlyBlobTxs bool // Return only blob transactions (block blob-space filling)
OnlyLocals bool // Return only txs from 'local' addresses.
NoLocals bool // Remove all txs from 'local' addresses
}
type TxTips struct {
From common.Address // sender
Tips uint256.Int // miner-fees earned by this transaction.
Time int64 // Time when the transaction was first seen
}
type TipList []*TxTips
func (f TipList) Len() int {
return len(f)
}
func (f TipList) Less(i, j int) bool {
cmp := f[i].Tips.Cmp(&f[j].Tips)
if cmp == 0 {
return f[i].Time < f[j].Time
}
return cmp > 0
}
func (f TipList) Swap(i, j int) {
f[i], f[j] = f[j], f[i]
}
func (f *TipList) Push(x any) {
*f = append(*f, x.(*TxTips))
}
func (f *TipList) Pop() any {
old := *f
n := len(old)
x := old[n-1]
old[n-1] = nil
*f = old[0 : n-1]
return x
}
type pendingSet struct {
Tails map[common.Address][]*LazyTransaction // Per account nonce-sorted list of transactions
Heads TipList // Next transaction for each unique account (price heap)
}
var EmptyPending = NewPendingSet(nil, nil)
func NewPendingSet(heads TipList, tails map[common.Address][]*LazyTransaction) *pendingSet {
if len(heads) != 0 {
heap.Init(&heads)
}
return &pendingSet{
Tails: tails,
Heads: heads,
}
}
// Shift replaces the current best head with the next one from the same account.
func (ps *pendingSet) Shift() {
acc := ps.Heads[0].From
if txs, ok := ps.Tails[acc]; ok && len(txs) > 1 {
ps.Heads[0].Tips = txs[1].Fees
ps.Heads[0].Time = txs[1].Time.UnixNano()
ps.Tails[acc] = txs[1:]
heap.Fix(&ps.Heads, 0)
return
}
heap.Pop(&ps.Heads)
}
// Peek returns the next transaction by price.
func (ps *pendingSet) Peek() (*LazyTransaction, *uint256.Int) {
if len(ps.Heads) == 0 {
return nil, nil
}
sender := ps.Heads[0].From
fees := ps.Heads[0].Tips
tx := ps.Tails[sender][0]
return tx, &fees
}
func (ps *pendingSet) Clear() {
ps.Heads = nil
ps.Tails = nil
}
func (ps *pendingSet) Empty() bool {
return len(ps.Heads) == 0
}
// Pop removes the best transaction, *not* replacing it with the next one from
// the same account. This should be used when a transaction cannot be executed
// and hence all subsequent ones should be discarded from the same account.
func (ps *pendingSet) Pop() {
heap.Pop(&ps.Heads)
}

151
core/txpool/pending_test.go Normal file
View file

@ -0,0 +1,151 @@
// Copyright 2024 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package txpool
import (
"container/heap"
"math/rand"
"testing"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/math"
"github.com/holiman/uint256"
)
func initLists() (heads TipList, tails map[common.Address][]*LazyTransaction) {
tails = make(map[common.Address][]*LazyTransaction)
for i := 0; i < 25; i++ {
var (
addr = common.Address{byte(i)}
tail []*LazyTransaction
first = true
)
for j := 0; j < 25; j++ {
tip := uint256.NewInt(uint64(100*i + j))
lazyTx := &LazyTransaction{
Pool: nil,
Hash: common.Hash{byte(i), byte(j)},
Fees: *tip,
Gas: uint64(i),
BlobGas: uint64(j),
}
if first {
first = false
heads = append(heads, &TxTips{
From: addr,
Tips: lazyTx.Fees,
Time: 0,
})
}
tail = append(tail, lazyTx)
}
if len(tail) > 0 {
tails[addr] = tail
}
}
// un-sort the heads
rand.Shuffle(len(heads), func(i, j int) {
heads[i], heads[j] = heads[j], heads[i]
})
return heads, tails
}
// Test the sorting and the Shift operation
func TestPendingSortAndShift(t *testing.T) {
// Create the pending-set
var (
heads, tails = initLists()
expectedCount = 25 * 25
txset = NewPendingSet(heads, tails)
haveCount = 0
prevFee = uint64(math.MaxInt64)
)
if txset.Empty() {
t.Fatalf("expected non-empty")
}
for {
ltx, fee := txset.Peek()
if ltx == nil {
break
}
haveCount++
if fee.Cmp(&ltx.Fees) != 0 {
t.Fatalf("error tx %d: %v != %v", haveCount, fee, ltx.Fees)
}
if fee.Uint64() > prevFee {
t.Fatalf("tx %d: fee %d > previous fee %d", haveCount, fee, prevFee)
}
txset.Shift()
}
if haveCount != expectedCount {
t.Errorf("expected %d transactions, found %d", expectedCount, haveCount)
}
}
// Test the sorting and the Pop operation
func TestPendingSortAndPop(t *testing.T) {
var (
heads, tails = initLists()
expectedCount = 25 * 1
txset = NewPendingSet(heads, tails)
haveCount = 0
prevFee = uint64(math.MaxInt64)
)
for {
ltx, fee := txset.Peek()
if ltx == nil {
break
}
haveCount++
if fee.Cmp(&ltx.Fees) != 0 {
t.Fatalf("error tx %d: %v != %v", haveCount, fee, ltx.Fees)
}
if fee.Uint64() > prevFee {
t.Fatalf("tx %d: fee %d > previous fee %d", haveCount, fee, prevFee)
}
txset.Pop()
}
if haveCount != expectedCount {
t.Errorf("expected %d transactions, found %d", expectedCount, haveCount)
}
}
// Tests that if multiple transactions have the same price, the ones seen earlier
// are prioritized to avoid network spam attacks aiming for a specific ordering.
func TestSortingByTime(t *testing.T) {
var heads TipList
for i := 0; i < 25; i++ {
addr := common.Address{byte(i)}
heads = append(heads, &TxTips{
From: addr,
Tips: *(uint256.NewInt(uint64(100))),
Time: int64(i),
})
}
// un-sort the heads
rand.Shuffle(len(heads), func(i, j int) {
heads[i], heads[j] = heads[j], heads[i]
})
heap.Init(&heads)
for want := int64(0); want < 25; want++ {
obj := heap.Pop(&heads).(*TxTips)
if have := obj.Time; have != want {
t.Fatalf("have %d want %d", have, want)
}
}
}

24
core/txpool/subpool.go
View file

@ -36,8 +36,7 @@ type LazyTransaction struct {
Tx *types.Transaction // Transaction if already resolved
Time time.Time // Time when the transaction was first seen
GasFeeCap *uint256.Int // Maximum fee per gas the transaction may consume
GasTipCap *uint256.Int // Maximum miner tip per gas the transaction can pay
Fees uint256.Int
Gas uint64 // Amount of gas required by the transaction
BlobGas uint64 // Amount of blob gas required by the transaction
@ -70,21 +69,6 @@ type LazyResolver interface {
// may request (and relinquish) exclusive access to certain addresses.
type AddressReserver func(addr common.Address, reserve bool) error
// PendingFilter is a collection of filter rules to allow retrieving a subset
// of transactions for announcement or mining.
//
// Note, the entries here are not arbitrary useful filters, rather each one has
// a very specific call site in mind and each one can be evaluated very cheaply
// by the pool implementations. Only add new ones that satisfy those constraints.
type PendingFilter struct {
MinTip *uint256.Int // Minimum miner tip required to include a transaction
BaseFee *uint256.Int // Minimum 1559 basefee needed to include a transaction
BlobFee *uint256.Int // Minimum 4844 blobfee needed to include a blob transaction
OnlyPlainTxs bool // Return only plain EVM transactions (peer-join announces, block space filling)
OnlyBlobTxs bool // Return only blob transactions (block blob-space filling)
}
// SubPool represents a specialized transaction pool that lives on its own (e.g.
// blob pool). Since independent of how many specialized pools we have, they do
// need to be updated in lockstep and assemble into one coherent view for block
@ -133,7 +117,11 @@ type SubPool interface {
//
// The transactions can also be pre-filtered by the dynamic fee components to
// reduce allocations and load on downstream subsystems.
Pending(filter PendingFilter) map[common.Address][]*LazyTransaction
Pending(filter PendingFilter) Pending
// PendingHashes retrieves the hashes of all currently processable transactions.
// The returned list is grouped by origin account and sorted by nonce
PendingHashes(filter PendingFilter) []common.Hash
// SubscribeTransactions subscribes to new transaction events. The subscriber
// can decide whether to receive notifications only for newly seen transactions

28
core/txpool/txpool.go
View file

@ -356,14 +356,30 @@ func (p *TxPool) Add(txs []*types.Transaction, local bool, sync bool) []error {
//
// The transactions can also be pre-filtered by the dynamic fee components to
// reduce allocations and load on downstream subsystems.
func (p *TxPool) Pending(filter PendingFilter) map[common.Address][]*LazyTransaction {
txs := make(map[common.Address][]*LazyTransaction)
for _, subpool := range p.subpools {
for addr, set := range subpool.Pending(filter) {
txs[addr] = set
func (p *TxPool) Pending(filter PendingFilter) Pending {
var pending Pending
// This code is not quite correct, it assumes that the filter requests
// _either_ blobs or legacy.
if !filter.OnlyBlobTxs && !filter.OnlyPlainTxs {
panic("please request either only-blobs or only-plain")
}
for _, sub := range p.subpools {
if pending = sub.Pending(filter); !pending.Empty() {
return pending
}
}
return txs
return pending
}
// PendingHashes retrieves the hashes of all currently processable transactions.
// The returned list is grouped by origin account and sorted by nonce
func (p *TxPool) PendingHashes(filter PendingFilter) []common.Hash {
var hashes []common.Hash
for _, sub := range p.subpools {
h := sub.PendingHashes(filter)
hashes = append(hashes, h...)
}
return hashes
}
// SubscribeTransactions registers a subscription for new transaction events,

9
eth/api_backend.go
View file

@ -289,12 +289,15 @@ func (b *EthAPIBackend) SendTx(ctx context.Context, signedTx *types.Transaction)
func (b *EthAPIBackend) GetPoolTransactions() (types.Transactions, error) {
pending := b.eth.txPool.Pending(txpool.PendingFilter{})
var txs types.Transactions
for _, batch := range pending {
for _, lazy := range batch {
for {
lazy, _ := pending.Peek()
if lazy == nil {
break
}
if tx := lazy.Resolve(); tx != nil {
txs = append(txs, tx)
}
}
pending.Shift()
}
return txs, nil
}

4
eth/catalyst/simulated_beacon.go
View file

@ -279,7 +279,7 @@ func (c *SimulatedBeacon) Rollback() {
// Fork sets the head to the provided hash.
func (c *SimulatedBeacon) Fork(parentHash common.Hash) error {
if len(c.eth.TxPool().Pending(txpool.PendingFilter{})) != 0 {
if len(c.eth.TxPool().PendingHashes(txpool.PendingFilter{})) != 0 {
return errors.New("pending block dirty")
}
parent := c.eth.BlockChain().GetBlockByHash(parentHash)
@ -291,7 +291,7 @@ func (c *SimulatedBeacon) Fork(parentHash common.Hash) error {
// AdjustTime creates a new block with an adjusted timestamp.
func (c *SimulatedBeacon) AdjustTime(adjustment time.Duration) error {
if len(c.eth.TxPool().Pending(txpool.PendingFilter{})) != 0 {
if len(c.eth.TxPool().PendingHashes(txpool.PendingFilter{})) != 0 {
return errors.New("could not adjust time on non-empty block")
}
parent := c.eth.BlockChain().CurrentBlock()

8
eth/handler.go
View file

@ -72,9 +72,11 @@ type txPool interface {
// Add should add the given transactions to the pool.
Add(txs []*types.Transaction, local bool, sync bool) []error
// Pending should return pending transactions.
// The slice should be modifiable by the caller.
Pending(filter txpool.PendingFilter) map[common.Address][]*txpool.LazyTransaction
// Pending return pending transactions.
Pending(filter txpool.PendingFilter) txpool.Pending
// PendingHashes returns a list of the hashes for all pending transactions.
PendingHashes(filter txpool.PendingFilter) []common.Hash
// SubscribeTransactions subscribes to new transaction events. The subscriber
// can decide whether to receive notifications only for newly seen transactions

56
eth/handler_test.go
View file

@ -92,7 +92,7 @@ func (p *testTxPool) Add(txs []*types.Transaction, local bool, sync bool) []erro
}
// Pending returns all the transactions known to the pool
func (p *testTxPool) Pending(filter txpool.PendingFilter) map[common.Address][]*txpool.LazyTransaction {
func (p *testTxPool) Pending(filter txpool.PendingFilter) txpool.Pending {
p.lock.RLock()
defer p.lock.RUnlock()
@ -104,21 +104,63 @@ func (p *testTxPool) Pending(filter txpool.PendingFilter) map[common.Address][]*
for _, batch := range batches {
sort.Sort(types.TxByNonce(batch))
}
pending := make(map[common.Address][]*txpool.LazyTransaction)
var (
tails = make(map[common.Address][]*txpool.LazyTransaction)
heads txpool.TipList
baseFee = new(uint256.Int)
)
if filter.BaseFee != nil {
baseFee = filter.BaseFee
}
for addr, batch := range batches {
for _, tx := range batch {
pending[addr] = append(pending[addr], &txpool.LazyTransaction{
var tail []*txpool.LazyTransaction
for i, tx := range batch {
gasTipCap := uint256.MustFromBig(tx.GasTipCap())
gasFeeCap := uint256.MustFromBig(tx.GasFeeCap())
tip := new(uint256.Int).Sub(gasFeeCap, baseFee)
if tip.Gt(gasTipCap) {
tip = gasTipCap
}
ltx := &txpool.LazyTransaction{
Hash: tx.Hash(),
Tx: tx,
Time: tx.Time(),
GasFeeCap: uint256.MustFromBig(tx.GasFeeCap()),
GasTipCap: uint256.MustFromBig(tx.GasTipCap()),
Fees: *tip,
Gas: tx.Gas(),
BlobGas: tx.BlobGas(),
}
tail = append(tail, ltx)
if i == 0 {
heads = append(heads, &txpool.TxTips{
From: addr,
Tips: *tip,
Time: ltx.Time.UnixNano(),
})
}
}
return pending
tails[addr] = tail
}
return txpool.NewPendingSet(heads, tails)
}
func (p *testTxPool) PendingHashes(filter txpool.PendingFilter) []common.Hash {
p.lock.RLock()
defer p.lock.RUnlock()
batches := make(map[common.Address][]*types.Transaction)
for _, tx := range p.pool {
from, _ := types.Sender(types.HomesteadSigner{}, tx)
batches[from] = append(batches[from], tx)
}
var hashes []common.Hash
for _, batch := range batches {
sort.Sort(types.TxByNonce(batch))
for _, tx := range batch {
hashes = append(hashes, tx.Hash())
}
}
return hashes
}
// SubscribeTransactions should return an event subscription of NewTxsEvent and

8
eth/sync.go
View file

@ -17,19 +17,13 @@
package eth
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/eth/protocols/eth"
)
// syncTransactions starts sending all currently pending transactions to the given peer.
func (h *handler) syncTransactions(p *eth.Peer) {
var hashes []common.Hash
for _, batch := range h.txpool.Pending(txpool.PendingFilter{OnlyPlainTxs: true}) {
for _, tx := range batch {
hashes = append(hashes, tx.Hash)
}
}
hashes := h.txpool.PendingHashes(txpool.PendingFilter{OnlyPlainTxs: true})
if len(hashes) == 0 {
return
}

166
miner/ordering.go
View file

@ -1,166 +0,0 @@
// Copyright 2014 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package miner
import (
"container/heap"
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
"github.com/holiman/uint256"
)
// txWithMinerFee wraps a transaction with its gas price or effective miner gasTipCap
type txWithMinerFee struct {
tx *txpool.LazyTransaction
from common.Address
fees *uint256.Int
}
// newTxWithMinerFee creates a wrapped transaction, calculating the effective
// miner gasTipCap if a base fee is provided.
// Returns error in case of a negative effective miner gasTipCap.
func newTxWithMinerFee(tx *txpool.LazyTransaction, from common.Address, baseFee *uint256.Int) (*txWithMinerFee, error) {
tip := new(uint256.Int).Set(tx.GasTipCap)
if baseFee != nil {
if tx.GasFeeCap.Cmp(baseFee) < 0 {
return nil, types.ErrGasFeeCapTooLow
}
tip = new(uint256.Int).Sub(tx.GasFeeCap, baseFee)
if tip.Gt(tx.GasTipCap) {
tip = tx.GasTipCap
}
}
return &txWithMinerFee{
tx: tx,
from: from,
fees: tip,
}, nil
}
// txByPriceAndTime implements both the sort and the heap interface, making it useful
// for all at once sorting as well as individually adding and removing elements.
type txByPriceAndTime []*txWithMinerFee
func (s txByPriceAndTime) Len() int { return len(s) }
func (s txByPriceAndTime) Less(i, j int) bool {
// If the prices are equal, use the time the transaction was first seen for
// deterministic sorting
cmp := s[i].fees.Cmp(s[j].fees)
if cmp == 0 {
return s[i].tx.Time.Before(s[j].tx.Time)
}
return cmp > 0
}
func (s txByPriceAndTime) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s *txByPriceAndTime) Push(x interface{}) {
*s = append(*s, x.(*txWithMinerFee))
}
func (s *txByPriceAndTime) Pop() interface{} {
old := *s
n := len(old)
x := old[n-1]
old[n-1] = nil
*s = old[0 : n-1]
return x
}
// transactionsByPriceAndNonce represents a set of transactions that can return
// transactions in a profit-maximizing sorted order, while supporting removing
// entire batches of transactions for non-executable accounts.
type transactionsByPriceAndNonce struct {
txs map[common.Address][]*txpool.LazyTransaction // Per account nonce-sorted list of transactions
heads txByPriceAndTime // Next transaction for each unique account (price heap)
signer types.Signer // Signer for the set of transactions
baseFee *uint256.Int // Current base fee
}
// newTransactionsByPriceAndNonce creates a transaction set that can retrieve
// price sorted transactions in a nonce-honouring way.
//
// Note, the input map is reowned so the caller should not interact any more with
// if after providing it to the constructor.
func newTransactionsByPriceAndNonce(signer types.Signer, txs map[common.Address][]*txpool.LazyTransaction, baseFee *big.Int) *transactionsByPriceAndNonce {
// Convert the basefee from header format to uint256 format
var baseFeeUint *uint256.Int
if baseFee != nil {
baseFeeUint = uint256.MustFromBig(baseFee)
}
// Initialize a price and received time based heap with the head transactions
heads := make(txByPriceAndTime, 0, len(txs))
for from, accTxs := range txs {
wrapped, err := newTxWithMinerFee(accTxs[0], from, baseFeeUint)
if err != nil {
delete(txs, from)
continue
}
heads = append(heads, wrapped)
txs[from] = accTxs[1:]
}
heap.Init(&heads)
// Assemble and return the transaction set
return &transactionsByPriceAndNonce{
txs: txs,
heads: heads,
signer: signer,
baseFee: baseFeeUint,
}
}
// Peek returns the next transaction by price.
func (t *transactionsByPriceAndNonce) Peek() (*txpool.LazyTransaction, *uint256.Int) {
if len(t.heads) == 0 {
return nil, nil
}
return t.heads[0].tx, t.heads[0].fees
}
// Shift replaces the current best head with the next one from the same account.
func (t *transactionsByPriceAndNonce) Shift() {
acc := t.heads[0].from
if txs, ok := t.txs[acc]; ok && len(txs) > 0 {
if wrapped, err := newTxWithMinerFee(txs[0], acc, t.baseFee); err == nil {
t.heads[0], t.txs[acc] = wrapped, txs[1:]
heap.Fix(&t.heads, 0)
return
}
}
heap.Pop(&t.heads)
}
// Pop removes the best transaction, *not* replacing it with the next one from
// the same account. This should be used when a transaction cannot be executed
// and hence all subsequent ones should be discarded from the same account.
func (t *transactionsByPriceAndNonce) Pop() {
heap.Pop(&t.heads)
}
// Empty returns if the price heap is empty. It can be used to check it simpler
// than calling peek and checking for nil return.
func (t *transactionsByPriceAndNonce) Empty() bool {
return len(t.heads) == 0
}
// Clear removes the entire content of the heap.
func (t *transactionsByPriceAndNonce) Clear() {
t.heads, t.txs = nil, nil
}

196
miner/ordering_test.go
View file

@ -1,196 +0,0 @@
// Copyright 2014 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package miner
import (
"crypto/ecdsa"
"math/big"
"math/rand"
"testing"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/holiman/uint256"
)
func TestTransactionPriceNonceSortLegacy(t *testing.T) {
t.Parallel()
testTransactionPriceNonceSort(t, nil)
}
func TestTransactionPriceNonceSort1559(t *testing.T) {
t.Parallel()
testTransactionPriceNonceSort(t, big.NewInt(0))
testTransactionPriceNonceSort(t, big.NewInt(5))
testTransactionPriceNonceSort(t, big.NewInt(50))
}
// Tests that transactions can be correctly sorted according to their price in
// decreasing order, but at the same time with increasing nonces when issued by
// the same account.
func testTransactionPriceNonceSort(t *testing.T, baseFee *big.Int) {
// Generate a batch of accounts to start with
keys := make([]*ecdsa.PrivateKey, 25)
for i := 0; i < len(keys); i++ {
keys[i], _ = crypto.GenerateKey()
}
signer := types.LatestSignerForChainID(common.Big1)
// Generate a batch of transactions with overlapping values, but shifted nonces
groups := map[common.Address][]*txpool.LazyTransaction{}
expectedCount := 0
for start, key := range keys {
addr := crypto.PubkeyToAddress(key.PublicKey)
count := 25
for i := 0; i < 25; i++ {
var tx *types.Transaction
gasFeeCap := rand.Intn(50)
if baseFee == nil {
tx = types.NewTx(&types.LegacyTx{
Nonce: uint64(start + i),
To: &common.Address{},
Value: big.NewInt(100),
Gas: 100,
GasPrice: big.NewInt(int64(gasFeeCap)),
Data: nil,
})
} else {
tx = types.NewTx(&types.DynamicFeeTx{
Nonce: uint64(start + i),
To: &common.Address{},
Value: big.NewInt(100),
Gas: 100,
GasFeeCap: big.NewInt(int64(gasFeeCap)),
GasTipCap: big.NewInt(int64(rand.Intn(gasFeeCap + 1))),
Data: nil,
})
if count == 25 && int64(gasFeeCap) < baseFee.Int64() {
count = i
}
}
tx, err := types.SignTx(tx, signer, key)
if err != nil {
t.Fatalf("failed to sign tx: %s", err)
}
groups[addr] = append(groups[addr], &txpool.LazyTransaction{
Hash: tx.Hash(),
Tx: tx,
Time: tx.Time(),
GasFeeCap: uint256.MustFromBig(tx.GasFeeCap()),
GasTipCap: uint256.MustFromBig(tx.GasTipCap()),
Gas: tx.Gas(),
BlobGas: tx.BlobGas(),
})
}
expectedCount += count
}
// Sort the transactions and cross check the nonce ordering
txset := newTransactionsByPriceAndNonce(signer, groups, baseFee)
txs := types.Transactions{}
for tx, _ := txset.Peek(); tx != nil; tx, _ = txset.Peek() {
txs = append(txs, tx.Tx)
txset.Shift()
}
if len(txs) != expectedCount {
t.Errorf("expected %d transactions, found %d", expectedCount, len(txs))
}
for i, txi := range txs {
fromi, _ := types.Sender(signer, txi)
// Make sure the nonce order is valid
for j, txj := range txs[i+1:] {
fromj, _ := types.Sender(signer, txj)
if fromi == fromj && txi.Nonce() > txj.Nonce() {
t.Errorf("invalid nonce ordering: tx #%d (A=%x N=%v) < tx #%d (A=%x N=%v)", i, fromi[:4], txi.Nonce(), i+j, fromj[:4], txj.Nonce())
}
}
// If the next tx has different from account, the price must be lower than the current one
if i+1 < len(txs) {
next := txs[i+1]
fromNext, _ := types.Sender(signer, next)
tip, err := txi.EffectiveGasTip(baseFee)
nextTip, nextErr := next.EffectiveGasTip(baseFee)
if err != nil || nextErr != nil {
t.Errorf("error calculating effective tip: %v, %v", err, nextErr)
}
if fromi != fromNext && tip.Cmp(nextTip) < 0 {
t.Errorf("invalid gasprice ordering: tx #%d (A=%x P=%v) < tx #%d (A=%x P=%v)", i, fromi[:4], txi.GasPrice(), i+1, fromNext[:4], next.GasPrice())
}
}
}
}
// Tests that if multiple transactions have the same price, the ones seen earlier
// are prioritized to avoid network spam attacks aiming for a specific ordering.
func TestTransactionTimeSort(t *testing.T) {
t.Parallel()
// Generate a batch of accounts to start with
keys := make([]*ecdsa.PrivateKey, 5)
for i := 0; i < len(keys); i++ {
keys[i], _ = crypto.GenerateKey()
}
signer := types.HomesteadSigner{}
// Generate a batch of transactions with overlapping prices, but different creation times
groups := map[common.Address][]*txpool.LazyTransaction{}
for start, key := range keys {
addr := crypto.PubkeyToAddress(key.PublicKey)
tx, _ := types.SignTx(types.NewTransaction(0, common.Address{}, big.NewInt(100), 100, big.NewInt(1), nil), signer, key)
tx.SetTime(time.Unix(0, int64(len(keys)-start)))
groups[addr] = append(groups[addr], &txpool.LazyTransaction{
Hash: tx.Hash(),
Tx: tx,
Time: tx.Time(),
GasFeeCap: uint256.MustFromBig(tx.GasFeeCap()),
GasTipCap: uint256.MustFromBig(tx.GasTipCap()),
Gas: tx.Gas(),
BlobGas: tx.BlobGas(),
})
}
// Sort the transactions and cross check the nonce ordering
txset := newTransactionsByPriceAndNonce(signer, groups, nil)
txs := types.Transactions{}
for tx, _ := txset.Peek(); tx != nil; tx, _ = txset.Peek() {
txs = append(txs, tx.Tx)
txset.Shift()
}
if len(txs) != len(keys) {
t.Errorf("expected %d transactions, found %d", len(keys), len(txs))
}
for i, txi := range txs {
fromi, _ := types.Sender(signer, txi)
if i+1 < len(txs) {
next := txs[i+1]
fromNext, _ := types.Sender(signer, next)
if txi.GasPrice().Cmp(next.GasPrice()) < 0 {
t.Errorf("invalid gasprice ordering: tx #%d (A=%x P=%v) < tx #%d (A=%x P=%v)", i, fromi[:4], txi.GasPrice(), i+1, fromNext[:4], next.GasPrice())
}
// Make sure time order is ascending if the txs have the same gas price
if txi.GasPrice().Cmp(next.GasPrice()) == 0 && txi.Time().After(next.Time()) {
t.Errorf("invalid received time ordering: tx #%d (A=%x T=%v) > tx #%d (A=%x T=%v)", i, fromi[:4], txi.Time(), i+1, fromNext[:4], next.Time())
}
}
}
}

83
miner/worker.go
View file

@ -271,7 +271,7 @@ func (miner *Miner) applyTransaction(env *environment, tx *types.Transaction) (*
return receipt, err
}
func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs *transactionsByPriceAndNonce, interrupt *atomic.Int32) error {
func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs txpool.Pending, interrupt *atomic.Int32) error {
gasLimit := env.header.GasLimit
if env.gasPool == nil {
env.gasPool = new(core.GasPool).AddGas(gasLimit)
@ -298,7 +298,7 @@ func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs *tran
// Retrieve the next transaction and abort if all done.
var (
ltx *txpool.LazyTransaction
txs *transactionsByPriceAndNonce
txs txpool.Pending
)
pltx, ptip := plainTxs.Peek()
bltx, btip := blobTxs.Peek()
@ -315,6 +315,7 @@ func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs *tran
txs, ltx = plainTxs, pltx
}
}
if ltx == nil {
break
}
@ -376,53 +377,61 @@ func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs *tran
// be customized with the plugin in the future.
func (miner *Miner) fillTransactions(interrupt *atomic.Int32, env *environment) error {
miner.confMu.RLock()
tip := miner.config.GasPrice
tip := uint256.MustFromBig(miner.config.GasPrice)
miner.confMu.RUnlock()
// Retrieve the pending transactions pre-filtered by the 1559/4844 dynamic fees
filter := txpool.PendingFilter{
MinTip: uint256.MustFromBig(tip),
}
var (
baseFee *uint256.Int
blobFee *uint256.Int
)
if env.header.BaseFee != nil {
filter.BaseFee = uint256.MustFromBig(env.header.BaseFee)
baseFee = uint256.MustFromBig(env.header.BaseFee)
}
if env.header.ExcessBlobGas != nil {
filter.BlobFee = uint256.MustFromBig(eip4844.CalcBlobFee(*env.header.ExcessBlobGas))
blobFee = uint256.MustFromBig(eip4844.CalcBlobFee(*env.header.ExcessBlobGas))
}
filter.OnlyPlainTxs, filter.OnlyBlobTxs = true, false
pendingPlainTxs := miner.txpool.Pending(filter)
localPlainTxs := miner.txpool.Pending(txpool.PendingFilter{
MinTip: nil, // no mintip for locals
BaseFee: baseFee,
BlobFee: blobFee,
OnlyPlainTxs: true,
OnlyBlobTxs: false,
OnlyLocals: true,
})
localBlobTxs := miner.txpool.Pending(txpool.PendingFilter{
MinTip: nil, // no mintip for locals
BaseFee: baseFee,
BlobFee: blobFee,
OnlyPlainTxs: false,
OnlyBlobTxs: true,
OnlyLocals: true,
})
filter.OnlyPlainTxs, filter.OnlyBlobTxs = false, true
pendingBlobTxs := miner.txpool.Pending(filter)
// Split the pending transactions into locals and remotes.
localPlainTxs, remotePlainTxs := make(map[common.Address][]*txpool.LazyTransaction), pendingPlainTxs
localBlobTxs, remoteBlobTxs := make(map[common.Address][]*txpool.LazyTransaction), pendingBlobTxs
for _, account := range miner.txpool.Locals() {
if txs := remotePlainTxs[account]; len(txs) > 0 {
delete(remotePlainTxs, account)
localPlainTxs[account] = txs
}
if txs := remoteBlobTxs[account]; len(txs) > 0 {
delete(remoteBlobTxs, account)
localBlobTxs[account] = txs
}
}
remotePlainTxs := miner.txpool.Pending(txpool.PendingFilter{
MinTip: tip,
BaseFee: baseFee,
BlobFee: blobFee,
OnlyPlainTxs: true,
OnlyBlobTxs: false,
OnlyLocals: false,
})
remoteBlobTxs := miner.txpool.Pending(txpool.PendingFilter{
MinTip: tip,
BaseFee: baseFee,
BlobFee: blobFee,
OnlyPlainTxs: false,
OnlyBlobTxs: true,
OnlyLocals: false,
})
// Fill the block with all available pending transactions.
if len(localPlainTxs) > 0 || len(localBlobTxs) > 0 {
plainTxs := newTransactionsByPriceAndNonce(env.signer, localPlainTxs, env.header.BaseFee)
blobTxs := newTransactionsByPriceAndNonce(env.signer, localBlobTxs, env.header.BaseFee)
if err := miner.commitTransactions(env, plainTxs, blobTxs, interrupt); err != nil {
if !localPlainTxs.Empty() || !localBlobTxs.Empty() {
if err := miner.commitTransactions(env, localPlainTxs, localBlobTxs, interrupt); err != nil {
return err
}
}
if len(remotePlainTxs) > 0 || len(remoteBlobTxs) > 0 {
plainTxs := newTransactionsByPriceAndNonce(env.signer, remotePlainTxs, env.header.BaseFee)
blobTxs := newTransactionsByPriceAndNonce(env.signer, remoteBlobTxs, env.header.BaseFee)
if err := miner.commitTransactions(env, plainTxs, blobTxs, interrupt); err != nil {
if !remotePlainTxs.Empty() || !remoteBlobTxs.Empty() {
if err := miner.commitTransactions(env, remotePlainTxs, remoteBlobTxs, interrupt); err != nil {
return err
}
}