Merge pull request #14737 from holiman/txpool_localaccounts

Txpool localaccounts

cmd/geth/main.go

@@ -66,6 +66,7 @@ var (
 		utils.EthashDatasetDirFlag,
 		utils.EthashDatasetsInMemoryFlag,
 		utils.EthashDatasetsOnDiskFlag,
+		utils.TxPoolNoLocalsFlag,
 		utils.TxPoolPriceLimitFlag,
 		utils.TxPoolPriceBumpFlag,
 		utils.TxPoolAccountSlotsFlag,

cmd/geth/usage.go

@@ -95,6 +95,7 @@ var AppHelpFlagGroups = []flagGroup{
 	{
 		Name: "TRANSACTION POOL",
 		Flags: []cli.Flag{
+			utils.TxPoolNoLocalsFlag,
 			utils.TxPoolPriceLimitFlag,
 			utils.TxPoolPriceBumpFlag,
 			utils.TxPoolAccountSlotsFlag,

cmd/utils/flags.go

@@ -209,6 +209,10 @@ var (
 		Value: eth.DefaultConfig.EthashDatasetsOnDisk,
 	}
 	// Transaction pool settings
+	TxPoolNoLocalsFlag = cli.BoolFlag{
+		Name:  "txpool.nolocals",
+		Usage: "Disables price exemptions for locally submitted transactions",
+	}
 	TxPoolPriceLimitFlag = cli.Uint64Flag{
 		Name:  "txpool.pricelimit",
 		Usage: "Minimum gas price limit to enforce for acceptance into the pool",
@@ -831,6 +835,9 @@ func setGPO(ctx *cli.Context, cfg *gasprice.Config) {
 }
 
 func setTxPool(ctx *cli.Context, cfg *core.TxPoolConfig) {
+	if ctx.GlobalIsSet(TxPoolNoLocalsFlag.Name) {
+		cfg.NoLocals = ctx.GlobalBool(TxPoolNoLocalsFlag.Name)
+	}
 	if ctx.GlobalIsSet(TxPoolPriceLimitFlag.Name) {
 		cfg.PriceLimit = ctx.GlobalUint64(TxPoolPriceLimitFlag.Name)
 	}

core/tx_list.go

@@ -420,18 +420,16 @@ 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 *txSet) types.Transactions {
+func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) 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(*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
 		}
@@ -440,7 +438,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions
 			break
 		}
 		// Non stale transaction found, discard unless local
-		if local.contains(hash) {
+		if local.containsTx(tx) {
 			save = append(save, tx)
 		} else {
 			drop = append(drop, tx)
@@ -454,9 +452,9 @@ func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions
 
 // Underpriced checks whether a transaction is cheaper than (or as cheap as) the
 // lowest priced transaction currently being tracked.
-func (l *txPricedList) Underpriced(tx *types.Transaction, local *txSet) bool {
+func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) bool {
 	// Local transactions cannot be underpriced
-	if local.contains(tx.Hash()) {
+	if local.containsTx(tx) {
 		return false
 	}
 	// Discard stale price points if found at the heap start
@@ -479,22 +477,20 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *txSet) bool {
 }
 
 // Discard finds a number of most underpriced transactions, removes them from the
-// priced list and returs them for further removal from the entire pool.
-func (l *txPricedList) Discard(count int, local *txSet) types.Transactions {
+// priced list and returns them for further removal from the entire pool.
+func (l *txPricedList) Discard(count int, local *accountSet) 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(*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(hash) {
+		if local.containsTx(tx) {
 			save = append(save, tx)
 		} else {
 			drop = append(drop, tx)

eth/api_backend.go

@@ -116,29 +116,18 @@ func (b *EthApiBackend) GetEVM(ctx context.Context, msg core.Message, state *sta
 }
 
 func (b *EthApiBackend) SendTx(ctx context.Context, signedTx *types.Transaction) error {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
-	b.eth.txPool.SetLocal(signedTx)
-	return b.eth.txPool.Add(signedTx)
+	return b.eth.txPool.AddLocal(signedTx)
 }
 
 func (b *EthApiBackend) RemoveTx(txHash common.Hash) {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
 	b.eth.txPool.Remove(txHash)
 }
 
 func (b *EthApiBackend) GetPoolTransactions() (types.Transactions, error) {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
 	pending, err := b.eth.txPool.Pending()
 	if err != nil {
 		return nil, err
 	}
-
 	var txs types.Transactions
 	for _, batch := range pending {
 		txs = append(txs, batch...)
@@ -147,30 +136,18 @@ func (b *EthApiBackend) GetPoolTransactions() (types.Transactions, error) {
 }
 
 func (b *EthApiBackend) GetPoolTransaction(hash common.Hash) *types.Transaction {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
 	return b.eth.txPool.Get(hash)
 }
 
 func (b *EthApiBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
 	return b.eth.txPool.State().GetNonce(addr), nil
 }
 
 func (b *EthApiBackend) Stats() (pending int, queued int) {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
 	return b.eth.txPool.Stats()
 }
 
 func (b *EthApiBackend) TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {
-	b.eth.txMu.Lock()
-	defer b.eth.txMu.Unlock()
-
 	return b.eth.TxPool().Content()
 }
 

eth/backend.go

@@ -63,7 +63,6 @@ type Ethereum struct {
 	stopDbUpgrade func()    // stop chain db sequential key upgrade
 	// Handlers
 	txPool          *core.TxPool
-	txMu            sync.Mutex
 	blockchain      *core.BlockChain
 	protocolManager *ProtocolManager
 	lesServer       LesServer

eth/handler.go

@@ -658,7 +658,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			}
 			p.MarkTransaction(tx.Hash())
 		}
-		pm.txpool.AddBatch(txs)
+		pm.txpool.AddRemotes(txs)
 
 	default:
 		return errResp(ErrInvalidMsgCode, "%v", msg.Code)

eth/helper_test.go

@@ -94,9 +94,9 @@ type testTxPool struct {
 	lock sync.RWMutex // Protects the transaction pool
 }
 
-// AddBatch appends a batch of transactions to the pool, and notifies any
+// AddRemotes appends a batch of transactions to the pool, and notifies any
 // listeners if the addition channel is non nil
-func (p *testTxPool) AddBatch(txs []*types.Transaction) error {
+func (p *testTxPool) AddRemotes(txs []*types.Transaction) error {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 

eth/protocol.go

@@ -94,8 +94,8 @@ var errorToString = map[int]string{
 }
 
 type txPool interface {
-	// AddBatch should add the given transactions to the pool.
-	AddBatch([]*types.Transaction) error
+	// AddRemotes should add the given transactions to the pool.
+	AddRemotes([]*types.Transaction) error
 
 	// Pending should return pending transactions.
 	// The slice should be modifiable by the caller.

eth/protocol_test.go

@@ -130,7 +130,7 @@ func testSendTransactions(t *testing.T, protocol int) {
 	for nonce := range alltxs {
 		alltxs[nonce] = newTestTransaction(testAccount, uint64(nonce), txsize)
 	}
-	pm.txpool.AddBatch(alltxs)
+	pm.txpool.AddRemotes(alltxs)
 
 	// Connect several peers. They should all receive the pending transactions.
 	var wg sync.WaitGroup

les/handler.go

@@ -87,8 +87,8 @@ type BlockChain interface {
 }
 
 type txPool interface {
-	// AddTransactions should add the given transactions to the pool.
-	AddBatch([]*types.Transaction) error
+	// AddRemotes should add the given transactions to the pool.
+	AddRemotes([]*types.Transaction) error
 }
 
 type ProtocolManager struct {
@@ -801,7 +801,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrRequestRejected, "")
 		}
 
-		if err := pm.txpool.AddBatch(txs); err != nil {
+		if err := pm.txpool.AddRemotes(txs); err != nil {
 			return errResp(ErrUnexpectedResponse, "msg: %v", err)
 		}