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。
- 新建目录,避免覆盖已有报告;重复实验时换一个目录名:
python3 --version
mkdir timing-demo
cd timing-demo
- 把下一节保存为当前目录下的
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. 运行:比较等待和计算
- 保持在练习目录中执行:
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
- 查看
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. 验证:检查结果、单位与失败行为
- 把下面代码保存为
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")
- 运行测试,再验证非法参数不会生成成功 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 排查工具,而不是直接加线程。
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/python-wall-cpu-stage-timing.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。