一个构建或内容处理任务通常有前置依赖：先取得资源，再解析、生成索引和缩略图，最后发布结果。把任务作为顶点，依赖作为有向边，就可以先检查是否可执行，再计算理论上的最早完成时间。

先写清调度模型

本项目每个任务有一个非负整数时长，任务启动前所有依赖必须完成；任务不能拆分，且假设执行资源无限。目标是输出一个合法拓扑序、每个任务的最早开始/结束时间和全部任务的最早完成时间。它没有实际启动进程，也没有模拟有限CPU、失败重试或网络争用。

以 fetch=2、parse=3、image=4、publish=1 为例，parse和image都依赖fetch，publish依赖前两项。fetch在0开始、2结束；两项中间工作并行，publish最早6开始、7结束。把四项时长直接相加得到10，是串行执行时间，不是当前模型的最优完成时间。

拓扑检查与最长路径在同一轮完成

维护每个任务的未完成依赖数量。零入度任务进入就绪堆，同一时刻用名字排序，让输出可复现。取出任务后，计算它的结束时间，再用该值更新每个后继的最早开始时间。后继最后一个前置任务处理完时，其最早开始时间就是所有前置结束时间的最大值。

import heapq


def plan_tasks(durations, dependencies):
    nodes = set(durations)
    if set(dependencies) - nodes:
        raise ValueError('dependencies mention an unknown task')
    if any(not isinstance(name, str) for name in nodes):
        raise ValueError('task IDs must be strings')
    if any(not isinstance(value, int) or value < 0 for value in durations.values()):
        raise ValueError('durations must be non-negative integers')
    outgoing = {node: set() for node in nodes}
    indegree = {}
    for node in nodes:
        parents = set(dependencies.get(node, ()))
        if parents - nodes:
            raise ValueError('unknown dependency for ' + node)
        indegree[node] = len(parents)
        for parent in parents:
            outgoing[parent].add(node)
    ready = [node for node in nodes if indegree[node] == 0]
    heapq.heapify(ready)
    order = []
    start = {node: 0 for node in nodes}
    finish = {}
    while ready:
        node = heapq.heappop(ready)
        order.append(node)
        finish[node] = start[node] + durations[node]
        for child in outgoing[node]:
            start[child] = max(start[child], finish[node])
            indegree[child] -= 1
            if indegree[child] == 0:
                heapq.heappush(ready, child)
    if len(order) != len(nodes):
        remaining = sorted(node for node in nodes if indegree[node])
        raise ValueError('cycle or blocked by cycle: ' + ', '.join(remaining))
    return {'order': order, 'start': start, 'finish': finish,
            'total': max(finish.values(), default=0)}


if __name__ == '__main__':
    print(plan_tasks({'fetch': 2, 'parse': 3, 'image': 4, 'publish': 1},
                     {'parse': ['fetch'], 'image': ['fetch'],
                      'publish': ['parse', 'image']}))

这个递推是DAG上的最长路径DP。对每个任务，沿某条依赖链的总时长都是无法绕过的下界；资源无限时，各任务在所有依赖完成后立刻启动，可以达到这些下界的最大值，所以得到最早完成时间。拓扑序只是合法处理顺序，不表示实际必须串行执行。

错误必须能定位

未知任务名通常意味着录入错误，代码直接拒绝。重复依赖先转换为集合，避免入度重复但出边只减一次。若最后仍有任务未处理，存在环或受到环阻塞的下游任务；错误消息列出未完成节点，这个列表不保证每个节点都位于环内。要显示精确环路，可增加三色DFS或强连通分量分析。

零时长任务合法，不应与缺失时长混为一谈。负时长拒绝。所有任务ID约定为可排序的字符串，输入时长为整数，这也让同分顺序与测试保持确定。实现的堆操作成本 O(V log V)，遍历边 O(E)，空间 O(V+E)。

从练习走向真实执行器

如果只有两个工作槽，两个已就绪任务不一定都能立刻启动，关键路径仍是下界，却不再总能达到。还需模拟资源、任务优先级、取消、失败传播和运行时观测。把这些能力直接塞入拓扑函数会混淆算法与执行状态，应该保留当前纯函数作为可测试的规划核心。

完成条件

先画出样例依赖图并手算7，再运行示例核对。测试空图、多条独立分支、重复依赖、自环、长环、未知依赖和零时长。对每条边检查 finish[parent] <= start[child]；检查所有任务恰好出现在顺序中一次。

进阶练习返回一条关键路径，并用该路径时长验证总时间。再增加两个工作槽，明确所用启发式是否保证最优，用小规模枚举验证它的局限。面试自查：拓扑排序为何能和DP合并，环内节点与受阻节点如何区分，关键路径下界为何不能直接等同于有限资源的实际工期。

本地运行

示例仅使用 Python 标准库，适用于 Python 3.10 及以上。把 example.py 与 test_example.py 保存到同一个目录，在该目录运行：

python example.py
python -m unittest -v test_example.py

测试检查本文列出的边界；新增练习仍需自己补充断言。

参考资料