性能分析 API

内存性能分析

用于 QPE 电路编译性能分析的内存测量基础设施。

提供多层内存测量:
  • resource.getrusage(RUSAGE_SELF/RUSAGE_CHILDREN) → 进程生命周期内的峰值 RSS

  • /proc/self/status VmRSS/VmPeak → 当前/峰值 RSS(Linux 内核层面)

  • tracemalloc → 仅 Python 堆

此模块无项目依赖(仅使用标准库)。

class q2m3.profiling.memory.MemorySnapshot(label, rss_mb, vm_peak_mb, maxrss_mb, maxrss_children_mb, tracemalloc_peak_mb, tracemalloc_current_mb, elapsed_s=0.0)[源代码]

基类:object

单个时间点的多层内存快照。

参数:
  • label (str)

  • rss_mb (float)

  • vm_peak_mb (float)

  • maxrss_mb (float)

  • maxrss_children_mb (float)

  • tracemalloc_peak_mb (float)

  • tracemalloc_current_mb (float)

  • elapsed_s (float)

class q2m3.profiling.memory.ProfileResult(molecule, n_system_qubits, n_estimation_wires, n_trotter, n_terms, ir_scale, mode='dynamic', phase_a=None, phase_b=None, phase_c=None, timeline_peak_mb=0.0, timeline_samples=<factory>, ir_analysis=<factory>, prob_sum=0.0, error=None)[源代码]

基类:object

一个参数组合的完整性能分析结果。

参数:
  • molecule (str)

  • n_system_qubits (int)

  • n_estimation_wires (int)

  • n_trotter (int)

  • n_terms (int)

  • ir_scale (int)

  • mode (str)

  • phase_a (MemorySnapshot | None)

  • phase_b (MemorySnapshot | None)

  • phase_c (MemorySnapshot | None)

  • timeline_peak_mb (float)

  • timeline_samples (list)

  • ir_analysis (list)

  • prob_sum (float)

  • error (str | None)

q2m3.profiling.memory.read_proc_status(pid='self')[源代码]

解析 /proc/[pid]/status 中的 VmRSS 和 VmPeak(仅限 Linux)。

参数:

pid (int | str) -- 要读取的进程 ID,或用 "self" 表示当前进程。

返回类型:

dict[str, float]

q2m3.profiling.memory.read_smaps_rollup(pid='self')[源代码]

解析 /proc/[pid]/smaps_rollup,按内存类型划分 RSS。

返回字典,包含以下值,单位为 MB:
  • Rss:驻留内存总量

  • Pss:按比例分摊的内存

  • Anonymous:堆 + 匿名 mmap(C++ 分配、LLVM JIT 代码)

  • LazyFree、AnonHugePages 等。

参数:

pid (int | str)

返回类型:

dict[str, float]

q2m3.profiling.memory.take_snapshot(label)[源代码]

从所有测量层获取完整的内存快照。

同时捕获 RUSAGE_SELF(Python 进程)和 RUSAGE_CHILDREN(catalyst 子进程),以检测 MLIR→LLVM 编译发生在子进程时的测量盲区。

参数:

label (str)

返回类型:

MemorySnapshot

class q2m3.profiling.memory.MemoryTimeline(interval_s=0.1)[源代码]

基类:object

后台守护线程,每 100ms 采样一次 /proc/self/status VmRSS。

同时在 RSS 峰值处捕获 /proc/self/smaps_rollup,以划分内存类型。

参数:

interval_s (float)

property peak_smaps: dict[str, float]

在 RSS 峰值时刻捕获的内存分类。

class q2m3.profiling.memory.ParentSideMonitor(pid, interval_s=0.1)[源代码]

基类:object

从父进程通过 /proc/PID/status 监测子进程的 RSS。

此方法提供独立测量,用于验证子进程自行报告的 ru_maxrss。它对于检测 getrusage 遗漏某些分配模式(mmap、LLVM JIT 等)时的测量盲区十分关键。

参数:
  • pid (int)

  • interval_s (float)

