From 57b459bc72092dd1457f90756e1113367355a3e3 Mon Sep 17 00:00:00 2001 From: ljz <425868052@qq.com> Date: Sat, 15 Aug 2026 20:48:04 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=8F=AF=E9=80=89=E6=80=A7?= =?UTF-8?q?=E8=83=BD=E5=9F=8B=E7=82=B9=E5=B9=B6=E5=AE=8C=E6=88=90=E7=89=A9?= =?UTF-8?q?=E6=80=A7=E6=95=88=E7=8E=87=E8=AF=84=E4=BC=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/main.py | 35 +- app/simulation/README.md | 18 + app/simulation/benchmark_performance.py | 262 +++++++ .../components/amesim/media/mediums.py | 26 + app/simulation/core/medium.py | 10 + app/simulation/core/peng_robinson.py | 53 ++ app/simulation/performance.py | 665 ++++++++++++++++++ app/simulation/solvers/algebraic.py | 2 + app/simulation/solvers/pneumatic_volume.py | 2 + app/simulation/solvers/signal.py | 2 + app/simulation/solvers/solver.py | 2 + app/simulation/solvers/stream.py | 30 +- app/simulation/systems/generic.py | 285 ++++---- app/system_xml.py | 2 + docs/README.md | 5 + docs/仿真性能评估-2026-08-15.md | 119 ++++ docs/后端求解逻辑与效率优化调研.md | 72 +- tests/test_performance_benchmark.py | 58 ++ ...st_property_performance_instrumentation.py | 141 ++++ tests/test_simulation_performance.py | 248 +++++++ tests/test_simulation_performance_pipeline.py | 171 +++++ 21 files changed, 2029 insertions(+), 179 deletions(-) create mode 100644 app/simulation/benchmark_performance.py create mode 100644 app/simulation/performance.py create mode 100644 docs/仿真性能评估-2026-08-15.md create mode 100644 tests/test_performance_benchmark.py create mode 100644 tests/test_property_performance_instrumentation.py create mode 100644 tests/test_simulation_performance.py create mode 100644 tests/test_simulation_performance_pipeline.py diff --git a/app/main.py b/app/main.py index 3cd4e45..07ef9bd 100644 --- a/app/main.py +++ b/app/main.py @@ -21,6 +21,7 @@ from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse from pydantic import BaseModel, ConfigDict, Field, ValidationError +from app.simulation.performance import performance_span, profile_phase, profile_run from app.system_xml import ( SystemXmlDocument, SystemXmlValidationReport, @@ -666,6 +667,26 @@ def run_system_xml_simulation( xml_bytes: bytes, progress_callback: SimulationProgressEmitter | None = None, cancel_check: Callable[[], bool] | None = None, +) -> dict[str, object]: + with profile_run() as trace: + result = _run_system_xml_simulation_profiled( + xml_bytes, + progress_callback, + cancel_check, + ) + + performance = trace.snapshot() + if performance.get("mode") != "off": + diagnostics = dict(result.get("diagnostics", {})) + diagnostics["performance"] = performance + result["diagnostics"] = diagnostics + return result + + +def _run_system_xml_simulation_profiled( + xml_bytes: bytes, + progress_callback: SimulationProgressEmitter | None = None, + cancel_check: Callable[[], bool] | None = None, ) -> dict[str, object]: from app.simulation.solvers.algebraic import AlgebraicSolveError from app.simulation.solvers.solver import SolveIVPConfig @@ -792,12 +813,13 @@ def run_system_xml_simulation( }, ) from exc - return { - "validation": report.as_dict(), - "simulation": document.as_model_data()["simulation"], - "model": network.as_interface_dict(), - **result.as_dict(), - } + with performance_span("simulation.response_assembly"): + return { + "validation": report.as_dict(), + "simulation": document.as_model_data()["simulation"], + "model": network.as_interface_dict(), + **result.as_dict(), + } def simulation_event_stream( @@ -1200,6 +1222,7 @@ def compile_reactflow_network( ) +@profile_phase("simulation.network_compilation") def compile_system_xml_network( document: SystemXmlDocument, *, diff --git a/app/simulation/README.md b/app/simulation/README.md index 278b53b..b0235e3 100644 --- a/app/simulation/README.md +++ b/app/simulation/README.md @@ -42,6 +42,8 @@ RESULT_VARIABLES / DISPLAY / create()`，再把类路径加入库清单。完整 - `components/amesim/media/`: AMESim 零端口介质物性定义元件；具体类型确定介质，`property_model` 下拉参数选择计算方法，当前提供空气理想气体和氦气 Peng-Robinson - `components/amesim/gases.py`: AMESim `gi` 介质物性实例注册表；`gi=0` 固定为空气（理想气体，内置默认），`gi=1..99` 引用画布中的显式介质定义 - `core/peng_robinson.py`: `test_mql` 与公开氦气介质共用的 Peng-Robinson 状态方程 +- `performance.py`: 默认关闭、按单次仿真隔离的阶段与物性性能埋点 +- `benchmark_performance.py`: System XML 主求解路径的可重复命令行基准工具 - `systems/network.py`: `SimulationNetwork`，负责组件注册、连接拓扑和状态向量拼装 - `solvers/solver.py`: `integrate_ode()`，优先走 `SciPy solve_ivp`，缺依赖时回退到内置 RK4，并支持 `t_start == t_stop` 的零时长返回 - `examples/testmodel/dynamic_pipe.py`: TestModel 专用单阻容管道近似，入口压降 + 出口直连内容腔 @@ -58,6 +60,22 @@ RESULT_VARIABLES / DISPLAY / create()`，再把类路径加入库清单。完整 - `examples/test_mql/run.py`: `test_mql` 结构运行与程序化执行入口 - `tests/`: 当前组件契约、XML、通用系统、AMESim 迁移和结果导出测试 +## 可选性能诊断 + +`SIMULATIONAPP_PROFILE` 支持 `off`（默认）、`standard` 和 `audit`。`standard` +只统计低频的大阶段；`audit` 才展开 RHS、代数闭合、stream 和物性调用，开销也 +明显更高。最终优化收益必须在 `off` 下复测。 + +```powershell +.venv-win\Scripts\python.exe -m app.simulation.benchmark_performance ` + --mode audit --warmups 1 --runs 3 ` + --factory "helium_step=tests.test_amesim_pnvo001_signal_xml:high_pressure_helium_step_project" ` + --output app/data/performance-evaluations/helium-step.json +``` + +基准原始 JSON 默认放到已忽略的 `app/data/` 下。指标字段、实测结果和使用边界见 +[`仿真性能评估 2026-08-15`](../../docs/仿真性能评估-2026-08-15.md)。 + ## 当前阶段进度 这一阶段原先有 4 件重点工作，现在的状态如下： diff --git a/app/simulation/benchmark_performance.py b/app/simulation/benchmark_performance.py new file mode 100644 index 0000000..616005f --- /dev/null +++ b/app/simulation/benchmark_performance.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +import argparse +import hashlib +import importlib +import json +import os +import platform +import statistics +import sys +from datetime import UTC, datetime +from math import ceil +from pathlib import Path +from time import perf_counter_ns, process_time_ns +from typing import Any + + +def _named_value(value: str, *, option: str) -> tuple[str, str]: + name, separator, target = value.partition("=") + if not separator or not name.strip() or not target.strip(): + raise ValueError( + f"{option} must use NAME=VALUE syntax, received {value!r}." + ) + return name.strip(), target.strip() + + +def _percentile(values: list[float], percentile: float) -> float: + ordered = sorted(values) + index = max(0, min(len(ordered) - 1, ceil(percentile * len(ordered)) - 1)) + return ordered[index] + + +def _duration_summary(values: list[float]) -> dict[str, object]: + return { + "samplesMs": values, + "minimumMs": min(values), + "medianMs": statistics.median(values), + "p95Ms": _percentile(values, 0.95), + "maximumMs": max(values), + } + + +def _load_factory_xml(specification: str) -> bytes: + module_name, separator, member_name = specification.partition(":") + if not separator or not module_name or not member_name: + raise ValueError( + "Factory specifications must use module.path:callable syntax." + ) + factory = getattr(importlib.import_module(module_name), member_name) + value = factory() + if isinstance(value, bytes): + return value + if isinstance(value, str): + return value.encode("utf-8") + + from app.main import build_reactflow_system_xml + + return build_reactflow_system_xml(value) + + +def _clear_property_caches() -> None: + from app.simulation.components.amesim.media.mediums import ( + AmesimHeliumPengRobinsonMedium, + ) + + for method_name in ( + "temperature_from_pressure_enthalpy", + "properties_from_mU", + ): + method = getattr(AmesimHeliumPengRobinsonMedium, method_name) + cache_clear = getattr(method, "cache_clear", None) + if cache_clear is not None: + cache_clear() + + +def _serialize_result_event(result: dict[str, object]) -> bytes: + """Render the final NDJSON payload shape used by the streaming endpoint.""" + + status = str(result.get("status", "completed")) + event = { + "event": "result", + "progress": 100 if status == "completed" else 0, + "phase": status, + "message": "仿真完成" if status == "completed" else "仿真任务结束", + "simulatedTime": result.get("simulatedUntil"), + "totalTime": result.get("requestedStopTime"), + "result": result, + } + return ( + json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n" + ).encode("utf-8") + + +def _run_case( + name: str, + xml_bytes: bytes, + *, + warmups: int, + runs: int, + cancellable_path: bool, + clear_property_cache: bool, + allow_failures: bool, +) -> dict[str, object]: + from app.main import run_system_xml_simulation + + cancel_check = (lambda: False) if cancellable_path else None + for _ in range(warmups): + if clear_property_cache: + _clear_property_caches() + result = run_system_xml_simulation(xml_bytes, cancel_check=cancel_check) + if not bool(result.get("success")) and not allow_failures: + raise RuntimeError(f"Warmup for {name!r} failed: {result.get('message')}") + + wall_samples_ms: list[float] = [] + cpu_samples_ms: list[float] = [] + serialization_samples_ms: list[float] = [] + serialized_sizes: list[int] = [] + profiles: list[dict[str, object]] = [] + final_result: dict[str, object] | None = None + for _ in range(runs): + if clear_property_cache: + _clear_property_caches() + wall_start = perf_counter_ns() + cpu_start = process_time_ns() + result = run_system_xml_simulation(xml_bytes, cancel_check=cancel_check) + cpu_samples_ms.append((process_time_ns() - cpu_start) / 1_000_000.0) + wall_samples_ms.append((perf_counter_ns() - wall_start) / 1_000_000.0) + if not bool(result.get("success")) and not allow_failures: + raise RuntimeError(f"Benchmark for {name!r} failed: {result.get('message')}") + diagnostics = result.get("diagnostics") + if isinstance(diagnostics, dict): + performance = diagnostics.get("performance") + if isinstance(performance, dict): + profiles.append(performance) + serialization_start = perf_counter_ns() + serialized_event = _serialize_result_event(result) + serialization_samples_ms.append( + (perf_counter_ns() - serialization_start) / 1_000_000.0 + ) + serialized_sizes.append(len(serialized_event)) + final_result = result + + assert final_result is not None + return { + "name": name, + "success": bool(final_result.get("success")), + "message": final_result.get("message"), + "inputBytes": len(xml_bytes), + "inputSha256": hashlib.sha256(xml_bytes).hexdigest(), + "status": final_result.get("status"), + "simulatedUntil": final_result.get("simulatedUntil"), + "requestedStopTime": final_result.get("requestedStopTime"), + "wall": _duration_summary(wall_samples_ms), + "cpu": _duration_summary(cpu_samples_ms), + "resultSerialization": _duration_summary(serialization_samples_ms), + "resultEventBytes": serialized_sizes, + "performanceRuns": profiles, + } + + +def _parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Benchmark the real System XML simulation path with optional profiling." + ) + parser.add_argument( + "--mode", + choices=("off", "standard", "audit"), + default="audit", + help="Instrumentation depth selected before importing the simulation modules.", + ) + parser.add_argument("--warmups", type=int, default=1) + parser.add_argument("--runs", type=int, default=5) + parser.add_argument( + "--xml", + action="append", + default=[], + metavar="NAME=PATH", + help="Add an XML file benchmark case.", + ) + parser.add_argument( + "--factory", + action="append", + default=[], + metavar="NAME=MODULE:CALLABLE", + help="Add a zero-argument factory returning XML or ReactFlowProjectPayload.", + ) + parser.add_argument( + "--direct-path", + action="store_true", + help="Do not pass a cancel callback; use the one-shot SciPy path when eligible.", + ) + parser.add_argument( + "--cold-property-cache", + action="store_true", + help="Clear the two helium property LRU caches before every warmup and measured run.", + ) + parser.add_argument( + "--allow-failures", + action="store_true", + help="Record failed simulation runs instead of aborting the benchmark.", + ) + parser.add_argument("--output", type=Path) + arguments = parser.parse_args(argv) + if arguments.warmups < 0: + parser.error("--warmups must not be negative.") + if arguments.runs <= 0: + parser.error("--runs must be positive.") + if not arguments.xml and not arguments.factory: + parser.error("At least one --xml or --factory case is required.") + return arguments + + +def main(argv: list[str] | None = None) -> int: + arguments = _parse_arguments(argv) + os.environ["SIMULATIONAPP_PROFILE"] = arguments.mode + + cases: list[tuple[str, bytes]] = [] + for raw_case in arguments.xml: + name, raw_path = _named_value(raw_case, option="--xml") + cases.append((name, Path(raw_path).read_bytes())) + for raw_case in arguments.factory: + name, specification = _named_value(raw_case, option="--factory") + cases.append((name, _load_factory_xml(specification))) + + report: dict[str, Any] = { + "generatedAt": datetime.now(UTC).isoformat(), + "profileMode": arguments.mode, + "cancellableSolverPath": not arguments.direct_path, + "coldPropertyCache": bool(arguments.cold_property_cache), + "allowFailures": bool(arguments.allow_failures), + "warmups": arguments.warmups, + "runs": arguments.runs, + "runtime": { + "python": sys.version, + "platform": platform.platform(), + "processor": platform.processor(), + }, + "cases": [ + _run_case( + name, + xml_bytes, + warmups=arguments.warmups, + runs=arguments.runs, + cancellable_path=not arguments.direct_path, + clear_property_cache=arguments.cold_property_cache, + allow_failures=arguments.allow_failures, + ) + for name, xml_bytes in cases + ], + } + text = json.dumps(report, ensure_ascii=False, indent=2) + if arguments.output is not None: + arguments.output.parent.mkdir(parents=True, exist_ok=True) + arguments.output.write_text(text + "\n", encoding="utf-8") + print(f"Performance report written to {arguments.output.resolve()}") + else: + print(text) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/app/simulation/components/amesim/media/mediums.py b/app/simulation/components/amesim/media/mediums.py index 084def4..70583b2 100644 --- a/app/simulation/components/amesim/media/mediums.py +++ b/app/simulation/components/amesim/media/mediums.py @@ -12,6 +12,7 @@ from app.simulation.core.medium import ( ThermodynamicProperties, ) from app.simulation.core.peng_robinson import HELIUM_PR, PengRobinsonFluid +from app.simulation.performance import profile_property, record_property_iterations @dataclass(frozen=True) @@ -69,6 +70,7 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): del T return self.cv + @profile_property("density") def density(self, p: float, T: float) -> float: return self.fluid.density(p, T) @@ -134,6 +136,7 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): ) return factor, exponent + @profile_property("isentropic_density_pressure_factor") def isentropic_density_pressure_factor( self, p: float, @@ -166,12 +169,14 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): raise ValueError("Volume must stay positive.") return self.fluid.pressure_from_density(T, m / V) + @profile_property("specific_internal_energy") def specific_internal_energy(self, T: float) -> float: return self.R_gas * ( (self.nasa_cp_over_R - 1.0) * T + self.nasa_enthalpy_constant_K ) + @profile_property("specific_internal_energy_at_pressure") def specific_internal_energy_at_pressure(self, p: float, T: float) -> float: density = self.density(p, T) return ( @@ -179,12 +184,14 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): + self.fluid.residual_specific_internal_energy_at_density(T, density) ) + @profile_property("specific_enthalpy") def specific_enthalpy(self, T: float) -> float: return self.R_gas * ( self.nasa_cp_over_R * T + self.nasa_enthalpy_constant_K ) + @profile_property("specific_enthalpy_at_pressure") def specific_enthalpy_at_pressure(self, p: float, T: float) -> float: return self.specific_enthalpy(T) + self.fluid.residual_specific_enthalpy(p, T) @@ -198,6 +205,7 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): h / self.R_gas - self.nasa_enthalpy_constant_K ) / self.nasa_cp_over_R + @profile_property("temperature_from_pressure_enthalpy", track_cache=True) @lru_cache(maxsize=8192) def temperature_from_pressure_enthalpy(self, p: float, h: float) -> float: temperature = max(self.temperature_from_enthalpy(h), 2.2) @@ -211,8 +219,18 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): temperature, 1.0, ): + record_property_iterations( + "temperature_from_pressure_enthalpy", + _iteration + 1, + True, + ) return next_temperature temperature = next_temperature + record_property_iterations( + "temperature_from_pressure_enthalpy", + 16, + False, + ) return temperature def temperature_from_mass_internal_energy(self, m: float, U: float) -> float: @@ -222,6 +240,7 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): ) return self.temperature_from_internal_energy(U / m) + @profile_property("properties_from_mU", track_cache=True) @lru_cache(maxsize=8192) def properties_from_mU( self, @@ -249,6 +268,7 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): self.temperature_from_internal_energy(target_internal_energy), 2.2, ) + converged = False for _iteration in range(16): residual_internal_energy = ( self.fluid.residual_specific_internal_energy_at_density( @@ -267,8 +287,14 @@ class AmesimHeliumPengRobinsonMedium(IdealGasMedium): 1.0, ): temperature = next_temperature + converged = True break temperature = next_temperature + record_property_iterations( + "properties_from_mU", + _iteration + 1, + converged, + ) pressure = self.fluid.pressure_from_density(temperature, density) return ThermodynamicProperties( p=pressure, diff --git a/app/simulation/core/medium.py b/app/simulation/core/medium.py index 614d418..f1210a1 100644 --- a/app/simulation/core/medium.py +++ b/app/simulation/core/medium.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from typing import Protocol from app.simulation.core.errors import RecoverableTrialStateError +from app.simulation.performance import profile_property @dataclass(frozen=True) @@ -108,9 +109,11 @@ class IdealGasMedium: def cv_at_temperature(self, T: float) -> float: return self.cp_at_temperature(T) - self.R_gas + @profile_property("density") def density(self, p: float, T: float) -> float: return p / (self.R_gas * T) + @profile_property("isentropic_density_pressure_factor") def isentropic_density_pressure_factor( self, p: float, @@ -123,6 +126,7 @@ class IdealGasMedium: cv = self.cv_at_temperature(T) return cv / cp + @profile_property("dynamic_viscosity") def dynamic_viscosity(self, T: float) -> float: """Return dynamic viscosity using the default air Sutherland law.""" @@ -135,6 +139,7 @@ class IdealGasMedium: / (T + self.sutherland_constant) ) + @profile_property("specific_internal_energy") def specific_internal_energy(self, T: float) -> float: delta_T = T - self.T_ref return ( @@ -143,10 +148,12 @@ class IdealGasMedium: + 0.5 * self.cp_slope * delta_T * delta_T ) + @profile_property("specific_internal_energy_at_pressure") def specific_internal_energy_at_pressure(self, p: float, T: float) -> float: del p return self.specific_internal_energy(T) + @profile_property("specific_enthalpy") def specific_enthalpy(self, T: float) -> float: delta_T = T - self.T_ref return ( @@ -155,6 +162,7 @@ class IdealGasMedium: + 0.5 * self.cp_slope * delta_T * delta_T ) + @profile_property("specific_enthalpy_at_pressure") def specific_enthalpy_at_pressure(self, p: float, T: float) -> float: del p return self.specific_enthalpy(T) @@ -191,6 +199,7 @@ class IdealGasMedium: delta_T = positive_root if abs(positive_root) <= abs(negative_root) else negative_root return self.T_ref + delta_T + @profile_property("temperature_from_pressure_enthalpy") def temperature_from_pressure_enthalpy(self, p: float, h: float) -> float: del p return self.temperature_from_enthalpy(h) @@ -207,6 +216,7 @@ class IdealGasMedium: raise ValueError("Volume must stay positive.") return m * self.R_gas * T / V + @profile_property("properties_from_mU") def properties_from_mU(self, m: float, U: float, V: float) -> ThermodynamicProperties: T = self.temperature_from_mass_internal_energy(m, U) p = self.pressure(m, T, V) diff --git a/app/simulation/core/peng_robinson.py b/app/simulation/core/peng_robinson.py index a711e3d..6e61996 100644 --- a/app/simulation/core/peng_robinson.py +++ b/app/simulation/core/peng_robinson.py @@ -5,6 +5,8 @@ from app.simulation.core.errors import RecoverableTrialStateError from dataclasses import dataclass from math import acos, cos, isfinite, log, pi, sqrt +from app.simulation.performance import profile_property + UNIVERSAL_GAS_CONSTANT = 8.31446261815324 # Simcenter Amesim 2404 ``sag_reinit_eos_`` keeps more digits than the # commonly printed Peng-Robinson constants 0.45724 and 0.07780. @@ -97,6 +99,11 @@ class PengRobinsonFluid: ) -> float: return self.a_parameter * self.alpha_temperature_second_derivative(temperature) + @profile_property( + "pressure_from_molar_volume", + layer="kernel", + minimum_mode="audit", + ) def pressure_from_molar_volume(self, temperature: float, molar_volume: float) -> float: self._validate_temperature(temperature) if molar_volume <= self.b_parameter: @@ -107,11 +114,21 @@ class PengRobinsonFluid: attractive = a_alpha / (molar_volume * (molar_volume + b) + b * (molar_volume - b)) return repulsive - attractive + @profile_property( + "pressure_from_density", + layer="kernel", + minimum_mode="audit", + ) def pressure_from_density(self, temperature: float, density: float) -> float: if density <= 0.0: raise ValueError("Density must be positive.") return self.pressure_from_molar_volume(temperature, self.molar_mass / density) + @profile_property( + "pressure_temperature_derivative_at_density", + layer="kernel", + minimum_mode="audit", + ) def pressure_temperature_derivative_at_density( self, temperature: float, @@ -132,6 +149,11 @@ class PengRobinsonFluid: - self.attractive_parameter_temperature_derivative(temperature) / denominator ) + @profile_property( + "pressure_density_derivative_at_temperature", + layer="kernel", + minimum_mode="audit", + ) def pressure_density_derivative_at_temperature( self, temperature: float, @@ -165,6 +187,11 @@ class PengRobinsonFluid: B = b * pressure / (UNIVERSAL_GAS_CONSTANT * temperature) return A, B + @profile_property( + "compressibility_roots", + layer="kernel", + minimum_mode="audit", + ) def compressibility_roots(self, pressure: float, temperature: float) -> tuple[float, ...]: A, B = self.reduced_parameters(pressure, temperature) coefficients = ( @@ -178,6 +205,11 @@ class PengRobinsonFluid: raise ValueError("Peng-Robinson cubic produced no physical compressibility root.") return physical_roots + @profile_property( + "compressibility_factor", + layer="kernel", + minimum_mode="audit", + ) def compressibility_factor( self, pressure: float, @@ -193,6 +225,11 @@ class PengRobinsonFluid: return roots[-1] raise ValueError(f"Unsupported phase selector: {phase!r}") + @profile_property( + "molar_volume", + layer="kernel", + minimum_mode="audit", + ) def molar_volume( self, pressure: float, @@ -202,6 +239,7 @@ class PengRobinsonFluid: z = self.compressibility_factor(pressure, temperature, phase=phase) return z * UNIVERSAL_GAS_CONSTANT * temperature / pressure + @profile_property("density", layer="kernel", minimum_mode="audit") def density( self, pressure: float, @@ -210,6 +248,11 @@ class PengRobinsonFluid: ) -> float: return self.molar_mass / self.molar_volume(pressure, temperature, phase=phase) + @profile_property( + "residual_specific_enthalpy", + layer="kernel", + minimum_mode="audit", + ) def residual_specific_enthalpy( self, pressure: float, @@ -239,6 +282,11 @@ class PengRobinsonFluid: ) return residual_molar_enthalpy / self.molar_mass + @profile_property( + "residual_specific_internal_energy_at_density", + layer="kernel", + minimum_mode="audit", + ) def residual_specific_internal_energy_at_density( self, temperature: float, @@ -268,6 +316,11 @@ class PengRobinsonFluid: ) * log(log_argument) / (2.0 * sqrt(2.0) * b) return residual_molar_internal_energy / self.molar_mass + @profile_property( + "residual_isochoric_heat_capacity_at_density", + layer="kernel", + minimum_mode="audit", + ) def residual_isochoric_heat_capacity_at_density( self, temperature: float, diff --git a/app/simulation/performance.py b/app/simulation/performance.py new file mode 100644 index 0000000..4cfa9a8 --- /dev/null +++ b/app/simulation/performance.py @@ -0,0 +1,665 @@ +"""Low-overhead, run-local performance instrumentation for simulations. + +The profiling mode is intentionally read once when this module is imported. +``standard`` records low-frequency pipeline stages, while ``audit`` also wraps +hot RHS/property operations and computes exact-input reuse metrics. With +profiling disabled, decorators return the original callable while classes are +being defined, so ordinary simulation calls do not pass through a wrapper. +""" + +from __future__ import annotations + +from collections.abc import Callable, Generator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar, Token +from dataclasses import dataclass, field +from functools import wraps +import inspect +import os +import struct +from time import perf_counter_ns +from typing import Any, Literal, ParamSpec, TypeVar, cast + + +ProfileMode = Literal["off", "standard", "audit"] + +_P = ParamSpec("_P") +_R = TypeVar("_R") +_MODE_RANK: Mapping[ProfileMode, int] = {"off": 0, "standard": 1, "audit": 2} + + +def _read_startup_mode() -> ProfileMode: + raw_mode = os.getenv("SIMULATIONAPP_PROFILE", "off").strip().lower() + aliases: dict[str, ProfileMode] = { + "": "off", + "0": "off", + "false": "off", + "no": "off", + "off": "off", + "1": "standard", + "true": "standard", + "yes": "standard", + "on": "standard", + "standard": "standard", + "audit": "audit", + } + try: + return aliases[raw_mode] + except KeyError as exc: + raise ValueError( + "SIMULATIONAPP_PROFILE must be one of: off, standard, audit." + ) from exc + + +PROFILE_MODE: ProfileMode = _read_startup_mode() + + +def _minimum_mode(value: str) -> ProfileMode: + normalized = value.strip().lower() + if normalized not in _MODE_RANK: + raise ValueError("minimum_mode must be one of: off, standard, audit.") + return cast(ProfileMode, normalized) + + +def _mode_enabled(minimum_mode: ProfileMode) -> bool: + # ``off`` is an unconditional zero-wrapper mode, even if a caller passes + # ``minimum_mode="off"`` by mistake. + return ( + PROFILE_MODE != "off" + and _MODE_RANK[PROFILE_MODE] >= _MODE_RANK[minimum_mode] + ) + + +@dataclass +class _TimingStats: + calls: int = 0 + inclusive_ns: int = 0 + self_ns: int = 0 + max_ns: int = 0 + errors: int = 0 + + def record(self, inclusive_ns: int, self_ns: int, error: bool) -> None: + self.calls += 1 + self.inclusive_ns += inclusive_ns + self.self_ns += self_ns + self.max_ns = max(self.max_ns, inclusive_ns) + if error: + self.errors += 1 + + def snapshot(self) -> dict[str, int]: + return { + "calls": self.calls, + "inclusiveNs": self.inclusive_ns, + "selfNs": self.self_ns, + "maxNs": self.max_ns, + "errors": self.errors, + } + + +@dataclass +class _PropertyStats(_TimingStats): + operation: str = "" + layer: str = "semantic" + medium: str = "unknown" + exact_input_unique: int = 0 + exact_input_repeats: int = 0 + iteration_calls: int = 0 + iteration_total: int = 0 + iteration_max: int = 0 + iteration_converged: int = 0 + iteration_nonconverged: int = 0 + cache_lookups: int = 0 + cache_hits: int = 0 + cache_misses: int = 0 + + def snapshot(self, *, audit: bool) -> dict[str, object]: + result: dict[str, object] = super().snapshot() + result.update( + { + "operation": self.operation, + "layer": self.layer, + "medium": self.medium, + "cacheLookups": self.cache_lookups, + "cacheHits": self.cache_hits, + "cacheMisses": self.cache_misses, + } + ) + if audit: + result.update( + { + "exactInputUnique": self.exact_input_unique, + "exactInputRepeats": self.exact_input_repeats, + "iterationCalls": self.iteration_calls, + "iterationTotal": self.iteration_total, + "iterationMax": self.iteration_max, + "iterationConverged": self.iteration_converged, + "iterationNonconverged": self.iteration_nonconverged, + } + ) + return result + + +@dataclass +class _ActiveSpan: + trace: PerformanceTrace + name: str + started_ns: int + property_key: str | None = None + property_operation: str | None = None + property_outermost: bool = False + child_ns: int = 0 + + +@dataclass +class PerformanceTrace: + """Mutable counters owned by exactly one :func:`profile_run` context.""" + + mode: ProfileMode + _phases: dict[str, _TimingStats] = field(default_factory=dict, repr=False) + _properties: dict[str, _PropertyStats] = field(default_factory=dict, repr=False) + _property_outermost_ns: int = field(default=0, repr=False) + + @property + def enabled(self) -> bool: + return self.mode != "off" + + def _property_stats( + self, + key: str, + *, + operation: str, + layer: str, + medium: str, + ) -> _PropertyStats: + stats = self._properties.get(key) + if stats is None: + stats = _PropertyStats( + operation=operation, + layer=layer, + medium=medium, + ) + self._properties[key] = stats + return stats + + def _record_span(self, frame: _ActiveSpan, elapsed_ns: int, error: bool) -> None: + self_ns = max(0, elapsed_ns - frame.child_ns) + if frame.property_key is None: + stats = self._phases.setdefault(frame.name, _TimingStats()) + else: + layer, medium, operation = frame.property_key.split("|", 2) + stats = self._property_stats( + frame.property_key, + operation=operation, + layer=layer, + medium=medium, + ) + if frame.property_outermost: + self._property_outermost_ns += elapsed_ns + stats.record(elapsed_ns, self_ns, error) + + def _record_exact_input( + self, + key: str, + *, + operation: str, + layer: str, + medium: str, + fingerprint: object, + ) -> None: + stats = self._property_stats( + key, + operation=operation, + layer=layer, + medium=medium, + ) + shadow_key = (key, fingerprint) + shadow = _PROPERTY_SHADOW.get() + if shadow is None: + # A trace normally installs its own set in ``profile_run``. Keep + # the ContextVar default immutable so no task can accidentally + # share a process-global shadow set. + shadow = set() + _PROPERTY_SHADOW.set(shadow) + if shadow_key in shadow: + stats.exact_input_repeats += 1 + else: + shadow.add(shadow_key) + stats.exact_input_unique += 1 + + def _record_iterations( + self, + key: str, + *, + operation: str, + layer: str, + medium: str, + iterations: int, + converged: bool, + ) -> None: + stats = self._property_stats( + key, + operation=operation, + layer=layer, + medium=medium, + ) + iteration_count = max(0, int(iterations)) + stats.iteration_calls += 1 + stats.iteration_total += iteration_count + stats.iteration_max = max(stats.iteration_max, iteration_count) + if converged: + stats.iteration_converged += 1 + else: + stats.iteration_nonconverged += 1 + + def _record_cache( + self, + key: str, + *, + operation: str, + layer: str, + medium: str, + hits: int, + misses: int, + ) -> None: + stats = self._property_stats( + key, + operation=operation, + layer=layer, + medium=medium, + ) + hit_delta = max(0, hits) + miss_delta = max(0, misses) + stats.cache_hits += hit_delta + stats.cache_misses += miss_delta + stats.cache_lookups += hit_delta + miss_delta + + def snapshot(self) -> dict[str, object]: + """Return a detached, JSON-serializable copy of all counters.""" + + audit = self.mode == "audit" + return { + "mode": self.mode, + "phases": { + name: self._phases[name].snapshot() + for name in sorted(self._phases) + }, + "properties": { + _public_property_key(key): self._properties[key].snapshot(audit=audit) + for key in sorted(self._properties) + }, + "propertyOutermostNs": self._property_outermost_ns, + } + + +_CURRENT_TRACE: ContextVar[PerformanceTrace | None] = ContextVar( + "simulation_performance_trace", + default=None, +) +_ACTIVE_SPANS: ContextVar[tuple[_ActiveSpan, ...]] = ContextVar( + "simulation_performance_spans", + default=(), +) +_PROPERTY_SHADOW: ContextVar[set[object] | None] = ContextVar( + "simulation_property_shadow", + default=None, +) + + +def _public_property_key(key: str) -> str: + layer, medium, operation = key.split("|", 2) + return f"{layer}.{medium}.{operation}" + + +@contextmanager +def _tracked_span( + trace: PerformanceTrace, + name: str, + *, + property_key: str | None = None, + property_operation: str | None = None, + reset_property_shadow: bool = False, +) -> Generator[None, None, None]: + stack = _ACTIVE_SPANS.get() + property_outermost = property_key is not None and not any( + item.property_key is not None for item in stack + ) + frame = _ActiveSpan( + trace=trace, + name=name, + started_ns=perf_counter_ns(), + property_key=property_key, + property_operation=property_operation, + property_outermost=property_outermost, + ) + stack_token = _ACTIVE_SPANS.set((*stack, frame)) + shadow_token: Token[set[object] | None] | None = None + if reset_property_shadow and trace.mode == "audit": + shadow_token = _PROPERTY_SHADOW.set(set()) + error = False + try: + yield + except BaseException: + error = True + raise + finally: + elapsed_ns = max(0, perf_counter_ns() - frame.started_ns) + _ACTIVE_SPANS.reset(stack_token) + if shadow_token is not None: + _PROPERTY_SHADOW.reset(shadow_token) + if stack: + stack[-1].child_ns += elapsed_ns + trace._record_span(frame, elapsed_ns, error) + + +@contextmanager +def profile_run() -> Generator[PerformanceTrace, None, None]: + """Create and bind an isolated trace for one simulation run. + + The yielded trace remains usable after the context exits, which lets the + caller attach ``trace.snapshot()`` to a result without exposing live state. + """ + + trace = PerformanceTrace(mode=PROFILE_MODE) + trace_token = _CURRENT_TRACE.set(trace) + spans_token = _ACTIVE_SPANS.set(()) + shadow_token = _PROPERTY_SHADOW.set(set()) + try: + if trace.enabled: + with _tracked_span(trace, "simulation.total"): + yield trace + else: + yield trace + finally: + _PROPERTY_SHADOW.reset(shadow_token) + _ACTIVE_SPANS.reset(spans_token) + _CURRENT_TRACE.reset(trace_token) + + +@contextmanager +def performance_span( + name: str, + minimum_mode: str = "standard", + reset_property_shadow: bool = False, +) -> Generator[None, None, None]: + """Time a block in the current run, or act as a no-op outside one.""" + + minimum = _minimum_mode(minimum_mode) + trace = _CURRENT_TRACE.get() + if trace is None or not _mode_enabled(minimum): + yield + return + with _tracked_span( + trace, + name, + reset_property_shadow=reset_property_shadow, + ): + yield + + +def profile_phase( + name: str, + minimum_mode: str = "standard", + reset_property_shadow: bool = False, +) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: + """Decorate a simulation phase while preserving the off-mode callable.""" + + minimum = _minimum_mode(minimum_mode) + + def decorate(function: Callable[_P, _R]) -> Callable[_P, _R]: + if not _mode_enabled(minimum): + return function + + if inspect.iscoroutinefunction(function): + + @wraps(function) + async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any: + trace = _CURRENT_TRACE.get() + if trace is None: + return await function(*args, **kwargs) + with _tracked_span( + trace, + name, + reset_property_shadow=reset_property_shadow, + ): + return await function(*args, **kwargs) + + return cast(Callable[_P, _R], async_wrapper) + + @wraps(function) + def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R: + trace = _CURRENT_TRACE.get() + if trace is None: + return function(*args, **kwargs) + with _tracked_span( + trace, + name, + reset_property_shadow=reset_property_shadow, + ): + return function(*args, **kwargs) + + return wrapper + + return decorate + + +def _medium_name(args: tuple[object, ...]) -> str: + if not args: + return "unknown" + owner = args[0] + configured_name = getattr(owner, "name", None) + if isinstance(configured_name, str) and configured_name: + return configured_name + return type(owner).__name__ + + +def _fingerprint(value: object) -> object: + """Build a hashable, bit-exact token without retaining arbitrary objects.""" + + if value is None or isinstance(value, (bool, int, str, bytes)): + return (type(value).__name__, value) + if isinstance(value, float): + return ("float64", struct.pack("!d", value)) + if isinstance(value, tuple): + return ("tuple", tuple(_fingerprint(item) for item in value)) + if isinstance(value, list): + return ("list", tuple(_fingerprint(item) for item in value)) + if isinstance(value, Mapping): + items = [(_fingerprint(key), _fingerprint(item)) for key, item in value.items()] + items.sort(key=repr) + return ("mapping", tuple(items)) + return ( + "object", + type(value).__module__, + type(value).__qualname__, + id(value), + ) + + +def _input_fingerprint( + signature: inspect.Signature | None, + args: tuple[object, ...], + kwargs: dict[str, object], +) -> object: + if signature is not None: + try: + bound = signature.bind(*args, **kwargs) + bound.apply_defaults() + return tuple( + (name, _fingerprint(value)) + for name, value in bound.arguments.items() + ) + except TypeError: + pass + return ( + _fingerprint(args), + tuple(sorted((name, _fingerprint(value)) for name, value in kwargs.items())), + ) + + +def _cache_counts(function: Callable[..., object]) -> tuple[int, int] | None: + cache_info = getattr(function, "cache_info", None) + if not callable(cache_info): + return None + try: + info = cache_info() + return int(info.hits), int(info.misses) + except (AttributeError, TypeError, ValueError): + return None + + +def _copy_cache_api(source: Callable[..., object], target: Callable[..., object]) -> None: + for attribute in ("cache_clear", "cache_info", "cache_parameters"): + value = getattr(source, attribute, None) + if value is not None: + setattr(target, attribute, value) + + +def profile_property( + operation: str, + layer: str = "semantic", + minimum_mode: str = "audit", + capture_inputs: bool = True, + track_cache: bool = False, +) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: + """Decorate one thermodynamic property operation.""" + + minimum = _minimum_mode(minimum_mode) + + def decorate(function: Callable[_P, _R]) -> Callable[_P, _R]: + if not _mode_enabled(minimum): + return function + try: + signature: inspect.Signature | None = inspect.signature(function) + except (TypeError, ValueError): + signature = None + + def prepare( + args: tuple[object, ...], + kwargs: dict[str, object], + ) -> tuple[PerformanceTrace | None, str, str, tuple[int, int] | None]: + trace = _CURRENT_TRACE.get() + medium = _medium_name(args) + key = f"{layer}|{medium}|{operation}" + if trace is not None and trace.mode == "audit" and capture_inputs: + trace._record_exact_input( + key, + operation=operation, + layer=layer, + medium=medium, + fingerprint=_input_fingerprint(signature, args, kwargs), + ) + before = _cache_counts(function) if trace is not None and track_cache else None + return trace, medium, key, before + + def finish_cache( + trace: PerformanceTrace | None, + medium: str, + key: str, + before: tuple[int, int] | None, + ) -> None: + if trace is None or before is None: + return + after = _cache_counts(function) + if after is None: + return + trace._record_cache( + key, + operation=operation, + layer=layer, + medium=medium, + hits=after[0] - before[0], + misses=after[1] - before[1], + ) + + if inspect.iscoroutinefunction(function): + + @wraps(function) + async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any: + object_args = cast(tuple[object, ...], args) + object_kwargs = cast(dict[str, object], kwargs) + trace, medium, key, before = prepare(object_args, object_kwargs) + try: + if trace is None: + return await function(*args, **kwargs) + with _tracked_span( + trace, + f"property.{_public_property_key(key)}", + property_key=key, + property_operation=operation, + ): + return await function(*args, **kwargs) + finally: + finish_cache(trace, medium, key, before) + + _copy_cache_api(function, async_wrapper) + return cast(Callable[_P, _R], async_wrapper) + + @wraps(function) + def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R: + object_args = cast(tuple[object, ...], args) + object_kwargs = cast(dict[str, object], kwargs) + trace, medium, key, before = prepare(object_args, object_kwargs) + try: + if trace is None: + return function(*args, **kwargs) + with _tracked_span( + trace, + f"property.{_public_property_key(key)}", + property_key=key, + property_operation=operation, + ): + return function(*args, **kwargs) + finally: + finish_cache(trace, medium, key, before) + + _copy_cache_api(function, wrapper) + return wrapper + + return decorate + + +def record_property_iterations( + operation: str, + iterations: int, + converged: bool, +) -> None: + """Record inverse-property solver iterations in audit mode.""" + + trace = _CURRENT_TRACE.get() + if trace is None or trace.mode != "audit": + return + for frame in reversed(_ACTIVE_SPANS.get()): + if ( + frame.property_key is not None + and frame.property_operation == operation + ): + layer, medium, _unused_operation = frame.property_key.split("|", 2) + trace._record_iterations( + frame.property_key, + operation=operation, + layer=layer, + medium=medium, + iterations=iterations, + converged=converged, + ) + return + key = f"semantic|unknown|{operation}" + trace._record_iterations( + key, + operation=operation, + layer="semantic", + medium="unknown", + iterations=iterations, + converged=converged, + ) + + +__all__ = [ + "PROFILE_MODE", + "PerformanceTrace", + "performance_span", + "profile_phase", + "profile_property", + "profile_run", + "record_property_iterations", +] diff --git a/app/simulation/solvers/algebraic.py b/app/simulation/solvers/algebraic.py index db3b9d5..50b4920 100644 --- a/app/simulation/solvers/algebraic.py +++ b/app/simulation/solvers/algebraic.py @@ -14,6 +14,7 @@ from app.simulation.components.amesim.flow.pipes import ( ) from app.simulation.core.equations import EquationResidual from app.simulation.core.ports import PortState, VariableRole +from app.simulation.performance import profile_phase from app.simulation.systems.network import SimulationNetwork @@ -1286,6 +1287,7 @@ class PressureFlowSolver: ), } + @profile_phase("simulation.pressure_flow", minimum_mode="audit") def solve( self, *, diff --git a/app/simulation/solvers/pneumatic_volume.py b/app/simulation/solvers/pneumatic_volume.py index 8c74551..cd5f78d 100644 --- a/app/simulation/solvers/pneumatic_volume.py +++ b/app/simulation/solvers/pneumatic_volume.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass from math import isfinite +from app.simulation.performance import profile_phase from app.simulation.systems.network import Endpoint, SimulationNetwork @@ -36,6 +37,7 @@ class PneumaticVolumeResolver: result[second] = first return result + @profile_phase("simulation.pneumatic_volume", minimum_mode="audit") def solve(self) -> PneumaticVolumeDiagnostics: for component in self.network.components.values(): for definition in component.active_port_definitions: diff --git a/app/simulation/solvers/signal.py b/app/simulation/solvers/signal.py index f0f467a..64d5bc4 100644 --- a/app/simulation/solvers/signal.py +++ b/app/simulation/solvers/signal.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from math import isfinite from typing import Protocol +from app.simulation.performance import profile_phase from app.simulation.systems.network import Endpoint, SimulationNetwork @@ -47,6 +48,7 @@ class SignalResolver: ] self.last_diagnostics: SignalSolveDiagnostics | None = None + @profile_phase("simulation.signal", minimum_mode="audit") def solve(self, time: float) -> SignalSolveDiagnostics: for component in self.network.components.values(): signal_output_values = getattr(component, "signal_output_values", None) diff --git a/app/simulation/solvers/solver.py b/app/simulation/solvers/solver.py index 3332de7..3113904 100644 --- a/app/simulation/solvers/solver.py +++ b/app/simulation/solvers/solver.py @@ -1,6 +1,7 @@ from __future__ import annotations from app.simulation.core.errors import RecoverableTrialStateError +from app.simulation.performance import profile_phase import math from dataclasses import dataclass @@ -1018,6 +1019,7 @@ def _integrate_scipy_stepwise( ) +@profile_phase("simulation.integration") def integrate_ode( rhs: Callable[[float, list[float]], list[float]], initial_state: list[float], diff --git a/app/simulation/solvers/stream.py b/app/simulation/solvers/stream.py index c017260..668f0b2 100644 --- a/app/simulation/solvers/stream.py +++ b/app/simulation/solvers/stream.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass from app.simulation.core.base import DynamicComponent +from app.simulation.performance import profile_phase from app.simulation.systems.network import Endpoint, SimulationNetwork @@ -83,14 +84,33 @@ class StreamResolver: ) return values + @profile_phase("simulation.refresh", minimum_mode="audit") + def _refresh_dynamic_components( + self, + components: list[DynamicComponent], + ) -> None: + for component in components: + component.refresh_thermodynamic_ports() + + @profile_phase("simulation.refresh", minimum_mode="audit") + def _refresh_stream_components( + self, + connected: dict[str, dict[str, float]], + ) -> None: + for component in self.network.components.values(): + if isinstance(component, DynamicComponent): + component.refresh_thermodynamic_ports() + else: + component.update_stream_outflows(connected[component.name]) + + @profile_phase("simulation.stream", minimum_mode="audit") def solve(self) -> tuple[StreamSolveDiagnostics, dict[str, dict[str, float]]]: dynamic_components = [ component for component in self.network.components.values() if isinstance(component, DynamicComponent) ] - for component in dynamic_components: - component.refresh_thermodynamic_ports() + self._refresh_dynamic_components(dynamic_components) max_delta = 0.0 for iteration in range(1, self.max_iterations + 1): @@ -100,11 +120,7 @@ class StreamResolver: for port_name, port in component.ports.items() } connected = self.connected_enthalpies() - for component in self.network.components.values(): - if isinstance(component, DynamicComponent): - component.refresh_thermodynamic_ports() - else: - component.update_stream_outflows(connected[component.name]) + self._refresh_stream_components(connected) deltas = [ abs(port.h_outflow - previous[(component.name, port_name)]) diff --git a/app/simulation/systems/generic.py b/app/simulation/systems/generic.py index 05cd2ae..189ba2c 100644 --- a/app/simulation/systems/generic.py +++ b/app/simulation/systems/generic.py @@ -7,6 +7,7 @@ from typing import Literal from app.simulation.core.base import DynamicComponent from app.simulation.core.metadata import ResultVariableMetadata +from app.simulation.performance import performance_span, profile_phase from app.simulation.solvers.algebraic import PressureFlowSolver from app.simulation.solvers.mechanical import ( MechanicalConstraintGroup, @@ -337,6 +338,7 @@ def simulation_sample_times( class GenericFluidSystem: """Topology-driven, semi-explicit fluid simulation for registered components.""" + @profile_phase("simulation.system_construction") def __init__(self, network: SimulationNetwork) -> None: issues = simulation_preparation_issues(network) if issues: @@ -465,14 +467,18 @@ class GenericFluidSystem: "colorGroupCount": group_count, } + @profile_phase( + "simulation.closure", + minimum_mode="audit", + reset_property_shadow=True, + ) def _close_current_state(self, time: float) -> dict[str, dict[str, float]]: signal = self.signal_resolver.solve(time) self.signal_propagation_count += signal.propagated self.pressure_flow_solver.propagate_equal_efforts(("x", "v")) pneumatic_volume = self.pneumatic_volume_resolver.solve() self.pneumatic_volume_propagation_count += pneumatic_volume.propagated - for component in self.dynamic_components: - component.refresh_thermodynamic_ports() + self._refresh_dynamic_components() algebraic = self.pressure_flow_solver.solve( effort_variables=("p",), ) @@ -549,18 +555,31 @@ class GenericFluidSystem: ) return connected_h + @profile_phase("simulation.refresh", minimum_mode="audit") + def _refresh_dynamic_components(self) -> None: + for component in self.dynamic_components: + component.refresh_thermodynamic_ports() + + @profile_phase("simulation.derivatives", minimum_mode="audit") + def _state_derivatives( + self, + connected_h: dict[str, dict[str, float]], + ) -> list[float]: + return self.pneumatic_storage_reducer.coupled_derivatives( + self.mechanical_state_reducer.state_derivatives(connected_h) + ) + def consistent_initial_state_vector(self, time: float = 0.0) -> list[float]: state = self.initial_state_vector() self.apply_state_vector(state) self._close_current_state(time) return state + @profile_phase("simulation.rhs", minimum_mode="audit") def rhs(self, _time: float, state_vector: list[float]) -> list[float]: self.apply_state_vector(state_vector) connected_h = self._close_current_state(_time) - return self.pneumatic_storage_reducer.coupled_derivatives( - self.mechanical_state_reducer.state_derivatives(connected_h) - ) + return self._state_derivatives(connected_h) def _append_current_state(self, series: dict[str, list[float]]) -> None: for component in self.network.components.values(): @@ -603,20 +622,26 @@ class GenericFluidSystem: progress_callback(last_reported_progress, phase) report_progress(0.0, "initializing", force=True) - integration_config = config - if isinstance(config.atol, (int, float)): - integration_config = replace( - config, - atol=self.mechanical_state_reducer.absolute_tolerances( - float(config.atol) - ), + with performance_span("simulation.sample_initialization"): + integration_config = config + if isinstance(config.atol, (int, float)): + integration_config = replace( + config, + atol=self.mechanical_state_reducer.absolute_tolerances( + float(config.atol) + ), + ) + t_eval = simulation_sample_times(config, sample_step) + signal_event_times = self.signal_resolver.event_times( + config.t_start, + config.t_stop, + ) + initial_state = self.consistent_initial_state_vector(config.t_start) + jac_sparsity = ( + self.jacobian_sparsity() + if integration_config.method in {"BDF", "Radau"} + else None ) - t_eval = simulation_sample_times(config, sample_step) - signal_event_times = self.signal_resolver.event_times( - config.t_start, - config.t_stop, - ) - initial_state = self.consistent_initial_state_vector(config.t_start) report_progress(0.0, "integrating", force=True) duration = config.t_stop - config.t_start furthest_solver_time = config.t_start @@ -651,11 +676,7 @@ class GenericFluidSystem: if self.mechanical_state_reducer.has_state_events else None ), - jac_sparsity=( - self.jacobian_sparsity() - if integration_config.method in {"BDF", "Radau"} - else None - ), + jac_sparsity=jac_sparsity, ) if isinstance(solution, ODESolution): run_status: SimulationRunStatus = solution.status @@ -719,112 +740,116 @@ class GenericFluidSystem: for segment in solver_segment_diagnostics ) - series: dict[str, list[float]] = {"time": []} - postprocessing_error: Exception | None = None - self.mechanical_state_reducer.reset_constraint_modes() - for time_index in range(len(times)): - if ( - run_status == "completed" - and cancel_check is not None - and cancel_check() - ): - run_status = "cancelled" - result_message = "Simulation was stopped while preparing partial results." - break - state = [ - float(solution.y[state_index][time_index]) - for state_index in range(len(solution.y)) - ] - try: - self.apply_state_vector(state) - self._close_current_state(times[time_index]) - self._append_current_state(series) - series["time"].append(times[time_index]) - except Exception as exc: - run_status = "failed" - result_message = str(exc) - postprocessing_error = exc - break - if len(series["time"]) < 2: - if postprocessing_error is not None: - raise postprocessing_error - if integration_error is not None: - raise integration_error + with performance_span("simulation.postprocessing"): + series: dict[str, list[float]] = {"time": []} + postprocessing_error: Exception | None = None + self.mechanical_state_reducer.reset_constraint_modes() + for time_index in range(len(times)): + if ( + run_status == "completed" + and cancel_check is not None + and cancel_check() + ): + run_status = "cancelled" + result_message = ( + "Simulation was stopped while preparing partial results." + ) + break + state = [ + float(solution.y[state_index][time_index]) + for state_index in range(len(solution.y)) + ] + try: + self.apply_state_vector(state) + self._close_current_state(times[time_index]) + self._append_current_state(series) + series["time"].append(times[time_index]) + except Exception as exc: + run_status = "failed" + result_message = str(exc) + postprocessing_error = exc + break + if len(series["time"]) < 2: + if postprocessing_error is not None: + raise postprocessing_error + if integration_error is not None: + raise integration_error - final = { - key: values[-1] - for key, values in series.items() - if key != "time" and values - } - diagnostics = { - "integration": { - "method": integration_config.method, - "jacobianSparsity": jacobian_diagnostics, - "segmentCount": len(solver_segment_diagnostics), - "segments": solver_segment_diagnostics, - "totals": solver_totals, - }, - "pressureFlow": { - "solveCount": self.algebraic_solve_count, - "maxScaledResidual": self.max_algebraic_residual, - "maxEvaluationsPerSolve": self.max_algebraic_evaluations, - "last": ( - self.pressure_flow_solver.last_diagnostics.as_dict() - if self.pressure_flow_solver.last_diagnostics is not None - else None + with performance_span("simulation.result_assembly"): + final = { + key: values[-1] + for key, values in series.items() + if key != "time" and values + } + diagnostics = { + "integration": { + "method": integration_config.method, + "jacobianSparsity": jacobian_diagnostics, + "segmentCount": len(solver_segment_diagnostics), + "segments": solver_segment_diagnostics, + "totals": solver_totals, + }, + "pressureFlow": { + "solveCount": self.algebraic_solve_count, + "maxScaledResidual": self.max_algebraic_residual, + "maxEvaluationsPerSolve": self.max_algebraic_evaluations, + "last": ( + self.pressure_flow_solver.last_diagnostics.as_dict() + if self.pressure_flow_solver.last_diagnostics is not None + else None + ), + }, + "stream": { + "maxIterationsPerSolve": self.max_stream_iterations, + "maxThermofluidIterations": self.max_thermofluid_iterations, + "last": ( + self.stream_resolver.last_diagnostics.as_dict() + if self.stream_resolver.last_diagnostics is not None + else None + ), + }, + "signal": { + "propagations": self.signal_propagation_count, + "eventTimes": list(signal_event_times), + "last": ( + self.signal_resolver.last_diagnostics.as_dict() + if self.signal_resolver.last_diagnostics is not None + else None + ), + }, + "pneumaticVolume": { + "propagations": self.pneumatic_volume_propagation_count, + "last": ( + self.pneumatic_volume_resolver.last_diagnostics.as_dict() + if self.pneumatic_volume_resolver.last_diagnostics is not None + else None + ), + }, + "stateCount": len(initial_state), + "sampleCount": len(series["time"]), + } + variables = tuple( + variable + for variable in self.network.result_variable_metadata() + if variable.key in series + ) + report_progress( + 1.0 if run_status == "completed" else max(0.0, last_reported_progress), + "complete" if run_status == "completed" else run_status, + force=True, + ) + return GenericSimulationResult( + success=run_status == "completed" and bool(solution.success), + status=run_status, + message=result_message, + simulated_until=( + float(series["time"][-1]) + if series["time"] + else float(config.t_start) ), - }, - "stream": { - "maxIterationsPerSolve": self.max_stream_iterations, - "maxThermofluidIterations": self.max_thermofluid_iterations, - "last": ( - self.stream_resolver.last_diagnostics.as_dict() - if self.stream_resolver.last_diagnostics is not None - else None - ), - }, - "signal": { - "propagations": self.signal_propagation_count, - "eventTimes": list(signal_event_times), - "last": ( - self.signal_resolver.last_diagnostics.as_dict() - if self.signal_resolver.last_diagnostics is not None - else None - ), - }, - "pneumaticVolume": { - "propagations": self.pneumatic_volume_propagation_count, - "last": ( - self.pneumatic_volume_resolver.last_diagnostics.as_dict() - if self.pneumatic_volume_resolver.last_diagnostics is not None - else None - ), - }, - "stateCount": len(initial_state), - "sampleCount": len(series["time"]), - } - variables = tuple( - variable - for variable in self.network.result_variable_metadata() - if variable.key in series - ) - report_progress( - 1.0 if run_status == "completed" else max(0.0, last_reported_progress), - "complete" if run_status == "completed" else run_status, - force=True, - ) - return GenericSimulationResult( - success=run_status == "completed" and bool(solution.success), - status=run_status, - message=result_message, - simulated_until=( - float(series["time"][-1]) - if series["time"] - else float(config.t_start) - ), - requested_stop_time=float(config.t_stop), - variables=variables, - series=series, - final=final, - diagnostics=diagnostics, - ) + requested_stop_time=float(config.t_stop), + variables=variables, + series=series, + final=final, + diagnostics=diagnostics, + ) diff --git a/app/system_xml.py b/app/system_xml.py index f90cf8e..7bc511b 100644 --- a/app/system_xml.py +++ b/app/system_xml.py @@ -10,6 +10,7 @@ from typing import Literal from lxml import etree from app.simulation.core.ports import PortDefinition +from app.simulation.performance import profile_phase from app.simulation.registry import COMPONENT_MODEL_REGISTRY, ParameterSpec from app.simulation.solvers.solver import SolveIVPConfig from app.simulation.systems.generic import ( @@ -191,6 +192,7 @@ class SystemXmlValidationReport: return result +@profile_phase("simulation.xml_validation") def validate_system_xml_document(source: bytes | str) -> SystemXmlValidationReport: xml_bytes = source.encode("utf-8") if isinstance(source, str) else source if not xml_bytes.strip(): diff --git a/docs/README.md b/docs/README.md index 43e1ab4..0da9383 100644 --- a/docs/README.md +++ b/docs/README.md @@ -9,6 +9,11 @@ - [更新日志 2026-08-15](更新日志-2026-08-15.md) +## 求解与性能 + +- [后端求解逻辑与效率优化调研](后端求解逻辑与效率优化调研.md) +- [仿真性能评估 2026-08-15](仿真性能评估-2026-08-15.md) + ## 模型开发 1. [组件模型建模规范 v1](component-model-authoring-spec-v1.md) diff --git a/docs/仿真性能评估-2026-08-15.md b/docs/仿真性能评估-2026-08-15.md new file mode 100644 index 0000000..5e7e8ca --- /dev/null +++ b/docs/仿真性能评估-2026-08-15.md @@ -0,0 +1,119 @@ +# SystemSimulationApp 仿真性能评估（2026-08-15） + +> 代码基线：`model-development@6a06489`，随后只加入本报告所述的可选埋点和基准工具。 +> 本次评估的是前端流式接口实际使用的 System XML 求解路径；所有时间均为本机实测，不代表其他机器的绝对性能。 + +## 1. 结论 + +1. **压力—流量闭合是当前首要热点。** 深度审计中，三个气动短算例有 71%～85% 的计时落在 `PressureFlowSolver.solve()` 的包含时间内。它同时包含残差组装及其触发的物性调用，不能与物性时间相加。 +2. **物性调用存在很高的完全相同输入重复率。** 按每次代数闭合重置精确输入影子集合后，空气链路、空气分支和氦气阶跃的重复率分别为 91.2%、96.5% 和 82.3%。空气公式很便宜，不能只凭重复率加缓存；Peng–Robinson 氦气更值得优化。 +3. **现有两项氦气 LRU 精确缓存有效。** 冷缓存审计中，`properties_from_mU` 命中率 95.5%，`temperature_from_pressure_enthalpy` 命中率 78.6%；21 次配对端到端测试中，暖缓存比每次清空缓存快约 7.9%。这些缓存已经存在于评估基线，本次没有新增或改变缓存算法。 +4. **长仿真的时间主要花在积分阶段。** 10 s 氦气均压算例耗时约 10.6～11.5 s，其中标准埋点测得积分占 90.5%，初始化约 4.2%，逐采样点后处理约 5.0%。 +5. **结果 JSON 暂不是这些算例的首要矛盾。** 四个短算例的最终 NDJSON 结果约 29～59 KiB，编码中位数约 0.4～1.2 ms；501 个采样点的长算例约 507 KiB，编码约 18.5 ms。 +6. **首次仿真有明显冷启动。** 新 Python 进程第一次短算例约 0.71 s，预热后同类算例约 0.06～0.13 s。剖析表明首次进入 SciPy 求解路径的惰性导入占了主要差额；这是服务首请求延迟，不是稳态吞吐。 +7. **用户提供的 demo XML 尚不能形成完整性能样本。** 它仍在 `0.000175 s` 左右因 `Initial guess is outside of provided bounds` 失败；深度埋点确认错误发生在压力—流量闭合。本批只记录失败路径，没有顺带改变求解器容错行为。 + +## 2. 埋点实现与污染控制 + +性能开关由进程启动环境变量 `SIMULATIONAPP_PROFILE` 决定： + +| 模式 | 用途 | 记录内容 | 适合场景 | +| --- | --- | --- | --- | +| `off` | 正常运行，默认值 | 不在响应中加入性能数据；装饰器在模块加载时直接返回原函数 | 正式仿真和最终性能对比 | +| `standard` | 低开销阶段统计 | XML 校验、网络编译、系统构造、初始化、积分、后处理、结果组装 | 日常定位“大阶段” | +| `audit` | 深度审计 | 再展开 RHS、代数闭合、压力流量、stream、刷新、导数和物性内核 | 短算例诊断、调用频率与缓存评估 | + +一次运行使用一个 `ContextVar` 隔离的 `PerformanceTrace`，不会把不同仿真任务的阶段计数混在一起。成功或失败的求解结果在 profiling 模式下都会把快照放入 `diagnostics.performance`。主要字段为： + +- 阶段：`calls`、`inclusiveNs`、`selfNs`、`maxNs`、`errors`； +- 物性：上述时间字段，以及介质、操作、缓存查询/命中/未命中； +- audit 专有：闭合内精确输入唯一数/重复数、逆解迭代总数/最大值/收敛与未收敛次数； +- `propertyOutermostNs`：只累计最外层物性调用，避免把嵌套 PR 内核时间重复相加。 + +标准模式只保留低频的大阶段计时。21 次氦气阶跃配对运行中，标准模式相对关闭模式的中位开销为 1.9%；四个短算例分开校准为 0.5%～2.9%。audit 会逐次生成精确指纹并计时，短算例可慢到约 2.5～5 倍，因此 audit 数据用于定位和计数，最终优化收益必须回到 `off` 模式复测。 + +将当前代码的 `off` 模式与备份提交 `6a06489` 同时运行 21 次氦气阶跃，墙钟中位数差为约 0.3%，处于本机噪声范围。也就是说，默认关闭时没有观察到稳定的热路径退化。 + +## 3. 测试方法 + +环境：Windows 11、Python 3.12.3、SciPy 1.18.0、64 位 Intel 处理器。仓库没有 PyInstaller/Nuitka 等可执行文件构建链，本次直接使用项目实际启动后端的 `.venv-win` 解释器。把同一 Python 代码再包成单文件只会混入解包和启动成本，不会使这里的求解内核更接近生产路径。 + +基准工具入口： + +```powershell +.venv-win\Scripts\python.exe -m app.simulation.benchmark_performance ` + --mode audit --warmups 1 --runs 3 ` + --factory "helium_step=tests.test_amesim_pnvo001_signal_xml:high_pressure_helium_step_project" ` + --output app/data/performance-evaluations/helium-step.json +``` + +工具默认传入取消检查回调，从而走与前端流式仿真相同的低层逐步积分路径。它记录墙钟、进程 CPU、最终 NDJSON 编码、输入 SHA-256 和完整性能快照。原始 JSON 写入被 Git 忽略的 `app/data/performance-evaluations/`，避免把机器相关的大量样本提交到仓库。 + +本次代表算例： + +| 算例 | 内容 | 暖机后 `off` 墙钟中位数 | 重复次数 | +| --- | --- | ---: | ---: | +| `air_chain` | 空气气缸—节流孔—管路—储罐 | 62.4 ms | 9 | +| `air_branched` | 空气分支网络 | 130.3 ms | 9 | +| `helium_step` | 高压 PR 氦气、信号阶跃阀 | 65.2 ms | 9 | +| `mechanical_contact` | MECMAS21/LSTP00A 弹性接触 | 33.0 ms | 9 | +| `helium_long` | 10 s PR 氦气均压、501 个输出点 | 10.63 s | 1 | + +短算例先暖机 2 次再测 9 次；缓存 A/B 使用两个同时启动的独立进程各暖机 5 次、测量 21 次，以尽量抵消瞬时系统负载。长算例只测 1 次，因此它只用于判断数量级与阶段占比。 + +## 4. 深度阶段结果 + +下表时间是 audit 中位数，会包含审计自身开销；调用数和相对热点比绝对时间更可靠。 + +| 算例 | RHS | 完整闭合 | 压力流量求解 | 压力流量包含时间占 audit 总时间 | 物性调用 | 闭合内精确重复率 | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | +| `air_chain` | 33 | 37 | 74 | 77.0% | 4,265 | 91.2% | +| `air_branched` | 23 | 26 | 56 | 84.6% | 7,378 | 96.5% | +| `helium_step` | 79 | 102 | 235 | 71.2% | 11,121 | 82.3% | +| `mechanical_contact` | 74 | 78 | 156 | 29.5% | 0 | 不适用 | + +当前热流耦合不是旧文档所写的固定 2～3 次压力求解。每次闭合先做 1 次压力求解，然后最多执行 25 轮 `stream → pressure-flow` 固定点；也就是说理论上最多 26 次。本次四个算例的平均压力求解次数/闭合分别为 2.00、2.15、2.30 和 2.00。优化时应编译依赖/脏标记并减少不必要的全网 pass，但不能直接删除第二轮，否则会重新引入求值历史依赖并破坏有限差分 Jacobian。 + +## 5. 物性调用与缓存结果 + +氦气阶跃的冷缓存 audit 代表运行： + +| 操作 | 调用 | 命中/未命中 | 命中率 | 真实逆解次数 | 平均迭代 | 最大迭代 | 未收敛 | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| `properties_from_mU` | 1,930 | 1,843 / 87 | 95.5% | 87 | 3.99 | 5 | 0 | +| `temperature_from_pressure_enthalpy` | 398 | 313 / 85 | 78.6% | 85 | 5.00 | 5 | 0 | + +暖机后以相同配置重复运行，这两项在代表快照中均为 100% 命中，说明当前精确 LRU 能跨同配置运行复用确定性轨迹。关闭埋点的端到端配对结果为：暖缓存中位数 77.05 ms，每次清空缓存为 83.69 ms；换算为暖缓存约快 7.9%。 + +audit 的自身时间排序还显示：`isentropic_density_pressure_factor` 调用 398 次，`density` 业务入口及 PR 密度内核各调用 1,198 次，PR `compressibility_roots` 调用 1,200 次。同一 `(p,T)` 周围存在“等熵因子内部求密度，随后流量公式再次求密度”的重复机会。这里应优先复用同一闭合内的精确结果或合并 API；不要用四舍五入/容差键缓存，否则会在残差函数中制造平台并影响 ODE/least-squares 的有限差分。 + +空气算例虽然精确重复率更高，但理想气体公式本身只有少量算术。对这些廉价函数增加字典查询可能比重算更慢，应先做专门 A/B，不应套用氦气结论。 + +## 6. 输出与失败样本 + +| 算例 | 最终结果大小 | NDJSON 编码中位数 | +| --- | ---: | ---: | +| `air_chain` | 29.2 KiB | 0.39 ms | +| `air_branched` | 59.1 KiB | 1.21 ms | +| `helium_step` | 55.5 KiB | 1.10 ms | +| `mechanical_contact` | 51.4 KiB | 0.62 ms | +| `helium_long` | 507.4 KiB | 18.48 ms | + +用户 demo 的输入 SHA-256 为 `27048a99da0a21922d75785b760c3b5d04be3349b8aef6fbfedfd811d87ef1d5`。audit 失败运行记录到 80 次 RHS、83 次闭合、165 次压力流量求解，最后一项各有 1 次错误；这与此前定位的低压试探态越过 `least_squares` 初值边界一致。由于没有到达 10 s 终点，不能把其 1.08 s 失败耗时当作完整模型性能。 + +## 7. 后续优化顺序 + +1. **先优化压力流量执行计划。** 继续预编译组件/连接残差归属、显式赋值顺序和热流耦合脏标记；增加快路径命中率、非线性 `nfev` 累计耗时，区分“全网扫描慢”与“非线性迭代慢”。 +2. **再减少 PR 物性重复。** 复用组件当前 `(m,U,V)` 的状态恢复结果，合并等熵因子与密度读取；沿用精确键、有界容量和按仿真隔离原则。现有 LRU 已带来约 8% 的短算例收益，不应回退。 +3. **处理冷启动。** 若首请求延迟重要，可在 worker 启动时显式导入 SciPy 求解模块或运行一个极小、无业务副作用的预热模型；不要把约 0.65 s 冷启动归因到每次仿真。 +4. **长算例再看后处理复用。** 当前代表长算例的积分占 90.5%，所以积分/闭合仍优先；当采样更密或变量更多时，再评估复用已接受状态闭合、按需变量和降采样。 +5. **把 demo 数值容错作为独立修复。** 统一压力可行域与初值投影、将试探态错误转为可恢复拒步，并在闭合前刷新外部容积缓存；该改动需要单独回归，不能混入性能优化提交。 + +## 8. 本次评估边界 + +- 没有固定 CPU 亲和性或关闭后台程序，短算例绝对时间存在数毫秒波动，因此以中位数和配对实验为主。 +- audit 会显著改变廉价函数的单次耗时；不能把 audit 的物性毫秒数直接当成关闭埋点后的真实占比。 +- 当前 LRU 命中/未命中来自调用前后的全局 `cache_info()` 差值；本报告均为单任务运行。多个 audit 仿真线程同时调用同一缓存时，阶段/物性调用仍按 trace 隔离，但缓存命中差值可能交错，不能用来做并发结论。 +- 直接抛出 `HTTPException` 的校验/执行异常会结束 trace，但当前不会把快照附到错误响应；demo 属于返回 `failed` 部分结果的路径，所以本报告能够取得其失败快照。 +- 本批没有测峰值 RSS、1/2/4 并发吞吐、浏览器解析/绘图或 8/32 单元拓扑扩展曲线。 +- 没有为评估引入新的近似缓存、容差调整或求解器算法变更；所有性能结论都与数值优化改动解耦。 diff --git a/docs/后端求解逻辑与效率优化调研.md b/docs/后端求解逻辑与效率优化调研.md index f85cc1e..8b1f41d 100644 --- a/docs/后端求解逻辑与效率优化调研.md +++ b/docs/后端求解逻辑与效率优化调研.md @@ -1,7 +1,7 @@ # SystemSimulationApp 后端求解逻辑与效率优化调研（通俗版） -> 调研基线：2026-08-12（System XML v3 迁移后），依据当前仓库代码、配置、说明文档与测试。 -> 本文所称“主求解路径”是当前前端实际调用的 System XML 流式接口；固定 TestModel 和 Test MQL 接口另行说明。文中没有把静态代码分析冒充 CPU、内存实测。 +> 调研基线：2026-08-15（System XML v3 迁移后），依据当前仓库代码、配置、说明文档、测试与阶段埋点。 +> 本文所称“主求解路径”是当前前端实际调用的 System XML 流式接口；固定 TestModel 和 Test MQL 接口另行说明。机器相关的实测结果单独见[仿真性能评估 2026-08-15](仿真性能评估-2026-08-15.md)。 ## 0. 三分钟读懂 @@ -74,7 +74,7 @@ 1. **[已实现] 当前主内核是“半显式 ODE + RHS 内代数闭合”。** 动态组件只把储能状态交给 ODE 积分器；每次计算导数前，系统先传播信号、刷新热力状态、求压力/流量代数网络、传播变容边界、迭代 stream 焓并更新机械加速度。它不是通用 DAE 求解器，也不等价于完整 Modelica `inStream/actualStream` 语义（`README.md:18-20`、`app/simulation/README.md:174-195`）。 2. **[已实现] 当前前端主链路是 System XML 流式仿真。** 浏览器生成 XML，经 `POST /api/system-xml/simulate-stream` 发送；后端以 NDJSON 返回心跳与进度，最后在一个 JSON 行中返回完整结果。不是 WebSocket 或标准 SSE。 3. **[已实现] XML v3 的 `sampleStep` 是输出采样间隔，不是固定积分步长。** 内部的 `max_step`（XML 为 `maxStep`）才是自适应积分步长上限；`BDF/Radau/LSODA/RK45/RK23/DOP853` 均受支持。流式运行因总是提供取消检查，会使用 SciPy 低层求解器逐个已接受步推进。 -4. **[已实现] 每次 RHS 的完整闭合至少调用 2 次、存在外部容积传播时最多调用 3 次压力流量求解。** 积分结束后，每个输出采样点又执行一次完整闭合并提取全部公开结果。这是当前最明确的单任务重复工作来源。 +4. **[已实现] 每次完整闭合先调用 1 次压力流量求解，再执行最多 25 轮 `stream → pressure-flow` 固定点。** 因而每次闭合至少 2 次、理论上最多 26 次压力求解；本次代表算例平均为 2.00～2.30 次。积分结束后，每个输出采样点又执行一次完整闭合并提取结果。 5. **[已实现] 代数求解已有因果化快路径。** 压力流量求解器预编译相等组与显式流量计划，种子残差足够小时不调用非线性优化；否则对全局未知向量调用 SciPy `least_squares`，当前未提供解析 Jacobian 或 `jac_sparsity`。 6. **[已实现] 当前启动脚本是一个 Uvicorn worker。** 每个流式任务再创建一个无并发上限的 daemon 线程和无界队列；没有进程池、集中任务队列、CPU/内存配额或持久化作业系统。同步仿真端点还会在 `async def` 中直接执行 CPU 密集代码。 7. **[推断] 优化应分两条线：** @@ -205,30 +205,25 @@ stream 焓 `h_outflow` 不直接强制相等；标量信号和气动外部容积 - 气体携带的能量按实际流向交给下游； - 若还有机械活塞或控制信号，它们在同一时刻也要一致。 -代码目前不是一次对完，而是按固定顺序做若干轮专项检查。因此一个“计算变化率”的请求会调用 **2 次压力/流量闭合**；涉及外部容积传播时会调用 **3 次**。这也是后文首要优化方向。 +代码目前不是一次对完，而是先建立初始压力解，再让 stream 焓和压力—流量相互迭代到同一个固定点。这样做是为了让当前 RHS 不依赖上一次调用留下的焓/流量历史，并保持有限差分 Jacobian 可重复。 -`GenericFluidSystem._close_current_state()` 的实际顺序见 `app/simulation/systems/generic.py:258-290`： +`GenericFluidSystem._close_current_state()` 的实际顺序见 `app/simulation/systems/generic.py`： | 顺序 | 操作 | 目的 | | ---: | --- | --- | | 1 | `SignalResolver.solve(time)` | 更新时间信号源并从 output 传播到 input | -| 2 | 刷新动态组件热力端口 | 由当前 `m/U/V` 恢复压力、温度、焓等 | -| 3 | 第一次 `PressureFlowSolver.solve()` | 闭合当前压力、流量、机械端口代数关系 | -| 4 | `PneumaticVolumeResolver.solve()` | 沿气动连接传播外部 `volume/volume_flow` | -| 5 | 若发生容积传播，再刷新热力状态并第二次求压力/流量 | 让变容边界进入气室状态关系 | -| 6 | `StreamResolver.solve()` | 按实际流向迭代传播/混合 `h_outflow` | -| 7 | 无条件再次求压力/流量 | 让依赖 stream/温度的构成关系重新闭合 | -| 8 | 更新机械约束加速度 | 为机械状态导数准备 `a` | +| 2 | 传播机械 `x/v` 等值关系 | 把当前机械状态同步到刚性连接端口 | +| 3 | `PneumaticVolumeResolver.solve()` | 沿气动连接传播外部 `volume/volume_flow` | +| 4 | 刷新动态组件热力端口 | 由当前 `m/U/V` 和最新体积恢复压力、温度、焓 | +| 5 | 第一次 `PressureFlowSolver.solve()` | 建立本轮热流固定点的初始压力和流量 | +| 6 | 最多 25 轮 `StreamResolver.solve()` 后再求压力流量 | 让焓、温度引用和构成流量同时收敛；按流量变化判停 | +| 7 | 更新机械约束加速度 | 为机械状态导数准备 `a` | -随后 `rhs()` 才收集各动态组件的导数（`app/simulation/systems/generic.py:298-301`）。 +随后 `rhs()` 才收集各动态组件的导数。 -因此： +因此每次闭合至少有 **2 次**、最多有 **26 次**压力流量求解。外部容积传播会影响初始热力状态，但不再用“有没有容积传播”直接决定固定次数。stream 自身仍有相对容差 `1e-9` 和最多 100 次内部迭代；外层热流固定点最多 25 轮，两层上限不能混为一个数。 -- 无外部容积传播时，每次闭合固定有 **2 次**压力流量求解； -- 有外部容积传播时固定有 **3 次**； -- stream 默认相对容差 `1e-9`、最多 100 次迭代，每轮复制端口焓并扫描组件/端口（`app/simulation/solvers/stream.py:29-119`）。 - -**[发现]** 这套固定顺序没有按模型实际能力裁剪。例如没有信号、没有外部容积源或 stream 不反向影响构成关系的网络，仍经过对应全网 pass。是否能安全删去某一 pass 必须由依赖关系和回归测试决定，不能只凭某个算例结果不变。 +**[发现]** 这套顺序仍没有按模型实际能力完全裁剪。例如没有信号、没有外部容积源或 stream 不反向影响构成关系的网络，仍经过对应全网 pass。是否能安全删去某一 pass 必须由依赖关系、脏标记和回归测试决定，不能只凭某个算例结果不变。 ## 6. 初始化、积分参数与推进方式 @@ -256,7 +251,7 @@ stream 焓 `h_outflow` 不直接强制相等；标量信号和气动外部容积 | `sampleStep`（内部 `sample_step`） | 默认 `0.1 s` | 仅生成输出 `t_eval` 采样网格；工程 JSON 仍暂名 `simulation.step` | | `maxStep`（内部 `max_step`） | 默认 `0.005 s` | 自适应求解器内部已接受步的上限 | | `method` | 默认 `BDF` | BDF、Radau、LSODA、RK45、RK23、DOP853 | -| `rtol` | 通用 XML 路径硬编码 `1e-5` | 外层 ODE 相对误差；用户不可配置 | +| `rtol` | 通用 XML 路径硬编码 `1e-6` | 外层 ODE 相对误差；用户不可配置 | | `atol` | `SolveIVPConfig` 标量默认 `1e-8` | 同时用于不同量纲的全部状态；用户不可配置 | | `first_step` | 默认 `None` | 交给 SciPy；用户不可配置 | | 代数残差容差 | `1e-7` | 压力流量快速路径/接受标准 | @@ -331,9 +326,9 @@ STEP0、UD00 等信号源提供离散事件时刻。积分器先推进到事件 - stream 最大迭代数； - 停止状态及部分错误上下文。 -**[发现]** `_close_current_state()` 中局部变量 `algebraic` 会被后续 pass 覆盖；最大残差/评估统计只采集每次闭合最后一次压力求解，而 `solveCount` 才累计了全部调用。现有响应还没有外层积分 `nfev/njev/nlu`、接受/拒绝步、重启次数、分阶段墙钟时间、热物性调用数、峰值 RSS、队列深度和结果字节数。 +**[已实现]** 积分诊断已经包含分段及汇总的 `nfev/njev/nlu`、已接受步、求解器启动、状态迁移和可恢复重试数。设置 `SIMULATIONAPP_PROFILE=standard|audit` 后，响应还会加入分阶段墙钟时间；audit 进一步记录物性调用、精确重复、缓存命中和逆解迭代。 -因此本文可静态定位重复工作，但不能用现有诊断精确量化每个热点的时间占比。 +**[发现]** `_close_current_state()` 中局部变量 `algebraic` 会被后续 pass 覆盖；最大残差/评估统计只采集每次闭合最后一次压力求解，而 `solveCount` 才累计了全部调用。仍缺少压力快路径命中率/累计 `nfev`、峰值 RSS、任务队列深度等服务级指标。结果字节和编码时间目前由离线基准工具测量，不进入常规 API 响应。 ## 9. 求解时前后端交流 @@ -440,11 +435,11 @@ O(组件 + 连接 + 代数结构) | 热点 | 代码证据 | 影响范围 | 判断 | | --- | --- | --- | --- | -| 每次闭合固定 2～3 次压力流量求解 | `generic.py:258-290` | 每个 RHS、初始化、每个结果采样点 | [已实现] 重复 pass；真实耗时待测 | +| 每次闭合执行 2～26 次压力流量求解 | `generic.py` 的热流固定点 | 每个 RHS、初始化、每个结果采样点 | [实测] 代表气动算例平均 2.00～2.30 次；audit 包含时间占 71%～85% | | stream 每轮复制/扫描并重复刷新 | `stream.py:29-119` | 每个闭合，最多 100 轮 | [已实现]；网络越大越明显 | | 非线性回退用全局有限差分 least-squares | `algebraic.py:927-947` | 快速路径失效时 | [已实现]；大非线性网络潜在陡增 | | 外层刚性积分器看不到显式稀疏 Jacobian | `solver.py:528-1009` | BDF/Radau 的每步/Newton | [已实现] | -| 热物性重复反算 | `mediums.py:199-266` 及各动态组件 refresh | 每个 RHS/闭合 pass | [推断] 需调用计数确认 | +| 热物性重复反算 | `mediums.py` 及各动态组件 refresh | 每个 RHS/闭合 pass | [实测] 闭合内精确重复率 82%～97%；PR 氦气缓存端到端收益约 8% | | 每采样点完整后处理闭合 | `generic.py:397-474` | 输出点 × 全网 | [已实现] | | 每个已接受步构造 dense output | `solver.py:528-914` | 流式逐步路径 | [已实现]；无跨样本/事件时可能浪费 | | 无界求解线程与任务结果驻留 | `main.py:491-520, 773-880` | 并发任务 | [已实现] 稳定性风险，不等于单算例变慢 | @@ -452,7 +447,7 @@ O(组件 + 连接 + 代数结构) ## 13. 优化建议排序 -以下按**预期综合收益**排序；同档位优先低风险、低难度项。收益是基于调用频率与复杂度的代码推断，不是基准测试结果。“单算例”指一个模型的墙钟时间，“吞吐”指多任务服务能力。 +以下按**预期综合收益**排序；同档位优先低风险、低难度项。排序同时参考代码结构和 2026-08-15 的阶段/物性实测，但尚未覆盖大规模拓扑与多任务吞吐。“单算例”指一个模型的墙钟时间，“吞吐”指多任务服务能力。 ### 13.1 先看人话版 @@ -460,7 +455,7 @@ O(组件 + 连接 + 代数结构) | 顺序 | 人话方案 | 为什么可能更快 | 主要风险 | | ---: | --- | --- | --- | -| 1 | 少做重复“瞬时对账” | 当前每次变化率计算固定做 2～3 次压力/流量闭合，调用频率最高 | 少做一轮可能漏掉真实耦合，必须按组件依赖裁剪 | +| 1 | 少做重复“瞬时对账” | 当前每次闭合做初始压力求解和热流固定点，实测压力流量层最热 | 少做一轮可能漏掉真实耦合，必须按组件依赖和脏标记裁剪 | | 2 | 先整理方程，再求解 | 合并重复未知量，把关联较弱的方程分组；大模型回退迭代时收益很高 | 连接、接触和跨域活塞会让分组出错 | | 3 | 给不同状态使用合适的“尺子” | 质量、内能、位置、速度量级差异很大；合理缩放可减少无效内部步 | 容差改变会影响精度和事件时刻 | | 4 | 相同输入不要重复查热物性 | 同一轮闭合中常以相同状态反算压力、温度等 | 缓存失效不严谨会产生错误结果 | @@ -486,15 +481,13 @@ O(组件 + 连接 + 代数结构) ### 13.2 推荐落地顺序 -在改算法前先增加低侵入观测，但不把“加指标”误列为直接加速： +低侵入阶段/物性观测、积分计数和代表算例首轮基准已经落地，但不把“加指标”误列为直接加速。下一步建议： -1. 记录每个闭合 pass 的调用数、墙钟时间、代数 `nfev` 与是否命中快路径； -2. 记录热物性调用数/迭代数、stream 迭代数； -3. 导出外层 `nfev/njev/nlu`、接受/拒绝步、事件与重启次数； -4. 记录状态/结果数组字节数、最终 JSON 字节、任务队列深度和进程 RSS； -5. 用小、中、大三类基准网络定位排名 1～5 的真实占比； -6. 先实施第 1、4、8、9 项的可回滚改造，再决定第 2、3、5 项的深度； -7. 服务并发需求明确后并行推进第 6、7 项。 +1. 给压力流量层补快路径命中、累计非线性 `nfev` 和残差装配时间，继续拆解本次确认的首要热点； +2. 用同一套基准对执行计划裁剪、组件级精确物性复用做 `off` 模式 A/B； +3. 补 1/8/32 单元规模曲线、1/2/4 并发吞吐和峰值 RSS； +4. 记录状态/结果数组字节、任务队列深度；结果 JSON 字节可继续由基准工具测量； +5. 再决定代数分块/Jacobian、结果按需计算和进程 worker 的实施深度。 每项算法改动都应继续验证质量/能量守恒、正反流、stream 混合、机械端挡、信号断点、取消部分结果和 AMESim/TestModel 基线。相关测试证据包括 `tests/test_generic_system_xml_simulation.py:244-533`、`tests/test_core_solver.py:19-508`、`tests/test_amesim_mechanical_public_components.py`。 @@ -516,9 +509,14 @@ O(组件 + 连接 + 代数结构) - 采样上限、超时、心跳和任务名义保留时长。 - 组件参数在单次运行中静态；时间变化通过信号源等模型表达。 -### 推断及必须实测 +### 已有初步实测、仍需扩大样本 + +- 压力流量闭合是当前代表气动短算例的首要热点；物性调用具有高精确重复率，现有 PR 缓存有可见端到端收益。 +- 长氦气代表算例的积分阶段占约 90.5%，后处理约 5%。 +- 上述结论仍需在更大拓扑、更多真实工程和固定硬件环境复测。 + +### 推断及必须继续实测 -- 哪一类闭合 pass、热物性或 Jacobian 估计占主要墙钟时间。 - 单个任务实际占用几个核心、SciPy/BLAS 原生线程数和多任务扩展曲线。 - 典型/最大工程的峰值 RSS、结果 JSON 大小、浏览器内存副本和 sessionStorage 成功率。 - 各优化的实际收益；表中排序应在观测数据出现后更新。 @@ -537,6 +535,8 @@ O(组件 + 连接 + 代数结构) | 气动外部容积 | `app/simulation/solvers/pneumatic_volume.py:21-93` | `PneumaticVolumeResolver` | | 机械因果化与事件 | `app/simulation/solvers/mechanical.py:225-627` | `MechanicalStateReducer` | | ODE 推进 | `app/simulation/solvers/solver.py:37-1009` | `SolveIVPConfig`、`integrate_ode()` | +| 可选性能埋点 | `app/simulation/performance.py` | `profile_run()`、`profile_phase()`、`profile_property()` | +| 可重复性能基准 | `app/simulation/benchmark_performance.py` | `python -m app.simulation.benchmark_performance` | | 前端流式协议 | `frontend/src/App.tsx` | `streamSystemSimulation()`、取消/轮询 | | 启动方式 | `start-backend.bat:17-21` | Uvicorn 单 worker 命令 | | 主路径回归测试 | `tests/test_generic_system_xml_simulation.py`、`tests/test_core_solver.py` | 通用仿真、事件、取消 | diff --git a/tests/test_performance_benchmark.py b/tests/test_performance_benchmark.py new file mode 100644 index 0000000..8d2db1c --- /dev/null +++ b/tests/test_performance_benchmark.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import unittest + +from app.simulation.benchmark_performance import ( + _duration_summary, + _load_factory_xml, + _named_value, + _serialize_result_event, +) + + +def sample_xml_factory() -> bytes: + return b"" + + +class PerformanceBenchmarkToolTests(unittest.TestCase): + def test_named_value_requires_an_explicit_name(self) -> None: + self.assertEqual( + _named_value("chain=tests.example:project", option="--factory"), + ("chain", "tests.example:project"), + ) + with self.assertRaisesRegex(ValueError, "NAME=VALUE"): + _named_value("tests.example:project", option="--factory") + + def test_duration_summary_reports_repeatable_order_statistics(self) -> None: + summary = _duration_summary([5.0, 1.0, 3.0, 2.0, 4.0]) + + self.assertEqual(summary["minimumMs"], 1.0) + self.assertEqual(summary["medianMs"], 3.0) + self.assertEqual(summary["p95Ms"], 5.0) + self.assertEqual(summary["maximumMs"], 5.0) + + def test_factory_loader_accepts_a_bytes_factory(self) -> None: + self.assertEqual( + _load_factory_xml( + "tests.test_performance_benchmark:sample_xml_factory" + ), + b"", + ) + + def test_result_event_serialization_uses_ndjson_shape(self) -> None: + payload = _serialize_result_event( + { + "success": True, + "status": "completed", + "simulatedUntil": 1.0, + "requestedStopTime": 1.0, + } + ) + + self.assertTrue(payload.endswith(b"\n")) + self.assertIn(b'"event":"result"', payload) + self.assertIn(b'"result":{"success":true', payload) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_property_performance_instrumentation.py b/tests/test_property_performance_instrumentation.py new file mode 100644 index 0000000..bd30ad5 --- /dev/null +++ b/tests/test_property_performance_instrumentation.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import unittest +from unittest.mock import Mock, patch + +from app.simulation.components.amesim.media.mediums import ( + AmesimHeliumPengRobinsonMedium, +) + + +class PropertyPerformanceInstrumentationTests(unittest.TestCase): + def setUp(self) -> None: + self.medium = AmesimHeliumPengRobinsonMedium() + self.medium.temperature_from_pressure_enthalpy.cache_clear() + self.medium.properties_from_mU.cache_clear() + + def tearDown(self) -> None: + self.medium.temperature_from_pressure_enthalpy.cache_clear() + self.medium.properties_from_mU.cache_clear() + + def test_temperature_iterations_are_recorded_only_on_cache_miss(self) -> None: + pressure = 15.3e6 + temperature = 293.15 + enthalpy = self.medium.specific_enthalpy_at_pressure( + pressure, + temperature, + ) + + with patch( + "app.simulation.components.amesim.media.mediums." + "record_property_iterations" + ) as record_iterations: + first = self.medium.temperature_from_pressure_enthalpy( + pressure, + enthalpy, + ) + second = self.medium.temperature_from_pressure_enthalpy( + pressure, + enthalpy, + ) + + self.assertEqual(second, first) + record_iterations.assert_called_once() + operation, iterations, converged = record_iterations.call_args.args + self.assertEqual(operation, "temperature_from_pressure_enthalpy") + self.assertEqual(iterations, 5) + self.assertTrue(converged) + + def test_state_recovery_iterations_are_recorded_only_on_cache_miss(self) -> None: + pressure = 15.3e6 + temperature = 293.15 + volume = 0.057 + mass = self.medium.density(pressure, temperature) * volume + internal_energy = mass * self.medium.specific_internal_energy_at_pressure( + pressure, + temperature, + ) + + with patch( + "app.simulation.components.amesim.media.mediums." + "record_property_iterations" + ) as record_iterations: + first = self.medium.properties_from_mU(mass, internal_energy, volume) + second = self.medium.properties_from_mU(mass, internal_energy, volume) + + self.assertIs(second, first) + record_iterations.assert_called_once() + operation, iterations, converged = record_iterations.call_args.args + self.assertEqual(operation, "properties_from_mU") + self.assertEqual(iterations, 5) + self.assertTrue(converged) + + def test_temperature_iteration_limit_is_reported_as_not_converged(self) -> None: + pressure = 123_456.0 + base_temperature = 300.0 + enthalpy = self.medium.specific_enthalpy(base_temperature) + fake_fluid = Mock() + fake_fluid.residual_specific_enthalpy.side_effect = ( + lambda _pressure, temperature: ( + self.medium.cp_ref * 10.0 + if temperature >= base_temperature + else -self.medium.cp_ref * 10.0 + ) + ) + + with ( + patch.object(AmesimHeliumPengRobinsonMedium, "fluid", fake_fluid), + patch( + "app.simulation.components.amesim.media.mediums." + "record_property_iterations" + ) as record_iterations, + ): + self.medium.temperature_from_pressure_enthalpy( + pressure, + enthalpy, + ) + + record_iterations.assert_called_once_with( + "temperature_from_pressure_enthalpy", + 16, + False, + ) + + def test_state_recovery_iteration_limit_is_reported_as_not_converged(self) -> None: + base_temperature = 300.0 + target_internal_energy = self.medium.specific_internal_energy( + base_temperature + ) + fake_fluid = Mock() + fake_fluid.residual_specific_internal_energy_at_density.side_effect = ( + lambda temperature, _density: ( + self.medium.cv * 10.0 + if temperature >= base_temperature + else -self.medium.cv * 10.0 + ) + ) + fake_fluid.pressure_from_density.return_value = 101_325.0 + fake_fluid.residual_specific_enthalpy.return_value = 0.0 + + with ( + patch.object(AmesimHeliumPengRobinsonMedium, "fluid", fake_fluid), + patch( + "app.simulation.components.amesim.media.mediums." + "record_property_iterations" + ) as record_iterations, + ): + self.medium.properties_from_mU( + 1.0, + target_internal_energy, + 1.0, + ) + + record_iterations.assert_called_once_with( + "properties_from_mU", + 16, + False, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_simulation_performance.py b/tests/test_simulation_performance.py new file mode 100644 index 0000000..67385e6 --- /dev/null +++ b/tests/test_simulation_performance.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import asyncio +from contextlib import contextmanager +from functools import lru_cache +import importlib +import os +import unittest +from unittest.mock import patch + +import app.simulation.performance as performance + + +@contextmanager +def profiling_mode(mode: str): + previous = os.environ.get("SIMULATIONAPP_PROFILE") + os.environ["SIMULATIONAPP_PROFILE"] = mode + module = importlib.reload(performance) + try: + yield module + finally: + if previous is None: + os.environ.pop("SIMULATIONAPP_PROFILE", None) + else: + os.environ["SIMULATIONAPP_PROFILE"] = previous + importlib.reload(performance) + + +class SimulationPerformanceTests(unittest.TestCase): + def test_off_mode_decorators_return_original_callables(self) -> None: + with profiling_mode("off") as module: + def phase_function(value: int) -> int: + return value + 1 + + def property_function(value: int) -> int: + return value * 2 + + self.assertIs( + module.profile_phase("phase")(phase_function), + phase_function, + ) + self.assertIs( + module.profile_phase("phase", minimum_mode="off")(phase_function), + phase_function, + ) + self.assertIs( + module.profile_property("property")(property_function), + property_function, + ) + self.assertIs( + module.profile_property( + "property", + minimum_mode="off", + )(property_function), + property_function, + ) + + with module.profile_run() as trace: + self.assertEqual(phase_function(2), 3) + + self.assertEqual( + trace.snapshot(), + { + "mode": "off", + "phases": {}, + "properties": {}, + "propertyOutermostNs": 0, + }, + ) + + def test_standard_mode_records_nested_inclusive_and_self_time(self) -> None: + with profiling_mode("standard") as module: + @module.profile_phase("inner") + def inner() -> None: + return None + + @module.profile_phase("outer") + def outer() -> None: + inner() + + clock = iter((0, 10, 20, 30, 50, 80)) + with patch.object(module, "perf_counter_ns", side_effect=clock): + with module.profile_run() as trace: + outer() + + snapshot = trace.snapshot() + self.assertEqual( + snapshot["phases"]["inner"], + { + "calls": 1, + "inclusiveNs": 10, + "selfNs": 10, + "maxNs": 10, + "errors": 0, + }, + ) + self.assertEqual(snapshot["phases"]["outer"]["inclusiveNs"], 40) + self.assertEqual(snapshot["phases"]["outer"]["selfNs"], 30) + self.assertEqual( + snapshot["phases"]["simulation.total"]["inclusiveNs"], + 80, + ) + self.assertEqual( + snapshot["phases"]["simulation.total"]["selfNs"], + 40, + ) + + def test_errors_are_recorded_and_propagated(self) -> None: + with profiling_mode("standard") as module: + @module.profile_phase("explode") + def explode() -> None: + raise RuntimeError("expected") + + with self.assertRaisesRegex(RuntimeError, "expected"): + with module.profile_run() as trace: + explode() + + snapshot = trace.snapshot() + self.assertEqual(snapshot["phases"]["explode"]["errors"], 1) + self.assertEqual( + snapshot["phases"]["simulation.total"]["errors"], + 1, + ) + + def test_manual_performance_span_uses_the_current_trace(self) -> None: + with profiling_mode("standard") as module: + clock = iter((0, 10, 20, 30)) + with patch.object(module, "perf_counter_ns", side_effect=clock): + with module.profile_run() as trace: + with module.performance_span("manual"): + pass + + snapshot = trace.snapshot() + self.assertEqual(snapshot["phases"]["manual"]["inclusiveNs"], 10) + self.assertEqual(snapshot["phases"]["simulation.total"]["selfNs"], 20) + + def test_nested_properties_only_add_outermost_time_once(self) -> None: + with profiling_mode("standard") as module: + class TestMedium: + name = "TestMedium" + + @module.profile_property("inner", minimum_mode="standard") + def inner(self) -> float: + return 1.0 + + @module.profile_property("outer", minimum_mode="standard") + def outer(self) -> float: + return self.inner() + + medium = TestMedium() + clock = iter((0, 10, 20, 30, 50, 80)) + with patch.object(module, "perf_counter_ns", side_effect=clock): + with module.profile_run() as trace: + self.assertEqual(medium.outer(), 1.0) + + snapshot = trace.snapshot() + outer = snapshot["properties"]["semantic.TestMedium.outer"] + inner = snapshot["properties"]["semantic.TestMedium.inner"] + self.assertEqual(outer["inclusiveNs"], 40) + self.assertEqual(outer["selfNs"], 30) + self.assertEqual(inner["inclusiveNs"], 10) + self.assertEqual(snapshot["propertyOutermostNs"], 40) + + def test_audit_resets_exact_input_shadow_and_records_iterations_and_cache(self) -> None: + with profiling_mode("audit") as module: + class TestMedium: + name = "AuditMedium" + + @module.profile_property("inverse", track_cache=True) + @lru_cache(maxsize=4) + def inverse(self, value: float) -> float: + module.record_property_iterations( + "inverse", + 3 if value == 1.0 else 4, + value == 1.0, + ) + return value * 2.0 + + medium = TestMedium() + + @module.profile_phase("closure", reset_property_shadow=True) + def closure() -> None: + medium.inverse(1.0) + medium.inverse(1.0) + medium.inverse(2.0) + + with module.profile_run() as trace: + closure() + closure() + + metric = trace.snapshot()["properties"][ + "semantic.AuditMedium.inverse" + ] + self.assertEqual(metric["calls"], 6) + self.assertEqual(metric["exactInputUnique"], 4) + self.assertEqual(metric["exactInputRepeats"], 2) + self.assertEqual(metric["iterationCalls"], 2) + self.assertEqual(metric["iterationTotal"], 7) + self.assertEqual(metric["iterationMax"], 4) + self.assertEqual(metric["iterationConverged"], 1) + self.assertEqual(metric["iterationNonconverged"], 1) + self.assertEqual(metric["cacheLookups"], 6) + self.assertEqual(metric["cacheHits"], 4) + self.assertEqual(metric["cacheMisses"], 2) + + self.assertTrue(callable(medium.inverse.cache_clear)) + self.assertTrue(callable(medium.inverse.cache_info)) + self.assertTrue(callable(medium.inverse.cache_parameters)) + medium.inverse.cache_clear() + self.assertEqual(medium.inverse.cache_info().currsize, 0) + + def test_audit_minimum_decorator_is_absent_in_standard_mode(self) -> None: + with profiling_mode("standard") as module: + def kernel(value: float) -> float: + return value + + self.assertIs( + module.profile_property( + "kernel", + layer="kernel", + minimum_mode="audit", + )(kernel), + kernel, + ) + + def test_async_profile_runs_are_context_isolated(self) -> None: + with profiling_mode("standard") as module: + @module.profile_phase("work") + async def work() -> None: + await asyncio.sleep(0) + + async def one_run() -> dict[str, object]: + with module.profile_run() as trace: + await work() + return trace.snapshot() + + async def exercise() -> list[dict[str, object]]: + return await asyncio.gather(one_run(), one_run()) + + snapshots = asyncio.run(exercise()) + self.assertEqual( + [snapshot["phases"]["work"]["calls"] for snapshot in snapshots], + [1, 1], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_simulation_performance_pipeline.py b/tests/test_simulation_performance_pipeline.py new file mode 100644 index 0000000..8802cc9 --- /dev/null +++ b/tests/test_simulation_performance_pipeline.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys +import textwrap +import unittest + + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + + +class SimulationPerformancePipelineTests(unittest.TestCase): + def test_off_mode_does_not_change_result_diagnostics(self) -> None: + script = textwrap.dedent( + """ + from app.main import ( + build_reactflow_system_xml, + run_system_xml_simulation, + ) + from tests.test_amesim_pnl00r_xml import amesim_pnl00r_project + + result = run_system_xml_simulation( + build_reactflow_system_xml(amesim_pnl00r_project()) + ) + assert result["success"], result["message"] + print("performance" in result["diagnostics"]) + """ + ) + environment = os.environ.copy() + environment["SIMULATIONAPP_PROFILE"] = "off" + completed = subprocess.run( + [sys.executable, "-c", script], + cwd=PROJECT_ROOT, + env=environment, + check=True, + capture_output=True, + text=True, + timeout=30, + ) + + self.assertEqual(completed.stdout.strip(), "False") + + def test_standard_mode_reports_coarse_pipeline_phases_per_run(self) -> None: + script = textwrap.dedent( + """ + import json + + from app.main import ( + build_reactflow_system_xml, + run_system_xml_simulation, + ) + from tests.test_amesim_pnl00r_xml import amesim_pnl00r_project + + xml = build_reactflow_system_xml(amesim_pnl00r_project()) + snapshots = [] + for _ in range(2): + result = run_system_xml_simulation(xml) + assert result["success"], result["message"] + performance = result["diagnostics"]["performance"] + snapshots.append( + { + "mode": performance["mode"], + "calls": { + name: metrics["calls"] + for name, metrics in performance["phases"].items() + }, + "propertyCount": len(performance["properties"]), + } + ) + print(json.dumps(snapshots)) + """ + ) + environment = os.environ.copy() + environment["SIMULATIONAPP_PROFILE"] = "standard" + completed = subprocess.run( + [sys.executable, "-c", script], + cwd=PROJECT_ROOT, + env=environment, + check=True, + capture_output=True, + text=True, + timeout=30, + ) + snapshots = json.loads(completed.stdout) + required_phases = { + "simulation.total", + "simulation.xml_validation", + "simulation.network_compilation", + "simulation.system_construction", + "simulation.sample_initialization", + "simulation.integration", + "simulation.postprocessing", + "simulation.result_assembly", + "simulation.response_assembly", + } + audit_only_phases = { + "simulation.rhs", + "simulation.closure", + "simulation.refresh", + "simulation.pressure_flow", + "simulation.pneumatic_volume", + "simulation.stream", + "simulation.signal", + "simulation.derivatives", + } + + self.assertEqual(len(snapshots), 2) + for snapshot in snapshots: + self.assertEqual(snapshot["mode"], "standard") + self.assertTrue(required_phases.issubset(snapshot["calls"])) + self.assertEqual(snapshot["calls"]["simulation.total"], 1) + self.assertEqual(snapshot["calls"]["simulation.xml_validation"], 1) + self.assertEqual(snapshot["calls"]["simulation.network_compilation"], 1) + self.assertEqual(snapshot["calls"]["simulation.integration"], 1) + self.assertEqual(snapshot["calls"]["simulation.postprocessing"], 1) + self.assertEqual(snapshot["calls"]["simulation.result_assembly"], 1) + self.assertEqual(snapshot["calls"]["simulation.response_assembly"], 1) + self.assertTrue(audit_only_phases.isdisjoint(snapshot["calls"])) + self.assertEqual(snapshot["propertyCount"], 0) + + def test_audit_mode_reports_hot_phases_and_property_metrics(self) -> None: + script = textwrap.dedent( + """ + import json + + from app.main import ( + build_reactflow_system_xml, + run_system_xml_simulation, + ) + from tests.test_amesim_pnl00r_xml import amesim_pnl00r_project + + result = run_system_xml_simulation( + build_reactflow_system_xml(amesim_pnl00r_project()) + ) + assert result["success"], result["message"] + print(json.dumps(result["diagnostics"]["performance"])) + """ + ) + environment = os.environ.copy() + environment["SIMULATIONAPP_PROFILE"] = "audit" + completed = subprocess.run( + [sys.executable, "-c", script], + cwd=PROJECT_ROOT, + env=environment, + check=True, + capture_output=True, + text=True, + timeout=30, + ) + snapshot = json.loads(completed.stdout) + + for phase in ( + "simulation.rhs", + "simulation.closure", + "simulation.refresh", + "simulation.pressure_flow", + "simulation.pneumatic_volume", + "simulation.stream", + "simulation.signal", + "simulation.derivatives", + ): + self.assertGreater(snapshot["phases"][phase]["calls"], 0) + self.assertGreater(len(snapshot["properties"]), 0) + self.assertGreater(snapshot["propertyOutermostNs"], 0) + + +if __name__ == "__main__": + unittest.main()