core: replace txpool priced map with a simple heap

core/tx_list.go

@@ -21,7 +21,6 @@ import (
 	"math"
 	"math/big"
 	"sort"
-	"unsafe"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -353,16 +352,16 @@ 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
+// priceHeap is a heap.Interface implementation over transactions for retrieving
+// price-sorted transactions to discard when the pool fills up.
+type priceHeap []*types.Transaction
 
 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) Less(i, j int) bool { return h[i].GasPrice().Cmp(h[j].GasPrice()) < 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))
+	*h = append(*h, x.(*types.Transaction))
 }
 
 func (h *priceHeap) Pop() interface{} {
@@ -373,190 +372,129 @@ func (h *priceHeap) Pop() interface{} {
 	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
+// txPricedList is a price-sorted heap to allow operating on transactions pool
+// contents in a price-incrementing way.
+type txPricedList struct {
+	all    *map[common.Hash]*types.Transaction // Pointer to the map of all transactions
+	items  *priceHeap                          // Heap of prices of all the stored transactions
+	stales int                                 // Number of stale price points to (re-heap trigger)
 }
 
-// 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),
+// newTxPricedList creates a new price-sorted transaction heap.
+func newTxPricedList(all *map[common.Hash]*types.Transaction) *txPricedList {
+	return &txPricedList{
+		all:   all,
+		items: 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)
+// Put inserts a new transaction into the heap.
+func (l *txPricedList) Put(tx *types.Transaction) {
+	heap.Push(l.items, tx)
 }
 
-// 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()
-			}
-		}
+// Removed notifies the prices transaction list that an old transaction dropped
+// from the pool. The list will just keep a counter of stale objects and update
+// the heap if a large enough ratio of transactions go stale.
+func (l *txPricedList) Removed() {
+	// Bump the stale counter, but exit if still too low (< 25%)
+	l.stales++
+	if l.stales <= len(*l.items)/4 {
+		return
 	}
-	m.logger.Trace("Removed old transaction", "hash", hash, "price", price)
-}
+	// Seems we've reached a critical number of stale transactions, reheap
+	reheap := make(priceHeap, 0, len(*l.all))
 
-// 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[:]))
-		}
+	l.stales, l.items = 0, &reheap
+	for _, tx := range *l.all {
+		*l.items = append(*l.items, tx)
 	}
-	heap.Init(m.index)
+	heap.Init(l.items)
 }
 
 // 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 {
+// 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 {
 	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)
+	for len(*l.items) > 0 {
+		// Discard stale transactions if found during cleanup
+		tx := heap.Pop(l.items).(*types.Transaction)
 
-		if len(m.items[key]) == 0 {
-			m.stales--
-			heap.Pop(m.index)
-			delete(m.items, key)
+		hash := tx.Hash()
+		if _, ok := (*l.all)[hash]; !ok {
+			l.stales--
 			continue
 		}
 		// Stop the discards if we've reached the threshold
-		if price.Cmp(threshold) >= 0 {
+		if tx.GasPrice().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)
-			}
+		// Non stale transaction found, discard unless local
+		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)
+		heap.Push(l.items, 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 {
+func (l *txPricedList) 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)
+	for len(*l.items) > 0 {
+		head := []*types.Transaction(*l.items)[0]
+		if _, ok := (*l.all)[head.Hash()]; !ok {
+			l.stales--
+			heap.Pop(l.items)
 			continue
 		}
 		break
 	}
 	// Check if the transaction is underpriced or not
-	if len(*m.index) == 0 {
+	if len(*l.items) == 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
+	cheapest := []*types.Transaction(*l.items)[0]
+	return cheapest.GasPrice().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 {
+// priced list and returs them for further removal from the entire pool.
+func (l *txPricedList) 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)
+	for len(*l.items) > 0 && count > 0 {
+		// Discard stale transactions if found during cleanup
+		tx := heap.Pop(l.items).(*types.Transaction)
 
-		if len(m.items[key]) == 0 {
-			m.stales--
-			heap.Pop(m.index)
-			delete(m.items, key)
+		hash := tx.Hash()
+		if _, ok := (*l.all)[hash]; !ok {
+			l.stales--
 			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)
+		// Non stale transaction found, discard unless local
+		if local.contains(hash) {
+			save = append(save, tx)
+		} else {
+			drop = append(drop, tx)
+			count--
 		}
 	}
 	for _, tx := range save {
-		m.Put(tx)
+		heap.Push(l.items, tx)
 	}
 	return drop
 }

core/tx_pool.go

@@ -101,7 +101,7 @@ type TxPool struct {
 	queue   map[common.Address]*txList         // Queued but non-processable transactions
 	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
+	priced  *txPricedList                      // All transactions sorted by price
 
 	wg   sync.WaitGroup // for shutdown sync
 	quit chan struct{}
@@ -117,7 +117,6 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState
 		queue:        make(map[common.Address]*txList),
 		beats:        make(map[common.Address]time.Time),
 		all:          make(map[common.Hash]*types.Transaction),
-		priced:       newTxPricedMap(),
 		eventMux:     eventMux,
 		currentState: currentStateFn,
 		gasLimit:     gasLimitFn,
@@ -127,7 +126,7 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState
 		events:       eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}),
 		quit:         make(chan struct{}),
 	}
