310 lines
10 KiB
Python
310 lines
10 KiB
Python
"""高级基准：条件评估/YAML 加载/通知器/run_iter/子图/取消/CPU+I/O 密集型任务.
|
||
|
||
这些基准覆盖核心调度引擎的扩展路径，与 bench_graph/bench_execution/bench_context/
|
||
bench_storage 互补，构成完整的性能度量体系。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import contextlib
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pyflowx as px
|
||
from benchmarks import print_results, time_it
|
||
from pyflowx import Graph, TaskSpec
|
||
from pyflowx.cancellation import CancelToken
|
||
from pyflowx.conditions import BuiltinConditions
|
||
from pyflowx.notification import CallbackNotifier, NotificationLevel
|
||
|
||
# ============================================================================
|
||
# 基准：条件评估
|
||
# ============================================================================
|
||
|
||
|
||
def bench_conditions() -> None:
|
||
"""条件评估基准（静态/上下文/复合）."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
# 静态条件（IS_LINUX）
|
||
cond_static = BuiltinConditions.IS_LINUX()
|
||
ctx: dict[str, Any] = {}
|
||
ms, ops = time_it(lambda c=cond_static, x=ctx: c(x), iterations=10000, warmup=500)
|
||
results.append(("static(IS_LINUX)", 10000, ms, ops))
|
||
|
||
# 上下文条件（DEP_EQUALS）
|
||
cond_dep = BuiltinConditions.DEP_EQUALS("a", 1)
|
||
ctx_dep = {"a": 1, "b": 2}
|
||
ms, ops = time_it(lambda c=cond_dep, x=ctx_dep: c(x), iterations=10000, warmup=500)
|
||
results.append(("DEP_EQUALS", 10000, ms, ops))
|
||
|
||
# 复合条件（AND(OR(NOT, DEP_TRUTHY), DEP_PRESENT)）
|
||
cond_complex = BuiltinConditions.AND(
|
||
BuiltinConditions.OR(
|
||
BuiltinConditions.NOT(BuiltinConditions.IS_WINDOWS()),
|
||
BuiltinConditions.DEP_TRUTHY("a"),
|
||
),
|
||
BuiltinConditions.DEP_PRESENT("b"),
|
||
)
|
||
ms, ops = time_it(lambda c=cond_complex, x=ctx_dep: c(x), iterations=10000, warmup=500)
|
||
results.append(("AND(OR(NOT,DEP_TRUTHY),DEP_PRESENT)", 10000, ms, ops))
|
||
|
||
print_results("条件评估", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：YAML 加载
|
||
# ============================================================================
|
||
|
||
|
||
_YAML_TEMPLATE = """\
|
||
jobs:
|
||
{jobs}
|
||
defaults:
|
||
retry:
|
||
max_attempts: 2
|
||
backoff: 0.0
|
||
"""
|
||
|
||
|
||
def _make_yaml(n: int) -> str:
|
||
"""生成 n 个任务的 YAML（链式依赖）."""
|
||
lines = []
|
||
for i in range(n):
|
||
deps = f"needs: [t{i - 1}]" if i > 0 else ""
|
||
lines.append(f" t{i}:\n cmd: ['true']\n {deps}".rstrip())
|
||
return _YAML_TEMPLATE.format(jobs="\n".join(lines))
|
||
|
||
|
||
def bench_yaml_load() -> None:
|
||
"""YAML 加载基准（解析 + Graph 构建）."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
tmp_dir = tempfile.mkdtemp()
|
||
|
||
for n in (10, 50, 100):
|
||
yaml_text = _make_yaml(n)
|
||
yaml_path = Path(tmp_dir) / f"bench_{n}.yaml"
|
||
yaml_path.write_text(yaml_text, encoding="utf-8")
|
||
ms, _ops = time_it(lambda p=yaml_path: Graph.from_yaml(p), iterations=20, warmup=2)
|
||
results.append((f"yaml_load({n})", 20, ms, 1.0 / (ms / 1000) if ms > 0 else 0))
|
||
|
||
print_results("YAML 加载 (from_yaml)", results)
|
||
|
||
import shutil
|
||
|
||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：通知器
|
||
# ============================================================================
|
||
|
||
|
||
def bench_notifiers() -> None:
|
||
"""通知器 notify 调用开销基准."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
# 构造一个真实 TaskEvent
|
||
from pyflowx.task import TaskEvent, TaskStatus
|
||
|
||
event = TaskEvent(task="t0", status=TaskStatus.SUCCESS, attempts=1)
|
||
|
||
# CallbackNotifier（无级别过滤）
|
||
calls: list[int] = [0]
|
||
|
||
def _cb(_e: Any) -> None:
|
||
calls[0] += 1
|
||
|
||
notifier = CallbackNotifier(_cb)
|
||
ms, ops = time_it(lambda e=event, n=notifier: n.notify(e), iterations=50000, warmup=1000)
|
||
results.append(("CallbackNotifier.notify", 50000, ms, ops))
|
||
|
||
# 级别过滤（仅 SUCCESS）
|
||
notifier_filtered = CallbackNotifier(_cb, levels={NotificationLevel.SUCCESS})
|
||
ms, ops = time_it(lambda e=event, n=notifier_filtered: n.notify(e), iterations=50000, warmup=1000)
|
||
results.append(("CallbackNotifier(filtered).notify", 50000, ms, ops))
|
||
|
||
print_results("通知器 notify", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：run_iter 流式 API
|
||
# ============================================================================
|
||
|
||
|
||
def bench_run_iter() -> None:
|
||
"""run_iter 流式执行基准."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
def noop() -> None:
|
||
pass
|
||
|
||
for n in (50, 200, 500):
|
||
specs = [TaskSpec(f"t{i}", fn=noop) for i in range(n)]
|
||
graph = Graph.from_specs(specs)
|
||
|
||
def _run_iter(g: Graph = graph) -> None:
|
||
list(px.run_iter(g, strategy="sequential"))
|
||
|
||
ms, ops = time_it(_run_iter, iterations=10, warmup=2)
|
||
results.append((f"run_iter(sequential,{n})", 10, ms, ops))
|
||
|
||
print_results("run_iter 流式执行", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：子图过滤
|
||
# ============================================================================
|
||
|
||
|
||
def bench_subgraph() -> None:
|
||
"""子图过滤基准（subgraph_with_deps 传递闭包计算）."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
for n in (100, 500, 1000):
|
||
specs = []
|
||
for i in range(n):
|
||
deps = (f"t{i - 1}",) if i > 0 else ()
|
||
specs.append(TaskSpec(f"t{i}", cmd=["true"], depends_on=deps))
|
||
graph = Graph.from_specs(specs)
|
||
# 取中点任务，触发前半部分传递闭包
|
||
target = f"t{n // 2}"
|
||
ms, ops = time_it(
|
||
lambda g=graph, t=target: g.subgraph_with_deps([t]),
|
||
iterations=50,
|
||
warmup=5,
|
||
)
|
||
results.append((f"subgraph_with_deps({n})", 50, ms, ops))
|
||
|
||
print_results("子图过滤 (subgraph_with_deps)", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：取消机制
|
||
# ============================================================================
|
||
|
||
|
||
def bench_cancellation() -> None:
|
||
"""取消机制基准（cancel_event 触发到返回）."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
def slow() -> None:
|
||
time.sleep(0.5)
|
||
|
||
for n in (50, 200):
|
||
specs = [TaskSpec(f"t{i}", fn=slow) for i in range(n)]
|
||
graph = Graph.from_specs(specs)
|
||
|
||
def _cancel_after(g: Graph = graph) -> None:
|
||
token = CancelToken()
|
||
# 立即取消，让所有任务变 SKIPPED
|
||
token.cancel()
|
||
with contextlib.suppress(Exception):
|
||
px.run(g, strategy="sequential", cancel_event=token)
|
||
|
||
ms, ops = time_it(_cancel_after, iterations=5, warmup=1)
|
||
results.append((f"cancel(immediate,{n})", 5, ms, ops))
|
||
|
||
print_results("取消机制 (cancel_event)", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：CPU 密集型任务
|
||
# ============================================================================
|
||
|
||
|
||
def _fib(n: int) -> int:
|
||
"""递归斐波那契（CPU 密集型）."""
|
||
if n < 2:
|
||
return n
|
||
return _fib(n - 1) + _fib(n - 2)
|
||
|
||
|
||
def bench_cpu_intensive() -> None:
|
||
"""CPU 密集型任务基准（递归斐波那契）."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
for n_tasks, fib_n in ((10, 20), (20, 20), (10, 25)):
|
||
specs = [TaskSpec(f"t{i}", fn=_fib, args=(fib_n,)) for i in range(n_tasks)]
|
||
graph = Graph.from_specs(specs)
|
||
|
||
# sequential
|
||
ms, ops = time_it(
|
||
lambda g=graph: px.run(g, strategy="sequential"),
|
||
iterations=3,
|
||
warmup=1,
|
||
)
|
||
results.append((f"cpu-seq({n_tasks}x fib{fib_n})", 3, ms, ops))
|
||
|
||
# thread（CPU 密集型受 GIL 限制，验证是否退化）
|
||
ms, ops = time_it(
|
||
lambda g=graph: px.run(g, strategy="thread", max_workers=4),
|
||
iterations=3,
|
||
warmup=1,
|
||
)
|
||
results.append((f"cpu-thread({n_tasks}x fib{fib_n})", 3, ms, ops))
|
||
|
||
print_results("CPU 密集型 (递归斐波那契)", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 基准：I/O 密集型任务
|
||
# ============================================================================
|
||
|
||
|
||
def bench_io_intensive() -> None:
|
||
"""I/O 密集型任务基准（sleep 模拟）."""
|
||
results: list[tuple[str, int, float, float]] = []
|
||
|
||
def io_sleep() -> None:
|
||
time.sleep(0.01) # 10ms 模拟 I/O
|
||
|
||
for n in (20, 50):
|
||
specs = [TaskSpec(f"t{i}", fn=io_sleep) for i in range(n)]
|
||
graph = Graph.from_specs(specs)
|
||
|
||
# sequential（总耗时 ≈ n * 10ms）
|
||
ms, ops = time_it(
|
||
lambda g=graph: px.run(g, strategy="sequential"),
|
||
iterations=3,
|
||
warmup=1,
|
||
)
|
||
results.append((f"io-seq({n})", 3, ms, ops))
|
||
|
||
# thread（I/O 密集型应能并行，总耗时 ≈ 10ms）
|
||
ms, ops = time_it(
|
||
lambda g=graph: px.run(g, strategy="thread", max_workers=8),
|
||
iterations=3,
|
||
warmup=1,
|
||
)
|
||
results.append((f"io-thread({n})", 3, ms, ops))
|
||
|
||
# async（I/O 密集型应能并行）
|
||
ms, ops = time_it(
|
||
lambda g=graph: px.run(g, strategy="async"),
|
||
iterations=3,
|
||
warmup=1,
|
||
)
|
||
results.append((f"io-async({n})", 3, ms, ops))
|
||
|
||
print_results("I/O 密集型 (sleep 10ms)", results)
|
||
|
||
|
||
# ============================================================================
|
||
# 主入口
|
||
# ============================================================================
|
||
|
||
|
||
def run_advanced() -> None:
|
||
"""运行全部高级基准."""
|
||
bench_conditions()
|
||
bench_yaml_load()
|
||
bench_notifiers()
|
||
bench_run_iter()
|
||
bench_subgraph()
|
||
bench_cancellation()
|
||
bench_cpu_intensive()
|
||
bench_io_intensive()
|