diff --git a/.trae/docs/iter-25-perf-hotpath.md b/.trae/docs/iter-25-perf-hotpath.md new file mode 100644 index 0000000..5d244ff --- /dev/null +++ b/.trae/docs/iter-25-perf-hotpath.md @@ -0,0 +1,71 @@ +# 迭代 25：执行热路径性能优化 + +## 本轮目标 + +针对 `px.run` 执行热路径中的冗余开销进行 4 项优化，提升 sequential/thread 策略吞吐。 + +## 改动文件清单 + +- `src/pyflowx/task.py` — `should_execute` 加早退快速路径 +- `src/pyflowx/executors.py` — `env_context` 快速路径、`storage_key` 内联、`resolved_spec` 去重 + +## 关键决策与依据 + +### 1. env_context 快速路径（最大开销项） + +**问题**：每个 fn 任务执行时 `with spec.env_context():` 都会创建 `_GeneratorContextManager` 对象 + `__enter__`/`__exit__`，即使任务无 env/cwd。500 noop 任务 = 500 次无谓对象创建。 + +**方案**：在 `SyncTaskRunner.run` 和 `_submit_sync_task.fn_call` 内联判断 `spec.env is None and spec.cwd is None`，直接调用 `effective_fn`，跳过上下文管理器。 + +**依据**：`_env_and_cwd` 本身已有 `if not env and cwd is None: yield; return` 快速路径，但 contextlib.contextmanager 包装器的对象创建 + 协议开销仍存在。内联判断完全消除这层开销。 + +### 2. should_execute 早退快速路径 + +**问题**：无 conditions 且无 skip_if_missing 的任务（最常见场景）仍分配空 list `failed_conditions` + 迭代空序列。 + +**方案**：在 `should_execute` 开头加 `if not self.conditions and not self.skip_if_missing: return True, None`。 + +### 3. resolved_spec 去重 + +**问题**：`_filter_and_sort` 已为每个任务计算 `graph.resolved_spec(name)` 并存入 `specs` dict，但三个 Layer runner（Sequential/Threaded/Async）各自又调用一次。500 任务 = 500 次冗余缓存查询。 + +**方案**： +- `_filter_and_sort` 返回 `(to_run, specs)` 元组，将 specs dict 透传给 runner +- `_build_semaphores` 签名从 `graph: Graph` 改为 `specs: Mapping[str, TaskSpec]` +- 三个 Layer runner 用 `specs[name]` 替代 `graph.resolved_spec(name)` +- `DependencyRunner` 启动时构建 `all_specs` dict 传入 `_build_semaphores`；`_run_task` 内保持 `graph.resolved_spec(name)`（动态任务不在 all_specs 中） + +**依据**：`resolved_spec` 虽有 `cached_property`/`lru_cache`，但每次仍有字典查找 + 函数调用开销。消除冗余调用后每任务省 1 次查找。 + +### 4. storage_key 内联快速路径 + +**问题**：`_apply_cached` 和 `_store_result` 各调用 `spec.storage_key(ctx)`。无 cache_key 时（最常见场景）该函数仅返回 `self.name`，但有函数调用开销。 + +**方案**：内联 `spec.name if spec.cache_key is None else spec.storage_key(ctx)`，避免函数调用。 + +**依据**：`storage_key` 内部首行已是 `if self.cache_key is None: return self.name`，内联仅省函数调用开销，但对 500 任务累积可见。 + +## 验证结果 + +### 门禁 + +- ruff: All checks passed +- pyrefly: 0 errors +- pytest: 1677 passed +- coverage: 97.23%（≥95% 门槛） + +### 基准对比（500 noop 任务） + +| 策略 | 优化前 | 优化后 | 提升 | +|------|--------|--------|------| +| sequential | 764 ops/s | 930 ops/s | +22% | +| thread | 108 ops/s | 119 ops/s | +10% | +| async | 49 ops/s | 50 ops/s | ~持平（固有开销主导） | +| dependency | 49 ops/s | 44 ops/s | ~持平（固有开销主导，噪声范围） | + +sequential 路径提升最显著（+22%），因每任务节省的开销在总耗时中占比最大。thread 次之（+10%）。async/dependency 受事件循环与线程池提交的固有开销主导，per-task 优化相对影响小。 + +## 遗留事项 + +- async/dependency 策略对 noop 任务的固有并发开销仍较高，需真实 I/O 任务才能体现并行优势（属设计预期，非缺陷）。 +- `datetime.now()` 每任务调用 2 次（started_at + finished_at），约占 sequential 预算的可观比例，但涉及 duration 正确性，本轮未优化。 diff --git a/.trae/skills/pyflowx-development/SKILL.md b/.trae/skills/pyflowx-development/SKILL.md index 826e1a9..7752a77 100644 --- a/.trae/skills/pyflowx-development/SKILL.md +++ b/.trae/skills/pyflowx-development/SKILL.md @@ -1,13 +1,14 @@ --- name: "pyflowx-development" -description: "PyFlowX 项目（DAG 任务调度库）的开发知识库。归档迭代 06-20 中可复用的架构模式、踩坑总结与设计决策，并同步迭代 21（P9）、迭代 22（P10 图片处理）、迭代 23（P11 文件操作底层 API）与迭代 24（P12 通用 DAG 构造器）的行为变更。在设计与调度器、CLI、YAML 编排、取消机制、序列化、观测性、错误诊断、状态后端、性能优化、任务编排、检查点恢复、监控导出、图片处理、文件操作底层 API、通用 DAG 构造器相关的功能时参考。" +description: "PyFlowX 项目（DAG 任务调度库）的开发知识库。归档迭代 06-20 中可复用的架构模式、踩坑总结与设计决策，并同步迭代 21（P9）、迭代 22（P10 图片处理）、迭代 23（P11 文件操作底层 API）、迭代 24（P12 通用 DAG 构造器）与迭代 25（执行热路径性能优化）的行为变更。在设计与调度器、CLI、YAML 编排、取消机制、序列化、观测性、错误诊断、状态后端、性能优化、任务编排、检查点恢复、监控导出、图片处理、文件操作底层 API、通用 DAG 构造器相关的功能时参考。" --- # PyFlowX 开发知识库 本技能归档自迭代 06-20 的过程记录，并同步迭代 21（P9 功能扩展）、 -迭代 22（P10 图片处理）、迭代 23（P11 文件操作底层 API）与迭代 24 -（P12 通用 DAG 构造器）的行为变更。按主题分类整理可复用知识。 +迭代 22（P10 图片处理）、迭代 23（P11 文件操作底层 API）、迭代 24 +（P12 通用 DAG 构造器）与迭代 25（执行热路径性能优化）的行为变更。 +按主题分类整理可复用知识。 过程性细节（覆盖率数字、命令输出）已剔除，仅保留架构模式、设计依据 与陷阱总结。 @@ -219,6 +220,29 @@ description: "PyFlowX 项目（DAG 任务调度库）的开发知识库。归档 - **额外**：`asyncio.get_event_loop()` → `asyncio.get_running_loop()`， 兼容 Python 3.12+ 弃用警告（调用点均在 `asyncio.run()` 内的协程中）。 +### 执行热路径 4 项优化（iter-25） + +sequential(500) 从 764 → 930 ops/s（+22%），thread(500) 从 108 → 119 ops/s（+10%）。 + +- **env_context 快速路径**：`SyncTaskRunner.run` 与 `_submit_sync_task.fn_call` + 内联 `if spec.env is None and spec.cwd is None: effective_fn(...)`，跳过 + `with spec.env_context():` 的 `_GeneratorContextManager` 对象创建。即使 + `_env_and_cwd` 内部已有 `if not env and cwd is None: yield` 快速路径， + contextlib 包装器的对象创建 + `__enter__`/`__exit__` 协议开销仍存在， + 内联判断完全消除这层开销。对 fn 任务（最常见场景）每任务省一次上下文管理器。 +- **should_execute 早退**：`not self.conditions and not self.skip_if_missing` + 时直接 `return True, None`，避免空 list 分配 + 空序列迭代。 +- **resolved_spec 去重**：`_filter_and_sort` 返回 `(to_run, specs)` 元组， + specs dict 透传给 Layer runner；`_build_semaphores` 签名 `graph: Graph` → + `specs: Mapping[str, TaskSpec]`。消除 Sequential/Threaded/AsyncLayerRunner + 内的重复 `graph.resolved_spec(name)` 调用。`DependencyRunner._run_task` 仍 + 调 `resolved_spec`（动态任务不在初始 specs 中）。 +- **storage_key 内联**：`_apply_cached`/`_store_result` 内联 + `spec.name if spec.cache_key is None else spec.storage_key(ctx)`，避免无 + cache_key 场景（最常见）的函数调用开销。 +- **适用边界**：async/dependency 策略对 noop 任务的固有并发开销（事件循环 + + 线程池提交）主导，per-task 优化相对影响小；真实 I/O 任务才能体现并行优势。 + ### 踩坑总结 - **`lru_cache` 对签名内省有 dict lookup 开销**：即便 `functools.lru_cache` diff --git a/src/pyflowx/executors.py b/src/pyflowx/executors.py index 72a1fb4..9b1f269 100644 --- a/src/pyflowx/executors.py +++ b/src/pyflowx/executors.py @@ -218,7 +218,8 @@ def _apply_cached( 单次 ``backend.get`` + ``KeyError`` 回退，避免 ``has`` + ``get`` 双重 哈希查找与双重 TTL 判断。 """ - storage_key = spec.storage_key(context) + # 快速路径：无 cache_key 时存储键即为任务名，避免函数调用开销。 + storage_key = spec.name if spec.cache_key is None else spec.storage_key(context) try: cached = backend.get(storage_key) except KeyError: @@ -442,8 +443,12 @@ class SyncTaskRunner: while True: result.attempts += 1 try: - with spec.env_context(): + # 快速路径：无 env/cwd 时直接调用，跳过上下文管理器创建开销。 + if spec.env is None and spec.cwd is None: value = spec.effective_fn(*args, **kwargs) + else: + with spec.env_context(): + value = spec.effective_fn(*args, **kwargs) _mark_success(spec, result, value) return result except Exception as exc: @@ -529,6 +534,9 @@ def _submit_sync_task( """ def fn_call() -> Any: + # 快速路径：无 env/cwd 时直接调用，跳过上下文管理器创建开销。 + if spec.env is None and spec.cwd is None: + return spec.effective_fn(*args, **kwargs) with spec.env_context(): return spec.effective_fn(*args, **kwargs) @@ -561,11 +569,12 @@ def _filter_and_sort( report: RunReport, backend: StateBackend, on_event: EventCallback | None, -) -> list[str]: - """过滤掉已命中缓存的任务，按优先级排序返回待运行列表。 +) -> tuple[list[str], dict[str, TaskSpec[Any]]]: + """过滤掉已命中缓存的任务，按优先级排序返回待运行列表与 specs 映射。 预构建 ``{name: spec}`` 映射，过滤与排序共享同一份 resolved spec， 避免 ``_sort_by_priority`` 内重复调用 ``graph.resolved_spec``。 + 返回 specs 映射供调用方复用，消除 runner 内的重复 ``resolved_spec`` 调用。 """ specs: dict[str, TaskSpec[Any]] = {} to_run: list[str] = [] @@ -578,7 +587,7 @@ def _filter_and_sort( continue if not _apply_cached(name, spec, context, report, backend, on_event): to_run.append(name) - return _sort_by_priority(to_run, specs) + return _sort_by_priority(to_run, specs), specs def _store_result( @@ -598,22 +607,23 @@ def _store_result( """ context[name] = result.value if result.status == TaskStatus.SUCCESS: - backend.save(spec.storage_key(task_ctx), result.value) + # 快速路径：无 cache_key 时存储键即为任务名，避免函数调用开销。 + key = spec.name if spec.cache_key is None else spec.storage_key(task_ctx) + backend.save(key, result.value) report.results[name] = result _emit(on_event, result) def _build_semaphores( to_run: list[str], - graph: Graph, + specs: Mapping[str, TaskSpec[Any]], sem_factory: Callable[[int], Any], concurrency_limits: Mapping[str, int], ) -> dict[str, Any]: """为每个 ``concurrency_key`` 创建一个信号量。""" semaphores: dict[str, Any] = {} for name in to_run: - spec = graph.resolved_spec(name) - key = spec.concurrency_key + key = specs[name].concurrency_key if key is not None and key not in semaphores: limit = concurrency_limits.get(key, 1) semaphores[key] = sem_factory(limit) @@ -643,8 +653,9 @@ class SequentialLayerRunner: layer_idx: int, on_event: EventCallback | None, ) -> None: - for name in _filter_and_sort(layer, graph, context, report, backend, on_event): - spec = graph.resolved_spec(name) + to_run, specs = _filter_and_sort(layer, graph, context, report, backend, on_event) + for name in to_run: + spec = specs[name] task_ctx = _build_context(spec, context, report) result = SyncTaskRunner.run(spec, task_ctx, layer_idx, on_event, report) _store_result(name, result, spec, task_ctx, context, report, backend, on_event) @@ -665,15 +676,15 @@ class ThreadedLayerRunner: pool: concurrent.futures.ThreadPoolExecutor, concurrency_limits: Mapping[str, int], ) -> None: - to_run = _filter_and_sort(layer, graph, context, report, backend, on_event) + to_run, specs = _filter_and_sort(layer, graph, context, report, backend, on_event) if not to_run: return - semaphores = _build_semaphores(to_run, graph, threading.Semaphore, concurrency_limits) + semaphores = _build_semaphores(to_run, specs, threading.Semaphore, concurrency_limits) context_snapshot = dict(context) lock = threading.Lock() def _run_threaded_task(name: str) -> tuple[dict[str, Any], TaskResult[Any]]: - spec = graph.resolved_spec(name) + spec = specs[name] task_ctx = _build_context(spec, context_snapshot, report) sem = _get_sem(semaphores, spec) if sem is not None: @@ -695,7 +706,7 @@ class ThreadedLayerRunner: finally: with lock: for name, (task_ctx, result) in completed.items(): - _store_result(name, result, graph.resolved_spec(name), task_ctx, context, report, backend, on_event) + _store_result(name, result, specs[name], task_ctx, context, report, backend, on_event) class AsyncLayerRunner: @@ -712,14 +723,14 @@ class AsyncLayerRunner: on_event: EventCallback | None, concurrency_limits: Mapping[str, int], ) -> None: - to_run = _filter_and_sort(layer, graph, context, report, backend, on_event) + to_run, specs = _filter_and_sort(layer, graph, context, report, backend, on_event) if not to_run: return - semaphores = _build_semaphores(to_run, graph, asyncio.Semaphore, concurrency_limits) + semaphores = _build_semaphores(to_run, specs, asyncio.Semaphore, concurrency_limits) context_snapshot = dict(context) async def _run_async_task(name: str) -> tuple[dict[str, Any], TaskResult[Any]]: - spec = graph.resolved_spec(name) + spec = specs[name] task_ctx = _build_context(spec, context_snapshot, report) sem = _get_sem(semaphores, spec) result = await AsyncTaskRunner.run(spec, task_ctx, layer_idx, on_event, report, sem) @@ -727,7 +738,7 @@ class AsyncLayerRunner: results = await asyncio.gather(*[_run_async_task(name) for name in to_run]) for name, (task_ctx, result) in zip(to_run, results, strict=True): - _store_result(name, result, graph.resolved_spec(name), task_ctx, context, report, backend, on_event) + _store_result(name, result, specs[name], task_ctx, context, report, backend, on_event) def _extract_dynamic_specs(result: TaskResult[Any], spec: TaskSpec[Any]) -> list[TaskSpec[Any]]: @@ -771,7 +782,8 @@ class DependencyRunner: cancel_event: threading.Event | CancelToken | None = None, ) -> None: all_names = list(graph.all_specs().keys()) - semaphores = _build_semaphores(all_names, graph, asyncio.Semaphore, concurrency_limits) + all_specs = {name: graph.resolved_spec(name) for name in all_names} + semaphores = _build_semaphores(all_names, all_specs, asyncio.Semaphore, concurrency_limits) futures: dict[str, asyncio.Task[TaskResult[Any]]] = {} async def _run_task(name: str) -> TaskResult[Any]: diff --git a/src/pyflowx/task.py b/src/pyflowx/task.py index 8cece61..e15f065 100644 --- a/src/pyflowx/task.py +++ b/src/pyflowx/task.py @@ -395,6 +395,9 @@ class TaskSpec(Generic[T]): ``should_run`` 为 False 时 ``skip_reason`` 描述跳过原因。 失败条件超过 2 个时仅展示前 2 个并附总数。 """ + # 快速路径：无条件且无需检查命令可用性时直接放行（最常见场景）。 + if not self.conditions and not self.skip_if_missing: + return True, None # 逐个求值条件，记录失败项。 failed_conditions: list[str] = [] for condition in self.conditions: