Coverage for src/taipanstack/core/optimizations.py: 100%
119 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 15:01 +0000
1"""
2Python Version-Specific Optimization Profiles.
4This module provides optimization strategies tailored to different Python
5versions, enabling performance improvements while maintaining stability.
7Following Stack pillars: Security, Stability, Simplicity, Scalability, Compatibility.
8"""
10import gc
11import logging
12import os
13from dataclasses import dataclass
15from taipanstack.core.compat import (
16 PY312,
17 PY313,
18 PY314,
19 get_features,
20 get_optimization_level,
21 is_experimental_enabled,
22)
24__all__ = [
25 "OptimizationProfile",
26 "OptimizationResult",
27 "apply_optimizations",
28 "get_optimization_profile",
29]
31logger = logging.getLogger(__name__)
34# =============================================================================
35# Optimization Profile
36# =============================================================================
39# Optimization Levels
40OPT_LEVEL_NONE = 0
41OPT_LEVEL_AGGRESSIVE = 2
44@dataclass(frozen=True, slots=True)
45class OptimizationProfile:
46 """Version-specific optimization settings.
48 This profile defines recommended settings based on Python version
49 and available features. All settings follow the stability-first principle.
50 """
52 # GC tuning
53 gc_threshold_0: int = 700 # Default: 700
54 gc_threshold_1: int = 10 # Default: 10
55 gc_threshold_2: int = 10 # Default: 10
56 gc_freeze_enabled: bool = False # Freeze objects after init
58 # Threading
59 thread_pool_multiplier: float = 1.0 # Multiplier for CPU count
60 max_thread_pool_size: int = 32 # Absolute maximum
62 # Memory
63 prefer_slots: bool = True # Use __slots__ in classes
64 use_frozen_dataclasses: bool = True # Prefer frozen dataclasses
66 # Code patterns
67 prefer_match_statements: bool = False # 3.10+
68 prefer_exception_groups: bool = False # 3.11+
69 prefer_type_params: bool = False # 3.12+
71 # Performance hints
72 enable_perf_hints: bool = False # JIT hints, etc.
73 aggressive_inlining: bool = False # More aggressive optimizations
75 # Experimental
76 enable_experimental: bool = False
79@dataclass(frozen=True, slots=True)
80class OptimizationResult:
81 """Result of applying optimizations."""
83 success: bool
84 applied: tuple[str, ...]
85 skipped: tuple[str, ...]
86 errors: tuple[str, ...]
88 def to_dict(self) -> dict[str, object]:
89 """Convert to dictionary."""
90 return {
91 "success": self.success,
92 "applied": list(self.applied),
93 "skipped": list(self.skipped),
94 "errors": list(self.errors),
95 }
98# =============================================================================
99# Version-Specific Profiles
100# =============================================================================
102# Python 3.11 - Stable baseline
103_PROFILE_311 = OptimizationProfile(
104 gc_threshold_0=700,
105 gc_threshold_1=10,
106 gc_threshold_2=10,
107 gc_freeze_enabled=False,
108 thread_pool_multiplier=1.0,
109 max_thread_pool_size=32,
110 prefer_slots=True,
111 use_frozen_dataclasses=True,
112 prefer_match_statements=True,
113 prefer_exception_groups=True,
114 prefer_type_params=False,
115 enable_perf_hints=False,
116 aggressive_inlining=False,
117 enable_experimental=False,
118)
120# Python 3.12 - Enhanced
121_PROFILE_312 = OptimizationProfile(
122 gc_threshold_0=800, # Slightly higher due to better GC
123 gc_threshold_1=10,
124 gc_threshold_2=10,
125 gc_freeze_enabled=True, # Safe to use
126 thread_pool_multiplier=1.0,
127 max_thread_pool_size=32,
128 prefer_slots=True,
129 use_frozen_dataclasses=True,
130 prefer_match_statements=True,
131 prefer_exception_groups=True,
132 prefer_type_params=True,
133 enable_perf_hints=False,
134 aggressive_inlining=False,
135 enable_experimental=False,
136)
138# Python 3.13 - Modern (with experimental options available)
139_PROFILE_313 = OptimizationProfile(
140 gc_threshold_0=900, # Higher with mimalloc
141 gc_threshold_1=15,
142 gc_threshold_2=15,
143 gc_freeze_enabled=True,
144 thread_pool_multiplier=1.5, # Can use more threads with better GIL
145 max_thread_pool_size=48,
146 prefer_slots=True,
147 use_frozen_dataclasses=True,
148 prefer_match_statements=True,
149 prefer_exception_groups=True,
150 prefer_type_params=True,
151 enable_perf_hints=True, # JIT hints available
152 aggressive_inlining=False,
153 enable_experimental=False, # Requires explicit opt-in
154)
156# Python 3.14 - Cutting edge
157_PROFILE_314 = OptimizationProfile(
158 gc_threshold_0=1000, # Optimized incremental GC
159 gc_threshold_1=20,
160 gc_threshold_2=20,
161 gc_freeze_enabled=True,
162 thread_pool_multiplier=2.0, # Free-threading ready
163 max_thread_pool_size=64,
164 prefer_slots=True,
165 use_frozen_dataclasses=True,
166 prefer_match_statements=True,
167 prefer_exception_groups=True,
168 prefer_type_params=True,
169 enable_perf_hints=True,
170 aggressive_inlining=True, # Tail-call interpreter
171 enable_experimental=False, # Requires explicit opt-in
172)
175_cached_optimization_profile: OptimizationProfile | None = None
178def _get_base_profile() -> OptimizationProfile:
179 """Select base profile by version."""
180 if PY314:
181 return _PROFILE_314
182 if PY313:
183 return _PROFILE_313
184 if PY312:
185 return _PROFILE_312
186 return _PROFILE_311
189def _build_aggressive_profile(profile: OptimizationProfile) -> OptimizationProfile:
190 """Build an aggressive optimization profile with experimental features."""
191 return OptimizationProfile(
192 gc_threshold_0=profile.gc_threshold_0,
193 gc_threshold_1=profile.gc_threshold_1,
194 gc_threshold_2=profile.gc_threshold_2,
195 gc_freeze_enabled=profile.gc_freeze_enabled,
196 thread_pool_multiplier=profile.thread_pool_multiplier,
197 max_thread_pool_size=profile.max_thread_pool_size,
198 prefer_slots=profile.prefer_slots,
199 use_frozen_dataclasses=profile.use_frozen_dataclasses,
200 prefer_match_statements=profile.prefer_match_statements,
201 prefer_exception_groups=profile.prefer_exception_groups,
202 prefer_type_params=profile.prefer_type_params,
203 enable_perf_hints=profile.enable_perf_hints,
204 aggressive_inlining=profile.aggressive_inlining,
205 enable_experimental=True,
206 )
209def get_optimization_profile(*, force_refresh: bool = False) -> OptimizationProfile:
210 """Get the optimization profile for the current Python version.
212 Args:
213 force_refresh: If True, re-detect instead of using cache.
215 Returns:
216 OptimizationProfile suitable for the runtime environment.
218 """
219 global _cached_optimization_profile # noqa: PLW0603
221 if _cached_optimization_profile is not None and not force_refresh:
222 return _cached_optimization_profile
224 _ = get_features(force_refresh=force_refresh) # Warm up cache, validate version
225 experimental = is_experimental_enabled(force_refresh=force_refresh)
226 opt_level = get_optimization_level(force_refresh=force_refresh)
228 profile = _get_base_profile()
230 # Adjust for optimization level
231 if opt_level == OPT_LEVEL_NONE:
232 _cached_optimization_profile = _PROFILE_311
233 elif opt_level == OPT_LEVEL_AGGRESSIVE and experimental:
234 _cached_optimization_profile = _build_aggressive_profile(profile)
235 else:
236 _cached_optimization_profile = profile
238 return _cached_optimization_profile
241# =============================================================================
242# Apply Optimizations
243# =============================================================================
246def _apply_gc_tuning(
247 profile: OptimizationProfile,
248 applied: list[str],
249 errors: list[str],
250) -> None:
251 """Apply Garbage Collector tuning."""
252 try:
253 current = gc.get_threshold()
254 gc.set_threshold(
255 profile.gc_threshold_0,
256 profile.gc_threshold_1,
257 profile.gc_threshold_2,
258 )
259 applied.append(
260 f"gc_threshold: {current} -> "
261 f"({profile.gc_threshold_0}, {profile.gc_threshold_1}, "
262 f"{profile.gc_threshold_2})",
263 )
264 except Exception as e:
265 errors.append(f"gc_threshold: {e}")
268def _apply_gc_freeze(
269 profile: OptimizationProfile,
270 freeze_after: bool,
271 applied: list[str],
272 skipped: list[str],
273 errors: list[str],
274) -> None:
275 """Apply GC freeze if supported."""
276 if not profile.gc_freeze_enabled:
277 return
278 if not PY312:
279 skipped.append("gc_freeze: requires Python 3.12+")
280 return
281 if not freeze_after:
282 return
284 try:
285 gc.freeze()
286 applied.append("gc_freeze: enabled")
287 except Exception as e:
288 errors.append(f"gc_freeze: {e}")
291def _apply_experimental(
292 profile: OptimizationProfile,
293 applied: list[str],
294 skipped: list[str],
295) -> None:
296 """Check and log experimental features."""
297 if profile.enable_experimental:
298 features = get_features()
299 if features.has_jit:
300 applied.append("jit: available")
301 if features.has_free_threading:
302 applied.append("free_threading: available")
303 logger.warning(
304 "EXPERIMENTAL FEATURES ENABLED: Stability and security may be affected.",
305 )
306 else:
307 skipped.append("experimental: requires STACK_ENABLE_EXPERIMENTAL=1")
310def _log_optimization_summary(
311 applied: list[str],
312 skipped: list[str],
313 errors: list[str],
314) -> None:
315 """Log the summary of applied optimizations."""
316 if applied:
317 logger.debug("Applied optimizations: %s", ", ".join(applied))
318 if skipped:
319 logger.debug("Skipped optimizations: %s", ", ".join(skipped))
320 if errors:
321 logger.warning("Optimization errors: %s", ", ".join(errors))
324def apply_optimizations(
325 *,
326 profile: OptimizationProfile | None = None,
327 apply_gc: bool = True,
328 freeze_after: bool = False,
329 force_refresh: bool = False,
330) -> OptimizationResult:
331 """Apply runtime optimizations based on profile.
333 This function applies safe, reversible optimizations to the Python
334 runtime. It is designed to be called once at application startup.
336 Args:
337 profile: Optimization profile to use (auto-detected if None).
338 apply_gc: Whether to apply GC tuning.
339 freeze_after: Whether to freeze objects after applying.
340 force_refresh: Whether to force re-detection of profile if none is provided.
342 Returns:
343 OptimizationResult with details of what was applied.
345 """
346 if profile is None:
347 profile = get_optimization_profile(force_refresh=force_refresh)
349 applied: list[str] = []
350 skipped: list[str] = []
351 errors: list[str] = []
353 # GC Tuning
354 if apply_gc:
355 _apply_gc_tuning(profile, applied, errors)
356 else:
357 skipped.append("gc_threshold: disabled")
359 # GC Freeze (3.12+)
360 _apply_gc_freeze(profile, freeze_after, applied, skipped, errors)
362 # Performance hints logging
363 if profile.enable_perf_hints:
364 applied.append("perf_hints: enabled (JIT-aware patterns)")
365 else:
366 skipped.append("perf_hints: disabled")
368 # Experimental features
369 _apply_experimental(profile, applied, skipped)
371 # Log summary
372 success = len(errors) == 0
373 _log_optimization_summary(applied, skipped, errors)
375 return OptimizationResult(
376 success=success,
377 applied=tuple(applied),
378 skipped=tuple(skipped),
379 errors=tuple(errors),
380 )
383# =============================================================================
384# Utility Functions
385# =============================================================================
388def get_recommended_thread_pool_size(*, force_refresh: bool = False) -> int:
389 """Get recommended thread pool size based on version and features.
391 Args:
392 force_refresh: If True, re-detect instead of using cache.
394 Returns:
395 Recommended number of threads for ThreadPoolExecutor.
397 """
398 profile = get_optimization_profile(force_refresh=force_refresh)
399 cpu_count = os.cpu_count() or 4
401 size = int(cpu_count * profile.thread_pool_multiplier)
402 return min(size, profile.max_thread_pool_size)