cmd, core, eth, miner: remove txpool gas price limits

cmd/utils/flags.go

@@ -237,7 +237,7 @@ var (
 	GasPriceFlag = BigFlag{
 		Name:  "gasprice",
 		Usage: "Minimal gas price to accept for mining a transactions",
-		Value: big.NewInt(20 * params.Shannon),
+		Value: eth.DefaultConfig.GasPrice,
 	}
 	ExtraDataFlag = cli.StringFlag{
 		Name:  "extradata",

core/events.go

@@ -17,8 +17,6 @@
 package core
 
 import (
-	"math/big"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 )
@@ -67,8 +65,6 @@ type ChainUncleEvent struct {
 
 type ChainHeadEvent struct{ Block *types.Block }
 
-type GasPriceChanged struct{ Price *big.Int }
-
 // Mining operation events
 type StartMining struct{}
 type TopMining struct{}

core/tx_list.go

@@ -21,8 +21,11 @@ import (
 	"math"
 	"math/big"
 	"sort"
+	"unsafe"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 // nonceHeap is a heap.Interface implementation over 64bit unsigned integers for
@@ -53,11 +56,11 @@ type txSortedMap struct {
 	cache types.Transactions            // Cache of the transactions already sorted
 }
 
-// newTxSortedMap creates a new sorted transaction map.
+// newTxSortedMap creates a new nonce-sorted transaction map.
 func newTxSortedMap() *txSortedMap {
 	return &txSortedMap{
 		items: make(map[uint64]*types.Transaction),
-		index: &nonceHeap{},
+		index: new(nonceHeap),
 	}
 }
 
@@ -340,3 +343,211 @@ func (l *txList) Empty() bool {
 func (l *txList) Flatten() types.Transactions {
 	return l.txs.Flatten()
 }
+
+// priceHeap is a heap.Interface implementation over big integers for retrieving
+// sorted transaction prices to discard when the pool fills up.
+type priceHeap []*big.Int
+
+func (h priceHeap) Len() int           { return len(h) }
+func (h priceHeap) Less(i, j int) bool { return h[i].Cmp(h[j]) < 0 }
+func (h priceHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
+
+func (h *priceHeap) Push(x interface{}) {
+	*h = append(*h, x.(*big.Int))
+}
+
+func (h *priceHeap) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}
+
+// priceKey is the hash map key type representing an unsigned big integer.
+type priceKey [256 / int(unsafe.Sizeof(uintptr(0)))]big.Word
+
+// newPriceKey converts a bit integer to a hash map key.
+func newPriceKey(x *big.Int) priceKey {
+	var key priceKey
+	copy(key[:], x.Bits())
+	return key
+}
+
+// txPricedMap is a price->transactions hash map with a heap based index to allow
+// iterating over the contents in a price-incrementing way.
+type txPricedMap struct {
+	items  map[priceKey]map[common.Hash]*types.Transaction // Hash map storing the transaction data
+	index  *priceHeap                                      // Heap of prices of all the stored transactions
+	stales int                                             // Number of stale price points to (re-heap trigger)
+	logger log.Logger                                      // Logger reporting basic price pool stats
+}
+
+// newTxPricedMap creates a new price-sorted transaction map.
+func newTxPricedMap() *txPricedMap {
+	m := &txPricedMap{
+		items: make(map[priceKey]map[common.Hash]*types.Transaction),
+		index: new(priceHeap),
+	}
+	m.logger = log.New(
+		"prices", log.Lazy{Fn: func() int { return len(*m.index) - m.stales }},
+		"stales", log.Lazy{Fn: func() int { return m.stales }},
+	)
+	return m
+}
+
+// Put inserts a new transaction into the map, also updating the map's price index.
+func (m *txPricedMap) Put(tx *types.Transaction) {
+	price, hash := tx.GasPrice(), tx.Hash()
+
+	// Generate the key and ensure we have a valid map for it
+	key := newPriceKey(price)
+	if m.items[key] == nil {
+		m.items[key] = make(map[common.Hash]*types.Transaction)
+		heap.Push(m.index, price)
+	} else if len(m.items[key]) == 0 {
+		m.stales--
+	}
+	// Add the transaction to the map
+	m.items[key][hash] = tx
+
+	m.logger.Trace("Accepted new transaction", "hash", hash, "price", price)
+}
+
+// Remove looks up a transaction and deletes it from the sorted map. Even if no
+// more transactions are left with the same price point, the price point itself
+// is retained to achieve hysteresis, until a staleness threshold is reached.
+func (m *txPricedMap) Remove(tx *types.Transaction) {
+	price, hash := tx.GasPrice(), tx.Hash()
+
+	key := newPriceKey(price)
+	if txs := m.items[key]; txs != nil {
+		if txs[hash] != nil {
+			// Transaction found, delete and update stale counter
+			delete(m.items[key], hash)
+			if len(m.items[key]) == 0 {
+				m.stales++
+			}
+			// If the number of stale entries reached a critical threshold (25%), re-heap and clean up
+			if m.stales > len(*m.index)/4 {
+				m.reheap()
+			}
+		}
+	}
+	m.logger.Trace("Removed old transaction", "hash", hash, "price", price)
+}
+
+// reheap discards the currently cached price point heap and regenerates it based
+// on the contents of the priced transaction map.
+func (m *txPricedMap) reheap() {
+	m.stales, m.index = 0, new(priceHeap)
+	for key, txs := range m.items {
+		if len(txs) == 0 {
+			delete(m.items, key)
+		} else {
+			*m.index = append(*m.index, new(big.Int).SetBits(key[:]))
+		}
+	}
+	heap.Init(m.index)
+}
+
+// Discard finds all the transactions below the given price threshold, drops them
+// from the priced map and returs them for further removal from the entire pool.
+func (m *txPricedMap) Cap(threshold *big.Int, local *txSet) types.Transactions {
+	drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop
+	save := make(types.Transactions, 0, 64)  // Local underpriced transactions to keep
+
+	for len(*m.index) > 0 {
+		// Discard stale price points if found during cleanup
+		price := []*big.Int(*m.index)[0]
+		key := newPriceKey(price)
+
+		if len(m.items[key]) == 0 {
+			m.stales--
+			heap.Pop(m.index)
+			delete(m.items, key)
+			continue
+		}
+		// Stop the discards if we've reached the threshold
+		if price.Cmp(threshold) >= 0 {
+			break
+		}
+		// Non stale price point found, discard its transactions
+		for hash, tx := range m.items[key] {
+			if local.contains(hash) {
+				save = append(save, tx)
+			} else {
+				drop = append(drop, tx)
+			}
+		}
+		m.items[key] = nil // will get removed in next iteration
+	}
+	for _, tx := range save {
+		m.Put(tx)
+	}
+	return drop
+}
+
+// Underpriced checks whether a transaction is cheaper than (or as cheap as) the
+// lowest priced transaction currently being tracked.
+func (m *txPricedMap) Underpriced(tx *types.Transaction, local *txSet) bool {
+	// Local transactions cannot be underpriced
+	if local.contains(tx.Hash()) {
+		return false
+	}
+	// Discard stale price points if found at the heap start
+	for len(*m.index) > 0 {
+		price := []*big.Int(*m.index)[0]
+		if key := newPriceKey(price); len(m.items[key]) == 0 {
+			m.stales--
+			heap.Pop(m.index)
+			delete(m.items, key)
+			continue
+		}
+		break
+	}
+	// Check if the transaction is underpriced or not
+	if len(*m.index) == 0 {
+		log.Error("Pricing query for empty pool") // This cannot happen, print to catch programming errors
+		return false
+	}
+	cheapest := []*big.Int(*m.index)[0]
+	return cheapest.Cmp(tx.GasPrice()) >= 0
+}
+
+// Discard finds a number of most underpriced transactions, removes them from the
+// priced map and returs them for further removal from the entire pool.
+func (m *txPricedMap) Discard(count int, local *txSet) types.Transactions {
+	drop := make(types.Transactions, 0, count) // Remote underpriced transactions to drop
+	save := make(types.Transactions, 0, 64)    // Local underpriced transactions to keep
+
+	for len(*m.index) > 0 && count > 0 {
+		// Discard stale price points if found during cleanup
+		price := []*big.Int(*m.index)[0]
+		key := newPriceKey(price)
+
+		if len(m.items[key]) == 0 {
+			m.stales--
+			heap.Pop(m.index)
+			delete(m.items, key)
+			continue
+		}
+		// Non stale price point found, discard its transactions
+		for hash, tx := range m.items[key] {
+			if count == 0 {
+				break
+			}
+			if local.contains(hash) {
+				save = append(save, tx)
+			} else {
+				drop = append(drop, tx)
+				count--
+			}
+			delete(m.items[key], hash)
+		}
+	}
+	for _, tx := range save {
+		m.Put(tx)
+	}
+	return drop
+}

core/tx_pool.go

@@ -36,23 +36,24 @@ import (
 
 var (
 	// Transaction Pool Errors
-	ErrInvalidSender     = errors.New("Invalid sender")
-	ErrNonce             = errors.New("Nonce too low")
-	ErrCheap             = errors.New("Gas price too low for acceptance")
-	ErrBalance           = errors.New("Insufficient balance")
-	ErrInsufficientFunds = errors.New("Insufficient funds for gas * price + value")
-	ErrIntrinsicGas      = errors.New("Intrinsic gas too low")
-	ErrGasLimit          = errors.New("Exceeds block gas limit")
-	ErrNegativeValue     = errors.New("Negative value")
+	ErrInvalidSender     = errors.New("invalid sender")
+	ErrNonce             = errors.New("nonce too low")
+	ErrUnderpriced       = errors.New("underpriced transaction")
+	ErrBalance           = errors.New("tnsufficient balance")
+	ErrInsufficientFunds = errors.New("tnsufficient funds for gas * price + value")
+	ErrIntrinsicGas      = errors.New("tntrinsic gas too low")
+	ErrGasLimit          = errors.New("exceeds block gas limit")
+	ErrNegativeValue     = errors.New("negative value")
 )
 
 var (
-	minPendingPerAccount = uint64(16)    // Min number of guaranteed transaction slots per address
-	maxPendingTotal      = uint64(4096)  // Max limit of pending transactions from all accounts (soft)
-	maxQueuedPerAccount  = uint64(64)    // Max limit of queued transactions per address
-	maxQueuedInTotal     = uint64(1024)  // Max limit of queued transactions from all accounts
-	maxQueuedLifetime    = 3 * time.Hour // Max amount of time transactions from idle accounts are queued
-	evictionInterval     = time.Minute   // Time interval to check for evictable transactions
+	minPendingPerAccount = uint64(16)      // Min number of guaranteed transaction slots per address
+	maxPendingTotal      = uint64(4096)    // Max limit of pending transactions from all accounts (soft)
+	maxQueuedPerAccount  = uint64(64)      // Max limit of queued transactions per address
+	maxQueuedTotal       = uint64(1024)    // Max limit of queued transactions from all accounts
+	maxQueuedLifetime    = 3 * time.Hour   // Max amount of time transactions from idle accounts are queued
+	evictionInterval     = time.Minute     // Time interval to check for evictable transactions
+	statsReportInterval  = 8 * time.Second // Time interval to report transaction pool stats
 )
 
 var (
@@ -69,7 +70,8 @@ var (
 	queuedNofundsCounter = metrics.NewCounter("txpool/queued/nofunds")   // Dropped due to out-of-funds
 
 	// General tx metrics
-	invalidTxCounter = metrics.NewCounter("txpool/invalid")
+	invalidTxCounter     = metrics.NewCounter("txpool/invalid")
+	underpricedTxCounter = metrics.NewCounter("txpool/underpriced")
 )
 
 type stateFn func() (*state.StateDB, error)
@@ -86,17 +88,18 @@ type TxPool struct {
 	currentState stateFn // The state function which will allow us to do some pre checks
 	pendingState *state.ManagedState
 	gasLimit     func() *big.Int // The current gas limit function callback
-	minGasPrice  *big.Int
+	gasPrice     *big.Int
 	eventMux     *event.TypeMux
 	events       *event.TypeMuxSubscription
-	localTx      *txSet
+	locals       *txSet
 	signer       types.Signer
 	mu           sync.RWMutex
 
 	pending map[common.Address]*txList         // All currently processable transactions
 	queue   map[common.Address]*txList         // Queued but non-processable transactions
-	all     map[common.Hash]*types.Transaction // All transactions to allow lookups
 	beats   map[common.Address]time.Time       // Last heartbeat from each known account
+	all     map[common.Hash]*types.Transaction // All transactions to allow lookups
+	priced  *txPricedMap                       // All transactions sorted by price
 
 	wg   sync.WaitGroup // for shutdown sync
 	quit chan struct{}
@@ -110,15 +113,16 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState
 		signer:       types.NewEIP155Signer(config.ChainId),
 		pending:      make(map[common.Address]*txList),
 		queue:        make(map[common.Address]*txList),
-		all:          make(map[common.Hash]*types.Transaction),
 		beats:        make(map[common.Address]time.Time),
+		all:          make(map[common.Hash]*types.Transaction),
+		priced:       newTxPricedMap(),
 		eventMux:     eventMux,
 		currentState: currentStateFn,
 		gasLimit:     gasLimitFn,
-		minGasPrice:  new(big.Int),
+		gasPrice:     big.NewInt(1),
 		pendingState: nil,
-		localTx:      newTxSet(),
-		events:       eventMux.Subscribe(ChainHeadEvent{}, GasPriceChanged{}, RemovedTransactionEvent{}),
+		locals:       newTxSet(),
+		events:       eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}),
 		quit:         make(chan struct{}),
 	}
 
@@ -134,27 +138,48 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState
 func (pool *TxPool) eventLoop() {
 	defer pool.wg.Done()
 
+	// Start a ticker and keep track of interesting pool stats to report
+	var prevPending, prevQueued, prevPrices, prevStales int
+
+	report := time.NewTicker(statsReportInterval)
+	defer report.Stop()
+
 	// Track chain events. When a chain events occurs (new chain canon block)
 	// we need to know the new state. The new state will help us determine
 	// the nonces in the managed state
-	for ev := range pool.events.Chan() {
-		switch ev := ev.Data.(type) {
-		case ChainHeadEvent:
-			pool.mu.Lock()
-			if ev.Block != nil {
-				if pool.config.IsHomestead(ev.Block.Number()) {
-					pool.homestead = true
+	for {
+		select {
+		// Handle any events fired by the system
+		case ev, ok := <-pool.events.Chan():
+			if !ok {
+				return
+			}
+			switch ev := ev.Data.(type) {
+			case ChainHeadEvent:
+				pool.mu.Lock()
+				if ev.Block != nil {
+					if pool.config.IsHomestead(ev.Block.Number()) {
+						pool.homestead = true
+					}
 				}
+				pool.resetState()
+				pool.mu.Unlock()
+
+			case RemovedTransactionEvent:
+				pool.AddBatch(ev.Txs)
 			}
 
-			pool.resetState()
-			pool.mu.Unlock()
-		case GasPriceChanged:
-			pool.mu.Lock()
-			pool.minGasPrice = ev.Price
-			pool.mu.Unlock()
-		case RemovedTransactionEvent:
-			pool.AddBatch(ev.Txs)
+		// Handle stats reporting ticks
+		case <-report.C:
+			pool.mu.RLock()
+			pending, queued := pool.stats()
+			prices, stales := len(pool.priced.items), pool.priced.stales
+			pool.mu.RUnlock()
+
+			if pending != prevPending || queued != prevQueued || prices != prevPrices || stales != prevStales {
+				log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "prices", prices-stales, "stales", stales)
+				prevPending, prevQueued, prevPrices, prevStales = pending, queued, prices, stales
+			}
 		}
 	}
 }
@@ -191,6 +216,27 @@ func (pool *TxPool) Stop() {
 	log.Info("Transaction pool stopped")
 }
 
+// GasPrice returns the current gas price enforced by the transaction pool.
+func (pool *TxPool) GasPrice() *big.Int {
+	pool.mu.RLock()
+	defer pool.mu.RUnlock()
+
+	return new(big.Int).Set(pool.gasPrice)
+}
+
+// SetGasPrice updates the minimum price required by the transaction pool for a
+// new transaction, and drops all transactions below this threshold.
+func (pool *TxPool) SetGasPrice(price *big.Int) {
+	pool.mu.Lock()
+	defer pool.mu.Unlock()
+
+	pool.gasPrice = price
+	for _, tx := range pool.priced.Cap(price, pool.locals) {
+		pool.removeTx(tx.Hash())
+	}
+	log.Info("Transaction pool price threshold updated", "price", price)
+}
+
 func (pool *TxPool) State() *state.ManagedState {
 	pool.mu.RLock()
 	defer pool.mu.RUnlock()
@@ -200,17 +246,25 @@ func (pool *TxPool) State() *state.ManagedState {
 
 // Stats retrieves the current pool stats, namely the number of pending and the
 // number of queued (non-executable) transactions.
-func (pool *TxPool) Stats() (pending int, queued int) {
+func (pool *TxPool) Stats() (int, int) {
 	pool.mu.RLock()
 	defer pool.mu.RUnlock()
 
+	return pool.stats()
+}
+
+// stats retrieves the current pool stats, namely the number of pending and the
+// number of queued (non-executable) transactions.
+func (pool *TxPool) stats() (int, int) {
+	pending := 0
 	for _, list := range pool.pending {
 		pending += list.Len()
 	}
+	queued := 0
 	for _, list := range pool.queue {
 		queued += list.Len()
 	}
-	return
+	return pending, queued
 }
 
 // Content retrieves the data content of the transaction pool, returning all the
@@ -260,16 +314,16 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) {
 func (pool *TxPool) SetLocal(tx *types.Transaction) {
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
-	pool.localTx.add(tx.Hash())
+	pool.locals.add(tx.Hash())
 }
 
 // validateTx checks whether a transaction is valid according
 // to the consensus rules.
 func (pool *TxPool) validateTx(tx *types.Transaction) error {
-	local := pool.localTx.contains(tx.Hash())
+	local := pool.locals.contains(tx.Hash())
 	// Drop transactions under our own minimal accepted gas price
-	if !local && pool.minGasPrice.Cmp(tx.GasPrice()) > 0 {
-		return ErrCheap
+	if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
+		return ErrUnderpriced
 	}
 
 	currentState, err := pool.currentState()
@@ -322,16 +376,32 @@ func (pool *TxPool) add(tx *types.Transaction) error {
 		log.Trace("Discarding already known transaction", "hash", hash)
 		return fmt.Errorf("known transaction: %x", hash)
 	}
-	// Otherwise ensure basic validation passes and queue it up
+	// If the transaction fails basic validation, discard it
 	if err := pool.validateTx(tx); err != nil {
 		log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
 		invalidTxCounter.Inc(1)
 		return err
 	}
+	// If the transaction pool is full, discard underpriced transactions
+	if uint64(len(pool.all)) >= maxPendingTotal+maxQueuedTotal {
+		// If the new transaction is underpriced, don't accept it
+		if pool.priced.Underpriced(tx, pool.locals) {
+			log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
+			underpricedTxCounter.Inc(1)
+			return ErrUnderpriced
+		}
+		// New transaction is better than our worse ones, make room for it
+		drop := pool.priced.Discard(len(pool.all)-int(maxPendingTotal+maxQueuedTotal-1), pool.locals)
+		for _, tx := range drop {
+			log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
+			underpricedTxCounter.Inc(1)
+			pool.removeTx(tx.Hash())
+		}
+	}
+	// Great, queue the transaction
 	pool.enqueueTx(hash, tx)
 
-	// Print a log message if low enough level is set
-	log.Debug("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(pool.signer, tx); return from }}, "to", tx.To())
+	log.Trace("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(pool.signer, tx); return from }}, "to", tx.To())
 	return nil
 }
 
@@ -352,9 +422,12 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) {
 	// Discard any previous transaction and mark this
 	if old != nil {
 		delete(pool.all, old.Hash())
+		pool.priced.Remove(old)
+
 		queuedReplaceCounter.Inc(1)
 	}
 	pool.all[hash] = tx
+	pool.priced.Put(tx)
 }
 
 // promoteTx adds a transaction to the pending (processable) list of transactions.
@@ -371,16 +444,23 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	if !inserted {
 		// An older transaction was better, discard this
 		delete(pool.all, hash)
+		pool.priced.Remove(tx)
+
 		pendingDiscardCounter.Inc(1)
 		return
 	}
 	// Otherwise discard any previous transaction and mark this
 	if old != nil {
 		delete(pool.all, old.Hash())
+		pool.priced.Remove(old)
+
 		pendingReplaceCounter.Inc(1)
 	}
-	pool.all[hash] = tx // Failsafe to work around direct pending inserts (tests)
-
+	// Failsafe to work around direct pending inserts (tests)
+	if pool.all[hash] == nil {
+		pool.all[hash] = tx
+		pool.priced.Put(tx)
+	}
 	// Set the potentially new pending nonce and notify any subsystems of the new tx
 	pool.beats[addr] = time.Now()
 	pool.pendingState.SetNonce(addr, tx.Nonce()+1)
@@ -395,7 +475,6 @@ func (pool *TxPool) Add(tx *types.Transaction) error {
 	if err := pool.add(tx); err != nil {
 		return err
 	}
-
 	state, err := pool.currentState()
 	if err != nil {
 		return err
@@ -467,6 +546,7 @@ func (pool *TxPool) removeTx(hash common.Hash) {
 
 	// Remove it from the list of known transactions
 	delete(pool.all, hash)
+	pool.priced.Remove(tx)
 
 	// Remove the transaction from the pending lists and reset the account nonce
 	if pending := pool.pending[addr]; pending != nil {
@@ -506,28 +586,31 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 		// Drop all transactions that are deemed too old (low nonce)
 		for _, tx := range list.Forward(state.GetNonce(addr)) {
 			hash := tx.Hash()
-			log.Debug("Removed old queued transaction", "hash", hash)
+			log.Trace("Removed old queued transaction", "hash", hash)
 			delete(pool.all, hash)
+			pool.priced.Remove(tx)
 		}
 		// Drop all transactions that are too costly (low balance)
 		drops, _ := list.Filter(state.GetBalance(addr))
 		for _, tx := range drops {
 			hash := tx.Hash()
-			log.Debug("Removed unpayable queued transaction", "hash", hash)
+			log.Trace("Removed unpayable queued transaction", "hash", hash)
 			delete(pool.all, hash)
+			pool.priced.Remove(tx)
 			queuedNofundsCounter.Inc(1)
 		}
 		// Gather all executable transactions and promote them
 		for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) {
 			hash := tx.Hash()
-			log.Debug("Promoting queued transaction", "hash", hash)
+			log.Trace("Promoting queued transaction", "hash", hash)
 			pool.promoteTx(addr, hash, tx)
 		}
 		// Drop all transactions over the allowed limit
 		for _, tx := range list.Cap(int(maxQueuedPerAccount)) {
 			hash := tx.Hash()
-			log.Debug("Removed cap-exceeding queued transaction", "hash", hash)
+			log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
 			delete(pool.all, hash)
+			pool.priced.Remove(tx)
 			queuedRLCounter.Inc(1)
 		}
 		queued += uint64(list.Len())
@@ -551,7 +634,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 			if uint64(list.Len()) > minPendingPerAccount {
 				// Skip local accounts as pools should maintain backlogs for themselves
 				for _, tx := range list.txs.items {
-					if !pool.localTx.contains(tx.Hash()) {
+					if !pool.locals.contains(tx.Hash()) {
 						spammers.Push(addr, float32(list.Len()))
 					}
 					break // Checking on transaction for locality is enough
@@ -593,7 +676,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 		pendingRLCounter.Inc(int64(pendingBeforeCap - pending))
 	}
 	// If we've queued more transactions than the hard limit, drop oldest ones
-	if queued > maxQueuedInTotal {
+	if queued > maxQueuedTotal {
 		// Sort all accounts with queued transactions by heartbeat
 		addresses := make(addresssByHeartbeat, 0, len(pool.queue))
 		for addr := range pool.queue {
@@ -602,7 +685,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 		sort.Sort(addresses)
 
 		// Drop transactions until the total is below the limit
-		for drop := queued - maxQueuedInTotal; drop > 0; {
+		for drop := queued - maxQueuedTotal; drop > 0; {
 			addr := addresses[len(addresses)-1]
 			list := pool.queue[addr.address]
 
@@ -639,20 +722,22 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
 		// Drop all transactions that are deemed too old (low nonce)
 		for _, tx := range list.Forward(nonce) {
 			hash := tx.Hash()
-			log.Debug("Removed old pending transaction", "hash", hash)
+			log.Trace("Removed old pending transaction", "hash", hash)
 			delete(pool.all, hash)
+			pool.priced.Remove(tx)
 		}
 		// Drop all transactions that are too costly (low balance), and queue any invalids back for later
 		drops, invalids := list.Filter(state.GetBalance(addr))
 		for _, tx := range drops {
 			hash := tx.Hash()
-			log.Debug("Removed unpayable pending transaction", "hash", hash)
+			log.Trace("Removed unpayable pending transaction", "hash", hash)
 			delete(pool.all, hash)
+			pool.priced.Remove(tx)
 			pendingNofundsCounter.Inc(1)
 		}
 		for _, tx := range invalids {
 			hash := tx.Hash()
-			log.Debug("Demoting pending transaction", "hash", hash)
+			log.Trace("Demoting pending transaction", "hash", hash)
 			pool.enqueueTx(hash, tx)
 		}
 		// Delete the entire queue entry if it became empty.

core/tx_pool_test.go

@@ -33,7 +33,11 @@ import (
 )
 
 func transaction(nonce uint64, gaslimit *big.Int, key *ecdsa.PrivateKey) *types.Transaction {
-	tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(100), gaslimit, big.NewInt(1), nil), types.HomesteadSigner{}, key)
+	return pricedTransaction(nonce, gaslimit, big.NewInt(1), key)
+}
+
+func pricedTransaction(nonce uint64, gaslimit, gasprice *big.Int, key *ecdsa.PrivateKey) *types.Transaction {
+	tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(100), gaslimit, gasprice, nil), types.HomesteadSigner{}, key)
 	return tx
 }
 
@@ -151,9 +155,9 @@ func TestInvalidTransactions(t *testing.T) {
 	}
 
 	tx = transaction(1, big.NewInt(100000), key)
-	pool.minGasPrice = big.NewInt(1000)
-	if err := pool.Add(tx); err != ErrCheap {
-		t.Error("expected", ErrCheap, "got", err)
+	pool.gasPrice = big.NewInt(1000)
+	if err := pool.Add(tx); err != ErrUnderpriced {
+		t.Error("expected", ErrUnderpriced, "got", err)
 	}
 
 	pool.SetLocal(tx)
@@ -557,8 +561,8 @@ func TestTransactionQueueAccountLimiting(t *testing.T) {
 // some threshold, the higher transactions are dropped to prevent DOS attacks.
 func TestTransactionQueueGlobalLimiting(t *testing.T) {
 	// Reduce the queue limits to shorten test time
-	defer func(old uint64) { maxQueuedInTotal = old }(maxQueuedInTotal)
-	maxQueuedInTotal = maxQueuedPerAccount * 3
+	defer func(old uint64) { maxQueuedTotal = old }(maxQueuedTotal)
+	maxQueuedTotal = maxQueuedPerAccount * 3
 
 	// Create the pool to test the limit enforcement with
 	db, _ := ethdb.NewMemDatabase()
@@ -578,7 +582,7 @@ func TestTransactionQueueGlobalLimiting(t *testing.T) {
 	// Generate and queue a batch of transactions
 	nonces := make(map[common.Address]uint64)
 
-	txs := make(types.Transactions, 0, 3*maxQueuedInTotal)
+	txs := make(types.Transactions, 0, 3*maxQueuedTotal)
 	for len(txs) < cap(txs) {
 		key := keys[rand.Intn(len(keys))]
 		addr := crypto.PubkeyToAddress(key.PublicKey)
@@ -596,8 +600,8 @@ func TestTransactionQueueGlobalLimiting(t *testing.T) {
 		}
 		queued += list.Len()
 	}
-	if queued > int(maxQueuedInTotal) {
-		t.Fatalf("total transactions overflow allowance: %d > %d", queued, maxQueuedInTotal)
+	if queued > int(maxQueuedTotal) {
+		t.Fatalf("total transactions overflow allowance: %d > %d", queued, maxQueuedTotal)
 	}
 }
 
@@ -791,6 +795,166 @@ func TestTransactionPendingMinimumAllowance(t *testing.T) {
 	}
 }
 
+// Tests that setting the transaction pool gas price to a higher value correctly
+// discards everything cheaper than that and moves any gapped transactions back
+// from the pending pool to the queue.
+//
+// Note, local transactions are never allowed to be dropped.
+func TestTransactionPoolRepricing(t *testing.T) {
+	// Create the pool to test the pricing enforcement with
+	db, _ := ethdb.NewMemDatabase()
+	statedb, _ := state.New(common.Hash{}, db)
+
+	pool := NewTxPool(params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
+	pool.resetState()
+
+	// Create a number of test accounts and fund them
+	state, _ := pool.currentState()
+
+	keys := make([]*ecdsa.PrivateKey, 3)
+	for i := 0; i < len(keys); i++ {
+		keys[i], _ = crypto.GenerateKey()
+		state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
+	}
+	// Generate and queue a batch of transactions, both pending and queued
+	txs := types.Transactions{}
+
+	txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(2), keys[0]))
+	txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[0]))
+	txs = append(txs, pricedTransaction(2, big.NewInt(100000), big.NewInt(2), keys[0]))
+
+	txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(2), keys[1]))
+	txs = append(txs, pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1]))
+	txs = append(txs, pricedTransaction(3, big.NewInt(100000), big.NewInt(2), keys[1]))
+
+	txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2]))
+	pool.SetLocal(txs[len(txs)-1]) // prevent this one from ever being dropped
+
+	// Import the batch and that both pending and queued transactions match up
+	pool.AddBatch(txs)
+
+	pending, queued := pool.stats()
+	if pending != 4 {
+		t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 4)
+	}
+	if queued != 3 {
+		t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 3)
+	}
+	// Reprice the pool and check that underpriced transactions get dropped
+	pool.SetGasPrice(big.NewInt(2))
+
+	pending, queued = pool.stats()
+	if pending != 2 {
+		t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
+	}
+	if queued != 3 {
+		t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 3)
+	}
+	// Check that we can't add the old transactions back
+	if err := pool.Add(pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[0])); err != ErrUnderpriced {
+		t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced)
+	}
+	if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced {
+		t.Fatalf("adding underpriced queued transaction error mismatch: have %v, want %v", err, ErrUnderpriced)
+	}
+	// However we can add local underpriced transactions
+	tx := pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[2])
+
+	pool.SetLocal(tx) // prevent this one from ever being dropped
+	if err := pool.Add(tx); err != nil {
+		t.Fatalf("failed to add underpriced local transaction: %v", err)
+	}
+	if pending, _ = pool.stats(); pending != 3 {
+		t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3)
+	}
+}
+
+// Tests that when the pool reaches its global transaction limit, underpriced
+// transactions are gradually shifted out for more expensive ones and any gapped
+// pending transactions are moved into te queue.
+//
+// Note, local transactions are never allowed to be dropped.
+func TestTransactionPoolUnderpricing(t *testing.T) {
+	// Reduce the queue limits to shorten test time
+	defer func(old uint64) { maxPendingTotal = old }(maxPendingTotal)
+	maxPendingTotal = 2
+
+	defer func(old uint64) { maxQueuedTotal = old }(maxQueuedTotal)
+	maxQueuedTotal = 2
+
+	// Create the pool to test the pricing enforcement with
+	db, _ := ethdb.NewMemDatabase()
+	statedb, _ := state.New(common.Hash{}, db)
+
+	pool := NewTxPool(params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
+	pool.resetState()
+
+	// Create a number of test accounts and fund them
+	state, _ := pool.currentState()
+
+	keys := make([]*ecdsa.PrivateKey, 3)
+	for i := 0; i < len(keys); i++ {
+		keys[i], _ = crypto.GenerateKey()
+		state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
+	}
+	// Generate and queue a batch of transactions, both pending and queued
+	txs := types.Transactions{}
+
+	txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[0]))
+	txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(2), keys[0]))
+
+	txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[1]))
+
+	txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2]))
+	pool.SetLocal(txs[len(txs)-1]) // prevent this one from ever being dropped
+
+	// Import the batch and that both pending and queued transactions match up
+	pool.AddBatch(txs)
+
+	pending, queued := pool.stats()
+	if pending != 3 {
+		t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3)
+	}
+	if queued != 1 {
+		t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1)
+	}
+	// Ensure that adding an underpriced transaction on block limit fails
+	if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced {
+		t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced)
+	}
+	// Ensure that adding high priced transactions drops cheap ones, but not own
+	if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(3), keys[1])); err != nil {
+		t.Fatalf("failed to add well priced transaction: %v", err)
+	}
+	if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(4), keys[1])); err != nil {
+		t.Fatalf("failed to add well priced transaction: %v", err)
+	}
+	if err := pool.Add(pricedTransaction(3, big.NewInt(100000), big.NewInt(5), keys[1])); err != nil {
+		t.Fatalf("failed to add well priced transaction: %v", err)
+	}
+	pending, queued = pool.stats()
+	if pending != 2 {
+		t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
+	}
+	if queued != 2 {
+		t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2)
+	}
+	// Ensure that adding local transactions can push out even higher priced ones
+	tx := pricedTransaction(1, big.NewInt(100000), big.NewInt(0), keys[2])
+
+	pool.SetLocal(tx) // prevent this one from ever being dropped
+	if err := pool.Add(tx); err != nil {
+		t.Fatalf("failed to add underpriced local transaction: %v", err)
+	}
+	pending, queued = pool.stats()
+	if pending != 2 {
+		t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
+	}
+	if queued != 2 {
+		t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2)
+	}
+}
+
 // Benchmarks the speed of validating the contents of the pending queue of the
 // transaction pool.
 func BenchmarkPendingDemotion100(b *testing.B)   { benchmarkPendingDemotion(b, 100) }

