14.2 排期与美术依赖

排期不是画一张甘特图,而是在依赖 DAG 上算关键路径。本节用《星屑旅店》九个美术任务的依赖图,算出最早开始、最晚结束与浮动时间(全整数),S 层 Python 与 P 层 TS 内核的 digest 逐位一致,并用三个负例证明检测器会拦下环、悬空依赖与紧截止导致的负浮动。

14.2 排期与美术依赖

14.1 产能度量:人天与吞吐 算的是总量——一个迭代有多少有效人天。总量够,不等于排得开:谁必须先做、谁可以并行、哪条链决定了交期,是另一张账。这一节把美术任务之间「谁依赖谁」建成一张有向无环图(DAG),在上面算关键路径。图本身怎么建、环怎么检测,4.2 环检测与悬空引用 已从依赖图角度讲过;本节换到排期角度,只算四个量:最早开始、最早结束、最晚开始、最晚结束,以及由它们推出的浮动时间。不变量:排期是整数题,且依赖必须显式声明。 真实排期还要叠加人员日历、供应商交期、评审往返,那些需真实环境(真实团队与合同),命令行不可复现;本节只做纯依赖结构上的关键路径,这是所有真实排期的骨架。

14.2.1 甘特图是画出来的,关键路径是算出来的

甘特图把每个任务画成一根横条,好看,但它本身不告诉你哪根条拖了整条链。真正决定交期的是那条最长的依赖链——关键路径(critical path):它上面的任何任务晚一天,项目就晚一天;不在它上面的任务晚一点,只要没吃掉自己的余量,项目纹丝不动。标准算法是 CPM,四步(由它们得到浮动时间 float = LS - ES,float == 0 即在关键路径上):整张表全整数,无浮点。

量含义递推
ES 最早开始所有前驱都完成后才能开始ES = max(前驱的 EF)
EF 最早结束最早开始 + 工期EF = ES + days
LF 最晚结束不影响后继的最晚收工LF = min(后继的 LS)
LS 最晚开始最晚结束 - 工期LS = LF - days

14.2.2 配置:九个美术任务的依赖图

下面是 D 层的一份示例排期:九个美术任务,每个有整数工期(人天)和它的前驱列表,覆盖《星屑旅店》一屏大厅从概念到集成验收的全部环节。

{
  "project": "stardust-inn",
  "chapter": 14,
  "note": "示例参数:本套构造的美术任务依赖 DAG(整数人天),非任何真实项目排期。",
  "deadline": 18,
  "tightDeadline": 16,
  "tasks": [
    { "id": "concept", "name": "概念与参考板", "days": 3, "deps": [] },
    { "id": "palette", "name": "色板与规范", "days": 2, "deps": ["concept"] },
    { "id": "furniture", "name": "家具绘制", "days": 6, "deps": ["palette"] },
    { "id": "portrait", "name": "角色头像", "days": 5, "deps": ["palette"] },
    { "id": "ui", "name": "UI 与图标", "days": 4, "deps": ["palette"] },
    { "id": "atlas", "name": "图集打包", "days": 2, "deps": ["furniture", "ui"] },
    { "id": "vfx", "name": "特效", "days": 3, "deps": ["ui"] },
    { "id": "scene", "name": "场景装配", "days": 4, "deps": ["furniture", "portrait", "ui"] },
    { "id": "integrate", "name": "集成验收", "days": 3, "deps": ["atlas", "vfx", "scene"] }
  ]
}

14.2.3 S 层:关键路径计算

下面这段是 S 层脚本,用 Kahn 拓扑排序定顺序,做前向(ES/EF)与后向(LS/LF)两遍,再算浮动、关键路径与 digest。negatives() 里内置三处负例断言:环、悬空依赖、紧截止负浮动。

#!/usr/bin/env python3
# file: sim/ta14_schedule.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ta14_schedule.py
"""14.2 关键路径(CPM)的 S 层实现(卷二第 14 章)。

读 config/ta14_schedule.json 的美术任务依赖 DAG(整数人天):前向 ES=max(前驱 EF)、EF=ES+days;
后向 LF=min(后继 LS)(汇点取 deadline,缺省取工期)、LS=LF-days;float=LS-ES,float==0 即关键路径;
duration=max(EF);minFloat<0 即不可行。Kahn 拓扑剩余非空即存在环(cycle_nodes)。
main() 里带三处负例断言(环 / 悬空依赖 / 紧截止负浮动)。全整数,行按 id 排序 → 与 TS 逐位一致。
"""
import hashlib
import json
import os
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CFG = os.path.join(ROOT, "config/ta14_schedule.json")
OUT = os.path.join(ROOT, "out")


