eth: implement new broadcast choice

eth/handler.go

@@ -17,9 +17,12 @@
 package eth
 
 import (
+	"bytes"
+	"crypto/sha256"
 	"errors"
+	"hash"
 	"maps"
-	"math"
+	gmath "math"
 	"math/big"
 	"slices"
 	"sync"
@@ -27,11 +30,11 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/math"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/eth/fetcher"
@@ -484,29 +487,11 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 
 		txset = make(map[*ethPeer][]common.Hash) // Set peer->hash to transfer directly
 		annos = make(map[*ethPeer][]common.Hash) // Set peer->hash to announce
+
+		signer = types.LatestSigner(h.chain.Config())
+		choice = newBroadcastChoice(h.peers.len())
 	)
-	// Broadcast transactions to a batch of peers not knowing about it
-	sqrtPeers := math.Sqrt(float64(h.peers.len())) // Approximate number of peers to broadcast to
 
-	// Use some hysteresis to avoid oscillating between two values, stabilising the modulus in the peer selection
-	// If the number of peers is small, use a minimum of 1 peer
-	var directInt int64
-	lastDirect := h.lastDirect.Load()
-	if int64(sqrtPeers) >= lastDirect {
-		directInt = max(int64(sqrtPeers), 1)
-	} else {
-		directInt = max(min(int64(sqrtPeers+directPeersHysteresis), lastDirect), 1)
-	}
-	h.lastDirect.Store(directInt)
-
-	direct := big.NewInt(directInt)            // Number of peers to send directly to
-	total := big.NewInt(directInt * directInt) // Stabilise total peer count a bit based on sqrt peers
-
-	var (
-		signer = types.LatestSigner(h.chain.Config()) // Don't care about chain status, we just need *a* sender
-		hasher = crypto.NewKeccakState()
-		hash   = make([]byte, 32)
-	)
 	for _, tx := range txs {
 		var maybeDirect bool
 		switch {
@@ -517,34 +502,22 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		default:
 			maybeDirect = true
 		}
-		// Send the transaction (if it's small enough) directly to a subset of
+
-		// the peers that have not received it yet, ensuring that the flow of
-		// transactions is grouped by account to (try and) avoid nonce gaps.
-		//
-		// To do this, we hash the local enode IW with together with a peer's
-		// enode ID together with the transaction sender and broadcast if
-		// `sha(self, peer, sender) mod peers < sqrt(peers)`.
 		for _, peer := range h.peers.peersWithoutTransaction(tx.Hash()) {
-			var broadcast bool
 			if maybeDirect {
-				hasher.Reset()
+				// Get transaction sender address. Here we can ignore any error
-				hasher.Write(h.nodeID.Bytes())
+				// since we're just interested in any value.
-				hasher.Write(peer.Node().ID().Bytes())
+				txSender, _ := types.Sender(signer, tx)
-
+				if choice.shouldBroadcastTx(h.nodeID, peer.Peer.Peer.ID(), txSender) {
-				from, _ := types.Sender(signer, tx) // Ignore error, we only use the addr as a propagation target splitter
+					// Send directly to peer.
-				hasher.Write(from.Bytes())
-
-				hasher.Read(hash)
-				if new(big.Int).Mod(new(big.Int).SetBytes(hash), total).Cmp(direct) < 0 {
-					broadcast = true
-				}
-			}
-			if broadcast {
 					txset[peer] = append(txset[peer], tx.Hash())
-			} else {
+					continue
-				annos[peer] = append(annos[peer], tx.Hash())
 				}
 			}
+
+			// Send announcement to peer.
+			annos[peer] = append(annos[peer], tx.Hash())
+		}
 	}
 	for peer, hashes := range txset {
 		directCount += len(hashes)
@@ -710,3 +683,44 @@ func (st *blockRangeState) stop() {
 func (st *blockRangeState) currentRange() eth.BlockRangeUpdatePacket {
 	return *st.next.Load()
 }
+
+// Send the transaction (if it's small enough) directly to a subset of
+// the peers that have not received it yet, ensuring that the flow of
+// transactions is grouped by account to (try and) avoid nonce gaps.
+//
+// To do this, we hash the local node ID together with a peer's
+// node ID together with the transaction sender and broadcast if
+// `sha(self, peer, sender) mod peers < sqrt(peers)`.
+
+type broadcastChoice struct {
+	hash      hash.Hash
+	threshold []byte
+	hashBytes []byte
+}
+
+func newBroadcastChoice(npeers int) *broadcastChoice {
+	var bc broadcastChoice
+	bc.hash = sha256.New()
+	bc.hashBytes = make([]byte, 32)
+
+	// compute the hash comparison threshold for one peer
+	unit := math.BigPow(2, 256)
+	unit.Sub(unit, common.Big1)
+	unit.Div(unit, big.NewInt(int64(max(1, npeers))))
+
+	// compute the threshold for n peers
+	sqrtp := max(gmath.Sqrt(float64(npeers)), 1)
+	thr := big.NewInt(int64(gmath.Ceil(sqrtp)))
+	thr.Mul(thr, unit)
+	bc.threshold = thr.FillBytes(make([]byte, 32))
+	return &bc
+}
+
+func (bc *broadcastChoice) shouldBroadcastTx(self, peer enode.ID, txSender common.Address) bool {
+	bc.hash.Reset()
+	bc.hash.Write(self[:])
+	bc.hash.Write(peer[:])
+	bc.hash.Write(txSender[:])
+	bc.hash.Sum(bc.hashBytes[:0])
+	return bytes.Compare(bc.hashBytes, bc.threshold) < 0
+}

eth/handler_test.go

@@ -18,8 +18,10 @@ package eth
 
 import (
 	"math/big"
+	"math/rand"
 	"sort"
 	"sync"
+	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/ethash"
@@ -31,6 +33,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/holiman/uint256"
@@ -212,3 +215,66 @@ func (b *testHandler) close() {
 	b.handler.Stop()
 	b.chain.Stop()
 }
+
+func TestBroadcastChoice(t *testing.T) {
+	choice49 := newBroadcastChoice(49)
+	choice50 := newBroadcastChoice(50)
+
+	var (
+		self      = enode.HexID("1111111111111111111111111111111111111111111111111111111111111111")
+		peers     = make([]enode.ID, 50)
+		txsenders = make([]common.Address, 400)
+		rand      = rand.New(rand.NewSource(33))
+	)
+	for i := range peers {
+		rand.Read(peers[i][:])
+	}
+	for i := range txsenders {
+		rand.Read(txsenders[i][:])
+	}
+
+	// Evaluate choice49 first.
+	var chosen49 = make([][]bool, len(txsenders))
+	for i, txSender := range txsenders {
+		chosen49[i] = make([]bool, len(peers))
+		for peerIndex, peer := range peers {
+			chosen49[i][peerIndex] = choice49.shouldBroadcastTx(self, peer, txSender)
+		}
+	}
+
+	// Sanity check choices.
+	for i := range chosen49 {
+		c := count(chosen49[i], true)
+		if c == 0 {
+			t.Errorf("for tx %d, choice49 chose zero peers", i)
+		}
+	}
+
+	// Evaluate choice50 for the same peers and transactions. It should always yield more
+	// peers than choice49, and the chosen set should be a superset of choice49's.
+	for i, txSender := range txsenders {
+		var chosen50 int
+		for peerIndex, peer := range peers {
+			send := choice50.shouldBroadcastTx(self, peer, txSender)
+			if chosen49[i][peerIndex] && !send {
+				t.Errorf("for tx %d, choice50 did not choose peer %d, but choice49 did", i, peerIndex)
+			}
+			if send {
+				chosen50++
+			}
+		}
+		if chosen50 < count(chosen49[i], true) {
+			t.Errorf("for tx %d, choice50 has less peers than choice49", i)
+		}
+	}
+}
+
+func count[T comparable](s []T, v T) int {
+	var c int
+	for _, elem := range s {
+		if elem == v {
+			c++
+		}
+	}
+	return c
+}