diff --git a/eth/handler.go b/eth/handler.go index 1c46c631ba..70ea674ebc 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -17,20 +17,18 @@ package eth import ( - "bytes" - "crypto/sha256" + "cmp" "errors" "hash" + "hash/fnv" "maps" gmath "math" - "math/big" "slices" "sync" "sync/atomic" "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" @@ -485,36 +483,38 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) { annos = make(map[*ethPeer][]common.Hash) // Set peer->hash to announce signer = types.LatestSigner(h.chain.Config()) - choice = newBroadcastChoice(h.peers.len()) + choice = newBroadcastChoice(h.nodeID) + peers = h.peers.all() ) for _, tx := range txs { - var maybeDirect bool + var directSet map[*ethPeer]struct{} switch { case tx.Type() == types.BlobTxType: blobTxs++ case tx.Size() > txMaxBroadcastSize: largeTxs++ default: - maybeDirect = true + // Get transaction sender address. Here we can ignore any error + // since we're just interested in any value. + txSender, _ := types.Sender(signer, tx) + directSet = choice.choosePeers(peers, txSender) } - for _, peer := range h.peers.peersWithoutTransaction(tx.Hash()) { - if maybeDirect { - // 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 - } + for _, peer := range peers { + if peer.KnownTransaction(tx.Hash()) { + continue + } + if _, ok := directSet[peer]; ok { + // Send direct. + txset[peer] = append(txset[peer], tx.Hash()) + } else { + // Send announcement. + 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) peer.AsyncSendTransactions(hashes) @@ -689,34 +689,49 @@ func (st *blockRangeState) currentRange() eth.BlockRangeUpdatePacket { // `sha(self, peer, sender) mod peers < sqrt(peers)`. type broadcastChoice struct { - hash hash.Hash - threshold []byte - hashBytes []byte + self enode.ID + hash hash.Hash64 + buffer map[*ethPeer]struct{} + tmp []broadcastPeer } -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 +type broadcastPeer struct { + p *ethPeer + score uint64 } -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 +func newBroadcastChoice(self enode.ID) *broadcastChoice { + return &broadcastChoice{ + self: self, + hash: fnv.New64(), + buffer: make(map[*ethPeer]struct{}), + } +} + +// choosePeers selects the peers that will receive a direct transaction broadcast message. +// +// Note the return value will only stay valid until the next call to choosePeers. +func (bc *broadcastChoice) choosePeers(peers []*ethPeer, txSender common.Address) map[*ethPeer]struct{} { + // Compute scores. + bc.tmp = bc.tmp[:0] + for _, peer := range peers { + bc.hash.Reset() + bc.hash.Write(bc.self[:]) + bc.hash.Write(peer.Peer.Peer.ID().Bytes()) + bc.hash.Write(txSender[:]) + bc.tmp = append(bc.tmp, broadcastPeer{peer, bc.hash.Sum64()}) + } + + // Sort by score. + slices.SortFunc(bc.tmp, func(a, b broadcastPeer) int { + return cmp.Compare(a.score, b.score) + }) + + // Take top n. + clear(bc.buffer) + n := int(gmath.Ceil(gmath.Sqrt(float64(len(bc.tmp))))) + for i := range n { + bc.buffer[bc.tmp[i].p] = struct{}{} + } + return bc.buffer } diff --git a/eth/handler_test.go b/eth/handler_test.go index 2bd538a3cc..218a167cc0 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -17,6 +17,7 @@ package eth import ( + "maps" "math/big" "math/rand" "sort" @@ -31,8 +32,10 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth/ethconfig" + "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" @@ -217,64 +220,63 @@ func (b *testHandler) close() { } func TestBroadcastChoice(t *testing.T) { - choice49 := newBroadcastChoice(49) - choice50 := newBroadcastChoice(50) + // Create choices. + self := enode.HexID("1111111111111111111111111111111111111111111111111111111111111111") + choice49 := newBroadcastChoice(self) + choice50 := newBroadcastChoice(self) + // Create test peers. var ( - self = enode.HexID("1111111111111111111111111111111111111111111111111111111111111111") - peers = make([]enode.ID, 50) - txsenders = make([]common.Address, 400) - rand = rand.New(rand.NewSource(33)) + rand = rand.New(rand.NewSource(33)) + peers = make([]*ethPeer, 50) ) for i := range peers { - rand.Read(peers[i][:]) + var id enode.ID + rand.Read(id[:]) + p2pPeer := p2p.NewPeer(id, "test", nil) + ep := eth.NewPeer(eth.ETH69, p2pPeer, nil, nil) + peers[i] = ðPeer{Peer: ep} } + defer func() { + for _, p := range peers { + p.Close() + } + }() + + // Create random tx sender addresses. + txsenders := make([]common.Address, 400) for i := range txsenders { rand.Read(txsenders[i][:]) } // Evaluate choice49 first. - var chosen49 = make([][]bool, len(txsenders)) + expectedCount := 7 // sqrt(49) + var chosen49 = make([]map[*ethPeer]struct{}, 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) - } - } + set := choice49.choosePeers(peers[:49], txSender) + chosen49[i] = maps.Clone(set) - // 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) + // Sanity check choices. Here we check that the function selects different peers + // for different transaction senders. + if len(set) != expectedCount { + t.Fatalf("choice49 produced wrong count %d, want %d", len(set), expectedCount) + } + if i > 0 && maps.Equal(set, chosen49[i-1]) { + t.Errorf("choice49 for tx %d is equal to tx %d", i, i-1) } } // 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) { + set := choice50.choosePeers(peers[:50], txSender) + if len(set) < len(chosen49[i]) { 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++ + for p := range chosen49[i] { + if _, ok := set[p]; !ok { + t.Errorf("for tx %d, choice50 did not choose peer %v, but choice49 did", i, p.ID()) + } } } - return c } diff --git a/eth/peerset.go b/eth/peerset.go index 6b0aff226c..e6f623f90c 100644 --- a/eth/peerset.go +++ b/eth/peerset.go @@ -19,9 +19,10 @@ package eth import ( "errors" "fmt" + "maps" + "slices" "sync" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/eth/protocols/snap" "github.com/ethereum/go-ethereum/p2p" @@ -191,19 +192,12 @@ func (ps *peerSet) peer(id string) *ethPeer { return ps.peers[id] } -// peersWithoutTransaction retrieves a list of peers that do not have a given -// transaction in their set of known hashes. -func (ps *peerSet) peersWithoutTransaction(hash common.Hash) []*ethPeer { +// all returns all current peers. +func (ps *peerSet) all() []*ethPeer { ps.lock.RLock() defer ps.lock.RUnlock() - list := make([]*ethPeer, 0, len(ps.peers)) - for _, p := range ps.peers { - if !p.KnownTransaction(hash) { - list = append(list, p) - } - } - return list + return slices.Collect(maps.Values(ps.peers)) } // len returns if the current number of `eth` peers in the set. Since the `snap`