def build(tasks):
    """返回 (ids, succ, indeg, deps)。deps 只保留存在的 id。"""
    ids = [t["id"] for t in tasks]
    idset = set(ids)
    succ = {i: [] for i in ids}
    indeg = {i: 0 for i in ids}
    deps = {t["id"]: [d for d in t["deps"] if d in idset] for t in tasks}
    for t in tasks:
        for d in deps[t["id"]]:
            succ[d].append(t["id"])
            indeg[t["id"]] += 1
    return ids, succ, indeg, deps


def topo(ids, succ, indeg):
    """Kahn 拓扑。返回 (order, remaining);remaining 非空即存在环。"""
    indeg = dict(indeg)
    ready = sorted(i for i in ids if indeg[i] == 0)
    order = []
    while ready:
        n = ready.pop(0)
        order.append(n)
        for m in succ[n]:
            indeg[m] -= 1
            if indeg[m] == 0:
                ready.append(m)
        ready.sort()
    return order, sorted(set(ids) - set(order))


def cycle_nodes(tasks):
    """处在环上或其下游的节点(Kahn 剩余);无环为空。"""
    ids, succ, indeg, _ = build(tasks)
    return topo(ids, succ, indeg)[1]


def dangling(tasks):
    """悬空依赖:deps 指向不存在的 id。"""
    idset = {t["id"] for t in tasks}
    return sorted(f"DANGLING:{t['id']}->{d}" for t in tasks for d in t["deps"] if d not in idset)


def schedule(tasks, deadline=None):
    """标准 CPM。有环返回 None。"""
    ids, succ, indeg, deps = build(tasks)
    days = {t["id"]: t["days"] for t in tasks}
    order, cyc = topo(ids, succ, indeg)
    if cyc:
        return None
    es, ef = {}, {}
    for i in order:
        es[i] = max([ef[d] for d in deps[i]], default=0)
        ef[i] = es[i] + days[i]
    duration = max(ef.values())
    cap = deadline if deadline is not None else duration
    ls, lf = {}, {}
    for i in reversed(order):
        lf[i] = min([ls[m] for m in succ[i]]) if succ[i] else cap
        ls[i] = lf[i] - days[i]
    rows = [{"id": i, "days": days[i], "es": es[i], "ef": ef[i], "ls": ls[i], "lf": lf[i],
             "float": ls[i] - es[i], "critical": ls[i] - es[i] == 0} for i in sorted(ids)]
    lines = [f"{r['id']}:{r['es']}:{r['ef']}:{r['ls']}:{r['lf']}:{r['float']}" for r in rows]
    return {"rows": rows, "duration": duration,
            "criticalPath": [i for i in order if ls[i] - es[i] == 0],
            "minFloat": min(r["float"] for r in rows),
            "digest": hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest()[:16]}


def negatives():
    """三处负例断言:环必须中止排期、悬空依赖必须被检出、紧截止必须给出负浮动。"""
    cyc = [{"id": "a", "days": 1, "deps": ["b"]}, {"id": "b", "days": 1, "deps": ["a"]}]
    assert cycle_nodes(cyc) == ["a", "b"] and schedule(cyc) is None
    assert dangling([{"id": "a", "days": 1, "deps": ["ghost"]}]) == ["DANGLING:a->ghost"]
    return {"cycleNodes": cycle_nodes(cyc), "dangling": "DANGLING:a->ghost"}


def main() -> int:
    cfg = json.load(open(CFG, encoding="utf-8"))
    tasks = cfg["tasks"]
    neg = negatives()
    m = schedule(tasks)
    tight = schedule(tasks, cfg["tightDeadline"])
    assert tight["minFloat"] < 0, "紧截止应给出负浮动"
    rep = {"project": cfg["project"], "deadline": cfg["deadline"],
           "tightDeadline": cfg["tightDeadline"], "duration": m["duration"],
           "minFloat": m["minFloat"], "criticalPath": m["criticalPath"], "digest": m["digest"],
           "rows": m["rows"], "dangling": dangling(tasks), "cycle": cycle_nodes(tasks),
           "tightMinFloat": tight["minFloat"], "tightDigest": tight["digest"]}
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ta14_schedule.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2, sort_keys=True)
        f.write("\n")
    print(f"duration={m['duration']} minFloat={m['minFloat']} "
          f"critical={'->'.join(m['criticalPath'])}")
    print(f"dangling={rep['dangling']} cycle={rep['cycle']}")
    for r in m["rows"]:
        print(f"  {r['id']:10s} d={r['days']} ES={r['es']:2d} EF={r['ef']:2d} LS={r['ls']:2d} "
              f"LF={r['lf']:2d} float={r['float']:2d}{' *' if r['critical'] else ''}")
    print(f"digest={m['digest']}")
    print(f"tightDeadline={cfg['tightDeadline']} tightMinFloat={tight['minFloat']} "
          f"tightDigest={tight['digest']}")
    print(f"negative: cycleNodes={neg['cycleNodes']} dangling={neg['dangling']}")
    print("wrote out/ta14_schedule.json")
    return 0


