distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""RouterThrottle — admission control for tunnel requests and messages.
2
3Ported from net.i2p.router.RouterThrottleImpl.
4
5Checks bandwidth, tunnel count, and job lag to decide whether to
6accept tunnel build requests and network messages.
7"""
8
9from __future__ import annotations
10
11import enum
12from typing import Protocol, runtime_checkable
13
14
15class TunnelRequestStatus(enum.Enum):
16 """Result of a tunnel build request evaluation."""
17 ACCEPT = "accept"
18 REJECT_BANDWIDTH = "reject_bandwidth"
19 REJECT_OVERLOADED = "reject_overloaded"
20 REJECT_TRANSIENT = "reject_transient"
21 REJECT_PROBABILISTIC = "reject_probabilistic"
22
23
24@runtime_checkable
25class RouterThrottle(Protocol):
26 """Protocol for router throttle implementations."""
27
28 def accept_tunnel_request(self) -> TunnelRequestStatus: ...
29 def accept_network_message(self) -> bool: ...
30 def get_status(self) -> str: ...
31
32
33class RouterThrottleImpl:
34 """Concrete throttle implementation using bandwidth, tunnel count, and job lag."""
35
36 MAX_PARTICIPATING_TUNNELS = 10_000
37 JOB_LAG_LIMIT_MS = 1500
38 MESSAGE_DELAY_LIMIT_MS = 3000
39 MIN_THROTTLE_TUNNELS = 100
40
41 def __init__(self, context=None) -> None:
42 self._context = context
43 self._participating_count: int = 0
44 self._bandwidth_load: float = 0.0
45 self._job_lag_ms: float = 0.0
46 self._message_delay_ms: float = 0.0
47
48 def set_participating_count(self, count: int) -> None:
49 self._participating_count = count
50
51 def set_bandwidth_load(self, load: float) -> None:
52 """Set bandwidth load as 0.0-1.0 fraction."""
53 self._bandwidth_load = max(0.0, min(1.0, load))
54
55 def set_job_lag_ms(self, lag_ms: float) -> None:
56 self._job_lag_ms = lag_ms
57
58 def set_message_delay_ms(self, delay_ms: float) -> None:
59 self._message_delay_ms = delay_ms
60
61 def accept_tunnel_request(self) -> TunnelRequestStatus:
62 """Evaluate whether to accept a tunnel build request."""
63 # Check tunnel count
64 result = self._check_tunnel_count()
65 if result is not None:
66 return result
67
68 # Check bandwidth
69 result = self._check_bandwidth()
70 if result is not None:
71 return result
72
73 # Check job lag
74 result = self._check_job_lag()
75 if result is not None:
76 return result
77
78 return TunnelRequestStatus.ACCEPT
79
80 def accept_network_message(self) -> bool:
81 """Return True if we should accept an inbound network message."""
82 if self._message_delay_ms > self.MESSAGE_DELAY_LIMIT_MS:
83 return False
84 if self._bandwidth_load > 0.95:
85 return False
86 return True
87
88 def get_status(self) -> str:
89 """Human-readable status string."""
90 return (
91 f"tunnels={self._participating_count}/{self.MAX_PARTICIPATING_TUNNELS} "
92 f"bw={self._bandwidth_load:.1%} "
93 f"lag={self._job_lag_ms:.0f}ms "
94 f"delay={self._message_delay_ms:.0f}ms"
95 )
96
97 def _check_tunnel_count(self) -> TunnelRequestStatus | None:
98 if self._participating_count >= self.MAX_PARTICIPATING_TUNNELS:
99 return TunnelRequestStatus.REJECT_OVERLOADED
100 return None
101
102 def _check_bandwidth(self) -> TunnelRequestStatus | None:
103 if self._bandwidth_load > 0.9:
104 return TunnelRequestStatus.REJECT_BANDWIDTH
105 return None
106
107 def _check_job_lag(self) -> TunnelRequestStatus | None:
108 if self._job_lag_ms > self.JOB_LAG_LIMIT_MS:
109 return TunnelRequestStatus.REJECT_TRANSIENT
110 return None