春江暮客

春江暮客的个人学习分享网站

Python 脚本耗时怎么测:用 perf_counter 与 process_time 区分等待和计算

2026-11-04 技术
Python 脚本耗时怎么测:用 perf_counter 与 process_time 区分等待和计算

脚本跑了十几秒,CPU 却不忙:到底是在计算,还是在等磁盘、网络或别的进程?只记一个“运行时间”,很容易把等待误当成算法慢。

本文面向初级到中级读者,给 Python 任务加上两种计时:经过时间和当前进程 CPU 时间。先用等待与计算两个受控例子验证,再把同一个函数放到实际处理阶段中;不用安装第三方包,也不修改 Linux 配置。

1. 方法:同一阶段记录两种时间

根据 Python 官方的 perf_counter 和 process_time 文档:

  • perf_counter_ns() 用两次读数之差测量经过时间,包含休眠。
  • process_time_ns() 测量当前进程消耗的用户态与内核态 CPU 时间,不包含休眠。
  • _ns 返回整数纳秒;单位不等于实际精度。记录日历时间仍应使用日期时间接口,不能把这些读数当 Unix 时间戳。

本例把每次运行的结果、两种秒数和三次运行的中位数一起写成 JSON。计时范围仅包含 action() 及少量读钟开销,不包含导入模块、参数解析和打印报告。

如果要决定“下一次巡检何时开始”,用固定间隔任务的 monotonic 调度方法;本文解决的是阶段耗时诊断,不是调度。

2. 准备:创建独立练习目录

需要已有的 Python 3.10 或更新版本。本文在 Linux、Python 3.11.4 下验证;命令适用于 bash 或 zsh。

  1. 新建目录,避免覆盖已有报告;重复实验时换一个目录名:
python3 --version
mkdir timing-demo
cd timing-demo
  1. 把下一节保存为当前目录下的 stage_timing.py。两个模式工作量固定:wait 休眠 0.2 秒,compute 累加 0 到 999999。--repeat 允许 1–10,默认 3。

3. 实现:计时函数与命令行入口

"""Measure elapsed and current-process CPU time. Python 3.10+."""
import argparse
import json
from statistics import median
import time


def measure(action):
    wall_start = time.perf_counter_ns()
    cpu_start = time.process_time_ns()
    result = action()
    cpu_ns = time.process_time_ns() - cpu_start
    wall_ns = time.perf_counter_ns() - wall_start
    return {
        "wall_seconds": wall_ns / 1_000_000_000,
        "cpu_seconds": cpu_ns / 1_000_000_000,
        "result": result,
    }


def compute():
    total = 0
    for number in range(1_000_000):
        total += number
    return total


def repeat_count(value):
    try:
        count = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("repeat must be an integer") from exc
    if not 1 <= count <= 10:
        raise argparse.ArgumentTypeError("repeat must be between 1 and 10")
    return count


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("mode", choices=("wait", "compute"))
    parser.add_argument("--repeat", type=repeat_count, default=3)
    args = parser.parse_args()
    action = (lambda: time.sleep(0.2)) if args.mode == "wait" else compute
    runs = [measure(action) for _ in range(args.repeat)]
    report = {
        "mode": args.mode,
        "runs": runs,
        "median_wall_seconds": median(row["wall_seconds"] for row in runs),
        "median_cpu_seconds": median(row["cpu_seconds"] for row in runs),
    }
    print(json.dumps(report, allow_nan=False, indent=2))


if __name__ == "__main__":
    main()

measure() 返回原始任务结果,便于确认没有为了加速而改坏语义。任务异常直接向上传递,不会产生成功报告。计时函数不会重试、设置超时或终止任务;外部命令的这些控制请参考subprocess 超时与日志。

4. 运行:比较等待和计算

  1. 保持在练习目录中执行:
python3 stage_timing.py --help
python3 stage_timing.py wait --repeat 3 > wait.json
python3 stage_timing.py compute --repeat 3 > compute.json
python3 -m json.tool wait.json
python3 -m json.tool compute.json
  1. 查看 median_wall_seconds 与 median_cpu_seconds。等待模式的经过时间通常约为 0.2 秒,CPU 时间很小;计算模式应返回 499999500000,并消耗可测量的 CPU 时间。具体秒数随机器负载变化,不能复制别人的数字当验收阈值。

CPU 时间小,不等于一定在等网络。 休眠、文件 I/O、锁、CPU 调度和子进程工作都可能造成类似差距。process_time_ns() 不统计子进程 CPU;当前进程有多个线程时也不是某一线程的独占计时。不要把 wall - cpu 命名为“精确 I/O 时间”,或把比值当作整台机器的 CPU 使用率。

要测真实任务,可以把 measure(compute) 换成 measure(lambda: process_batch()),但先保留同一输入与结果校验。确定计算阶段值得分析后,再用cProfile 定位函数热点。

