From 84f45def3a289dee5be5df69d5b3242c024803bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 10 May 2017 16:14:27 +0300 Subject: [PATCH] core: replace txpool priced map with a simple heap --- core/tx_list.go | 206 +++++++++++++++++------------------------------- core/tx_pool.go | 35 ++++---- 2 files changed, 89 insertions(+), 152 deletions(-) diff --git a/core/tx_list.go b/core/tx_list.go index 760ac46ec1..eb380da0bc 100644 --- a/core/tx_list.go +++ b/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 } diff --git a/core/tx_pool.go b/core/tx_pool.go index ae8c6b92d3..a0373ca7da 100644 --- a/core/tx_pool.go +++ b/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 {