forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 07:06:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core/txpool: refactor Reserver into tracker and named handles

Browse source
This commit is contained in:
lightclient 2025-04-07 21:24:20 -06:00
parent 710bcc3d80
commit 38783b7e1b
No known key found for this signature in database
GPG key ID: 657913021EF45A6A
7 changed files with 89 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -47,8 +47,6 @@ import (
) )
const ( const (
poolName = "blobpool" // The name of legacy txpool
// blobSize is the protocol constrained byte size of a single blob in a // blobSize is the protocol constrained byte size of a single blob in a
// transaction. There can be multiple of these embedded into a single tx. // transaction. There can be multiple of these embedded into a single tx.
blobSize = params.BlobTxFieldElementsPerBlob * params.BlobTxBytesPerFieldElement blobSize = params.BlobTxFieldElementsPerBlob * params.BlobTxBytesPerFieldElement
@ -503,7 +501,7 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
return err return err
} }
if _, ok := p.index[sender]; !ok { if _, ok := p.index[sender]; !ok {
if err := p.reserver.Hold(sender, poolName); err != nil { if err := p.reserver.Hold(sender); err != nil {
return err return err
} }
p.index[sender] = []*blobTxMeta{} p.index[sender] = []*blobTxMeta{}
@ -558,7 +556,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
if inclusions != nil { // only during reorgs will the heap be initialized if inclusions != nil { // only during reorgs will the heap be initialized
heap.Remove(p.evict, p.evict.index[addr]) heap.Remove(p.evict, p.evict.index[addr])
} }
p.reserver.Release(addr, poolName) p.reserver.Release(addr)
if gapped { if gapped {
log.Warn("Dropping dangling blob transactions", "from", addr, "missing", next, "drop", nonces, "ids", ids) log.Warn("Dropping dangling blob transactions", "from", addr, "missing", next, "drop", nonces, "ids", ids)
@ -711,7 +709,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
if inclusions != nil { // only during reorgs will the heap be initialized if inclusions != nil { // only during reorgs will the heap be initialized
heap.Remove(p.evict, p.evict.index[addr]) heap.Remove(p.evict, p.evict.index[addr])
} }
p.reserver.Release(addr, poolName) p.reserver.Release(addr)
} else { } else {
p.index[addr] = txs p.index[addr] = txs
} }
@ -1010,7 +1008,7 @@ func (p *BlobPool) reinject(addr common.Address, txhash common.Hash) error {
// Update the indices and metrics // Update the indices and metrics
meta := newBlobTxMeta(id, tx.Size(), p.store.Size(id), tx) meta := newBlobTxMeta(id, tx.Size(), p.store.Size(id), tx)
if _, ok := p.index[addr]; !ok { if _, ok := p.index[addr]; !ok {
if err := p.reserver.Hold(addr, poolName); err != nil { if err := p.reserver.Hold(addr); err != nil {
log.Warn("Failed to reserve account for blob pool", "tx", tx.Hash(), "from", addr, "err", err) log.Warn("Failed to reserve account for blob pool", "tx", tx.Hash(), "from", addr, "err", err)
return err return err
} }
@ -1070,7 +1068,7 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {
delete(p.spent, addr) delete(p.spent, addr)
heap.Remove(p.evict, p.evict.index[addr]) heap.Remove(p.evict, p.evict.index[addr])
p.reserver.Release(addr, poolName) p.reserver.Release(addr)
} }
// Clear out the transactions from the data store // Clear out the transactions from the data store
log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids) log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids)
@ -1409,7 +1407,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
// only by this subpool until all transactions are evicted // only by this subpool until all transactions are evicted
from, _ := types.Sender(p.signer, tx) // already validated above from, _ := types.Sender(p.signer, tx) // already validated above
if _, ok := p.index[from]; !ok { if _, ok := p.index[from]; !ok {
if err := p.reserver.Hold(from, poolName); err != nil { if err := p.reserver.Hold(from); err != nil {
addNonExclusiveMeter.Mark(1) addNonExclusiveMeter.Mark(1)
return err return err
} }
@ -1421,7 +1419,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
// by a return statement before running deferred methods. Take care with // by a return statement before running deferred methods. Take care with
// removing or subscoping err as it will break this clause. // removing or subscoping err as it will break this clause.
if err != nil { if err != nil {
p.reserver.Release(from, poolName) p.reserver.Release(from)
} }
}() }()
} }
@ -1553,7 +1551,7 @@ func (p *BlobPool) drop() {
if last { if last {
delete(p.index, from) delete(p.index, from)
delete(p.spent, from) delete(p.spent, from)
p.reserver.Release(from, poolName) p.reserver.Release(from)
} else { } else {
txs[len(txs)-1] = nil txs[len(txs)-1] = nil
txs = txs[:len(txs)-1] txs = txs[:len(txs)-1]
@ -1829,7 +1827,7 @@ func (p *BlobPool) Clear() {
// can't happen until Clear releases the reservation lock. Clear cannot // can't happen until Clear releases the reservation lock. Clear cannot
// acquire the subpool lock until the transaction addition is completed. // acquire the subpool lock until the transaction addition is completed.
for acct := range p.index { for acct := range p.index {
p.reserver.Release(acct, poolName) p.reserver.Release(acct)
} }
p.lookup = newLookup() p.lookup = newLookup()
p.index = make(map[common.Address][]*blobTxMeta) p.index = make(map[common.Address][]*blobTxMeta)

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

@ -405,6 +405,10 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) {
} }
} }
func newReserver() *txpool.Reserver {
return txpool.NewReservationTracker().NewHandle(42)
}
// Tests that transactions can be loaded from disk on startup and that they are // Tests that transactions can be loaded from disk on startup and that they are
// correctly discarded if invalid. // correctly discarded if invalid.
// //
@ -672,7 +676,7 @@ func TestOpenDrops(t *testing.T) {
statedb: statedb, statedb: statedb,
} }
pool := New(Config{Datadir: storage}, chain, nil) pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
t.Fatalf("failed to create blob pool: %v", err) t.Fatalf("failed to create blob pool: %v", err)
} }
defer pool.Close() defer pool.Close()
@ -790,7 +794,7 @@ func TestOpenIndex(t *testing.T) {
statedb: statedb, statedb: statedb,
} }
pool := New(Config{Datadir: storage}, chain, nil) pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
t.Fatalf("failed to create blob pool: %v", err) t.Fatalf("failed to create blob pool: %v", err)
} }
defer pool.Close() defer pool.Close()
@ -891,7 +895,7 @@ func TestOpenHeap(t *testing.T) {
statedb: statedb, statedb: statedb,
} }
pool := New(Config{Datadir: storage}, chain, nil) pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
t.Fatalf("failed to create blob pool: %v", err) t.Fatalf("failed to create blob pool: %v", err)
} }
defer pool.Close() defer pool.Close()
@ -970,7 +974,7 @@ func TestOpenCap(t *testing.T) {
statedb: statedb, statedb: statedb,
} }
pool := New(Config{Datadir: storage, Datacap: datacap}, chain, nil) pool := New(Config{Datadir: storage, Datacap: datacap}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
t.Fatalf("failed to create blob pool: %v", err) t.Fatalf("failed to create blob pool: %v", err)
} }
// Verify that enough transactions have been dropped to get the pool's size // Verify that enough transactions have been dropped to get the pool's size
@ -1071,7 +1075,7 @@ func TestChangingSlotterSize(t *testing.T) {
statedb: statedb, statedb: statedb,
} }
pool := New(Config{Datadir: storage}, chain, nil) pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
t.Fatalf("failed to create blob pool: %v", err) t.Fatalf("failed to create blob pool: %v", err)
} }
@ -1514,7 +1518,7 @@ func TestAdd(t *testing.T) {
statedb: statedb, statedb: statedb,
} }
pool := New(Config{Datadir: storage}, chain, nil) pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
t.Fatalf("test %d: failed to create blob pool: %v", i, err) t.Fatalf("test %d: failed to create blob pool: %v", i, err)
} }
verifyPoolInternals(t, pool) verifyPoolInternals(t, pool)
@ -1613,7 +1617,7 @@ func benchmarkPoolPending(b *testing.B, datacap uint64) {
pool = New(Config{Datadir: ""}, chain, nil) pool = New(Config{Datadir: ""}, chain, nil)
) )
if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
b.Fatalf("failed to create blob pool: %v", err) b.Fatalf("failed to create blob pool: %v", err)
} }
// Make the pool not use disk (just drop everything). This test never reads // Make the pool not use disk (just drop everything). This test never reads

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

