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

1"""Configuration for schedule decorator. 

2 

3Example: 

4 >>> from qreward.utils.scheduler.config import ScheduleConfig 

5 >>> config = ScheduleConfig(timeout=30, retry_times=3, retry_interval=2.0) 

6 >>> config.timeout 

7 30 

8""" 

9 

10import threading (empty)

11import warnings (empty)

12from dataclasses import dataclass, field (empty)

13from typing import Any, Callable, List, Tuple, Type, Union (empty)

14 

15from .priority_queue import Priority (empty)

16 

17# Sentinel for distinguishing None from unset 

18_sentinel_none = object() (empty)

19 

20# Minimum allowed proportion for hedged requests 

21MIN_HEDGE_PROPORTION = 1e-6 (empty)

22 

23# Minimum fallback wait time (seconds) when computed wait is zero or negative 

24MIN_WAIT_TIME = 0.01 (empty)

25 

26 

27@dataclass (empty)

28class ScheduleConfig: (empty)

29 """Configuration for schedule decorator. 

30 

31 Attributes: 

32 timeout: Timeout in seconds, 0 means no timeout 

33 hedged_request_time: Time to trigger hedged requests in seconds 

34 hedged_request_proportion: Max proportion of hedged requests 

35 hedged_request_max_times: Max number of hedged requests 

36 speed_up_max_multiply: Max speed up multiplier 

37 retry_times: Max retry times (not including first call) 

38 retry_interval: Retry interval in seconds 

39 limit_size: Rate limit size (requests per window) 

40 limit_window: Rate limit window in seconds 

41 key_func: Function to generate key for pool identification 

42 exception_types: Exception types to catch and retry 

43 default_result: Default result on failure (callable or value) 

44 debug: Enable debug logging 

45 adaptive_limit: Enable adaptive rate limiting (default False). 

46 adaptive_limit_min: Minimum rate limit floor. 

47 adaptive_limit_max: Maximum rate limit ceiling. 

48 adaptive_error_threshold: Error rate threshold to trigger slowdown. 

49 adaptive_latency_threshold: Avg latency threshold (seconds). 

50 adaptive_window_seconds: Sliding window duration for statistics. 

51 priority: Default task priority (Priority.HIGH / NORMAL / LOW or 0-9). 

52 """ 

53 

54 timeout: float = 0 (empty)

55 hedged_request_time: float = 0 (empty)

56 hedged_request_proportion: float = 0.05 (empty)

57 hedged_request_max_times: int = 2 (empty)

58 speed_up_max_multiply: int = 5 (empty)

59 retry_times: int = 0 (empty)

60 retry_interval: float = 1 (empty)

61 limit_size: int = 0 (empty)

62 limit_window: float = 1.0 (empty)

63 key_func: Any = field(default=_sentinel_none) (empty)

64 exception_types: Union[ (empty)

65 Type[BaseException], Tuple[Type[BaseException], ...] 

66 ] = (BaseException,) 

67 default_result: Any = field(default=_sentinel_none) (empty)

68 debug: bool = False (empty)

69 adaptive_limit: bool = False (empty)

70 adaptive_limit_min: int = 10 (empty)

71 adaptive_limit_max: int = 500 (empty)

72 adaptive_error_threshold: float = 0.3 (empty)

73 adaptive_latency_threshold: float = 5.0 (empty)

74 adaptive_window_seconds: float = 10.0 (empty)

75 priority: int = Priority.NORMAL (empty)

76 

77 def __post_init__(self): (empty)

78 # Validate timeout 

79 if self.timeout < 0: (empty)

80 raise ValueError("timeout must be >= 0") (empty)

81 

82 # Validate retry_interval 

83 if self.retry_interval < 0: (empty)

84 raise ValueError("retry_interval must be >= 0") (empty)

85 

86 # Normalize exception_types to tuple 

87 if self.exception_types is None: (empty)

88 self.exception_types = (BaseException,) (empty)