eth/api.go

@@ -153,6 +153,8 @@ func (api *PrivateMinerAPI) Start(threads *int) error {
 	}
 	// Start the miner and return
 	if !api.e.IsMining() {
+		// Propagate the initial price point to the transaction pool
+		api.e.txPool.SetGasPrice(api.e.gasPrice)
 		return api.e.StartMining(true)
 	}
 	return nil
@@ -180,7 +182,7 @@ func (api *PrivateMinerAPI) SetExtra(extra string) (bool, error) {
 
 // SetGasPrice sets the minimum accepted gas price for the miner.
 func (api *PrivateMinerAPI) SetGasPrice(gasPrice hexutil.Big) bool {
-	api.e.Miner().SetGasPrice((*big.Int)(&gasPrice))
+	api.e.txPool.SetGasPrice((*big.Int)(&gasPrice))
 	return true
 }
 

eth/backend.go

@@ -20,6 +20,7 @@ package eth
 import (
 	"errors"
 	"fmt"
+	"math/big"
 	"runtime"
 	"sync"
 	"sync/atomic"
@@ -76,6 +77,7 @@ type Ethereum struct {
 	ApiBackend *EthApiBackend
 
 	miner        *miner.Miner
+	gasPrice     *big.Int
 	Mining       bool
 	MinerThreads int
 	etherbase    common.Address
@@ -167,7 +169,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	}
 
 	eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine)
-	eth.miner.SetGasPrice(config.GasPrice)
+	eth.gasPrice = config.GasPrice
 	eth.miner.SetExtra(makeExtraData(config.ExtraData))
 
 	eth.ApiBackend = &EthApiBackend{eth, nil}

eth/config.go

@@ -42,7 +42,7 @@ var DefaultConfig = Config{
 	NetworkId:            1,
 	LightPeers:           20,
 	DatabaseCache:        128,
-	GasPrice:             big.NewInt(20 * params.Shannon),
+	GasPrice:             big.NewInt(18 * params.Shannon),
 
 	GPO: gasprice.Config{
 		Blocks:     10,

ethstats/ethstats.go

@@ -18,6 +18,7 @@
 package ethstats
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -639,7 +640,8 @@ func (s *Service) reportStats(conn *websocket.Conn) error {
 		sync := s.eth.Downloader().Progress()
 		syncing = s.eth.BlockChain().CurrentHeader().Number.Uint64() >= sync.HighestBlock
 
-		gasprice = int(s.eth.Miner().GasPrice().Uint64())
+		price, _ := s.eth.ApiBackend.SuggestPrice(context.Background())
+		gasprice = int(price.Uint64())
 	} else {
 		sync := s.les.Downloader().Progress()
 		syncing = s.les.BlockChain().CurrentHeader().Number.Uint64() >= sync.HighestBlock

miner/miner.go

@@ -19,7 +19,6 @@ package miner
 
 import (
 	"fmt"
-	"math/big"
 	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/accounts"
@@ -104,18 +103,6 @@ out:
 	}
 }
 
-func (m *Miner) GasPrice() *big.Int {
-	return new(big.Int).Set(m.worker.gasPrice)
-}
-
-func (m *Miner) SetGasPrice(price *big.Int) {
-	// FIXME block tests set a nil gas price. Quick dirty fix
-	if price == nil {
-		return
-	}
-	m.worker.setGasPrice(price)
-}
-
 func (self *Miner) Start(coinbase common.Address) {
 	atomic.StoreInt32(&self.shouldStart, 1)
 	self.worker.setEtherbase(coinbase)

miner/worker.go

@@ -59,14 +59,12 @@ type Work struct {
 	config *params.ChainConfig
 	signer types.Signer
 
-	state         *state.StateDB // apply state changes here
-	ancestors     *set.Set       // ancestor set (used for checking uncle parent validity)
-	family        *set.Set       // family set (used for checking uncle invalidity)
-	uncles        *set.Set       // uncle set
-	tcount        int            // tx count in cycle
-	ownedAccounts *set.Set
-	lowGasTxs     types.Transactions
-	failedTxs     types.Transactions
+	state     *state.StateDB // apply state changes here
+	ancestors *set.Set       // ancestor set (used for checking uncle parent validity)
+	family    *set.Set       // family set (used for checking uncle invalidity)
+	uncles    *set.Set       // uncle set
+	tcount    int            // tx count in cycle
+	failedTxs types.Transactions
 
 	Block *types.Block // the new block
 
@@ -103,7 +101,6 @@ type worker struct {
 	chainDb ethdb.Database
 
 	coinbase common.Address
-	gasPrice *big.Int
 	extra    []byte
 
 	currentMu sync.Mutex
@@ -132,7 +129,6 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
 		mux:            mux,
 		chainDb:        eth.ChainDb(),
 		recv:           make(chan *Result, resultQueueSize),
-		gasPrice:       new(big.Int),
 		chain:          eth.BlockChain(),
 		proc:           eth.BlockChain().Validator(),
 		possibleUncles: make(map[common.Hash]*types.Block),
@@ -252,7 +248,7 @@ func (self *worker) update() {
 				txs := map[common.Address]types.Transactions{acc: {ev.Tx}}
 				txset := types.NewTransactionsByPriceAndNonce(txs)
 
-				self.current.commitTransactions(self.mux, txset, self.gasPrice, self.chain, self.coinbase)
+				self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase)
 				self.currentMu.Unlock()
 			}
 		}
@@ -375,22 +371,10 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
 	}
 	// Keep track of transactions which return errors so they can be removed
 	work.tcount = 0
-	work.ownedAccounts = accountAddressesSet(accounts)
 	self.current = work
 	return nil
 }
 
-func (w *worker) setGasPrice(p *big.Int) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-
-	// calculate the minimal gas price the miner accepts when sorting out transactions.
-	const pct = int64(90)
-	w.gasPrice = gasprice(p, pct)
-
-	w.mux.Post(core.GasPriceChanged{Price: w.gasPrice})
-}
-
 func (self *worker) commitNewWork() {
 	self.mu.Lock()
 	defer self.mu.Unlock()
@@ -460,9 +444,8 @@ func (self *worker) commitNewWork() {
 		return
 	}
 	txs := types.NewTransactionsByPriceAndNonce(pending)
-	work.commitTransactions(self.mux, txs, self.gasPrice, self.chain, self.coinbase)
+	work.commitTransactions(self.mux, txs, self.chain, self.coinbase)
 
-	self.eth.TxPool().RemoveBatch(work.lowGasTxs)
 	self.eth.TxPool().RemoveBatch(work.failedTxs)
 
 	// compute uncles for the new block.
@@ -515,7 +498,7 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
 	return nil
 }
 
-func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, gasPrice *big.Int, bc *core.BlockChain, coinbase common.Address) {
+func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) {
 	gp := new(core.GasPool).AddGas(env.header.GasLimit)
 
 	var coalescedLogs []*types.Log
@@ -539,17 +522,6 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
 			txs.Pop()
 			continue
 		}
