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。
282 lines
9.2 KiB
Python
282 lines
9.2 KiB
Python
"""协作式取消令牌的行为测试。
|
|
|
|
这里必须用**真实线程**：令牌的价值恰恰在于跨线程正确握手（任务线程阻塞在
|
|
暂停点、调度线程请求停止并唤醒它）。用假的线程模型测不出竞态。
|
|
|
|
因此这些用例带小超时并断言线程确实退出——**同时也在验证没有死锁**。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from chongming_task.errors import TaskCancelled
|
|
from chongming_task.runtime.cancellation import CancellationToken
|
|
|
|
_JOIN_TIMEOUT = 5.0
|
|
|
|
|
|
def _run_in_thread(target: object) -> threading.Thread:
|
|
thread = threading.Thread(target=target) # type: ignore[arg-type]
|
|
thread.daemon = True
|
|
thread.start()
|
|
return thread
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 停止
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_checkpoint_is_noop_without_any_request() -> None:
|
|
"""没有请求时 checkpoint 必须立即返回，否则每轮循环都在付代价。"""
|
|
CancellationToken().checkpoint()
|
|
|
|
|
|
def test_checkpoint_raises_when_stop_requested() -> None:
|
|
token = CancellationToken()
|
|
token.request_stop("发布窗口")
|
|
with pytest.raises(TaskCancelled) as info:
|
|
token.checkpoint()
|
|
assert info.value.reason == "发布窗口"
|
|
|
|
|
|
def test_stop_is_idempotent_and_keeps_the_first_reason() -> None:
|
|
"""首个原因通常最接近真实起因，后续请求不应覆盖它。"""
|
|
token = CancellationToken()
|
|
token.request_stop("first")
|
|
token.request_stop("second")
|
|
with pytest.raises(TaskCancelled) as info:
|
|
token.checkpoint()
|
|
assert info.value.reason == "first"
|
|
|
|
|
|
def test_raise_if_stopped_ignores_pause() -> None:
|
|
"""用于清理路径：只响应停止，绝不阻塞。"""
|
|
token = CancellationToken()
|
|
token.request_pause()
|
|
token.raise_if_stopped() # 不应阻塞
|
|
token.request_stop()
|
|
with pytest.raises(TaskCancelled):
|
|
token.raise_if_stopped()
|
|
|
|
|
|
def test_is_stop_requested_reflects_state() -> None:
|
|
token = CancellationToken()
|
|
assert not token.is_stop_requested
|
|
token.request_stop()
|
|
assert token.is_stop_requested
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 暂停
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_pause_blocks_until_resume() -> None:
|
|
token = CancellationToken()
|
|
token.request_pause() # 先请求，消除"任务先跑到检查点"的竞态
|
|
observed: list[str] = []
|
|
|
|
def worker() -> None:
|
|
token.checkpoint()
|
|
observed.append("resumed")
|
|
|
|
thread = _run_in_thread(worker)
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is True
|
|
assert observed == [], "暂停请求生效前任务不应继续执行"
|
|
|
|
token.resume()
|
|
thread.join(_JOIN_TIMEOUT)
|
|
assert not thread.is_alive(), "resume() 必须唤醒阻塞中的任务"
|
|
assert observed == ["resumed"]
|
|
|
|
|
|
def test_stop_wakes_a_paused_task() -> None:
|
|
"""暂停中的任务必须能被停止，否则"暂停"就成了不可撤销的挂起。"""
|
|
token = CancellationToken()
|
|
token.request_pause()
|
|
reasons: list[str | None] = []
|
|
|
|
def worker() -> None:
|
|
try:
|
|
token.checkpoint()
|
|
except TaskCancelled as exc:
|
|
reasons.append(exc.reason)
|
|
|
|
thread = _run_in_thread(worker)
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is True
|
|
|
|
token.request_stop("运维介入")
|
|
thread.join(_JOIN_TIMEOUT)
|
|
assert not thread.is_alive(), "停止请求必须能唤醒暂停点"
|
|
assert reasons == ["运维介入"]
|
|
|
|
|
|
def test_repeated_pause_does_not_clear_acknowledgement() -> None:
|
|
"""重复请求暂停不应把"已确认"重置掉，否则 wait_paused 会误判为未生效。"""
|
|
token = CancellationToken()
|
|
token.request_pause()
|
|
|
|
def worker() -> None:
|
|
token.checkpoint()
|
|
|
|
thread = _run_in_thread(worker)
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is True
|
|
token.request_pause()
|
|
assert token.is_pause_acknowledged is True
|
|
|
|
token.resume()
|
|
thread.join(_JOIN_TIMEOUT)
|
|
|
|
|
|
def test_wait_paused_times_out_when_task_never_checkpoints() -> None:
|
|
"""调度器必须能区分"已暂停"与"暂停没生效"——后者要回滚为 RUNNING 并告警。"""
|
|
token = CancellationToken()
|
|
token.request_pause()
|
|
assert token.wait_paused(timeout=0.05) is False
|
|
assert token.is_pause_acknowledged is False
|
|
|
|
|
|
def test_wait_paused_reports_false_when_stopped_instead() -> None:
|
|
"""请求暂停后又被停止：这不是"暂停成功"，必须返回 False。"""
|
|
token = CancellationToken()
|
|
token.request_pause()
|
|
token.request_stop()
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is False
|
|
|
|
|
|
def test_resume_without_pause_is_noop() -> None:
|
|
token = CancellationToken()
|
|
token.resume()
|
|
assert not token.is_pause_requested
|
|
token.checkpoint() # 仍应立即返回
|
|
|
|
|
|
def _start_pausing_worker(token: CancellationToken) -> threading.Event:
|
|
"""起一个持续调用 ``checkpoint()`` 的工作线程，返回"完成过一轮"的事件。
|
|
|
|
线程必须**持续**调用 checkpoint——只调一次就退出的话，之后的暂停请求自然没人
|
|
确认，那是测试写错而不是缺陷。
|
|
"""
|
|
|
|
def worker() -> None:
|
|
# 收尾时的 stop 会让 checkpoint 抛 TaskCancelled：那正是预期行为，
|
|
# 但线程里未捕获的异常会被 pytest 转成错误。
|
|
with contextlib.suppress(TaskCancelled):
|
|
while True:
|
|
token.checkpoint()
|
|
ran.set()
|
|
|
|
ran = threading.Event()
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
return ran
|
|
|
|
|
|
def test_pause_is_acknowledged_again_after_a_resume() -> None:
|
|
"""**回归用例**：暂停 → 恢复 → 再暂停里，第二次暂停曾经永久失效。
|
|
|
|
``checkpoint()`` 原先只在**进入**暂停分支时设置 ``_ack``。``resume()`` 把
|
|
``_pause`` 置回 False 并通知后，任务线程还没来得及重新取到锁，新的
|
|
``request_pause()`` 就把 ``_pause`` 又置成了 True——线程重新检查循环条件时发现
|
|
"还要暂停"，于是继续等待，而它已经越过了设置 ``_ack`` 的那一行。
|
|
|
|
结果 ``wait_paused()`` 永远等不到确认，暂停超时回滚。表现出来就是"暂停一次成功，
|
|
之后再也暂停不了"。
|
|
"""
|
|
token = CancellationToken()
|
|
ran = _start_pausing_worker(token)
|
|
assert ran.wait(_JOIN_TIMEOUT), "工作线程没有跑起来"
|
|
|
|
for round_index in range(5):
|
|
token.request_pause()
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is True, (
|
|
f"第 {round_index + 1} 轮暂停没有得到确认"
|
|
)
|
|
|
|
ran.clear()
|
|
token.resume()
|
|
assert ran.wait(_JOIN_TIMEOUT), f"第 {round_index + 1} 轮恢复后任务没有继续推进"
|
|
|
|
token.request_stop()
|
|
|
|
|
|
def test_a_pause_right_after_a_resume_is_still_acknowledged() -> None:
|
|
"""恢复与再次暂停之间不留任何间隔——最容易踩中的时序。"""
|
|
token = CancellationToken()
|
|
ran = _start_pausing_worker(token)
|
|
assert ran.wait(_JOIN_TIMEOUT)
|
|
|
|
token.request_pause()
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is True
|
|
|
|
# 立刻恢复再立刻暂停：不给任务线程任何重新取锁的机会。
|
|
token.resume()
|
|
token.request_pause()
|
|
assert token.wait_paused(timeout=_JOIN_TIMEOUT) is True, "紧接着的第二次暂停丢了"
|
|
|
|
token.resume()
|
|
token.request_stop()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 异步路径
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_acheckpoint_fast_path_does_not_offload_to_thread() -> None:
|
|
"""紧密循环里的每次检查都丢线程会拖垮事件循环，快速路径必须留在本线程。"""
|
|
|
|
async def main() -> str:
|
|
token = CancellationToken()
|
|
for _ in range(2000):
|
|
await token.acheckpoint()
|
|
return "done"
|
|
|
|
assert asyncio.run(main()) == "done"
|
|
|
|
|
|
def test_acheckpoint_raises_when_stopped() -> None:
|
|
token = CancellationToken()
|
|
token.request_stop("已取消")
|
|
|
|
async def main() -> None:
|
|
await token.acheckpoint()
|
|
|
|
with pytest.raises(TaskCancelled) as info:
|
|
asyncio.run(main())
|
|
assert info.value.reason == "已取消"
|
|
|
|
|
|
def test_acheckpoint_suspends_and_resumes_without_blocking_loop() -> None:
|
|
"""暂停期间事件循环必须仍能推进——这正是异步版本存在的理由。"""
|
|
token = CancellationToken()
|
|
token.request_pause()
|
|
|
|
async def main() -> list[str]:
|
|
events: list[str] = []
|
|
|
|
async def heartbeat() -> None:
|
|
while True:
|
|
events.append("tick")
|
|
await asyncio.sleep(0.01)
|
|
|
|
ticker = asyncio.create_task(heartbeat())
|
|
waiter = asyncio.create_task(token.acheckpoint())
|
|
|
|
while not token.is_pause_acknowledged:
|
|
await asyncio.sleep(0.01)
|
|
ticks_while_paused = len(events)
|
|
assert ticks_while_paused > 0, "暂停期间事件循环不应被阻塞"
|
|
|
|
token.resume()
|
|
await asyncio.wait_for(waiter, timeout=_JOIN_TIMEOUT)
|
|
ticker.cancel()
|
|
return events
|
|
|
|
assert asyncio.run(main())
|