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。
75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""示例 2：运行期干预 —— 暂停、恢复、停止。
|
||
|
||
演示协作式暂停（D6 定案的语义）与"停不掉时会如实上报"。
|
||
|
||
运行::
|
||
|
||
uv run python examples/02_control.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
|
||
from chongming_task.run import TaskRun
|
||
|
||
|
||
def long_running(ctx) -> str:
|
||
"""持续做小步工作，每一步都经过协作点。"""
|
||
for _ in range(10_000):
|
||
ctx.checkpoint()
|
||
time.sleep(0.01)
|
||
return "跑完了"
|
||
|
||
|
||
def uncooperative(ctx) -> str:
|
||
"""刻意**不**调用 checkpoint：这样的任务既停不掉也暂停不了。
|
||
|
||
它约 4 秒后自行结束，以免在示例里留下后台线程。
|
||
"""
|
||
for _ in range(400):
|
||
time.sleep(0.01)
|
||
return "无视了停止请求"
|
||
|
||
|
||
def demo_pause_resume() -> None:
|
||
print("== 暂停 / 恢复 ==")
|
||
run = TaskRun(long_running, task_id="demo-pause")
|
||
run.start()
|
||
time.sleep(0.2)
|
||
|
||
print(f" 暂停请求已受理：{run.pause(timeout=5)}；当前状态 {run.status.value}")
|
||
time.sleep(0.2)
|
||
print(f" 暂停期间状态保持：{run.status.value}")
|
||
|
||
run.resume()
|
||
print(f" 恢复后状态 ：{run.status.value}")
|
||
|
||
run.stop(reason="演示主动停止")
|
||
outcome = run.wait(timeout=10)
|
||
print(f" 停止后终态 ：{outcome.status.value}")
|
||
print(f" 失败归类 ：{outcome.failure.kind.value if outcome.failure else '-'}")
|
||
print(f" 可重试 ：{outcome.failure.retryable if outcome.failure else '-'}")
|
||
|
||
|
||
def demo_unstoppable() -> None:
|
||
print("\n== 停不掉的任务：如实上报 ==")
|
||
run = TaskRun(uncooperative, task_id="demo-stubborn", graceful_stop=0.3)
|
||
run.start()
|
||
time.sleep(0.2)
|
||
run.stop(reason="试一下")
|
||
|
||
outcome = run.wait(timeout=10)
|
||
print(f" 终态 ：{outcome.status.value}")
|
||
print(f" 失败归类 ：{outcome.failure.kind.value if outcome.failure else '-'}")
|
||
print(" 含义 ：线程无法强杀，任务可能仍在后台产生副作用，因此不敢说'已停止'")
|
||
|
||
|
||
def main() -> None:
|
||
demo_pause_resume()
|
||
demo_unstoppable()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|