property peak_hwm_mb: float

内核记录的 RSS 历史最高值(最可靠)。

类型:

VmHWM

q2m3.profiling.memory.measure_command(command, directory, *, wall_s=900.0, rss_mb=8192.0, cwd=None, env=None)[源代码]

运行命令,并记录实际耗时、退出状态和进程树 RSS。

此监测器支持具有 /proc 的 Linux 系统。每 0.01 秒轮询一次命令进程及其当前后代进程,不包含观测进程。RSS 使用十进制兆字节。轮询样本无法为任意瞬态分配提供上界。

参数:
  • command (list[str]) -- 非空 argv 列表。每项都是不含 NUL 的字符串。

  • directory (Path) -- 新的证据目录。相对路径以调用方的当前目录为基准。

  • wall_s (float) -- 有限且为正的实际耗时预算,单位为秒。

  • rss_mb (float) -- 有限且为正的进程树 RSS 预算,单位为十进制兆字节。

  • cwd (Path | None) -- 命令的工作目录。None 表示调用方的当前目录。父进程不改变目录。

  • env (dict[str, str] | None) -- 覆盖合并到默认子进程环境的字符串映射。

返回:

一条记录,包含 command、exit_code、reason、wall_s、peak_tree_rss_mb、poll_s、rss_budget_mb、wall_budget_s、scope 和 limitation。

抛出:
  • ValueError -- 输入值无效。函数会在创建输出目录或启动进程之前拒绝此输入。

  • FileExistsError -- 输出目录已存在。

  • RuntimeError -- 平台不支持 Linux /proc。

  • OSError -- 工作目录不存在,或命令无法启动。

返回类型:

dict[str, Any]

计时工具

q2m3 的通用计时工具。

提供轻量性能分析工具,用于测量代码段和函数的执行时间。这些工具独立于 QPE 专用性能分析,可用于代码库的任意位置。

q2m3.profiling.timing.profile_section(name, verbose=True)[源代码]

用于代码段性能分析的上下文管理器。

参数:
  • name (str) -- 被分析代码段的名称

  • verbose (bool) -- 是否打印计时信息

生成器:

dict -- 随已用时间更新的计时信息字典

示例:

with profile_section("QPE calculation") as timing:
    result = run_qpe()
print(f"Elapsed: {timing['elapsed']:.3f}s")
q2m3.profiling.timing.profile_function(func=None, *, verbose=True)[源代码]

用于分析函数执行时间的装饰器。

参数:
  • func (Callable) -- 要分析的函数

  • verbose (bool) -- 是否打印计时信息

返回:

跟踪执行时间的包装函数

返回类型:

Callable

示例:

@profile_function
def compute_energy():
    ...

@profile_function(verbose=False)
def silent_compute():
    ...
class q2m3.profiling.timing.ProfilingStats(name='default')[源代码]

基类:object

累积多次运行的性能统计信息。

参数:

name (str)

record(elapsed)[源代码]

记录一次计时测量。

参数:

elapsed (float)

返回类型:

None

property total: float

总耗时。

property count: int

记录的计时次数。

property mean: float

平均耗时。

property min: float

最短耗时。

property max: float

最长耗时。

summary()[源代码]

返回性能统计的摘要字符串。

返回类型:

str

Catalyst IR 分析

Catalyst IR 各阶段的分析工具。

警告:ir_output_dir() 使用 os.chdir(),会修改全局进程状态。它不具备线程安全性。不得从多个线程并发调用。ir_output_dir() 必须是包围 @qjit 装饰的最外层上下文管理器。

q2m3.profiling.catalyst_ir.ir_output_dir(path=None)[源代码]

管理 Catalyst IR 工作目录。

此上下文管理器必须包围 @qjit 装饰,而不只是函数调用。Catalyst 在装饰时捕获 os.getcwd(),以确定 IR 输出位置。

参数:

path (str | None) -- 若提供目录,IR 文件写入此处并在退出后保留。若为 None,则使用临时目录,并在退出时自动清理。

