Coverage for qreward / utils / scheduler / adaptive_limiter.py: 100.00%

1"""Adaptive rate limiter that dynamically adjusts throughput based on 

2error rate and latency.""" 

3 

4import logging (empty)

5import threading (empty)

6import time (empty)

7from collections import deque (empty)

8from dataclasses import dataclass (empty)

9 

10logger = logging.getLogger(__name__) (empty)

11 

12# Default slowdown factor when error rate exceeds threshold 

13DEFAULT_SLOWDOWN_FACTOR = 0.5 (empty)

14 

15# Default speedup factor when metrics are healthy 

16DEFAULT_SPEEDUP_FACTOR = 1.1 (empty)

17 

18# Minimum cooldown (seconds) between consecutive adjustments 

19DEFAULT_COOLDOWN_SECONDS = 5.0 (empty)

20 

21 

22@dataclass(frozen=True) (empty)

23class _RequestRecord: (empty)

24 """Immutable record of a single request outcome.""" 

25 

26 timestamp: float (empty)

27 latency_seconds: float (empty)

28 success: bool (empty)

29 

30 

31class AdaptiveRateLimiter: (empty)

32 """Dynamically adjusts rate limit based on observed error rate and latency. 

33 

34 Uses a sliding time window to collect request outcomes, then periodically 

35 evaluates whether to slow down (reduce limit) or speed up (increase limit). 

36 

37 Example: 

38 >>> limiter = AdaptiveRateLimiter( 

39 ... initial_limit=100, limit_min=10, limit_max=500, 

40 ... error_threshold=0.3, latency_threshold=5.0, 

41 ... ) 

42 >>> limiter.current_limit 

43 100 

44 """ 

45 

46 def __init__( (empty)

47 self, 

48 initial_limit: int, 

49 limit_min: int = 10, 

50 limit_max: int = 500, 

51 error_threshold: float = 0.3, 

52 latency_threshold: float = 5.0, 

53 window_seconds: float = 10.0, 

54 slowdown_factor: float = DEFAULT_SLOWDOWN_FACTOR, 

55 speedup_factor: float = DEFAULT_SPEEDUP_FACTOR, 

56 cooldown_seconds: float = DEFAULT_COOLDOWN_SECONDS, 

57 ): 

58 """Initialize adaptive rate limiter. 

59 

60 Args: 

61 initial_limit: Starting rate limit value. 

62 limit_min: Floor for the rate limit. 

63 limit_max: Ceiling for the rate limit. 

64 error_threshold: Error rate (0-1) above which to slow down. 

65 latency_threshold: Average latency (seconds) above which to 

66 slow down. 

67 window_seconds: Sliding window duration for statistics. 

68 slowdown_factor: Multiplier applied on slowdown (< 1). 

69 speedup_factor: Multiplier applied on speedup (> 1). 

70 cooldown_seconds: Minimum interval between adjustments. 

71 """ 

72 if limit_min > limit_max: (empty)

73 raise ValueError("limit_min must be <= limit_max") (empty)

74 if not (0 < error_threshold <= 1): (empty)

75 raise ValueError("error_threshold must be in (0, 1]") (empty)

76 

77 self._current_limit = float( (empty)

78 max(limit_min, min(initial_limit, limit_max)) 

79 ) 

80 self._limit_min = limit_min (empty)

81 self._limit_max = limit_max (empty)

82 self._error_threshold = error_threshold (empty)

83 self._latency_threshold = latency_threshold (empty)

84 self._window_seconds = window_seconds (empty)

85 self._slowdown_factor = slowdown_factor (empty)

86 self._speedup_factor = speedup_factor (empty)

87 self._cooldown_seconds = cooldown_seconds (empty)

88 

89 self._records: deque[_RequestRecord] = deque() (empty)

90 self._lock = threading.Lock() (empty)

91 self._last_adjust_time = 0.0 (empty)

92 

93 @property (empty)

94 def current_limit(self) -> int: (empty)

95 """Current effective rate limit (rounded to nearest integer).""" 

96 with self._lock: (empty)

97 return int(self._current_limit) (empty)

98 

99 def record(self, latency_seconds: float, success: bool) -> None: (empty)

100 """Record a request outcome and possibly adjust the limit. 

101 

102 Args: 

103 latency_seconds: How long the request took. 

104 success: Whether the request succeeded. 

105 """ 

106 now = time.monotonic() (empty)

107 entry = _RequestRecord( (empty)

108 timestamp=now, latency_seconds=latency_seconds, success=success 

109 ) 

110 with self._lock: (empty)

111 self._records.append(entry) (empty)

112 self._evict_expired(now) (empty)

113 self._maybe_adjust(now) (empty)

114 

115 def _evict_expired(self, now: float) -> None: (empty)

116 """Remove records older than the sliding window. Must hold lock.""" 

117 cutoff = now - self._window_seconds (empty)

118 while self._records and self._records[0].timestamp < cutoff: (empty)

119 self._records.popleft() (empty)

120 

121 def _maybe_adjust(self, now: float) -> None: (empty)

122 """Evaluate metrics and adjust limit if cooldown has elapsed. 

123 Must hold lock.""" 

124 if now - self._last_adjust_time < self._cooldown_seconds: (empty)

125 return (empty)

126 

127 total = len(self._records) (empty)

128 if total == 0: (empty)

129 return (empty)

130 

131 failures = sum( (empty)

132 1 for record in self._records if not record.success 

133 ) 

134 error_rate = failures / total (empty)

135 avg_latency = ( (empty)

136 sum(record.latency_seconds for record in self._records) / total 

137 ) 

138 

139 previous_limit = self._current_limit (empty)

140 

141 if ( (empty)

142 error_rate > self._error_threshold 

143 or avg_latency > self._latency_threshold 

144 ): 

145 self._current_limit = max( (empty)

146 self._limit_min, self._current_limit * self._slowdown_factor 

147 ) 

148 elif error_rate < self._error_threshold * 0.5: (empty)

149 self._current_limit = min( (empty)

150 self._limit_max, self._current_limit * self._speedup_factor 

151 ) 

152 

153 if self._current_limit != previous_limit: (empty)

154 self._last_adjust_time = now (empty)

155 logger.debug( (empty)

156 "Adaptive limiter adjusted: %.0f -> %.0f " 

157 "(error_rate=%.2f, avg_latency=%.3fs)", 

158 previous_limit, 

159 self._current_limit, 

160 error_rate, 

161 avg_latency, 

162 ) 

163 

164 def snapshot(self) -> dict: (empty)

165 """Return a snapshot of current statistics. 

166 

167 Returns: 

168 Dict with current_limit, total_records, error_rate, 

169 avg_latency. 

170 """ 

171 with self._lock: (empty)

172 total = len(self._records) (empty)

173 if total == 0: (empty)

174 return { (empty)

175 "current_limit": int(self._current_limit), 

176 "total_records": 0, 

177 "error_rate": 0.0, 

178 "avg_latency": 0.0, 

179 } 

180 failures = sum( (empty)

181 1 for record in self._records if not record.success 

182 ) 

183 avg_latency = ( (empty)

184 sum(record.latency_seconds for record in self._records) / total 

185 ) 

186 return { (empty)

187 "current_limit": int(self._current_limit), 

188 "total_records": total, 

189 "error_rate": failures / total, 

190 "avg_latency": avg_latency, 

191 }