89 elif not isinstance(self.exception_types, tuple): (empty)

90 self.exception_types = (self.exception_types,) (empty)

91 

92 # Validate hedged_request_proportion 

93 if self.hedged_request_time > 0 and ( (empty)

94 self.hedged_request_proportion <= MIN_HEDGE_PROPORTION 

95 or self.hedged_request_proportion > 1.0 

96 ): 

97 raise ValueError( (empty)

98 f"hedged_request_proportion must be in " 

99 f"({MIN_HEDGE_PROPORTION}, 1]" 

100 ) 

101 

102 @property (empty)

103 def hedged_request_multiply(self) -> float: (empty)

104 """Calculate multiply factor for hedged requests.""" 

105 if self.hedged_request_time > 0 and self.hedged_request_proportion > 0: (empty)

106 return 1 / self.hedged_request_proportion - 1 (empty)

107 return 0 (empty)

108 

109 def adjust_wait_time( (empty)

110 self, basic_wait_time: float, has_wait_time: float, 

111 max_wait_time: float 

112 ) -> float: 

113 """Calculate max wait time considering timeout. 

114 

115 Args: 

116 basic_wait_time: Basic wait time 

117 has_wait_time: Already waited time 

118 max_wait_time: Maximum wait time (timeout) 

119 

120 Returns: 

121 Adjusted wait time 

122 """ 

123 if basic_wait_time < 0: (empty)

124 basic_wait_time = MIN_WAIT_TIME (empty)

125 if ( (empty)

126 max_wait_time <= 0 

127 or basic_wait_time + has_wait_time < max_wait_time 

128 ): 

129 return basic_wait_time (empty)

130 if has_wait_time > max_wait_time: (empty)

131 return MIN_WAIT_TIME (empty)

132 return max_wait_time - has_wait_time (empty)

133 

134 def get_max_wait_time( (empty)

135 self, basic_wait_time: float, has_wait_time: float, 

136 max_wait_time: float 

137 ) -> float: 

138 warnings.warn( (empty)

139 "get_max_wait_time is deprecated, use adjust_wait_time instead", 

140 DeprecationWarning, 

141 stacklevel=2, 

142 ) 

143 return self.adjust_wait_time( (empty)

144 basic_wait_time, has_wait_time, max_wait_time 

145 ) 

146 

147 # --- Hot-reload support --- 

148 

149 _change_callbacks: List[Callable[["ScheduleConfig"], None]] = field( (empty)

150 default_factory=list, repr=False, compare=False, 

151 ) 

152 _update_lock: threading.Lock = field( (empty)

153 default_factory=threading.Lock, repr=False, compare=False, 

154 ) 

155 

156 def on_change(self, callback: Callable[["ScheduleConfig"], None]) -> None: (empty)

157 """Register a callback invoked after config is updated. 

158 

159 Args: 

160 callback: Function receiving the updated config instance. 

161 """ 

162 self._change_callbacks.append(callback) (empty)

163 

164 def update(self, **kwargs: Any) -> None: (empty)

165 """Atomically update config fields and notify listeners. 

166 

167 Only known dataclass fields are updated; unknown keys are ignored. 

168 Validation is re-run after update. 

169 

170 Args: 

171 **kwargs: Field name → new value pairs. 

172 """ 

173 with self._update_lock: (empty)

174 for key, value in kwargs.items(): (empty)

175 if hasattr(self, key) and key not in ( (empty)

176 "_change_callbacks", 

177 "_update_lock", 

178 ): 

179 setattr(self, key, value) (empty)

180 self.__post_init__() (empty)

181 for cb in self._change_callbacks: (empty)

182 cb(self) (empty)

183 

184 def snapshot(self) -> dict[str, Any]: (empty)

185 """Return a dict snapshot of the current config values.""" 

186 from dataclasses import fields as dc_fields (empty)

187 return { (empty)

188 f.name: getattr(self, f.name) 

189 for f in dc_fields(self) 

190 if not f.name.startswith("_") 

191 }