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 工业化与质量的一致性 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。