Perf async priced

cmd/geth/main.go

@@ -79,6 +79,8 @@ var (
 		utils.TxPoolRejournalFlag,
 		utils.TxPoolPriceLimitFlag,
 		utils.TxPoolPriceBumpFlag,
+		utils.TxPoolDisablePricedFlag,
+		utils.TxPoolEnableAsyncPricedFlag,
 		utils.TxPoolAccountSlotsFlag,
 		utils.TxPoolGlobalSlotsFlag,
 		utils.TxPoolAccountQueueFlag,

cmd/utils/flags.go

@@ -404,6 +404,18 @@ var (
 		Value:    ethconfig.Defaults.TxPool.PriceBump,
 		Category: flags.TxPoolCategory,
 	}
+	TxPoolDisablePricedFlag = &cli.BoolFlag{
+		Name:     "txpool.disablepriced",
+		Usage:    "disable priced-sorted list for txpool",
+		Value:    false,
+		Category: flags.TxPoolCategory,
+	}
+	TxPoolEnableAsyncPricedFlag = &cli.BoolFlag{
+		Name:     "txpool.asyncpriced",
+		Usage:    "enable async-priced-sorted list for txpool",
+		Value:    false,
+		Category: flags.TxPoolCategory,
+	}
 	TxPoolAccountSlotsFlag = &cli.Uint64Flag{
 		Name:     "txpool.accountslots",
 		Usage:    "Minimum number of executable transaction slots guaranteed per account",
@@ -1704,6 +1716,12 @@ func setTxPool(ctx *cli.Context, cfg *legacypool.Config) {
 	if ctx.IsSet(TxPoolPriceBumpFlag.Name) {
 		cfg.PriceBump = ctx.Uint64(TxPoolPriceBumpFlag.Name)
 	}
+	if ctx.IsSet(TxPoolDisablePricedFlag.Name) {
+		cfg.DisablePriced = ctx.Bool(TxPoolDisablePricedFlag.Name)
+	}
+	if ctx.IsSet(TxPoolEnableAsyncPricedFlag.Name) {
+		cfg.EnableAsyncPriced = ctx.Bool(TxPoolEnableAsyncPricedFlag.Name)
+	}
 	if ctx.IsSet(TxPoolAccountSlotsFlag.Name) {
 		cfg.AccountSlots = ctx.Uint64(TxPoolAccountSlotsFlag.Name)
 	}

core/blockchain.go

@@ -1682,14 +1682,14 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 		return 0, nil
 	}
 
-	minerMode := false
-	if len(chain) == 1 {
-		block := chain[0]
-		_, receiptExist := bc.miningReceiptsCache.Get(block.Hash())
-		_, logExist := bc.miningTxLogsCache.Get(block.Hash())
-		_, stateExist := bc.miningStateCache.Get(block.Hash())
-		minerMode = receiptExist && logExist && stateExist
-	}
+	minerMode := true
+	// if len(chain) == 1 {
+	// 	block := chain[0]
+	// 	_, receiptExist := bc.miningReceiptsCache.Get(block.Hash())
+	// 	_, logExist := bc.miningTxLogsCache.Get(block.Hash())
+	// 	_, stateExist := bc.miningStateCache.Get(block.Hash())
+	// 	minerMode = receiptExist && logExist && stateExist
+	// }
 
 	// Start a parallel signature recovery (signer will fluke on fork transition, minimal perf loss)
 	SenderCacher.RecoverFromBlocks(types.MakeSigner(bc.chainConfig, chain[0].Number(), chain[0].Time()), chain)
@@ -1941,7 +1941,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 
 		vstart := time.Now()
 		// Async validate if minerMode
-		asyncValidateStateCh := make(chan error, 1)
+		// asyncValidateStateCh := make(chan error, 1)
 		if minerMode {
 			header := block.Header()
 			// Can not validate root concurrently
@@ -1951,9 +1951,9 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 				followupInterrupt.Store(true)
 				return it.index, err
 			}
-			go func() {
-				asyncValidateStateCh <- bc.validator.ValidateState(block, statedb, receipts, usedGas, true)
-			}()
+			// go func() {
+			// 	asyncValidateStateCh <- bc.validator.ValidateState(block, statedb, receipts, usedGas, true)
+			// }()
 		} else {
 			if err := bc.validator.ValidateState(block, statedb, receipts, usedGas, false); err != nil {
 				bc.reportBlock(block, receipts, err)
@@ -1998,11 +1998,11 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 		if err != nil {
 			return it.index, err
 		}
-		if minerMode {
-			if err := <-asyncValidateStateCh; err != nil {
-				panic(fmt.Errorf("self mined block(hash: %x number %v) async verify state err: %w", block.Hash(), block.NumberU64(), err))
-			}
-		}
+		// if minerMode {
+		// 	if err := <-asyncValidateStateCh; err != nil {
+		// 		panic(fmt.Errorf("self mined block(hash: %x number %v) async verify state err: %w", block.Hash(), block.NumberU64(), err))
+		// 	}
+		// }
 		bc.CacheBlock(block.Hash(), block)
 		log.Info("perf-trace insertChain debug2", "duration", common.PrettyDuration(time.Since(wstart)), "hash", block.Hash())
 

core/txpool/legacypool/async_priced_list.go

@@ -0,0 +1,183 @@
+package legacypool
+
+import (
+	"math/big"
+	"sync"
+	"sync/atomic"
+
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+var _ pricedListInterface = &asyncPricedList{}
+
+type addEvent struct {
+	tx    *types.Transaction
+	local bool
+}
+
+type discardEvent struct {
+	slots int
+	force bool
+	done  chan *discardResult
+}
+type discardResult struct {
+	discardTxs types.Transactions
+	succ       bool
+}
+
+type asyncPricedList struct {
+	priced         *pricedList
+	floatingLowest atomic.Value
+	urgentLowest   atomic.Value
+	mu             sync.Mutex
+
+	// events
+	quit       chan struct{}
+	reheap     chan struct{}
+	add        chan *addEvent
+	remove     chan int
+	discard    chan *discardEvent
+	setBaseFee chan *big.Int
+}
+
+func newAsyncPricedList(all *lookup) *asyncPricedList {
+	a := &asyncPricedList{
+		priced:     newPricedList(all),
+		quit:       make(chan struct{}),
+		reheap:     make(chan struct{}),
+		add:        make(chan *addEvent),
+		remove:     make(chan int),
+		discard:    make(chan *discardEvent),
+		setBaseFee: make(chan *big.Int),
+	}
+	go a.run()
+	return a
+}
+
+// run is a loop that handles async operations:
+//   - reheap: reheap the whole priced list, to get the lowest gas price
+//   - put: add a transaction to the priced list
+//   - remove: remove transactions from the priced list
+//   - discard: remove transactions to make room for new ones
+func (a *asyncPricedList) run() {
+	var reheap bool
+	var newOnes []*types.Transaction
+	var toRemove int = 0
+	// current loop state
+	var currentDone chan struct{} = nil
+	var baseFee *big.Int = nil
+	for {
+		if currentDone == nil {
+			currentDone = make(chan struct{})
+			go func(reheap bool, newOnes []*types.Transaction, toRemove int, baseFee *big.Int) {
+				a.handle(reheap, newOnes, toRemove, baseFee, currentDone)
+				<-currentDone
+				currentDone = nil
+			}(reheap, newOnes, toRemove, baseFee)
+
+			reheap, newOnes, toRemove, baseFee = false, nil, 0, nil
+		}
+		select {
+		case <-a.reheap:
+			reheap = true
+
+		case add := <-a.add:
+			newOnes = append(newOnes, add.tx)
+
+		case remove := <-a.remove:
+			toRemove += remove
+
+		case baseFee = <-a.setBaseFee:
+
+		case <-a.quit:
+			return
+		}
+	}
+}
+
+func (a *asyncPricedList) handle(reheap bool, newOnes []*types.Transaction, toRemove int, baseFee *big.Int, finished chan struct{}) {
+	defer close(finished)
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	// add new transactions to the priced list
+	for _, tx := range newOnes {
+		a.priced.Put(tx, false)
+	}
+	// remove staled transactions from the priced list
+	a.priced.Removed(toRemove)
+	// reheap if needed
+	if reheap {
+		a.priced.Reheap()
+		// set the lowest priced transaction when reheap is done
+		var emptyTx *types.Transaction = nil
+		if len(a.priced.floating.list) > 0 {
+			a.floatingLowest.Store(a.priced.floating.list[0])
+		} else {
+			a.floatingLowest.Store(emptyTx)
+		}
+		if len(a.priced.urgent.list) > 0 {
+			a.urgentLowest.Store(a.priced.urgent.list[0])
+		} else {
+			a.urgentLowest.Store(emptyTx)
+		}
+	}
+	if baseFee != nil {
+		a.priced.SetBaseFee(baseFee)
+	}
+}
+
+func (a *asyncPricedList) Put(tx *types.Transaction, local bool) {
+	a.add <- &addEvent{tx, local}
+}
+
+func (a *asyncPricedList) Removed(count int) {
+	a.remove <- count
+}
+
+func (a *asyncPricedList) Underpriced(tx *types.Transaction) bool {
+	var urgentLowest, floatingLowest *types.Transaction = nil, nil
+	ul, fl := a.urgentLowest.Load(), a.floatingLowest.Load()
+	if ul != nil {
+		// be careful that ul might be nil
+		urgentLowest = ul.(*types.Transaction)
+	}
+	if fl != nil {
+		// be careful that fl might be nil
+		floatingLowest = fl.(*types.Transaction)
+	}
+	return (urgentLowest == nil || a.priced.urgent.cmp(urgentLowest, tx) >= 0) &&
+		(floatingLowest == nil || a.priced.floating.cmp(floatingLowest, tx) >= 0) &&
+		(floatingLowest != nil || urgentLowest != nil)
+}
+
+// Disacard cleans staled transactions to make room for new ones
+func (a *asyncPricedList) Discard(slots int, force bool) (types.Transactions, bool) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.priced.Discard(slots, force)
+}
+
+func (a *asyncPricedList) NeedReheap(currHead *types.Header) bool {
+	return false
+}
+
+func (a *asyncPricedList) Reheap() {
+	a.reheap <- struct{}{}
+}
+
+func (a *asyncPricedList) SetBaseFee(baseFee *big.Int) {
+	a.setBaseFee <- baseFee
+	a.reheap <- struct{}{}
+}
+
+func (a *asyncPricedList) SetHead(currHead *types.Header) {
+	//do nothing
+}
+
+func (a *asyncPricedList) GetBaseFee() *big.Int {
+	return a.priced.floating.baseFee
+}
+
+func (a *asyncPricedList) Stop() {
+	close(a.quit)
+}