chakcy
f4de43c743
单任务受控执行（15 态状态机 + 审计轨迹）、DAG 管道、Controller 门面。
执行器三个：线程（默认，含 async def 任务）、内联（同步 / 测试）、进程
（spawn 子进程：硬性停止、绕开 GIL、崩溃隔离）。
调度：并发额度 + 每桶限流 + 具名资源信号量（按标签）+ 优先级排队。
持久层：TaskStore 契约 + 内存 / SQLite / Postgres 实现，契约测试随包发布。
跨进程与多节点：写穿 / 读穿、指令通道与租约、owned_buckets 划界、reconcile。
CLI、12 个可直接运行的示例、docs/ 下 5 份文档。
门禁：ruff / ruff format / mypy(strict) 干净；
pytest 563 passed（44 skipped 是需要真实 Postgres 的那批）；
带一台可连的 Postgres 时 607 passed。
172 lines
6.0 KiB
Python
172 lines
6.0 KiB
Python
"""进程执行器的端到端接线：``TaskRun`` / ``Controller`` / 存储。
|
|
|
|
单元测试盯的是句柄本身（暂停、强杀、心跳、崩溃）。这里盯的是**接进主线之后**的
|
|
三件承诺：
|
|
|
|
1. 快照里真的记下了子进程的 pid（排障时 `ps` 对得上号的就是它）；
|
|
2. 不响应停止的任务会被**超时机制真的杀掉**，而不是留一个卡死的线程（线程执行器
|
|
只能把它记成 `LOST`，这是进程执行器存在的理由）；
|
|
3. 这套东西写进存储之后仍然对得上（pid 是 store 契约里的字段，不是内存里的临时值）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import multiprocessing
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from chongming_task import (
|
|
Controller,
|
|
FailureKind,
|
|
FixedBackoff,
|
|
ProcessExecutor,
|
|
RetryPolicy,
|
|
TaskStatus,
|
|
Timeout,
|
|
run_task,
|
|
)
|
|
from chongming_task.stores.sqlite import SqliteTaskStore
|
|
|
|
_WINDOW = 20.0
|
|
|
|
|
|
def _add(ctx: Any, left: int, right: int) -> int:
|
|
"""两数相加。"""
|
|
del ctx
|
|
return left + right
|
|
|
|
|
|
def _boom(ctx: Any) -> None:
|
|
"""总是失败：用来让任务反复重试。"""
|
|
del ctx
|
|
raise RuntimeError("总是失败")
|
|
|
|
|
|
def _deaf_loop(ctx: Any) -> str:
|
|
"""**不**调用 checkpoint：停不掉，只能强杀。"""
|
|
del ctx
|
|
end = time.monotonic() + 120
|
|
while time.monotonic() < end:
|
|
pass
|
|
return "never"
|
|
|
|
|
|
def test_a_task_run_records_the_child_pid_in_its_snapshot() -> None:
|
|
outcome = run_task(
|
|
_add,
|
|
2,
|
|
40,
|
|
task_id="proc-1",
|
|
executor=ProcessExecutor(),
|
|
timeout=Timeout(per_attempt=_WINDOW),
|
|
)
|
|
|
|
assert outcome.ok, outcome.failure
|
|
assert outcome.value == 42
|
|
snapshot = outcome.snapshot
|
|
assert snapshot.executor == "process"
|
|
assert snapshot.pid is not None
|
|
assert snapshot.pid != os.getpid(), "任务跑在另一个进程里"
|
|
|
|
|
|
def test_an_idle_task_is_actually_killed_by_the_timeout() -> None:
|
|
"""线程执行器下这条只能"放弃等待"并落 `LOST`（线程杀不掉）；进程执行器**真的**把它停掉。
|
|
|
|
超时仍然如实记成超时（不是"已停止"）——但载体确实没了，资源也确实释放了。
|
|
"""
|
|
started = time.monotonic()
|
|
outcome = run_task(
|
|
_deaf_loop,
|
|
task_id="proc-deaf",
|
|
executor=ProcessExecutor(),
|
|
timeout=Timeout(idle=1),
|
|
retry=RetryPolicy(max_attempts=1),
|
|
graceful_stop=0.5,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
|
|
assert outcome.status is TaskStatus.FAILED
|
|
assert outcome.failure is not None
|
|
assert outcome.failure.kind is FailureKind.TIMEOUT_IDLE
|
|
assert outcome.snapshot.pid is not None
|
|
|
|
# 它自己那个循环要跑两分钟，这里几秒就回来了——只有"真的被停掉"解释得通。
|
|
assert elapsed < _WINDOW
|
|
# 载体不在了：那个 pid 已经不再是本进程活着的子进程。
|
|
alive = {child.pid for child in multiprocessing.active_children()}
|
|
assert outcome.snapshot.pid not in alive, "子进程必须真的被终止并回收"
|
|
|
|
|
|
def test_a_controller_can_run_tasks_in_child_processes(tmp_path: Path) -> None:
|
|
store = SqliteTaskStore(tmp_path / "proc.db")
|
|
controller = Controller(store=store, executor=ProcessExecutor(), owner="node-a")
|
|
try:
|
|
controller.register("add", _add)
|
|
controller.submit("add", 20, 22, task_id="t1")
|
|
outcome = controller.wait("t1", timeout=_WINDOW)
|
|
|
|
assert outcome.ok and outcome.value == 42
|
|
assert outcome.snapshot.executor == "process"
|
|
|
|
# 落库之后仍然对得上：pid 是 store 契约里的字段，不是内存里的临时值。
|
|
stored = store.get("t1")
|
|
assert stored is not None
|
|
assert stored.pid == outcome.snapshot.pid
|
|
assert stored.executor == "process"
|
|
finally:
|
|
controller.close(timeout=_WINDOW, close_store=True)
|
|
|
|
|
|
def test_a_target_that_cannot_cross_the_boundary_fails_fast() -> None:
|
|
"""不可 pickle 的目标：任务落 ``failed`` 并带上诊断，而不是让调用方干等。
|
|
|
|
（直接调 ``ProcessExecutor.launch()`` 会当场抛 ``ConfigurationError``；这条测的是
|
|
走 ``run_task`` 那条路——异常在那里必须变成任务的结论。）
|
|
"""
|
|
|
|
def local_closure(ctx: Any) -> str: # 局部函数：子进程里找不到它
|
|
del ctx
|
|
return "nope"
|
|
|
|
started = time.monotonic()
|
|
outcome = run_task(local_closure, task_id="bad-1", executor=ProcessExecutor())
|
|
elapsed = time.monotonic() - started
|
|
|
|
assert outcome.status is TaskStatus.FAILED
|
|
assert outcome.failure is not None
|
|
assert "可 pickle" in outcome.failure.message
|
|
assert outcome.failure.retryable is False
|
|
assert elapsed < _WINDOW, "它必须立刻有结论，而不是等到天荒地老"
|
|
|
|
|
|
def test_a_retry_does_not_carry_the_previous_attempts_pid() -> None:
|
|
"""重试之间采一次快照：上一次的进程号必须已经清掉。
|
|
|
|
留着它的危害很具体：运维看到一个 pid 就去 `ps`，而对上的可能是一个早在别处跑着的
|
|
无关进程——比"没有 pid"更容易误导。
|
|
"""
|
|
controller = Controller(executor=ProcessExecutor())
|
|
controller.register("boom", _boom)
|
|
try:
|
|
controller.submit(
|
|
"boom",
|
|
task_id="t-retry",
|
|
retry=RetryPolicy(max_attempts=3, backoff=FixedBackoff(2.0)),
|
|
)
|
|
deadline = time.monotonic() + _WINDOW
|
|
while time.monotonic() < deadline:
|
|
snapshot = controller.get("t-retry")
|
|
if snapshot is not None and snapshot.status is TaskStatus.RETRY_WAIT:
|
|
break
|
|
time.sleep(0.02)
|
|
else: # pragma: no cover - 没等到就说明退避策略没生效
|
|
raise AssertionError("任务没有进入 retry_wait")
|
|
|
|
assert snapshot is not None
|
|
assert snapshot.pid is None, "上一次尝试的进程号必须清掉"
|
|
assert snapshot.executor == "process", "执行器是这一轮的配置，不随尝试变化"
|
|
finally:
|
|
controller.close(timeout=_WINDOW)
|