if __name__ == "__main__":
    sys.exit(main())

实跑 /opt/homebrew/bin/python3 sim/ta14_schedule.py:

duration=18 minFloat=0 critical=concept->palette->furniture->scene->integrate
dangling=[] cycle=[]
  atlas      d=2 ES=11 EF=13 LS=13 LF=15 float= 2
  concept    d=3 ES= 0 EF= 3 LS= 0 LF= 3 float= 0 *
  furniture  d=6 ES= 5 EF=11 LS= 5 LF=11 float= 0 *
  integrate  d=3 ES=15 EF=18 LS=15 LF=18 float= 0 *
  palette    d=2 ES= 3 EF= 5 LS= 3 LF= 5 float= 0 *
  portrait   d=5 ES= 5 EF=10 LS= 6 LF=11 float= 1
  scene      d=4 ES=11 EF=15 LS=11 LF=15 float= 0 *
  ui         d=4 ES= 5 EF= 9 LS= 7 LF=11 float= 2
  vfx        d=3 ES= 9 EF=12 LS=12 LF=15 float= 3
digest=b1a1398e77e216b2
tightDeadline=16 tightMinFloat=-2 tightDigest=41a9464176288912
negative: cycleNodes=['a', 'b'] dangling=DANGLING:a->ghost
wrote out/ta14_schedule.json

工期 18 天,关键路径是 concept → palette → furniture → scene → integrate(3+2+6+4+3=18)。三条并行分支里 furniture 用了 6 天最长,所以 portrait(float=1)与 ui(float=2)各有余量;vfx(float=3)最松。这些浮动值就是排期的调度空间。

14.2.4 P 层:同一套 CPM 的 TS 内核

下面这段是 P 层 TS 内核,与上面的 Python 逐条对应——同一套 Kahn 拓扑、同一套前向后向、同一个 digest 口径:

// file: core/src/ta14/schedule.ts
// tier: P
// verify: node --test 'core/test/ta14_*.test.ts'
// 14.2 关键路径(CPM)的 P 层内核。与 sim/ta14_schedule.py 同一套算法与 digest 口径:
// Kahn 拓扑 + 前向 ES/EF + 后向 LS/LF + 浮动。全整数、无随机 → 与 Python 逐位一致。
// 只允许可擦除 TS 语法(禁 enum / namespace / 参数属性 / 装饰器)。
import { createHash } from "node:crypto";

export type Task = { id: string; days: number; deps: string[] };
export type Row = {
  id: string; days: number; es: number; ef: number;
  ls: number; lf: number; float: number; critical: boolean;
};

export function build(tasks: Task[]) {
  const ids = tasks.map((t) => t.id);
  const idset = new Set(ids);
  const succ: Record<string, string[]> = {};
  const indeg: Record<string, number> = {};
  const deps: Record<string, string[]> = {};
  for (const i of ids) { succ[i] = []; indeg[i] = 0; }
  for (const t of tasks) {
    deps[t.id] = t.deps.filter((d) => idset.has(d));
    for (const d of deps[t.id]) { succ[d].push(t.id); indeg[t.id] += 1; }
  }
  return { ids, succ, indeg, deps };
}

/** Kahn 拓扑。remaining 非空即存在环。 */
export function topo(ids: string[], succ: Record<string, string[]>, indegIn: Record<string, number>) {
  const indeg = { ...indegIn };
  const ready = ids.filter((i) => indeg[i] === 0).sort();
  const order: string[] = [];
  while (ready.length) {
    const n = ready.shift() as string;
    order.push(n);
    for (const m of succ[n]) { indeg[m] -= 1; if (indeg[m] === 0) { ready.push(m); ready.sort(); } }
  }
  return { order, remaining: ids.filter((i) => !order.includes(i)).sort() };
}

/** 处在环上或其下游的节点(Kahn 剩余);无环为空。 */
export function cycleNodes(tasks: Task[]): string[] {
  const { ids, succ, indeg } = build(tasks);
  return topo(ids, succ, indeg).remaining;
}

