Accept transactions from peers on writeblock and send all transaction…

…s to all peers (including queued transactions)
ayhanfener · Mar 17, 2023 · c06c5a2 · c06c5a2
1 parent 0b09e87
commit c06c5a2
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 19 deletions.
diff --git a/core/tx_pool.go b/core/tx_pool.go
@@ -738,7 +738,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 		localGauge.Inc(1)
 	}
 	pool.journalTx(from, tx)
-
+	pool.queueTxEvent(tx)
 	log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
 	return replaced, nil
 }

diff --git a/core/types/internal_tx.go b/core/types/internal_tx.go
@@ -82,23 +82,23 @@ func (tx *InternalTx) copy() TxData {
 }
 
 // accessors for innerTx.
-func (tx *InternalTx) txType() byte                { return InternalTxType }
-func (tx *InternalTx) chainID() *big.Int           { return tx.ChainID }
-func (tx *InternalTx) protected() bool             { return true }
-func (tx *InternalTx) accessList() AccessList      { return tx.AccessList }
-func (tx *InternalTx) data() []byte                { return tx.Data }
-func (tx *InternalTx) gas() uint64                 { return tx.Gas }
-func (tx *InternalTx) gasFeeCap() *big.Int         { return tx.GasFeeCap }
-func (tx *InternalTx) gasTipCap() *big.Int         { return tx.GasTipCap }
-func (tx *InternalTx) gasPrice() *big.Int          { return tx.GasFeeCap }
-func (tx *InternalTx) value() *big.Int             { return tx.Value }
-func (tx *InternalTx) nonce() uint64               { return tx.Nonce }
-func (tx *InternalTx) to() *common.Address         { return tx.To }
-func (tx *InternalTx) etxGasLimit() uint64         { panic("internal TX does not have etxGasLimit") }
-func (tx *InternalTx) etxGasPrice() *big.Int       { panic("internal TX does not have etxGasPrice") }
-func (tx *InternalTx) etxGasTip() *big.Int         { panic("internal TX does not have etxGasTip") }
-func (tx *InternalTx) etxData() []byte	           { panic("internal TX does not have etxData") }
-func (tx *InternalTx) etxAccessList() AccessList   { panic("internal TX does not have etxAccessList") }
+func (tx *InternalTx) txType() byte              { return InternalTxType }
+func (tx *InternalTx) chainID() *big.Int         { return tx.ChainID }
+func (tx *InternalTx) protected() bool           { return true }
+func (tx *InternalTx) accessList() AccessList    { return tx.AccessList }
+func (tx *InternalTx) data() []byte              { return tx.Data }
+func (tx *InternalTx) gas() uint64               { return tx.Gas }
+func (tx *InternalTx) gasFeeCap() *big.Int       { return tx.GasFeeCap }
+func (tx *InternalTx) gasTipCap() *big.Int       { return tx.GasTipCap }
+func (tx *InternalTx) gasPrice() *big.Int        { return tx.GasFeeCap }
+func (tx *InternalTx) value() *big.Int           { return tx.Value }
+func (tx *InternalTx) nonce() uint64             { return tx.Nonce }
+func (tx *InternalTx) to() *common.Address       { return tx.To }
+func (tx *InternalTx) etxGasLimit() uint64       { panic("internal TX does not have etxGasLimit") }
+func (tx *InternalTx) etxGasPrice() *big.Int     { panic("internal TX does not have etxGasPrice") }
+func (tx *InternalTx) etxGasTip() *big.Int       { panic("internal TX does not have etxGasTip") }
+func (tx *InternalTx) etxData() []byte           { panic("internal TX does not have etxData") }
+func (tx *InternalTx) etxAccessList() AccessList { panic("internal TX does not have etxAccessList") }
 
 func (tx *InternalTx) rawSignatureValues() (v, r, s *big.Int) {
 	return tx.V, tx.R, tx.S

diff --git a/eth/handler.go b/eth/handler.go
@@ -21,6 +21,7 @@ import (
 	"math"
 	"math/rand"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/dominant-strategies/go-quai/common"
@@ -59,6 +60,9 @@ const (
 	// sqrt of len(peers) is less than minPeerRequest we make the body request
 	// to as much as minPeerSend peers otherwise send it to sqrt of len(peers).
 	minPeerRequest = 3
+
+	// minPeerSendTx is the minimum number of peers that will receive a new transaction.
+	minPeerSendTx = 2
 )
 
 // txPool defines the methods needed from a transaction pool implementation to
@@ -165,6 +169,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
 	}
 	// writeBlock writes the block to the DB
 	writeBlock := func(block *types.Block) {
+		atomic.StoreUint32(&h.acceptTxs, 1)
 		h.core.WriteBlock(block)
 	}
 	h.blockFetcher = fetcher.NewBlockFetcher(h.core.GetBlockByHash, writeBlock, validator, h.BroadcastBlock, heighter, h.removePeer)
@@ -388,7 +393,12 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		peers := h.peers.peersWithoutTransaction(tx.Hash())
 		// Send the tx unconditionally to a subset of our peers
 		numDirect := int(math.Sqrt(float64(len(peers))))
-		for _, peer := range peers[:numDirect] {
+		subset := peers[:numDirect]
+		if len(subset) < minPeerSendTx {
+			// If our subset is less than the minimum, send to the minimum
+			subset = peers[:minPeerSendTx+1] // The high bound is exclusive
+		}
+		for _, peer := range subset {
 			txset[peer] = append(txset[peer], tx.Hash())
 		}
 		// For the remaining peers, send announcement only