|
| 1 | +// Copyright 2025 The go-ethereum Authors |
| 2 | +// This file is part of the go-ethereum library. |
| 3 | +// |
| 4 | +// The go-ethereum library is free software: you can redistribute it and/or modify |
| 5 | +// it under the terms of the GNU Lesser General Public License as published by |
| 6 | +// the Free Software Foundation, either version 3 of the License, or |
| 7 | +// (at your option) any later version. |
| 8 | +// |
| 9 | +// The go-ethereum library is distributed in the hope that it will be useful, |
| 10 | +// but WITHOUT ANY WARRANTY; without even the implied warranty of |
| 11 | +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| 12 | +// GNU Lesser General Public License for more details. |
| 13 | +// |
| 14 | +// You should have received a copy of the GNU Lesser General Public License |
| 15 | +// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. |
| 16 | + |
| 17 | +package eth |
| 18 | + |
| 19 | +import ( |
| 20 | + mrand "math/rand" |
| 21 | + "slices" |
| 22 | + "sync" |
| 23 | + "time" |
| 24 | + |
| 25 | + "github.com/ethereum/go-ethereum/common" |
| 26 | + "github.com/ethereum/go-ethereum/common/mclock" |
| 27 | + "github.com/ethereum/go-ethereum/log" |
| 28 | + "github.com/ethereum/go-ethereum/metrics" |
| 29 | + "github.com/ethereum/go-ethereum/p2p" |
| 30 | +) |
| 31 | + |
| 32 | +const ( |
| 33 | + // Interval between peer drop events (uniform between min and max) |
| 34 | + peerDropIntervalMin = 3 * time.Minute |
| 35 | + // Interval between peer drop events (uniform between min and max) |
| 36 | + peerDropIntervalMax = 7 * time.Minute |
| 37 | + // Avoid dropping peers for some time after connection |
| 38 | + doNotDropBefore = 10 * time.Minute |
| 39 | + // How close to max should we initiate the drop timer. O should be fine, |
| 40 | + // dropping when no more peers can be added. Larger numbers result in more |
| 41 | + // aggressive drop behavior. |
| 42 | + peerDropThreshold = 0 |
| 43 | +) |
| 44 | + |
| 45 | +var ( |
| 46 | + // droppedInbound is the number of inbound peers dropped |
| 47 | + droppedInbound = metrics.NewRegisteredMeter("eth/dropper/inbound", nil) |
| 48 | + // droppedOutbound is the number of outbound peers dropped |
| 49 | + droppedOutbound = metrics.NewRegisteredMeter("eth/dropper/outbound", nil) |
| 50 | +) |
| 51 | + |
| 52 | +// dropper monitors the state of the peer pool and makes changes as follows: |
| 53 | +// - during sync the Downloader handles peer connections, so dropper is disabled |
| 54 | +// - if not syncing and the peer count is close to the limit, it drops peers |
| 55 | +// randomly every peerDropInterval to make space for new peers |
| 56 | +// - peers are dropped separately from the inboud pool and from the dialed pool |
| 57 | +type dropper struct { |
| 58 | + maxDialPeers int // maximum number of dialed peers |
| 59 | + maxInboundPeers int // maximum number of inbound peers |
| 60 | + peersFunc getPeersFunc |
| 61 | + syncingFunc getSyncingFunc |
| 62 | + |
| 63 | + // peerDropTimer introduces churn if we are close to limit capacity. |
| 64 | + // We handle Dialed and Inbound connections separately |
| 65 | + peerDropTimer *time.Timer |
| 66 | + |
| 67 | + wg sync.WaitGroup // wg for graceful shutdown |
| 68 | + shutdownCh chan struct{} |
| 69 | +} |
| 70 | + |
| 71 | +// Callback type to get the list of connected peers. |
| 72 | +type getPeersFunc func() []*p2p.Peer |
| 73 | + |
| 74 | +// Callback type to get syncing status. |
| 75 | +// Returns true while syncing, false when synced. |
| 76 | +type getSyncingFunc func() bool |
| 77 | + |
| 78 | +func newDropper(maxDialPeers, maxInboundPeers int) *dropper { |
| 79 | + cm := &dropper{ |
| 80 | + maxDialPeers: maxDialPeers, |
| 81 | + maxInboundPeers: maxInboundPeers, |
| 82 | + peerDropTimer: time.NewTimer(randomDuration(peerDropIntervalMin, peerDropIntervalMax)), |
| 83 | + shutdownCh: make(chan struct{}), |
| 84 | + } |
| 85 | + if peerDropIntervalMin > peerDropIntervalMax { |
| 86 | + panic("peerDropIntervalMin duration must be less than or equal to peerDropIntervalMax duration") |
| 87 | + } |
| 88 | + return cm |
| 89 | +} |
| 90 | + |
| 91 | +// Start the dropper. |
| 92 | +func (cm *dropper) Start(srv *p2p.Server, syncingFunc getSyncingFunc) { |
| 93 | + cm.peersFunc = srv.Peers |
| 94 | + cm.syncingFunc = syncingFunc |
| 95 | + cm.wg.Add(1) |
| 96 | + go cm.loop() |
| 97 | +} |
| 98 | + |
| 99 | +// Stop the dropper. |
| 100 | +func (cm *dropper) Stop() { |
| 101 | + cm.peerDropTimer.Stop() |
| 102 | + close(cm.shutdownCh) |
| 103 | + cm.wg.Wait() |
| 104 | +} |
| 105 | + |
| 106 | +// dropRandomPeer selects one of the peers randomly and drops it from the peer pool. |
| 107 | +func (cm *dropper) dropRandomPeer() bool { |
| 108 | + peers := cm.peersFunc() |
| 109 | + var numInbound int |
| 110 | + for _, p := range peers { |
| 111 | + if p.Inbound() { |
| 112 | + numInbound++ |
| 113 | + } |
| 114 | + } |
| 115 | + numDialed := len(peers) - numInbound |
| 116 | + |
| 117 | + selectDoNotDrop := func(p *p2p.Peer) bool { |
| 118 | + // Avoid dropping trusted and static peers, or recent peers. |
| 119 | + // Only drop peers if their respective category (dialed/inbound) |
| 120 | + // is close to limit capacity. |
| 121 | + return p.Trusted() || p.StaticDialed() || |
| 122 | + p.Lifetime() < mclock.AbsTime(doNotDropBefore) || |
| 123 | + (p.DynDialed() && cm.maxDialPeers-numDialed > peerDropThreshold) || |
| 124 | + (p.Inbound() && cm.maxInboundPeers-numInbound > peerDropThreshold) |
| 125 | + } |
| 126 | + |
| 127 | + droppable := slices.DeleteFunc(peers, selectDoNotDrop) |
| 128 | + if len(droppable) > 0 { |
| 129 | + p := droppable[mrand.Intn(len(droppable))] |
| 130 | + log.Debug("Dropping random peer", "inbound", p.Inbound(), |
| 131 | + "id", p.ID(), "duration", common.PrettyDuration(p.Lifetime()), "peercountbefore", len(peers)) |
| 132 | + p.Disconnect(p2p.DiscUselessPeer) |
| 133 | + if p.Inbound() { |
| 134 | + droppedInbound.Mark(1) |
| 135 | + } else { |
| 136 | + droppedOutbound.Mark(1) |
| 137 | + } |
| 138 | + return true |
| 139 | + } |
| 140 | + return false |
| 141 | +} |
| 142 | + |
| 143 | +// randomDuration generates a random duration between min and max. |
| 144 | +func randomDuration(min, max time.Duration) time.Duration { |
| 145 | + if min > max { |
| 146 | + panic("min duration must be less than or equal to max duration") |
| 147 | + } |
| 148 | + return time.Duration(mrand.Int63n(int64(max-min)) + int64(min)) |
| 149 | +} |
| 150 | + |
| 151 | +// loop is the main loop of the connection dropper. |
| 152 | +func (cm *dropper) loop() { |
| 153 | + defer cm.wg.Done() |
| 154 | + |
| 155 | + for { |
| 156 | + select { |
| 157 | + case <-cm.peerDropTimer.C: |
| 158 | + // Drop a random peer if we are not syncing and the peer count is close to the limit. |
| 159 | + if !cm.syncingFunc() { |
| 160 | + cm.dropRandomPeer() |
| 161 | + } |
| 162 | + cm.peerDropTimer.Reset(randomDuration(peerDropIntervalMin, peerDropIntervalMax)) |
| 163 | + case <-cm.shutdownCh: |
| 164 | + return |
| 165 | + } |
| 166 | + } |
| 167 | +} |
0 commit comments