|
1 | 1 | //! Pieces shared across the blocking runtimes. |
2 | 2 |
|
| 3 | +use std::collections::HashMap; |
3 | 4 | use std::io::{self, Read, Write}; |
4 | | -use std::net::TcpStream; |
5 | | -use std::sync::Mutex; |
| 5 | +use std::net::{IpAddr, TcpStream}; |
| 6 | +use std::sync::{Arc, Mutex}; |
6 | 7 | use std::time::{Duration, Instant}; |
7 | 8 |
|
8 | 9 | use crate::error::Result; |
9 | 10 | use crate::session::Session; |
10 | 11 |
|
| 12 | +/// Bounds concurrent connections per client IP so one source cannot consume the |
| 13 | +/// whole global connection budget. A `max_per_ip` of 0 disables limiting. |
| 14 | +pub(crate) struct PeerLimiter { |
| 15 | + counts: Mutex<HashMap<IpAddr, u32>>, |
| 16 | + max_per_ip: u32, |
| 17 | +} |
| 18 | + |
| 19 | +impl PeerLimiter { |
| 20 | + pub(crate) fn new(max_per_ip: u32) -> Arc<PeerLimiter> { |
| 21 | + Arc::new(PeerLimiter { |
| 22 | + counts: Mutex::new(HashMap::new()), |
| 23 | + max_per_ip, |
| 24 | + }) |
| 25 | + } |
| 26 | + |
| 27 | + /// Admit a connection from `ip`. `ip == None` (peer address unavailable) |
| 28 | + /// always admits — it cannot be attributed, so the global caps govern it. |
| 29 | + /// Returns a guard that releases the slot on drop, or `None` when `ip` is at |
| 30 | + /// the per-IP cap (caller must drop the connection). |
| 31 | + pub(crate) fn admit(self: &Arc<Self>, ip: Option<IpAddr>) -> Option<PeerGuard> { |
| 32 | + let Some(ip) = ip.filter(|_| self.max_per_ip > 0) else { |
| 33 | + return Some(PeerGuard { limiter: None }); |
| 34 | + }; |
| 35 | + let mut counts = self.counts.lock().unwrap_or_else(|e| e.into_inner()); |
| 36 | + let slot = counts.entry(ip).or_insert(0); |
| 37 | + if *slot >= self.max_per_ip { |
| 38 | + return None; |
| 39 | + } |
| 40 | + *slot += 1; |
| 41 | + Some(PeerGuard { |
| 42 | + limiter: Some((Arc::clone(self), ip)), |
| 43 | + }) |
| 44 | + } |
| 45 | +} |
| 46 | + |
| 47 | +/// Releases a per-IP connection slot when dropped. A `None` limiter is a no-op |
| 48 | +/// guard (limiting disabled or peer unattributable). |
| 49 | +pub(crate) struct PeerGuard { |
| 50 | + limiter: Option<(Arc<PeerLimiter>, IpAddr)>, |
| 51 | +} |
| 52 | + |
| 53 | +impl Drop for PeerGuard { |
| 54 | + fn drop(&mut self) { |
| 55 | + if let Some((limiter, ip)) = &self.limiter { |
| 56 | + let mut counts = limiter.counts.lock().unwrap_or_else(|e| e.into_inner()); |
| 57 | + if let Some(slot) = counts.get_mut(ip) { |
| 58 | + *slot -= 1; |
| 59 | + if *slot == 0 { |
| 60 | + counts.remove(ip); // don't leak entries for departed IPs |
| 61 | + } |
| 62 | + } |
| 63 | + } |
| 64 | + } |
| 65 | +} |
| 66 | + |
11 | 67 | /// Read buffer size used by the blocking drive loop. |
12 | 68 | pub(crate) const READ_BUF: usize = 16 * 1024; |
13 | 69 |
|
@@ -167,3 +223,60 @@ pub(crate) fn serve_blocking_prefed<S: Read + Write>( |
167 | 223 | } |
168 | 224 | Ok(()) |
169 | 225 | } |
| 226 | + |
| 227 | +#[cfg(test)] |
| 228 | +mod tests { |
| 229 | + use super::PeerLimiter; |
| 230 | + use std::net::{IpAddr, Ipv4Addr}; |
| 231 | + |
| 232 | + fn ip(n: u8) -> Option<IpAddr> { |
| 233 | + Some(IpAddr::V4(Ipv4Addr::new(10, 0, 0, n))) |
| 234 | + } |
| 235 | + |
| 236 | + #[test] |
| 237 | + fn caps_per_ip_and_releases_on_drop() { |
| 238 | + let limiter = PeerLimiter::new(2); |
| 239 | + |
| 240 | + // A cap of 2 admits two connections from the same IP... |
| 241 | + let g1 = limiter.admit(ip(1)).expect("first admits"); |
| 242 | + let g2 = limiter.admit(ip(1)).expect("second admits"); |
| 243 | + // ...then rejects the third. |
| 244 | + assert!(limiter.admit(ip(1)).is_none(), "third is over the cap"); |
| 245 | + |
| 246 | + // A different IP is unaffected by the first IP's usage. |
| 247 | + let other = limiter.admit(ip(2)).expect("distinct IP admits"); |
| 248 | + |
| 249 | + // Dropping a guard frees a slot for that IP. |
| 250 | + drop(g1); |
| 251 | + let g3 = limiter.admit(ip(1)).expect("freed slot re-admits"); |
| 252 | + |
| 253 | + drop((g2, g3, other)); |
| 254 | + |
| 255 | + // The map does not retain zero-count entries: once every guard for an IP |
| 256 | + // is dropped, admitting again must start from a fresh count. |
| 257 | + assert!( |
| 258 | + limiter.counts.lock().unwrap().is_empty(), |
| 259 | + "no zero-count entries retained after all guards drop" |
| 260 | + ); |
| 261 | + } |
| 262 | + |
| 263 | + #[test] |
| 264 | + fn cap_zero_always_admits() { |
| 265 | + let limiter = PeerLimiter::new(0); |
| 266 | + // Unlimited: many connections from one IP all admit, and no entries are |
| 267 | + // tracked at all. |
| 268 | + let guards: Vec<_> = (0..100).map(|_| limiter.admit(ip(1)).unwrap()).collect(); |
| 269 | + assert_eq!(guards.len(), 100); |
| 270 | + assert!(limiter.counts.lock().unwrap().is_empty()); |
| 271 | + } |
| 272 | + |
| 273 | + #[test] |
| 274 | + fn none_ip_always_admits() { |
| 275 | + let limiter = PeerLimiter::new(1); |
| 276 | + // An unattributable peer (no address) always admits, even past the cap, |
| 277 | + // and is not recorded in the map. |
| 278 | + let _a = limiter.admit(None).expect("None admits"); |
| 279 | + let _b = limiter.admit(None).expect("None admits again"); |
| 280 | + assert!(limiter.counts.lock().unwrap().is_empty()); |
| 281 | + } |
| 282 | +} |
0 commit comments