Skip to content

Commit

Permalink
core, eth, les: polish txpool API around local/remote txs
Browse files Browse the repository at this point in the history
  • Loading branch information
@karalabe
karalabe committed Jul 5, 2017
1 parent a633a2d commit 48ee7f9
Show file tree
Hide file tree
Showing 9 changed files with 153 additions and 149 deletions.
16 changes: 6 additions & 10 deletions core/tx_list.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -420,7 +420,7 @@ func (l *txPricedList) Removed() {
heap.Init(l.items)
}

// Discard finds all the transactions below the given price threshold, drops them
// Cap 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.
func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transactions {
drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop
Expand All @@ -429,9 +429,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
for len(*l.items) > 0 {
// Discard stale transactions if found during cleanup
tx := heap.Pop(l.items).(*types.Transaction)

hash := tx.Hash()
if _, ok := (*l.all)[hash]; !ok {
if _, ok := (*l.all)[tx.Hash()]; !ok {
l.stales--
continue
}
Expand All @@ -440,7 +438,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
break
}
// Non stale transaction found, discard unless local
if local.contains(tx) {
if local.containsTx(tx) {
save = append(save, tx)
} else {
drop = append(drop, tx)
Expand All @@ -456,7 +454,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
// lowest priced transaction currently being tracked.
func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) bool {
// Local transactions cannot be underpriced
if local.contains(tx) {
if local.containsTx(tx) {
return false
}
// Discard stale price points if found at the heap start
Expand Down Expand Up @@ -487,14 +485,12 @@ func (l *txPricedList) Discard(count int, local *accountSet) types.Transactions
for len(*l.items) > 0 && count > 0 {
// Discard stale transactions if found during cleanup
tx := heap.Pop(l.items).(*types.Transaction)

hash := tx.Hash()
if _, ok := (*l.all)[hash]; !ok {
if _, ok := (*l.all)[tx.Hash()]; !ok {
l.stales--
continue
}
// Non stale transaction found, discard unless local
if local.contains(tx) {
if local.containsTx(tx) {
save = append(save, tx)
} else {
drop = append(drop, tx)
Expand Down
157 changes: 86 additions & 71 deletions core/tx_pool.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -176,12 +176,12 @@ type TxPool struct {
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
config = (&config).sanitize()
signer := types.NewEIP155Signer(chainconfig.ChainId)

// Create the transaction pool with its initial settings
pool := &TxPool{
config: config,
chainconfig: chainconfig,
signer: signer,
signer: types.NewEIP155Signer(chainconfig.ChainId),
pending: make(map[common.Address]*txList),
queue: make(map[common.Address]*txList),
beats: make(map[common.Address]time.Time),
Expand All @@ -191,10 +191,10 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, eventMux *e
gasLimit: gasLimitFn,
gasPrice: new(big.Int).SetUint64(config.PriceLimit),
pendingState: nil,
locals: newAccountSet(signer),
events: eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}),
quit: make(chan struct{}),
}
pool.locals = newAccountSet(pool.signer)
pool.priced = newTxPricedList(&pool.all)
pool.resetState()

Expand Down Expand Up @@ -237,7 +237,7 @@ func (pool *TxPool) eventLoop() {
pool.mu.Unlock()

case RemovedTransactionEvent:
pool.AddBatch(ev.Txs)
pool.addTxs(ev.Txs, false)
}

// Handle stats reporting ticks
Expand Down Expand Up @@ -371,52 +371,40 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) {
return pending, nil
}

// SetLocal marks a transaction as local, skipping gas price
// check against local miner minimum in the future
func (pool *TxPool) SetLocal(tx *types.Transaction) {
pool.mu.Lock()
defer pool.mu.Unlock()
pool.locals.add(tx)
}

// validateTx checks whether a transaction is valid according
// to the consensus rules.
func (pool *TxPool) validateTx(tx *types.Transaction) error {

// validateTx checks whether a transaction is valid according to the consensus
// rules and adheres to some heuristic limits of the local node (price and size).
func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
// Heuristic limit, reject transactions over 32KB to prevent DOS attacks
if tx.Size() > 32*1024 {
return ErrOversizedData
}
// Transactions can't be negative. This may never happen using RLP decoded
// transactions but may occur if you create a transaction using the RPC.
if tx.Value().Sign() < 0 {
return ErrNegativeValue
}
// Ensure the transaction doesn't exceed the current block limit gas.
if pool.gasLimit().Cmp(tx.Gas()) < 0 {
return ErrGasLimit
}
// Make sure the transaction is signed properly
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 non-local transactions under our own minimal accepted gas price
local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network
if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
return ErrUnderpriced
}

// Ensure the transaction adheres to nonce ordering
currentState, err := pool.currentState()
if err != nil {
return err
}

// Last but not least check for nonce errors
if currentState.GetNonce(from) > tx.Nonce() {
return ErrNonceTooLow
}

// Check the transaction doesn't exceed the current
// block limit gas.
if pool.gasLimit().Cmp(tx.Gas()) < 0 {
return ErrGasLimit
}

// Transactions can't be negative. This may never happen
// using RLP decoded transactions but may occur if you create
// a transaction using the RPC for example.
if tx.Value().Sign() < 0 {
return ErrNegativeValue
}

// Transactor should have enough funds to cover the costs
// cost == V + GP * GL
if currentState.GetBalance(from).Cmp(tx.Cost()) < 0 {
Expand All @@ -426,27 +414,26 @@ func (pool *TxPool) validateTx(tx *types.Transaction) error {
if tx.Gas().Cmp(intrGas) < 0 {
return ErrIntrinsicGas
}

// Heuristic limit, reject transactions over 32KB to prevent DOS attacks
if tx.Size() > 32*1024 {
return ErrOversizedData
}
return nil
}

// add validates a transaction and inserts it into the non-executable queue for
// later pending promotion and execution. If the transaction is a replacement for
// an already pending or queued one, it overwrites the previous and returns this
// so outer code doesn't uselessly call promote.
func (pool *TxPool) add(tx *types.Transaction) (bool, error) {
//
// If a newly added transaction is marked as local, its sending account will be
// whitelisted, preventing any associated transaction from being dropped out of
// the pool due to pricing constraints.
func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
// If the transaction is already known, discard it
hash := tx.Hash()
if pool.all[hash] != nil {
log.Trace("Discarding already known transaction", "hash", hash)
return false, fmt.Errorf("known transaction: %x", hash)
}
// If the transaction fails basic validation, discard it
if err := pool.validateTx(tx); err != nil {
if err := pool.validateTx(tx, local); err != nil {
log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
invalidTxCounter.Inc(1)
return false, err
Expand Down Expand Up @@ -488,11 +475,14 @@ func (pool *TxPool) add(tx *types.Transaction) (bool, error) {
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
return old != nil, nil
}
// New transaction isn't replacing a pending one, push into queue
// New transaction isn't replacing a pending one, push into queue and potentially mark local
replace, err := pool.enqueueTx(hash, tx)
if err != nil {
return false, err
}
if local {
pool.locals.add(from)
}
log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
return replace, nil
}
Expand Down Expand Up @@ -560,13 +550,41 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
go pool.eventMux.Post(TxPreEvent{tx})
}

// Add queues a single transaction in the pool if it is valid.
func (pool *TxPool) Add(tx *types.Transaction) error {
// AddLocal enqueues a single transaction into the pool if it is valid, marking
// the sender as a local one in the mean time, ensuring it goes around the local
// pricing constraints.
func (pool *TxPool) AddLocal(tx *types.Transaction) error {
return pool.addTx(tx, true)
}

// AddRemote enqueues a single transaction into the pool if it is valid. If the
// sender is not among the locally tracked ones, full pricing constraints will
// apply.
func (pool *TxPool) AddRemote(tx *types.Transaction) error {
return pool.addTx(tx, false)
}

// AddLocals enqueues a batch of transactions into the pool if they are valid,
// marking the senders as a local ones in the mean time, ensuring they go around
// the local pricing constraints.
func (pool *TxPool) AddLocals(txs []*types.Transaction) error {
return pool.addTxs(txs, true)
}

// AddRemotes enqueues a batch of transactions into the pool if they are valid.
// If the senders are not among the locally tracked ones, full pricing constraints
// will apply.
func (pool *TxPool) AddRemotes(txs []*types.Transaction) error {
return pool.addTxs(txs, false)
}

// addTx enqueues a single transaction into the pool if it is valid.
func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
pool.mu.Lock()
defer pool.mu.Unlock()

// Try to inject the transaction and update any state
replace, err := pool.add(tx)
replace, err := pool.add(tx, local)
if err != nil {
return err
}
Expand All @@ -582,15 +600,15 @@ func (pool *TxPool) Add(tx *types.Transaction) error {
return nil
}

// AddBatch attempts to queue a batch of transactions.
func (pool *TxPool) AddBatch(txs []*types.Transaction) error {
// addTxs attempts to queue a batch of transactions if they are valid.
func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) error {
pool.mu.Lock()
defer pool.mu.Unlock()

// Add the batch of transaction, tracking the accepted ones
dirty := make(map[common.Address]struct{})
for _, tx := range txs {
if replace, err := pool.add(tx); err == nil {
if replace, err := pool.add(tx, local); err == nil {
if !replace {
from, _ := types.Sender(pool.signer, tx) // already validated
dirty[from] = struct{}{}
Expand Down Expand Up @@ -725,7 +743,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A
pool.promoteTx(addr, hash, tx)
}
// Drop all transactions over the allowed limit
if !pool.locals.containsAddress(addr) {
if !pool.locals.contains(addr) {
for _, tx := range list.Cap(int(pool.config.AccountQueue)) {
hash := tx.Hash()
delete(pool.all, hash)
Expand All @@ -752,7 +770,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A
spammers := prque.New()
for addr, list := range pool.pending {
// Only evict transactions from high rollers
if !pool.locals.containsAddress(addr) && uint64(list.Len()) > pool.config.AccountSlots {
if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots {
spammers.Push(addr, float32(list.Len()))
}
}
Expand Down Expand Up @@ -818,7 +836,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB, accounts []common.A
addresses := make(addresssByHeartbeat, 0, len(pool.queue))
for addr := range pool.queue {
// Don't drop locals
if !pool.locals.containsAddress(addr) {
if !pool.locals.contains(addr) {
addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]})
}
}
Expand Down Expand Up @@ -930,41 +948,38 @@ 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) Swap(i, j int) { a[i], a[j] = a[j], a[i] }

// accountSet is simply a map of addresses, and a signer, to be able
// to determine the address from a tx
// accountSet is simply a set of addresses to check for existance, and a signer
// capable of deriving addresses from transactions.
type accountSet struct {
accounts map[common.Address]struct{}
signer types.Signer
}

// newAccountSet creates a new address set with an associated signer for sender
// derivations.
func newAccountSet(signer types.Signer) *accountSet {
return &accountSet{
accounts: make(map[common.Address]struct{}),
signer: signer,
}
}

// containsAddress checks if a given address is within the set
func (as *accountSet) containsAddress(address common.Address) bool {
_, exist := as.accounts[address]
// contains checks if a given address is contained within the set.
func (as *accountSet) contains(addr common.Address) bool {
_, exist := as.accounts[addr]
return exist
}

// contains checks if the sender of a given tx is within the set
func (as *accountSet) contains(tx *types.Transaction) bool {
if address, err := types.Sender(as.signer, tx); err == nil {
return as.containsAddress(address)
// containsTx checks if the sender of a given tx is within the set. If the sender
// cannot be derived, this method returns false.
func (as *accountSet) containsTx(tx *types.Transaction) bool {
if addr, err := types.Sender(as.signer, tx); err == nil {
return as.contains(addr)
}
return false
}

// add a transaction sender to the set
// if sender can't be derived, this is a no-op (no errors returned)
func (as *accountSet) add(tx *types.Transaction) {
if address, err := types.Sender(as.signer, tx); err == nil {
if _, exist := as.accounts[address]; !exist {
as.accounts[address] = struct{}{}
}
}

// add inserts a new address into the set to track.
func (as *accountSet) add(addr common.Address) {
as.accounts[addr] = struct{}{}
}
Loading

0 comments on commit 48ee7f9

Please sign in to comment.