トポロジカルソートと循環の検出
入次数を更新し、前提を満たした処理だけを順に取り出す方法を学びます。
説明
処理Bを始める前にAを終える必要がある関係をA→Bで表します。全ての辺で前提が先に来る順序をトポロジカル順序と呼びます。循環した依存があると、その循環内の処理を最初に開始できません。
fetchの後にcompileとlintを実行し、両方が終わってからpackageを行う例を使います。未完了の前提数である入次数を減らし、0になった処理を待ち行列へ追加します。最後に依存を一本追加して循環を作り、止まった状態を調べます。
1処理と依存を定義
各キーのリストは、その処理が終わると前提が一つ満たされる後続処理です。
tasks = {'fetch': ['compile', 'lint'], 'compile': ['package'], 'lint': ['package'], 'package': []}前提から後続への辺; incoming = {v: sum(v in nexts for nexts in tasks.values()) for v in tasks}入次数を集計; incoming必要な前提数fetchは0、compileとlintは1、packageは2です。packageには二つの前提があるため、一方が終わっただけでは開始できません。
2依存先のない処理を取り出す
最初に実行できるのはfetchだけです。順序を記録し、後続のcompileとlintの入次数を一つ減らします。
from collections import deque待ち行列; ready = deque(v for v in tasks if incoming[v] == 0)前提0を登録; order = []; current = ready.popleft(); order.append(current)fetchを処理して記録; incoming['compile'] -= 1; incoming['lint'] -= 1後続の前提を一つ減らす; current, incoming処理済みと残り前提数compileとlintがともに0になりました。前提のfetchが終わったため、どちらから進めても依存条件を満たします。
3片方の完了ではpackageを始めない
compileとlintをこの順で待機させ、compileを処理します。packageの未完了前提はまだ1個あります。
ready.extend(['compile', 'lint'])可能になった処理を登録; current = ready.popleft(); order.append(current)compileを処理; incoming['package'] -= 1packageの前提を減らす; order, list(ready), incoming['package']順序・待機・残り前提orderはfetch・compile、待機はlint、packageの前提数は1です。ここでpackageを追加するとlintの完了を待たない順序になってしまいます。
4全前提が終わったら追加
lintを処理するとpackageの前提が0になります。ここで初めて待ち行列へ追加します。
current = ready.popleft(); order.append(current)lintを処理; incoming['package'] -= 1最後の前提を完了; ready.append('package')0になったpackageを登録; order, list(ready), incoming更新した状態compileとlintが完了したためpackageを実行できます。入次数は「何本の辺があったか」から「まだ満たされていない前提が何個あるか」へ更新されています。
5全辺で前後関係を検査
packageを取り出して順序を完成させます。各辺の出発点が到着点より先かを照合します。
order.append(ready.popleft())packageを完了; positions = {v: i for i, v in enumerate(order)}処理位置の表; order完成した順序, all(positions[start] < positions[end] for start, ends in tasks.items() for end in ends)全依存の順序を検査fetch・compile・lint・packageで全辺の条件を満たします。compileとlintを逆にしても有効なので、トポロジカル順序は一つとは限りません。
6循環する辺を追加
packageが終わらないとcompileを始められない辺を追加します。compile→package→compileという循環になります。
tasks['package'].append('compile')逆向きの依存を追加; degree = {v: sum(v in ends for ends in tasks.values()) for v in tasks}変更後の入次数; degree循環を含む初期値compileの入次数は2です。fetchを終えてもpackage待ちが残り、packageもcompile待ちなので、その二つを完了できません。
7可能な処理だけ進める
fetchとlintは依存を満たして処理できます。その二つが終わった後の未完了前提数を計算します。
degree['compile'] -= 1; degree['lint'] -= 1fetch完了の反映; degree['package'] -= 1lint完了の反映; done = ['fetch', 'lint']; remaining = [v for v in tasks if v not in done]未処理を抽出; [(v, degree[v]) for v in remaining]残った前提数compileとpackageがともに1で残りました。待機できる入次数0の頂点がないのに未処理があるので、全体のトポロジカル順序を作れません。
8循環と無関係の後続も残り得る
packageの後にpublishを追加すると、publish自身は循環を構成しませんが、前提packageが終わらないため残ります。
tasks['package'].append('publish'); tasks['publish'] = []循環の後続を追加; blocked = remaining + ['publish']開始できない処理; [(v, tasks[v]) for v in blocked]残る接続を表示未処理集合には循環上の頂点だけでなく、その後続も含まれます。「残った全頂点が循環そのもの」とは判断できません。原因の循環を特定するには、残った辺をさらに調べます。
まとめ
入次数0の処理を完了し、後続の未完了前提数を減らすと依存を守る順序を作れます。
待機列が空でも未処理が残る場合、循環のため全体の順序を作れません。
未処理集合には循環の後続も含まれるため、循環そのものと停止の影響範囲を分けて調べます。
操作履歴と解説を振り返れます。
Python 3
tasks = {'fetch': ['compile', 'lint'], 'compile': ['package'], 'lint': ['package'], 'package': []}; incoming = {v: sum(v in nexts for nexts in tasks.values()) for v in tasks}; incoming{'fetch': 0, 'compile': 1, 'lint': 1, 'package': 2}from collections import deque; ready = deque(v for v in tasks if incoming[v] == 0); order = []; current = ready.popleft(); order.append(current); incoming['compile'] -= 1; incoming['lint'] -= 1; current, incoming('fetch', {'fetch': 0, 'compile': 0, 'lint': 0, 'package': 2})ready.extend(['compile', 'lint']); current = ready.popleft(); order.append(current); incoming['package'] -= 1; order, list(ready), incoming['package'](['fetch', 'compile'], ['lint'], 1)
current = ready.popleft(); order.append(current); incoming['package'] -= 1; ready.append('package'); order, list(ready), incoming(['fetch', 'compile', 'lint'], ['package'], {'fetch': 0, 'compile': 0, 'lint': 0, 'package': 0})order.append(ready.popleft()); positions = {v: i for i, v in enumerate(order)}; order, all(positions[start] < positions[end] for start, ends in tasks.items() for end in ends)(['fetch', 'compile', 'lint', 'package'], True)
tasks['package'].append('compile'); degree = {v: sum(v in ends for ends in tasks.values()) for v in tasks}; degree{'fetch': 0, 'compile': 2, 'lint': 1, 'package': 2}degree['compile'] -= 1; degree['lint'] -= 1; degree['package'] -= 1; done = ['fetch', 'lint']; remaining = [v for v in tasks if v not in done]; [(v, degree[v]) for v in remaining][('compile', 1), ('package', 1)]tasks['package'].append('publish'); tasks['publish'] = []; blocked = remaining + ['publish']; [(v, tasks[v]) for v in blocked][('compile', ['package']), ('package', ['compile', 'publish']), ('publish', [])]演習完了