Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

core: fix write concurrency in txpool #19835

Merged
merged 3 commits into from
Jul 17, 2019
Merged

core: fix write concurrency in txpool #19835

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3f60e5b
core: fix write coucurrency in txpool
rjl493456442 Jul 14, 2019
b49f3bf
core: add rlock for pendingState read access
rjl493456442 Jul 16, 2019
0baea4d
core: address comments
rjl493456442 Jul 17, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions core/tx_noncer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
package core

import (
"sync"

"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
)
Expand All @@ -27,6 +29,7 @@ import (
type txNoncer struct {
fallback *state.StateDB
nonces map[common.Address]uint64
lock sync.Mutex
}

// newTxNoncer creates a new virtual state database to track the pool nonces.
Expand All @@ -40,6 +43,11 @@ func newTxNoncer(statedb *state.StateDB) *txNoncer {
// get returns the current nonce of an account, falling back to a real state
// database if the account is unknown.
func (txn *txNoncer) get(addr common.Address) uint64 {
// We use mutex for get operation is the underlying
// state will mutate db even for read access.
txn.lock.Lock()
defer txn.lock.Unlock()

if _, ok := txn.nonces[addr]; !ok {
txn.nonces[addr] = txn.fallback.GetNonce(addr)
}
Expand All @@ -49,5 +57,23 @@ func (txn *txNoncer) get(addr common.Address) uint64 {
// set inserts a new virtual nonce into the virtual state database to be returned
// whenever the pool requests it instead of reaching into the real state database.
func (txn *txNoncer) set(addr common.Address, nonce uint64) {
txn.lock.Lock()
defer txn.lock.Unlock()

txn.nonces[addr] = nonce
}

// compareAndSet inserts or updates a new virtual nonce into the virtual state
// database if the compare callback is true.
func (txn *txNoncer) compareAndSet(addr common.Address, nonce uint64, compare func(uint64, uint64) bool) {
Copy link
Member

@karalabe karalabe Jul 17, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Passing a compare method in here seems really convoluted, especially since all instances use the same logic anyway. Let's just rename compareAndSet to setIfLower, get rid of compare altogether and do the comparison internally.

txn.lock.Lock()
defer txn.lock.Unlock()

if _, ok := txn.nonces[addr]; !ok {
txn.nonces[addr] = txn.fallback.GetNonce(addr)
}
if compare == nil || !compare(txn.nonces[addr], nonce) {
return
}
txn.nonces[addr] = nonce
}
12 changes: 3 additions & 9 deletions core/tx_pool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -854,9 +854,7 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
pool.enqueueTx(tx.Hash(), tx)
}
// Update the account nonce if needed
if nonce := tx.Nonce(); pool.pendingNonces.get(addr) > nonce {
pool.pendingNonces.set(addr, nonce)
}
pool.pendingNonces.compareAndSet(addr, tx.Nonce(), func(old uint64, new uint64) bool { return old > new })
// Reduce the pending counter
pendingCounter.Dec(int64(1 + len(invalids)))
return
Expand Down Expand Up @@ -1232,9 +1230,7 @@ func (pool *TxPool) truncatePending() {
pool.all.Remove(hash)

// Update the account nonce to the dropped transaction
if nonce := tx.Nonce(); pool.pendingNonces.get(offenders[i]) > nonce {
pool.pendingNonces.set(offenders[i], nonce)
}
pool.pendingNonces.compareAndSet(offenders[i], tx.Nonce(), func(old uint64, new uint64) bool { return old > new })
log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
}
pool.priced.Removed(len(caps))
Expand All @@ -1261,9 +1257,7 @@ func (pool *TxPool) truncatePending() {
pool.all.Remove(hash)

// Update the account nonce to the dropped transaction
if nonce := tx.Nonce(); pool.pendingNonces.get(addr) > nonce {
pool.pendingNonces.set(addr, nonce)
}
pool.pendingNonces.compareAndSet(addr, tx.Nonce(), func(old uint64, new uint64) bool { return old > new })
log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
}
pool.priced.Removed(len(caps))
Expand Down