-
-		// Ignore any transactions (and accounts subsequently) with low gas limits
-		if tx.GasPrice().Cmp(gasPrice) < 0 && !env.ownedAccounts.Has(from) {
-			// Pop the current low-priced transaction without shifting in the next from the account
-			log.Warn("Transaction below gas price", "sender", from, "hash", tx.Hash(), "have", tx.GasPrice(), "want", gasPrice)
-
-			env.lowGasTxs = append(env.lowGasTxs, tx)
-			txs.Pop()
-
-			continue
-		}
 		// Start executing the transaction
 		env.state.StartRecord(tx.Hash(), common.Hash{}, env.tcount)
 
@@ -607,25 +579,3 @@ func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, c
 
 	return nil, receipt.Logs
 }
-
-// TODO: remove or use
-func (self *worker) HashRate() int64 {
-	return 0
-}
-
-// gasprice calculates a reduced gas price based on the pct
-// XXX Use big.Rat?
-func gasprice(price *big.Int, pct int64) *big.Int {
-	p := new(big.Int).Set(price)
-	p.Div(p, big.NewInt(100))
-	p.Mul(p, big.NewInt(pct))
-	return p
-}
-
-func accountAddressesSet(accounts []accounts.Account) *set.Set {
-	accountSet := set.New()
-	for _, account := range accounts {
-		accountSet.Add(account.Address)
-	}
-	return accountSet
-}