core: Change local-handling to use sender-account instead of tx hashes

This commit is contained in:
Martin Holst Swende 2017-06-30 22:43:26 +02:00
parent a0aa071ca6
commit 67aff49822
2 changed files with 49 additions and 60 deletions

@ -422,7 +422,7 @@ func (l *txPricedList) Removed() {
// Discard finds all the transactions below the given price threshold, drops them // Discard finds all the transactions below the given price threshold, drops them
// from the priced list and returs them for further removal from the entire pool. // from the priced list and returs them for further removal from the entire pool.
func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions { func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transactions {
drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop
save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep
@ -440,7 +440,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions
break break
} }
// Non stale transaction found, discard unless local // Non stale transaction found, discard unless local
if local.contains(hash) { if local.contains(tx) {
save = append(save, tx) save = append(save, tx)
} else { } else {
drop = append(drop, tx) drop = append(drop, tx)
@ -454,9 +454,9 @@ func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions
// Underpriced checks whether a transaction is cheaper than (or as cheap as) the // Underpriced checks whether a transaction is cheaper than (or as cheap as) the
// lowest priced transaction currently being tracked. // lowest priced transaction currently being tracked.
func (l *txPricedList) Underpriced(tx *types.Transaction, local *txSet) bool { func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) bool {
// Local transactions cannot be underpriced // Local transactions cannot be underpriced
if local.contains(tx.Hash()) { if local.contains(tx) {
return false return false
} }
// Discard stale price points if found at the heap start // Discard stale price points if found at the heap start
@ -480,7 +480,7 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *txSet) bool {
// Discard finds a number of most underpriced transactions, removes them from the // Discard finds a number of most underpriced transactions, removes them from the
// priced list and returs them for further removal from the entire pool. // priced list and returs them for further removal from the entire pool.
func (l *txPricedList) Discard(count int, local *txSet) types.Transactions { func (l *txPricedList) Discard(count int, local *accountSet) types.Transactions {
drop := make(types.Transactions, 0, count) // Remote underpriced transactions to drop drop := make(types.Transactions, 0, count) // Remote underpriced transactions to drop
save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep
@ -494,7 +494,7 @@ func (l *txPricedList) Discard(count int, local *txSet) types.Transactions {
continue continue
} }
// Non stale transaction found, discard unless local // Non stale transaction found, discard unless local
if local.contains(hash) { if local.contains(tx) {
save = append(save, tx) save = append(save, tx)
} else { } else {
drop = append(drop, tx) drop = append(drop, tx)

@ -155,7 +155,7 @@ type TxPool struct {
gasPrice *big.Int gasPrice *big.Int
eventMux *event.TypeMux eventMux *event.TypeMux
events *event.TypeMuxSubscription events *event.TypeMuxSubscription
locals *txSet locals *accountSet
signer types.Signer signer types.Signer
mu sync.RWMutex mu sync.RWMutex
@ -176,12 +176,12 @@ type TxPool struct {
func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, eventMux *event.TypeMux, currentStateFn stateFn, gasLimitFn func() *big.Int) *TxPool { func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, eventMux *event.TypeMux, currentStateFn stateFn, gasLimitFn func() *big.Int) *TxPool {
// Sanitize the input to ensure no vulnerable gas prices are set // Sanitize the input to ensure no vulnerable gas prices are set
config = (&config).sanitize() config = (&config).sanitize()
signer := types.NewEIP155Signer(chainconfig.ChainId)
// Create the transaction pool with its initial settings // Create the transaction pool with its initial settings
pool := &TxPool{ pool := &TxPool{
config: config, config: config,
chainconfig: chainconfig, chainconfig: chainconfig,
signer: types.NewEIP155Signer(chainconfig.ChainId), signer: signer,
pending: make(map[common.Address]*txList), pending: make(map[common.Address]*txList),
queue: make(map[common.Address]*txList), queue: make(map[common.Address]*txList),
beats: make(map[common.Address]time.Time), beats: make(map[common.Address]time.Time),
@ -191,7 +191,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, eventMux *e
gasLimit: gasLimitFn, gasLimit: gasLimitFn,
gasPrice: new(big.Int).SetUint64(config.PriceLimit), gasPrice: new(big.Int).SetUint64(config.PriceLimit),
pendingState: nil, pendingState: nil,
locals: newTxSet(), locals: newAccountSet(signer),
events: eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}), events: eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}),
quit: make(chan struct{}), quit: make(chan struct{}),
} }
@ -376,13 +376,19 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) {
func (pool *TxPool) SetLocal(tx *types.Transaction) { func (pool *TxPool) SetLocal(tx *types.Transaction) {
pool.mu.Lock() pool.mu.Lock()
defer pool.mu.Unlock() defer pool.mu.Unlock()
pool.locals.add(tx.Hash()) pool.locals.add(tx)
} }
// validateTx checks whether a transaction is valid according // validateTx checks whether a transaction is valid according
// to the consensus rules. // to the consensus rules.
func (pool *TxPool) validateTx(tx *types.Transaction) error { func (pool *TxPool) validateTx(tx *types.Transaction) error {
local := pool.locals.contains(tx.Hash())
from, err := types.Sender(pool.signer, tx)
if err != nil {
return ErrInvalidSender
}
local := pool.locals.containsAddress(from)
// Drop transactions under our own minimal accepted gas price // Drop transactions under our own minimal accepted gas price
if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
return ErrUnderpriced return ErrUnderpriced
@ -393,10 +399,6 @@ func (pool *TxPool) validateTx(tx *types.Transaction) error {
return err return err
} }
from, err := types.Sender(pool.signer, tx)
if err != nil {
return ErrInvalidSender
}
// Last but not least check for nonce errors // Last but not least check for nonce errors
if currentState.GetNonce(from) > tx.Nonce() { if currentState.GetNonce(from) > tx.Nonce() {
return ErrNonceTooLow return ErrNonceTooLow
@ -748,15 +750,9 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A
spammers := prque.New() spammers := prque.New()
for addr, list := range pool.pending { for addr, list := range pool.pending {
// Only evict transactions from high rollers // Only evict transactions from high rollers
if uint64(list.Len()) > pool.config.AccountSlots { if !pool.locals.containsAddress(addr) && uint64(list.Len()) > pool.config.AccountSlots {
// Skip local accounts as pools should maintain backlogs for themselves
for _, tx := range list.txs.items {
if !pool.locals.contains(tx.Hash()) {
spammers.Push(addr, float32(list.Len())) spammers.Push(addr, float32(list.Len()))
} }
break // Checking on transaction for locality is enough
}
}
} }
// Gradually drop transactions from offenders // Gradually drop transactions from offenders
offenders := []common.Address{} offenders := []common.Address{}
@ -929,48 +925,41 @@ func (a addresssByHeartbeat) Len() int { return len(a) }
func (a addresssByHeartbeat) Less(i, j int) bool { return a[i].heartbeat.Before(a[j].heartbeat) } func (a addresssByHeartbeat) Less(i, j int) bool { return a[i].heartbeat.Before(a[j].heartbeat) }
func (a addresssByHeartbeat) Swap(i, j int) { a[i], a[j] = a[j], a[i] } func (a addresssByHeartbeat) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
// txSet represents a set of transaction hashes in which entries // accountSet is simply a map of addresses, and a signer, to be able
// are automatically dropped after txSetDuration time // to determine the address from a tx
type txSet struct { type accountSet struct {
txMap map[common.Hash]struct{} accounts map[common.Address]struct{}
txOrd map[uint64]txOrdType signer types.Signer
addPtr, delPtr uint64
} }
const txSetDuration = time.Hour * 2 func newAccountSet(signer types.Signer) *accountSet {
return &accountSet{
// txOrdType represents an entry in the time-ordered list of transaction hashes accounts: make(map[common.Address]struct{}),
type txOrdType struct { signer: signer,
hash common.Hash
time time.Time
}
// newTxSet creates a new transaction set
func newTxSet() *txSet {
return &txSet{
txMap: make(map[common.Hash]struct{}),
txOrd: make(map[uint64]txOrdType),
} }
} }
// contains returns true if the set contains the given transaction hash // containsAddress checks if a given address is within the set
// (not thread safe, should be called from a locked environment) func (as *accountSet) containsAddress(address common.Address) bool {
func (ts *txSet) contains(hash common.Hash) bool { _, exist := as.accounts[address]
_, ok := ts.txMap[hash] return exist
return ok
} }
// add adds a transaction hash to the set, then removes entries older than txSetDuration // contains checks if the sender of a given tx is within the set
// (not thread safe, should be called from a locked environment) func (as *accountSet) contains(tx *types.Transaction) bool {
func (ts *txSet) add(hash common.Hash) { if address, err := types.Sender(as.signer, tx); err == nil {
ts.txMap[hash] = struct{}{} return as.containsAddress(address)
now := time.Now() }
ts.txOrd[ts.addPtr] = txOrdType{hash: hash, time: now} return false
ts.addPtr++ }
delBefore := now.Add(-txSetDuration)
for ts.delPtr < ts.addPtr && ts.txOrd[ts.delPtr].time.Before(delBefore) { // add a transaction sender to the set
delete(ts.txMap, ts.txOrd[ts.delPtr].hash) // if sender can't be derived, this is a no-op (no errors returned)
delete(ts.txOrd, ts.delPtr) func (as *accountSet) add(tx *types.Transaction) {
ts.delPtr++ if address, err := types.Sender(as.signer, tx); err == nil {
if _, exist := as.accounts[address]; !exist {
as.accounts[address] = struct{}{}
} }
} }
}