/** 悬空依赖:deps 指向不存在的 id。 */
export function danglingDeps(tasks: Task[]): string[] {
  const idset = new Set(tasks.map((t) => t.id));
  const out: string[] = [];
  for (const t of tasks) for (const d of t.deps) if (!idset.has(d)) out.push(`DANGLING:${t.id}->${d}`);
  return out.sort();
}

export type ScheduleResult = {
  rows: Row[]; duration: number; criticalPath: string[];
  minFloat: number; digest: string;
};

/** 标准 CPM。有环返回 null。 */
export function schedule(tasks: Task[], deadline?: number): ScheduleResult | null {
  const { ids, succ, indeg, deps } = build(tasks);
  const days: Record<string, number> = {};
  for (const t of tasks) days[t.id] = t.days;
  const { order, remaining } = topo(ids, succ, indeg);
  if (remaining.length) return null;
  const es: Record<string, number> = {}, ef: Record<string, number> = {};
  for (const i of order) {
    const vals = deps[i].map((d) => ef[d]);
    es[i] = vals.length ? Math.max(...vals) : 0;
    ef[i] = es[i] + days[i];
  }
  const duration = Math.max(...order.map((i) => ef[i]));
  const cap = deadline !== undefined ? deadline : duration;
  const ls: Record<string, number> = {}, lf: Record<string, number> = {};
  for (let k = order.length - 1; k >= 0; k -= 1) {
    const i = order[k];
    lf[i] = succ[i].length ? Math.min(...succ[i].map((m) => ls[m])) : cap;
    ls[i] = lf[i] - days[i];
  }
  const rows: Row[] = [...ids].sort().map((i) => ({
    id: i, days: days[i], es: es[i], ef: ef[i], ls: ls[i], lf: lf[i],
    float: ls[i] - es[i], critical: ls[i] - es[i] === 0,
  }));
  const lines = rows.map((r) => `${r.id}:${r.es}:${r.ef}:${r.ls}:${r.lf}:${r.float}`);
  return {
    rows, duration, criticalPath: order.filter((i) => ls[i] - es[i] === 0),
    minFloat: Math.min(...rows.map((r) => r.float)),
    digest: createHash("sha256").update(lines.join("\n"), "utf8").digest("hex").slice(0, 16),
  };
}

14.2.5 跨语言对照与三个负例

P 层测试读的是 Python 落盘的 out/ta14_schedule.json(不是同一份源码),做跨语言对照,并带三个负例断言。实跑 node --test 'core/test/ta14_*.test.ts':

✔ 跨语言:ES/EF/LS/LF/float/critical/digest 与 Python 产物逐位一致 (1.414833ms)
✔ 负例:环 → cycleNodes 报出、schedule 返回 null;悬空依赖被检出 (0.184958ms)
✔ 负例:紧截止 16 < 工期 18 → 关键路径出现负浮动,排期不可行 (0.13275ms)
ℹ tests 3
ℹ pass 3
ℹ fail 0

两侧的排期 digest 都是 b1a1398e77e216b2,紧截止 digest 都是 41a9464176288912——这是整数类,逐位一致(不是「容差内一致」;容差那套只适用于卷一第 4 章的色彩派生量)。ES/EF/LS/LF/float/critical 六列同样逐位一致。三个负例各拦一类真实错误:环(a→b→a)让排期无法终止,cycleNodes 报出、schedule 返回 null;悬空依赖被 danglingDeps 检出;紧截止(交期从 18 压到 16)让关键路径出现 -2 的浮动——负浮动就是「这份排期不可能」的机器判据。

14.2.6 浮动时间怎么用

浮动时间不是「可以偷懒的天数」,而是调度自由度:

  • float = 0 的任务不能动——关键链上任何一天延误都直接推后交期,资源优先保它们;vfx 有 3 天浮动,可让特效师先支援关键任务 furniture 再回来做特效。
  • 负浮动是可行性报警。紧截止 16 天时 minFloat = -2,意思是「要么砍 2 天工作量,要么接受延期 2 天」——排期会上该拍板的,不是藏着等它爆。

14.2.7 X 边界

本节只做纯依赖结构上的整数关键路径。真实排期还要叠加人员日历(谁哪天请假)、外部供应商交期、评审往返的随机等待——这些需真实环境(真实团队与合同),命令行不可复现,本节的 DAG 是它们的骨架而非替代。数字是示例参数,绝对数值不可移植。

阅读导航:上一节:14.1 产能度量:人天与吞吐 · 下一节:14.3 工业化与质量的一致性 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「gameart」更多文章

  1. 附录 A:风格参数表、量化规则与数据集索引
  2. 9.3 等距资产的制作约束
  3. 9.2 等距 tile 与遮挡规则