-
+	pool.priced = newTxPricedList(&pool.all)
 	pool.resetState()
 
 	pool.wg.Add(2)
@@ -141,7 +140,7 @@ 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
+	var prevPending, prevQueued, prevStales int
 
 	report := time.NewTicker(statsReportInterval)
 	defer report.Stop()
@@ -175,12 +174,12 @@ func (pool *TxPool) eventLoop() {
 		case <-report.C:
 			pool.mu.RLock()
 			pending, queued := pool.stats()
-			prices, stales := len(pool.priced.items), pool.priced.stales
+			stales := 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
+			if pending != prevPending || queued != prevQueued || stales != prevStales {
+				log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales)
+				prevPending, prevQueued, prevStales = pending, queued, stales
 			}
 		}
 	}
@@ -414,7 +413,7 @@ func (pool *TxPool) add(tx *types.Transaction) (bool, error) {
 		// New transaction is better, replace old one
 		if old != nil {
 			delete(pool.all, old.Hash())
-			pool.priced.Remove(old)
+			pool.priced.Removed()
 			pendingReplaceCounter.Inc(1)
 		}
 		pool.all[tx.Hash()] = tx
@@ -450,7 +449,7 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
 	// Discard any previous transaction and mark this
 	if old != nil {
 		delete(pool.all, old.Hash())
-		pool.priced.Remove(old)
+		pool.priced.Removed()
 		queuedReplaceCounter.Inc(1)
 	}
 	pool.all[hash] = tx
@@ -472,7 +471,7 @@ 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)
+		pool.priced.Removed()
 
 		pendingDiscardCounter.Inc(1)
 		return
@@ -480,7 +479,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	// Otherwise discard any previous transaction and mark this
 	if old != nil {
 		delete(pool.all, old.Hash())
-		pool.priced.Remove(old)
+		pool.priced.Removed()
 
 		pendingReplaceCounter.Inc(1)
 	}
@@ -583,7 +582,7 @@ func (pool *TxPool) removeTx(hash common.Hash) {
 
 	// Remove it from the list of known transactions
 	delete(pool.all, hash)
-	pool.priced.Remove(tx)
+	pool.priced.Removed()
 
 	// Remove the transaction from the pending lists and reset the account nonce
 	if pending := pool.pending[addr]; pending != nil {
@@ -625,7 +624,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 			hash := tx.Hash()
 			log.Trace("Removed old queued transaction", "hash", hash)
 			delete(pool.all, hash)
-			pool.priced.Remove(tx)
+			pool.priced.Removed()
 		}
 		// Drop all transactions that are too costly (low balance)
 		drops, _ := list.Filter(state.GetBalance(addr))
@@ -633,7 +632,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 			hash := tx.Hash()
 			log.Trace("Removed unpayable queued transaction", "hash", hash)
 			delete(pool.all, hash)
-			pool.priced.Remove(tx)
+			pool.priced.Removed()
 			queuedNofundsCounter.Inc(1)
 		}
 		// Gather all executable transactions and promote them
@@ -647,7 +646,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
 			hash := tx.Hash()
 			log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
 			delete(pool.all, hash)
-			pool.priced.Remove(tx)
+			pool.priced.Removed()
 			queuedRLCounter.Inc(1)
 		}
 		queued += uint64(list.Len())
@@ -761,7 +760,7 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
 			hash := tx.Hash()
 			log.Trace("Removed old pending transaction", "hash", hash)
 			delete(pool.all, hash)
-			pool.priced.Remove(tx)
+			pool.priced.Removed()
 		}
 		// Drop all transactions that are too costly (low balance), and queue any invalids back for later
 		drops, invalids := list.Filter(state.GetBalance(addr))
@@ -769,7 +768,7 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
 			hash := tx.Hash()
 			log.Trace("Removed unpayable pending transaction", "hash", hash)
 			delete(pool.all, hash)
-			pool.priced.Remove(tx)
+			pool.priced.Removed()
 			pendingNofundsCounter.Inc(1)
 		}
 		for _, tx := range invalids {