生成器:

工作目录路径(用户指定的目录或自动创建的临时目录)。

返回类型:

Generator[str, None, None]

q2m3.profiling.catalyst_ir.analyze_ir_stages(compiled_fn, stages=None)[源代码]

导出各编译阶段的 IR 文本,并测量大小与行数。

参数:
  • compiled_fn (Any) -- 经 Catalyst @qjit 编译的函数。

  • stages (list[str] | None) -- 用于覆盖默认值的阶段名称列表。若为 None,则使用 COMPILATION_STAGES。

返回:

(stage_name, size_kb, n_lines) 列表。若 Catalyst 不可用或所有阶段均失败,则为空列表。

返回类型:

list[tuple[str, float, int]]

QPE 性能分析

三阶段 QPE 编译性能分析工作流。

提供 profile_* 函数,用于测量以下阶段的内存占用与耗时:
  • 阶段 A:PySCF → PennyLane 哈密顿量构建

  • 阶段 B:@qjit QPE 电路编译(H_dynamic 与 H_fixed 模式)

  • 阶段 C:重复执行已编译的电路

Catalyst 是可选依赖。若未安装 catalyst,profile_qjit_compilation[_fixed] 在调用时抛出 ImportError。

q2m3.profiling.qpe_profiler.profile_hamiltonian_build(mol, n_est, n_trotter)[源代码]

分析阶段 A:PySCF → PennyLane 哈密顿量构建。

返回:

(snapshot, ops, coeffs, hf_state, circuit_params) 元组

参数:
返回类型:

tuple[MemorySnapshot, list, list[float], ndarray, dict]

q2m3.profiling.qpe_profiler.profile_qjit_compilation(ops, coeffs, hf_state, circuit_params, ir_dir=None, keep_intermediate=True)[源代码]

分析阶段 B:H_dynamic 模式的 @qjit QPE 电路编译。

这是关键阶段:首次调用触发 MLIR→LLVM 编译。

重要:在 ir_output_dir 上下文内部,以函数调用形式应用 @qjit,因为 Catalyst 在装饰时捕获 os.getcwd(),以确定 IR 工作目录位置。

参数:
  • ops (list) -- 用于构建运行时哈密顿量的 PennyLane 算符。

  • coeffs (list[float]) -- 运行时提供的哈密顿量系数。

  • hf_state (ndarray) -- 用于系统态制备的 Hartree-Fock 占据向量。

  • circuit_params (dict) -- 来自 profile_hamiltonian_build() 的电路元数据。

  • ir_dir (str | None) -- 若提供路径,IR 文件保留在该处。若为 None,则使用临时目录,并在 IR 分析后自动清理。

  • keep_intermediate (bool) -- 若为 True,在内存中保留全部 6 个 IR 阶段以供分析。若为 False,仅保留最后阶段(用于测试内存影响)。

返回:

(snapshot, timeline, ir_analysis, compiled_fn) 元组

返回类型:

tuple[MemorySnapshot, MemoryTimeline, list, Any]

q2m3.profiling.qpe_profiler.profile_qjit_compilation_fixed(ops, coeffs, hf_state, circuit_params, ir_dir=None, keep_intermediate=True)[源代码]

分析 H_fixed 模式的阶段 B:在 @qjit 外构建哈密顿量。

系数为 Python 浮点数,因此 Catalyst 可将其常量折叠到 MLIR 中。编译后的函数不接收参数。

重要:在 ir_output_dir 上下文内部,以函数调用形式应用 @qjit,因为 Catalyst 在装饰时捕获 os.getcwd()。

参数:
  • ops (list) -- 用于构建固定哈密顿量的 PennyLane 算符。

  • coeffs (list[float]) -- 在 @qjit 之前以 Python 浮点数嵌入的哈密顿量系数。

  • hf_state (ndarray) -- 用于系统态制备的 Hartree-Fock 占据向量。

  • circuit_params (dict) -- 来自 profile_hamiltonian_build() 的电路元数据。

  • ir_dir (str | None) -- 若提供路径,IR 文件保留在该处。若为 None,则使用临时目录,并在 IR 分析后自动清理。

  • keep_intermediate (bool) -- 若为 True,保留全部 6 个 IR 阶段以供分析。