5. 验证:检查结果、单位与失败行为

  1. 把下面代码保存为 check_timing.py。它既检查真实报告,也用模拟时钟核对单位换算,不依赖“睡眠一定比计算慢”之类脆弱断言:
import json
from pathlib import Path
from unittest.mock import patch
import stage_timing

wait = json.loads(Path("wait.json").read_text(encoding="utf-8"))
compute = json.loads(Path("compute.json").read_text(encoding="utf-8"))
assert wait["mode"] == "wait" and compute["mode"] == "compute"
for report in (wait, compute):
    assert len(report["runs"]) == 3
    assert all(row["wall_seconds"] >= 0 for row in report["runs"])
    assert all(row["cpu_seconds"] >= 0 for row in report["runs"])
assert all(row["result"] is None for row in wait["runs"])
assert all(row["result"] == 499999500000 for row in compute["runs"])
assert stage_timing.compute() == 499999500000

# Deterministic arithmetic test, independent of server load.
with patch.object(stage_timing.time, "perf_counter_ns", side_effect=[10, 210]), \
     patch.object(stage_timing.time, "process_time_ns", side_effect=[20, 70]):
    measured = stage_timing.measure(lambda: "done")
assert measured == {"wall_seconds": 200 / 1e9,
                    "cpu_seconds": 50 / 1e9, "result": "done"}

# Failed work must propagate its exception, not return a success measurement.
def fail():
    raise RuntimeError("demo failure")

try:
    stage_timing.measure(fail)
except RuntimeError as exc:
    assert str(exc) == "demo failure"
else:
    raise AssertionError("failure was swallowed")
print("PASS: results, clock arithmetic, and failure propagation")
  1. 运行测试,再验证非法参数不会生成成功 JSON:
python3 check_timing.py
if python3 stage_timing.py wait --repeat 0 > invalid.json 2> invalid.err; then
    echo "unexpected success" >&2
    exit 1
else
    code=$?
    test "$code" -eq 2 || exit 1
    test ! -s invalid.json || exit 1
    grep -F "repeat must be between 1 and 10" invalid.err
fi

测试应输出 PASS: results, clock arithmetic, and failure propagation。非法参数由 argparse 返回退出码 2。shell 重定向会预先创建或截断报告,所以文件存在不代表成功;接入流水线时还要检查退出码。

6. 排错与测量边界

现象 下一步
CPU 很小,经过时间很长 逐段加计时,检查文件、网络、锁和外部命令;两种时间不能直接指出等待来源。
同一输入的数字波动 在相近环境重复测量,记录输入规模与机器负载;中位数也不能消除缓存、节能或邻近任务的影响。
CPU 时间大于经过时间 检查当前进程是否有并行工作的原生线程;两个计时器也有边界读数开销。不要强制把 CPU 时间裁剪到经过时间。
很短的函数看不到差别 用 timeit 批量重复短表达式;不要把计时器调用成本当任务成本。
任务出错后报告为空 先看 stderr 和退出码,修复任务后重跑;不要把空报告交给下游。

这里没有采集内存、系统全局 CPU 或所有子进程资源,也没有基准测试隔离。该工具适合在优化前缩小排查范围,不适合发布固定加速倍数。

小结

对同一个阶段同时记录经过时间和当前进程 CPU 时间,先分清计算与等待的线索。保留原始结果,并用真实输入与模拟时钟分别验证语义和单位。定位到具体阶段后,再选择性能分析器或 I/O 排查工具,而不是直接加线程。

标签

1024 12306 2fa ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn chatgpt checkpoint cli cloudflare codex cofoldarena concurrency configparser configuration copy cprofile cpu监控 cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging decimal deployment desktop-app devtools difflib disk-space disown dns docker dovecot download electron encoding esm esm2 esm3 esmc esmfold2 etag evaluation faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io file-management filecmp filenames files flashppi flask folium frontend game generator git github-actions google google-research gpt-6.1-sol grep gzip harness heapq hls html http hugo imports indexnow ipaddress jailbreak javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp memory mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics multiprocessing mutation mysql nanobert nanobodies networking networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 performance permissions phaser pillow pip postfix preprocessing print prompt-engineering protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security selenium seo sequence-identity sha256 shell signals sitemap sklearn socket solana somaticsignatures sorting spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tarfile tcp tempfile tensorflow tkinter tls toml totp tracemalloc tron tronpy troubleshooting turtle typesafe-ai ubuntu unicode urllib usdt utf-8 uv venv vhhbert vite web-tools webp wordcloud wordpress workflow xml yaml zipfile zoneinfo 后台 寓言 概率 经济 贸易 迅雷解析 钱包

友情链接

其它