第 19 章 · 任务规划与调度

# ❌ 有重叠：两个检查都会跑到 missing_rate
tasks = [
    Task("结构检查", checks=["required_vars", "missing_rate"]),
    Task("完整性检查", checks=["missing_rate", "date_pairs"]),
]

# ❌ 有遗漏：ADSL 有 5 项必查，只写了 3 项
tasks = [
    Task("检查 ADSL", checks=["required_vars", "missing_rate", "key_unique"]),
]

# ✅ MECE：每一项检查恰好出现在一个任务里
ALL_CHECKS = ["required_vars", "missing_rate", "key_unique",
              "date_pairs", "codelist", "range"]
tasks = [Task(f"检查 {c}", checks=[c]) for c in ALL_CHECKS]

# clinic/agent_planner.py（节选）
@dataclass
class Task:
    """一个可调度的任务。

    刻意**不持有函数引用** —— 只描述"要做什么"，
    由执行器通过工具注册表去调用。这样任务图可以序列化、可以重放。
    """
    id: str
    title: str
    tool: str                         # 要调用的工具名
    args: dict = field(default_factory=dict)
    depends_on: tuple[str, ...] = ()  # 依赖的任务 id
    optional: bool = False            # 失败是否阻塞整条流水线
    status: str = "pending"           # pending/running/done/failed/skipped
    result: Any = None
    error: str | None = None
    elapsed: float = 0.0

def topological_order(tasks: list[Task]) -> list[Task]:
    """Kahn 拓扑排序。有环时抛 PlanCycleError。"""
    by_id = {t.id: t for t in tasks}
    indeg = {t.id: len(t.depends_on) for t in tasks}
    children: dict[str, list[str]] = {t.id: [] for t in tasks}

    for t in tasks:
        for dep in t.depends_on:
            if dep not in by_id:
                raise PlanError(f"任务 {t.id} 依赖了不存在的任务 {dep}")
            children[dep].append(t.id)

    queue = [tid for tid, d in indeg.items() if d == 0]
    order: list[str] = []
    while queue:
        tid = queue.pop(0)
        order.append(tid)
        for ch in children[tid]:
            indeg[ch] -= 1
            if indeg[ch] == 0:
                queue.append(ch)

    if len(order) != len(tasks):
        stuck = sorted(set(indeg) - set(order))
        raise PlanCycleError(f"任务图存在循环依赖：{stuck}")
    return [by_id[tid] for tid in order]

def levels(tasks: list[Task]) -> list[list[Task]]:
    """把任务按"最早可执行轮次"分层。同一层内的任务互不依赖，可以并行。"""
    by_id = {t.id: t for t in tasks}
    depth: dict[str, int] = {}

    def d(tid: str) -> int:
        if tid not in depth:
            depth[tid] = 0 if not by_id[tid].depends_on else \
                1 + max(d(x) for x in by_id[tid].depends_on)
        return depth[tid]

    out: dict[int, list[Task]] = {}
    for t in tasks:
        out.setdefault(d(t.id), []).append(t)
    return [out[k] for k in sorted(out)]

# 别信任何人的结论，包括本书的 —— 在你的数据上量一次
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def bench(fn, items, workers=4):
    t0 = time.perf_counter(); [fn(x) for x in items]           # 串行
    serial = time.perf_counter() - t0

    t0 = time.perf_counter()
    with ThreadPoolExecutor(workers) as ex: list(ex.map(fn, items))
    thread = time.perf_counter() - t0

    print(f"串行 {serial:6.2f}s  多线程 {thread:6.2f}s  "
          f"加速 {serial / thread:.2f}x")

# ❌ 反例：把数据传过去
with ProcessPoolExecutor(4) as ex:
    ex.map(process, [df1, df2, df3, df4])       # 每个 df 都会被 pickle 一遍

# ✅ 正例：把路径传过去，让子进程自己读
with ProcessPoolExecutor(4) as ex:
    ex.map(process_file, ["data/samples/adsl.csv",
                          "data/samples/adae.csv", ...])

def execute_plan(plan: Plan, registry, policy,
                 max_workers: int = 4,
                 on_task: Callable[[Task], None] | None = None) -> Plan:
    """按层执行：同层并行，层间串行。"""
    plan = copy.deepcopy(plan)

    for layer in levels(plan.tasks):
        ready = [t for t in layer if all_deps_ok(t, plan)]
        if not ready:
            continue
        if len(ready) == 1 or max_workers <= 1:
            for t in ready:
                _run_one(t, registry, policy, on_task)
        else:
            with ThreadPoolExecutor(max_workers=min(max_workers, len(ready))) as ex:
                list(ex.map(lambda t: _run_one(t, registry, policy, on_task), ready))
    return plan

RETRYABLE = {"transient"}

def _run_one(task: Task, registry, policy, on_task=None) -> None:
    for attempt in range(1, policy.max_retries + 1):
        t0 = time.perf_counter()
        res = registry.execute(task.tool, task.args, policy)
        task.elapsed = time.perf_counter() - t0

        if res.ok:
            task.status, task.result = "done", res.data
            break

        if res.error_type in RETRYABLE and attempt < policy.max_retries:
            time.sleep(policy.backoff_base ** attempt)     # 指数退避 2s, 4s, 8s
            continue

        task.status, task.error = "failed", res.error
        break

    if on_task:
        on_task(task)

def summarize(plan: Plan) -> str:
    done = [t for t in plan.tasks if t.status == "done"]
    failed = [t for t in plan.tasks if t.status == "failed"]
    skipped = [t for t in plan.tasks if t.status == "skipped"]

    if failed:
        # ⚠️ 绝不能说"检查完成" —— 必须说清楚哪几项没做
        return (f"完成 {len(done)}/{len(plan.tasks)} 项；"
                f"**{len(failed)} 项失败**（{', '.join(t.title for t in failed)}）；"
                f"{len(skipped)} 项跳过。结论仅覆盖已完成部分。")
    return f"全部 {len(done)} 项完成。"

@dataclass
class PlanPolicy:
    max_workers: int = 4
    max_retries: int = 2
    backoff_base: float = 2.0
    hitl_before: tuple[str, ...] = ("write_report", "write_xpt", "send_summary")
    """命中这些工具时暂停，等待人工确认后才继续。"""

python cases/case10_TLF生成流水线.py                    # 4 并发
python cases/case10_TLF生成流水线.py --workers 1        # 对照：串行
python cases/case10_TLF生成流水线.py --fail t3          # 故意让一个任务失败，看汇总
python cases/case10_TLF生成流水线.py --dry-run          # 只打印计划，不执行

/* SAS 的"任务调度"通常就是按顺序跑 */
%include "01_adsl.sas";
%include "02_table1.sas";
%include "03_table2.sas";
%include "04_table3.sas";
%include "05_export.sas";