diff --git a/eth/handler.go b/eth/handler.go index 6404104a79..20a3a5af68 100644 --- a/eth/handler.go +++ b/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,33 +502,21 @@ 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() - hasher.Write(h.nodeID.Bytes()) - hasher.Write(peer.Node().ID().Bytes()) - - from, _ := types.Sender(signer, tx) // Ignore error, we only use the addr as a propagation target splitter - hasher.Write(from.Bytes()) - - hasher.Read(hash) - if new(big.Int).Mod(new(big.Int).SetBytes(hash), total).Cmp(direct) < 0 { - broadcast = true + // Get transaction sender address. Here we can ignore any error + // since we're just interested in any value. + txSender, _ := types.Sender(signer, tx) + if choice.shouldBroadcastTx(h.nodeID, peer.Peer.Peer.ID(), txSender) { + // Send directly to peer. + txset[peer] = append(txset[peer], tx.Hash()) + continue } } - if broadcast { - txset[peer] = append(txset[peer], tx.Hash()) - } else { - annos[peer] = append(annos[peer], tx.Hash()) - } + + // Send announcement to peer. + annos[peer] = append(annos[peer], tx.Hash()) } } for peer, hashes := range txset { @@ -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 +} diff --git a/eth/handler_test.go b/eth/handler_test.go index d0da098430..2bd538a3cc 100644 --- a/eth/handler_test.go +++ b/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 +}