@ -45,8 +45,6 @@ import (
) )
const ( const (
poolName = "general" // The name of legacy txpool
// txSlotSize is used to calculate how many data slots a single transaction // txSlotSize is used to calculate how many data slots a single transaction
// takes up based on its size. The slots are used as DoS protection, ensuring // takes up based on its size. The slots are used as DoS protection, ensuring
// that validating a new transaction remains a constant operation (in reality // that validating a new transaction remains a constant operation (in reality
@ -693,7 +691,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
_, hasQueued = pool.queue[from] _, hasQueued = pool.queue[from]
) )
if !hasPending && !hasQueued { if !hasPending && !hasQueued {
if err := pool.reserver.Hold(from, poolName); err != nil { if err := pool.reserver.Hold(from); err != nil {
return false, err return false, err
} }
defer func() { defer func() {
@ -704,7 +702,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
// by a return statement before running deferred methods. Take care with // by a return statement before running deferred methods. Take care with
// removing or subscoping err as it will break this clause. // removing or subscoping err as it will break this clause.
if err != nil { if err != nil {
pool.reserver.Release(from, poolName) pool.reserver.Release(from)
} }
}() }()
} }
@ -1097,7 +1095,7 @@ func (pool *LegacyPool) removeTx(hash common.Hash, outofbound bool, unreserve bo
_, hasQueued = pool.queue[addr] _, hasQueued = pool.queue[addr]
) )
if !hasPending && !hasQueued { if !hasPending && !hasQueued {
pool.reserver.Release(addr, poolName) pool.reserver.Release(addr)
} }
}() }()
} }
@ -1477,7 +1475,7 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
delete(pool.queue, addr) delete(pool.queue, addr)
delete(pool.beats, addr) delete(pool.beats, addr)
if _, ok := pool.pending[addr]; !ok { if _, ok := pool.pending[addr]; !ok {
pool.reserver.Release(addr, poolName) pool.reserver.Release(addr)
} }
} }
} }
@ -1663,7 +1661,7 @@ func (pool *LegacyPool) demoteUnexecutables() {
if list.Empty() { if list.Empty() {
delete(pool.pending, addr) delete(pool.pending, addr)
if _, ok := pool.queue[addr]; !ok { if _, ok := pool.queue[addr]; !ok {
pool.reserver.Release(addr, poolName) pool.reserver.Release(addr)
} }
} }
} }
@ -1908,7 +1906,7 @@ func (pool *LegacyPool) Clear() {
// acquire the subpool lock until the transaction addition is completed. // acquire the subpool lock until the transaction addition is completed.
for _, tx := range pool.all.txs { for _, tx := range pool.all.txs {
senderAddr, _ := types.Sender(pool.signer, tx) senderAddr, _ := types.Sender(pool.signer, tx)
pool.reserver.Release(senderAddr, poolName) pool.reserver.Release(senderAddr)
} }
pool.all = newLookup() pool.all = newLookup()
pool.priced = newPricedList(pool.all) pool.priced = newPricedList(pool.all)

9
core/txpool/legacypool/legacypool2_test.go
View file

@ -24,7 +24,6 @@ import (
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/tracing" "github.com/ethereum/go-ethereum/core/tracing"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
@ -87,7 +86,7 @@ func TestTransactionFutureAttack(t *testing.T) {
config.GlobalQueue = 100 config.GlobalQueue = 100
config.GlobalSlots = 100 config.GlobalSlots = 100
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
fillPool(t, pool) fillPool(t, pool)
pending, _ := pool.Stats() pending, _ := pool.Stats()
@ -121,7 +120,7 @@ func TestTransactionFuture1559(t *testing.T) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a number of test accounts, fund them and make transactions // Create a number of test accounts, fund them and make transactions
@ -154,7 +153,7 @@ func TestTransactionZAttack(t *testing.T) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a number of test accounts, fund them and make transactions // Create a number of test accounts, fund them and make transactions
fillPool(t, pool) fillPool(t, pool)
@ -225,7 +224,7 @@ func BenchmarkFutureAttack(b *testing.B) {
config.GlobalQueue = 100 config.GlobalQueue = 100
config.GlobalSlots = 100 config.GlobalSlots = 100
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
fillPool(b, pool) fillPool(b, pool)

38
core/txpool/legacypool/legacypool_test.go
View file

@ -171,13 +171,17 @@ func setupPool() (*LegacyPool, *ecdsa.PrivateKey) {
return setupPoolWithConfig(params.TestChainConfig) return setupPoolWithConfig(params.TestChainConfig)
} }
func newReserver() *txpool.Reserver {
return txpool.NewReservationTracker().NewHandle(42)
}
func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.PrivateKey) { func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.PrivateKey) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
blockchain := newTestBlockChain(config, 10000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(config, 10000000, statedb, new(event.Feed))
key, _ := crypto.GenerateKey() key, _ := crypto.GenerateKey()
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()); err != nil { if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()); err != nil {
panic(err) panic(err)
} }
// wait for the pool to initialize // wait for the pool to initialize
@ -310,7 +314,7 @@ func TestStateChangeDuringReset(t *testing.T) {
tx1 := transaction(1, 100000, key) tx1 := transaction(1, 100000, key)
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
nonce := pool.Nonce(address) nonce := pool.Nonce(address)
@ -727,7 +731,7 @@ func TestPostponing(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create two test accounts to produce different gap profiles with // Create two test accounts to produce different gap profiles with
@ -937,7 +941,7 @@ func TestQueueGlobalLimiting(t *testing.T) {
config.GlobalQueue = config.AccountQueue*3 - 1 // reduce the queue limits to shorten test time (-1 to make it non divisible) config.GlobalQueue = config.AccountQueue*3 - 1 // reduce the queue limits to shorten test time (-1 to make it non divisible)
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a number of test accounts and fund them (last one will be the local) // Create a number of test accounts and fund them (last one will be the local)
@ -989,7 +993,7 @@ func TestQueueTimeLimiting(t *testing.T) {
config.Lifetime = time.Second config.Lifetime = time.Second
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a test account to ensure remotes expire // Create a test account to ensure remotes expire
@ -1150,7 +1154,7 @@ func TestPendingGlobalLimiting(t *testing.T) {
config.GlobalSlots = config.AccountSlots * 10 config.GlobalSlots = config.AccountSlots * 10
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a number of test accounts and fund them // Create a number of test accounts and fund them
@ -1249,7 +1253,7 @@ func TestCapClearsFromAll(t *testing.T) {
config.GlobalSlots = 8 config.GlobalSlots = 8
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a number of test accounts and fund them // Create a number of test accounts and fund them
@ -1282,7 +1286,7 @@ func TestPendingMinimumAllowance(t *testing.T) {
config.GlobalSlots = 1 config.GlobalSlots = 1
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a number of test accounts and fund them // Create a number of test accounts and fund them
@ -1326,7 +1330,7 @@ func TestRepricing(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Keep track of transaction events to ensure all executables get announced // Keep track of transaction events to ensure all executables get announced
@ -1431,7 +1435,7 @@ func TestMinGasPriceEnforced(t *testing.T) {
txPoolConfig := DefaultConfig txPoolConfig := DefaultConfig
txPoolConfig.NoLocals = true txPoolConfig.NoLocals = true
pool := New(txPoolConfig, blockchain) pool := New(txPoolConfig, blockchain)
pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
key, _ := crypto.GenerateKey() key, _ := crypto.GenerateKey()
@ -1580,7 +1584,7 @@ func TestUnderpricing(t *testing.T) {
config.GlobalQueue = 2 config.GlobalQueue = 2
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Keep track of transaction events to ensure all executables get announced // Keep track of transaction events to ensure all executables get announced
@ -1670,7 +1674,7 @@ func TestStableUnderpricing(t *testing.T) {
config.GlobalQueue = 0 config.GlobalQueue = 0
pool := New(config, blockchain) pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Keep track of transaction events to ensure all executables get announced // Keep track of transaction events to ensure all executables get announced
@ -1873,7 +1877,7 @@ func TestDeduplication(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create a test account to add transactions with // Create a test account to add transactions with
@ -1940,7 +1944,7 @@ func TestReplacement(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Keep track of transaction events to ensure all executables get announced // Keep track of transaction events to ensure all executables get announced
@ -2131,7 +2135,7 @@ func TestStatusCheck(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create the test accounts to check various transaction statuses with // Create the test accounts to check various transaction statuses with
@ -2204,7 +2208,7 @@ func TestSetCodeTransactions(t *testing.T) {
blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create the test accounts // Create the test accounts
@ -2428,7 +2432,7 @@ func TestSetCodeTransactionsReorg(t *testing.T) {
blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed)) blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain) pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
defer pool.Close() defer pool.Close()
// Create the test accounts // Create the test accounts

61
core/txpool/reserver.go
View file

@ -35,40 +35,51 @@ var (
reservationsGaugeName = "txpool/reservations" reservationsGaugeName = "txpool/reservations"
) )
// Reserver is a struct shared between different subpools. It is used to reserve // ReservationTracker is a struct shared between different subpools. It is used to reserve
// the account and ensure that one address cannot initiate transactions, authorizations, // the account and ensure that one address cannot initiate transactions, authorizations,
// and other state-changing behaviors in different pools at the same time. // and other state-changing behaviors in different pools at the same time.
type Reserver struct { type ReservationTracker struct {
accounts map[common.Address]string accounts map[common.Address]int
lock sync.RWMutex lock sync.RWMutex
} }
// NewReserver initializes the account reserver. // NewReservationTracker initializes the account reservation tracker.
func NewReserver() *Reserver { func NewReservationTracker() *ReservationTracker {
return &Reserver{ return &ReservationTracker{
accounts: make(map[common.Address]string), accounts: make(map[common.Address]int),
} }
} }
// NewHandle creates a named handle on the ReservationTracker.
func (r *ReservationTracker) NewHandle(id int) *Reserver {
return &Reserver{r, id}
}
// Reserver is a named handle on ReservationTracker.
type Reserver struct {
tracker *ReservationTracker
id int
}
// Hold attempts to reserve the specified account address for the given pool. // Hold attempts to reserve the specified account address for the given pool.
// Returns an error if the account is already reserved. // Returns an error if the account is already reserved.
func (r *Reserver) Hold(addr common.Address, id string) error { func (h *Reserver) Hold(addr common.Address) error {
r.lock.Lock() h.tracker.lock.Lock()
defer r.lock.Unlock() defer h.tracker.lock.Unlock()
// Double reservations are forbidden even from the same pool to // Double reservations are forbidden even from the same pool to
// avoid subtle bugs in the long term. // avoid subtle bugs in the long term.
owner, exists := r.accounts[addr] owner, exists := h.tracker.accounts[addr]
if exists { if exists {
if owner == id { if owner == h.id {
log.Error("pool attempted to reserve already-owned address", "address", addr) log.Error("pool attempted to reserve already-owned address", "address", addr)
return nil // Ignore fault to give the pool a chance to recover while the bug gets fixed return nil // Ignore fault to give the pool a chance to recover while the bug gets fixed
} }
return ErrAlreadyReserved return ErrAlreadyReserved
} }
r.accounts[addr] = id h.tracker.accounts[addr] = h.id
if metrics.Enabled() { if metrics.Enabled() {
m := fmt.Sprintf("%s/%s", reservationsGaugeName, id) m := fmt.Sprintf("%s/%d", reservationsGaugeName, h.id)
metrics.GetOrRegisterGauge(m, nil).Inc(1) metrics.GetOrRegisterGauge(m, nil).Inc(1)
} }
return nil return nil
@ -76,34 +87,34 @@ func (r *Reserver) Hold(addr common.Address, id string) error {
// Release attempts to release the reservation for the specified account. // Release attempts to release the reservation for the specified account.
// Returns an error if the address is not reserved or is reserved by another pool. // Returns an error if the address is not reserved or is reserved by another pool.
func (r *Reserver) Release(addr common.Address, id string) error { func (h *Reserver) Release(addr common.Address) error {
r.lock.Lock() h.tracker.lock.Lock()
defer r.lock.Unlock() defer h.tracker.lock.Unlock()
// Ensure subpools only attempt to unreserve their own owned addresses, // Ensure subpools only attempt to unreserve their own owned addresses,
// otherwise flag as a programming error. // otherwise flag as a programming error.
owner, exists := r.accounts[addr] owner, exists := h.tracker.accounts[addr]
if !exists { if !exists {
log.Error("pool attempted to unreserve non-reserved address", "address", addr) log.Error("pool attempted to unreserve non-reserved address", "address", addr)
return errors.New("address not reserved") return errors.New("address not reserved")
} }
if id != owner { if owner != h.id {
log.Error("pool attempted to unreserve non-owned address", "address", addr) log.Error("pool attempted to unreserve non-owned address", "address", addr)
return errors.New("address not owned") return errors.New("address not owned")
} }
delete(r.accounts, addr) delete(h.tracker.accounts, addr)
if metrics.Enabled() { if metrics.Enabled() {
m := fmt.Sprintf("%s/%s", reservationsGaugeName, id) m := fmt.Sprintf("%s/%d", reservationsGaugeName, h.id)
metrics.GetOrRegisterGauge(m, nil).Dec(1) metrics.GetOrRegisterGauge(m, nil).Dec(1)
} }
return nil return nil
} }
// Has returns a flag indicating if the address has been reserved or not. // Has returns a flag indicating if the address has been reserved or not.
func (r *Reserver) Has(address common.Address) bool { func (h *Reserver) Has(address common.Address) bool {
r.lock.RLock() h.tracker.lock.RLock()
defer r.lock.RUnlock() defer h.tracker.lock.RUnlock()
_, exists := r.accounts[address] _, exists := h.tracker.accounts[address]
return exists return exists
} }

4
core/txpool/txpool.go
View file

@ -105,9 +105,9 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
term: make(chan struct{}), term: make(chan struct{}),
sync: make(chan chan error), sync: make(chan chan error),
} }
reserver := NewReserver() reserver := NewReservationTracker()
for i, subpool := range subpools { for i, subpool := range subpools {
if err := subpool.Init(gasTip, head, reserver); err != nil { if err := subpool.Init(gasTip, head, reserver.NewHandle(i)); err != nil {
for j := i - 1; j >= 0; j-- { for j := i - 1; j >= 0; j-- {
subpools[j].Close() subpools[j].Close()
} }