eth, p2p: reserve half peer slots for snap peers during snap sync (#22171)

* eth, p2p: reserve half peer slots for snap peers during snap sync

* eth: less logging

* eth: rework the eth/snap peer reservation logic

* eth: rework the eth/snap peer reservation logic (again)

eth/handler.go

@@ -250,9 +250,20 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
 		peer.Log().Debug("Ethereum handshake failed", "err", err)
 		return err
 	}
+	reject := false // reserved peer slots
+	if atomic.LoadUint32(&h.snapSync) == 1 && !peer.SupportsCap("snap", 1) {
+		// If we are running snap-sync, we want to reserve roughly half the peer
+		// slots for peers supporting the snap protocol.
+		// The logic here is; we only allow up to 5 more non-snap peers than snap-peers.
+		if all, snp := h.peers.Len(), h.peers.SnapLen(); all-snp > snp+5 {
+			reject = true
+		}
+	}
 	// Ignore maxPeers if this is a trusted peer
-	if h.peers.Len() >= h.maxPeers && !peer.Peer.Info().Network.Trusted {
-		return p2p.DiscTooManyPeers
+	if !peer.Peer.Info().Network.Trusted {
+		if reject || h.peers.Len() >= h.maxPeers {
+			return p2p.DiscTooManyPeers
+		}
 	}
 	peer.Log().Debug("Ethereum peer connected", "name", peer.Name())
 

eth/peerset.go

@@ -259,7 +259,7 @@ func (ps *peerSet) ethPeersWithoutTransaction(hash common.Hash) []*ethPeer {
 }
 
 // Len returns if the current number of `eth` peers in the set. Since the `snap`
-// peers are tied to the existnce of an `eth` connection, that will always be a
+// peers are tied to the existence of an `eth` connection, that will always be a
 // subset of `eth`.
 func (ps *peerSet) Len() int {
 	ps.lock.RLock()
@@ -268,6 +268,15 @@ func (ps *peerSet) Len() int {
 	return len(ps.ethPeers)
 }
 
+// SnapLen returns if the current number of `snap` peers in the set. Since the `snap`
+// peers are tied to the existence of an `eth` connection, that will always be a
+// subset of `eth`.
+func (ps *peerSet) SnapLen() int {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+	return len(ps.snapPeers)
+}
+
 // ethPeerWithHighestTD retrieves the known peer with the currently highest total
 // difficulty.
 func (ps *peerSet) ethPeerWithHighestTD() *eth.Peer {

eth/sync.go

@@ -329,6 +329,10 @@ func (h *handler) doSync(op *chainSyncOp) error {
 		log.Info("Fast sync complete, auto disabling")
 		atomic.StoreUint32(&h.fastSync, 0)
 	}
+	if atomic.LoadUint32(&h.snapSync) == 1 {
+		log.Info("Snap sync complete, auto disabling")
+		atomic.StoreUint32(&h.snapSync, 0)
+	}
 	// If we've successfully finished a sync cycle and passed any required checkpoint,
 	// enable accepting transactions from the network.
 	head := h.chain.CurrentBlock()

p2p/peer.go

@@ -158,6 +158,16 @@ func (p *Peer) Caps() []Cap {
 	return p.rw.caps
 }
 
+// SupportsCap returns true if the peer supports the given protocol/version
+func (p *Peer) SupportsCap(protocol string, version uint) bool {
+	for _, cap := range p.rw.caps {
+		if cap.Name == protocol {
+			return version <= cap.Version
+		}
+	}
+	return false
+}
+
 // RemoteAddr returns the remote address of the network connection.
 func (p *Peer) RemoteAddr() net.Addr {
 	return p.rw.fd.RemoteAddr()