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