返回:

(snapshot, timeline, ir_analysis, compiled_fn) 元组

返回类型:

tuple[MemorySnapshot, MemoryTimeline, list, Any]

q2m3.profiling.qpe_profiler.profile_execution(compiled_fn, coeffs, n_calls=5, is_fixed=False)[源代码]

分析阶段 C:重复执行已编译的电路。

返回:

(snapshot, prob_sum_from_last_call) 元组

参数:
  • compiled_fn (Any)

  • coeffs (list[float])

  • n_calls (int)

  • is_fixed (bool)

返回类型:

tuple[MemorySnapshot, float]

性能分析编排器

用于 QPE 编译性能分析的子进程编排和参数扫描。

提供在隔离子进程中运行性能分析的函数(避免 ru_maxrss 累积),以及遍历参数网格的函数。

所有进度报告均使用可注入的 on_progress 回调。进度报告不直接输出到控制台(Rich、print 等)。因此,调用方可使用任意界面层。

q2m3.profiling.orchestrator.run_single_profile(mol, n_est, n_trotter, mode='both', ir_dir=None, on_progress=None)[源代码]

为一个参数组合执行全部三个性能分析阶段。

参数:
  • mol (MoleculeConfig) -- 分子配置。

  • n_est (int) -- 估计线路数。

  • n_trotter (int) -- Trotter 分解阶数。

  • mode (str) -- "dynamic"、"fixed" 或 "both"。

  • ir_dir (str | None) -- 用于保留 IR 文件的目录。None 表示使用临时目录。

  • on_progress (Callable[[str], None] | None) -- 用于进度消息的可选回调。

返回:

"dynamic"/"fixed" 模式返回单个 ProfileResult;mode="both" 时返回 (fixed_result, dynamic_result) 元组。

返回类型:

ProfileResult | tuple[ProfileResult, ProfileResult]

q2m3.profiling.orchestrator.run_single_profile_in_subprocess(mol_key, n_est, n_trotter, queue, mode='dynamic', ir_dir=None)[源代码]

在子进程中运行一次性能分析(避免 ru_maxrss 累积)。

此函数将结果放入 queue,不直接返回结果。

参数:
  • mol_key (str)

  • n_est (int)

  • n_trotter (int)

  • queue (Queue)

  • mode (str)

  • ir_dir (str | None)

返回类型:

None

q2m3.profiling.orchestrator.run_both_modes(mol_key, n_est, n_trotter, ir_dir=None, on_progress=None)[源代码]

在隔离子进程中运行 fixed 与 dynamic 模式,并从父进程侧监测。

返回:

(result_fixed, result_dynamic, parent_data_fixed, parent_data_dynamic) 元组。

参数:
  • mol_key (str)

  • n_est (int)

  • n_trotter (int)

  • ir_dir (str | None)

  • on_progress (Callable[[str], None] | None)

返回类型:

tuple[ProfileResult, ProfileResult, dict, dict]

q2m3.profiling.orchestrator.run_sweep(mol_key='h2', mode='dynamic', grid=None, on_progress=None)[源代码]

使用子进程隔离进行参数扫描。

参数:
  • mol_key (str) -- MOLECULES 字典的键。

  • mode (str) -- "dynamic" 或 "fixed"。

  • grid (list[tuple[int, int]] | None) -- (n_est, n_trotter) 对列表。默认为 H2_SWEEP_GRID。

  • on_progress (Callable[[str], None] | None) -- 用于进度消息的可选回调。

返回:

将 (n_est, n_trotter) 映射到成功运行的 ProfileResult 的字典。失败的网格点会被跳过(以 warning 级别记录日志),且不会纳入结果。

返回类型:

dict[tuple[int, int], ProfileResult]