202 lines
7.6 KiB
Python
202 lines
7.6 KiB
Python
|
"""示例 12：进程执行器 —— 硬性停止、绕开 GIL、崩溃隔离。
|
|||
|
|
|
|||
|
|
uv run python examples/12_process_executor.py
|
|||
|
|
|
|||
|
|
绝大多数任务该用默认的**线程执行器**：共享内存、没有 pickle 约束、能直接复用连接池。
|
|||
|
|
但有三件事线程做不到，只有子进程能：
|
|||
|
|
|
|||
|
|
1. **硬性停止保证**：任务不响应停止（不调 ``ctx.checkpoint()``）时，线程只能放弃等待
|
|||
|
|
并把结果记成 ``lost``（未知）；子进程会被 ``terminate()`` → ``kill()`` 真的停掉；
|
|||
|
|
2. **绕开 GIL**：CPU 密集的计算能真正并行；
|
|||
|
|
3. **崩溃隔离**：任务里加载的原生扩展段错误，只会带走子进程。
|
|||
|
|
|
|||
|
|
本示例把 1 和 3 跑给你看，并把代价说清楚。
|
|||
|
|
|
|||
|
|
这个文件本身也是"目标函数必须是模块级"的演示：下面的任务都定义在模块顶层，因为
|
|||
|
|
spawn 起的子进程要通过"模块 + 函数名"把它们找回来。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from chongming_task import (
|
|||
|
|
ConfigurationError,
|
|||
|
|
FailureKind,
|
|||
|
|
ProcessExecutor,
|
|||
|
|
RetryPolicy,
|
|||
|
|
TaskStatus,
|
|||
|
|
Timeout,
|
|||
|
|
run_task,
|
|||
|
|
)
|
|||
|
|
from chongming_task.runtime.executors.base import ExecutionRequest
|
|||
|
|
|
|||
|
|
# 子进程不继承父进程的 logging 配置（spawn 起的是全新解释器），日志想看就得自己配。
|
|||
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# -----------------------------------------------------------------------------
|
|||
|
|
# 任务定义（模块级：spawn 的子进程要靠"模块 + 函数名"找回来）
|
|||
|
|
# -----------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
def cpu_burn(ctx: Any, rounds: int = 6) -> dict[str, object]:
|
|||
|
|
"""CPU 密集的活：子进程里不受父进程 GIL 的影响。"""
|
|||
|
|
total = 0
|
|||
|
|
for _ in range(rounds):
|
|||
|
|
ctx.checkpoint()
|
|||
|
|
for index in range(400_000):
|
|||
|
|
total += index % 7
|
|||
|
|
return {"rounds": rounds, "checksum": total, "pid": os.getpid()}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def deaf_loop(ctx: Any) -> str:
|
|||
|
|
"""**不**调用 checkpoint 的任务：停止请求它根本看不见。
|
|||
|
|
|
|||
|
|
这正是不配合的任务的样子——拿线程执行器它只会落 ``lost``；拿进程执行器它会被真的杀掉。
|
|||
|
|
"""
|
|||
|
|
del ctx
|
|||
|
|
end = time.monotonic() + 60
|
|||
|
|
while time.monotonic() < end:
|
|||
|
|
pass
|
|||
|
|
return "never"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def crash_hard(ctx: Any) -> str:
|
|||
|
|
"""模拟原生扩展段错误：直接消失，一句话都不留。"""
|
|||
|
|
del ctx
|
|||
|
|
os._exit(139)
|
|||
|
|
return "never" # pragma: no cover - 走不到这里
|
|||
|
|
|
|||
|
|
|
|||
|
|
# -----------------------------------------------------------------------------
|
|||
|
|
# 演示
|
|||
|
|
# -----------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
def demo_cpu_bound() -> None:
|
|||
|
|
print("=" * 72)
|
|||
|
|
print("一、CPU 密集：任务跑在子进程里（父进程的 GIL 管不着它）")
|
|||
|
|
print("=" * 72)
|
|||
|
|
outcome = run_task(
|
|||
|
|
cpu_burn,
|
|||
|
|
6,
|
|||
|
|
task_id="cpu-1",
|
|||
|
|
executor=ProcessExecutor(),
|
|||
|
|
timeout=Timeout(per_attempt=30),
|
|||
|
|
)
|
|||
|
|
assert outcome.ok, outcome.failure
|
|||
|
|
payload = outcome.value
|
|||
|
|
print(f" 结果：{payload}")
|
|||
|
|
print(f" 监督进程 pid={os.getpid()}，任务进程 pid={outcome.snapshot.pid}")
|
|||
|
|
print(f" 快照里的 executor={outcome.snapshot.executor}（写进存储，事后可查）")
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def demo_hard_stop() -> None:
|
|||
|
|
print("=" * 72)
|
|||
|
|
print("二、硬性停止：不响应停止的任务被真的杀掉（线程执行器只能放弃等待）")
|
|||
|
|
print("=" * 72)
|
|||
|
|
started = time.monotonic()
|
|||
|
|
outcome = run_task(
|
|||
|
|
deaf_loop,
|
|||
|
|
task_id="deaf-1",
|
|||
|
|
executor=ProcessExecutor(),
|
|||
|
|
timeout=Timeout(idle=1), # 一秒没有心跳就判卡死
|
|||
|
|
retry=RetryPolicy(max_attempts=1),
|
|||
|
|
graceful_stop=0.5, # 优雅窗口：0.5 秒内不响应就强杀
|
|||
|
|
)
|
|||
|
|
elapsed = time.monotonic() - started
|
|||
|
|
print(f" 任务自己的循环要跑 60 秒，实际用了 {elapsed:.1f} 秒")
|
|||
|
|
print(f" 终态：{outcome.status.value}；失败类型：{outcome.failure.kind.value}")
|
|||
|
|
print(f" 子进程 pid={outcome.snapshot.pid} —— 它已经不在了（ps 查不到）")
|
|||
|
|
assert outcome.status is TaskStatus.FAILED
|
|||
|
|
assert outcome.failure.kind is FailureKind.TIMEOUT_IDLE, "超时如实上报，不谎报'已停止'"
|
|||
|
|
assert elapsed < 30, "不该等到任务自己结束"
|
|||
|
|
print(" 注意：超时仍然记成超时（不是'已停止'）——但资源确实释放了。")
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def demo_crash_isolation() -> None:
|
|||
|
|
print("=" * 72)
|
|||
|
|
print("三、崩溃隔离：子进程崩了，监督进程活着并如实记一笔")
|
|||
|
|
print("=" * 72)
|
|||
|
|
outcome = run_task(
|
|||
|
|
crash_hard,
|
|||
|
|
task_id="crash-1",
|
|||
|
|
executor=ProcessExecutor(),
|
|||
|
|
timeout=Timeout(per_attempt=30),
|
|||
|
|
)
|
|||
|
|
assert not outcome.ok
|
|||
|
|
print(f" 监督进程还活着（pid={os.getpid()}），任务终态：{outcome.status.value}")
|
|||
|
|
print(f" 失败类型：{outcome.failure.kind.value}；退出码：{outcome.failure.exit_code}")
|
|||
|
|
print(f" 说明：{outcome.failure.message}")
|
|||
|
|
print(" 换成线程执行器，这个崩溃会带走整个监督进程。")
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def demo_picklable_guard() -> None:
|
|||
|
|
print("=" * 72)
|
|||
|
|
print("四、代价：不可 pickle 的目标在**提交那一刻**就被拦下")
|
|||
|
|
print("=" * 72)
|
|||
|
|
|
|||
|
|
def local_closure(ctx: Any) -> str: # 局部函数：子进程里找不到它
|
|||
|
|
del ctx
|
|||
|
|
return "nope"
|
|||
|
|
|
|||
|
|
# 1) 直接交给执行器：当场抛错，并告诉你该改成什么。
|
|||
|
|
request = ExecutionRequest(task_id="bad-0", name="bad", target=local_closure, args=())
|
|||
|
|
try:
|
|||
|
|
ProcessExecutor().launch(request)
|
|||
|
|
except ConfigurationError as exc:
|
|||
|
|
print(f" launch() 直接报错 {type(exc).__name__}：")
|
|||
|
|
for line in str(exc).splitlines():
|
|||
|
|
print(f" {line}")
|
|||
|
|
else: # pragma: no cover - 应当抛错
|
|||
|
|
raise AssertionError("不可 pickle 的目标必须被拦下")
|
|||
|
|
|
|||
|
|
# 2) 走 run_task：任务**落 failed 并带上这条诊断**，调用方不会干等，也不会被
|
|||
|
|
# 说成"框架内部错误"——那是把任务侧的问题指向错误的排查方向。
|
|||
|
|
outcome = run_task(local_closure, task_id="bad-1", executor=ProcessExecutor())
|
|||
|
|
assert outcome.status is TaskStatus.FAILED
|
|||
|
|
assert outcome.failure is not None
|
|||
|
|
print(f"\n 走 run_task：终态 {outcome.status.value}，retryable={outcome.failure.retryable}")
|
|||
|
|
print(f" 失败信息首行：{outcome.failure.message.splitlines()[0]}")
|
|||
|
|
print(" 配置类错误不重试——重试多少次都是同一个结果，只会白烧重试预算。")
|
|||
|
|
assert outcome.failure.retryable is False
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
print()
|
|||
|
|
demo_cpu_bound()
|
|||
|
|
demo_hard_stop()
|
|||
|
|
demo_crash_isolation()
|
|||
|
|
demo_picklable_guard()
|
|||
|
|
|
|||
|
|
print("=" * 72)
|
|||
|
|
print("选型：默认线程，需要这三种能力时才换进程")
|
|||
|
|
print("=" * 72)
|
|||
|
|
print(
|
|||
|
|
"""
|
|||
|
|
线程（默认） 进程（ProcessExecutor）
|
|||
|
|
------------------------ ------------------------------------------
|
|||
|
|
共享内存、可复用连接池 参数与返回值必须可 pickle
|
|||
|
|
闭包 / lambda 都能用 目标必须是模块级函数，调用方要有 __main__ 守卫
|
|||
|
|
停止/暂停即时生效 停止/暂停有约 20ms 轮询延迟
|
|||
|
|
不配合的任务只能放弃（lost） 不配合的任务被真的杀掉
|
|||
|
|
受 GIL 限制 真并行
|
|||
|
|
原生扩展段错误带走整个进程 只带走子进程
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
print("全部通过 ✅")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|