· 開発

トポロジカルソートと循環検出：入次数で順序を決め、詰まった箇所を見つける

依存関係を実行順序に変えるトポロジカルソートを入次数の方法で実装し、循環があるときに何が残るのか、Python graphlibのTopologicalSorterとCycleErrorの挙動を整理します。

algorithmsgraphtopological-sortpython

ビルド手順、パッケージのインストール、データベースのマイグレーションのように「Aが終わらないとBを始められない」という制約がいくつも絡むと、すべての制約を守る実行順序を一つ決める必要があります。これがトポロジカルソート（topological sort）です。制約が循環していればそのような順序は存在しないので、実務ではトポロジカルソートと循環検出をセットで扱うのが現実的です。

この記事では、入次数（in-degree）を数える方法でトポロジカルソートを実装し、循環があるときにアルゴリズムがどこで止まるかを確認したうえで、Python標準ライブラリのgraphlib.TopologicalSorterが同じ問題をどのようなAPIで扱うかを見ていきます。Pythonの挙動は3.13、3.14、3.15の公式ドキュメントに基づいています。

トポロジカル順序の定義

有向グラフで、すべての辺 u → v について u が v より前に来るように頂点を一列に並べたものをトポロジカル順序と呼びます。完全なトポロジカル順序が存在するのは、グラフに有向閉路がない場合、つまりDAG（directed acyclic graph）の場合に限られます。

先に二つの点を押さえておきます。

  • 答えは一つとは限りません。互いに依存しない二つのタスクは、どちらが先でもかまいません。
  • 辺の向きの意味を先に決めておく必要があります。この記事では u → v を「u は v より先に来なければならない」という意味で使います。

入次数で実装する

入次数が0の頂点、つまり待つべき先行タスクがない頂点は、すぐに出力できます。頂点を一つ出力してそこから出る辺を取り除くと、新たに入次数が0になる頂点が現れます。出力できる頂点がなくなるまでこれを繰り返します。

from collections import deque

def topo_sort(nodes, edges):
    # edges の (u, v) は「u が v より先」を意味する
    indegree = {n: 0 for n in nodes}
    succ = {n: [] for n in nodes}
    for u, v in edges:
        succ[u].append(v)
        indegree[v] += 1

    queue = deque(n for n in nodes if indegree[n] == 0)
    order = []
    while queue:
        u = queue.popleft()
        order.append(u)
        for v in succ[u]:
            indegree[v] -= 1
            if indegree[v] == 0:
                queue.append(v)
    return order  # len(order) < len(nodes) なら循環がある

各頂点がキューに入るのは高々一回で、各辺について入次数を一回ずつ減らすので、計算量はO(V + E)です。入次数の表と隣接リストを合わせた追加領域もO(V + E)です。

循環があるとどこで止まるか

循環上の頂点は互いを待っているので、どれも入次数が0になりません。そのためキューが先に空になり、orderの長さが頂点数より短くなります。上のコードはこの差で循環の有無を判定できます。逆に循環がなければ、残った頂点の中に入次数0の頂点が必ず一つはあるので、すべての頂点が出力されます。

ただし、残った頂点がすべて循環に属しているわけではありません。aとbが互いを待っていて、b → c という辺があれば、cも残ります。しかしcは循環の一部ではなく、循環の後ろで止まっているだけです。ユーザーに「これらのノードが循環している」と伝えたいなら、残った頂点の中から実際の循環を別に探す必要があります。

やり方は単純です。残った頂点はどれも、残った先行頂点を少なくとも一つ持っています（入次数が0より大きいからです）。したがって、任意の残った頂点から出発して残った先行頂点をたどり続けると、頂点数は有限なので、いずれ一度通った頂点に戻ってきます。その間の区間が循環です。

def find_cycle(nodes, edges, order):
    done = set(order)
    preds = {n: [] for n in nodes}
    for u, v in edges:
        if u not in done and v not in done:
            preds[v].append(u)

    n = next(x for x in nodes if x not in done)
    seen = {}
    path = []
    while n not in seen:
        seen[n] = len(path)
        path.append(n)
        n = preds[n][0]
    # path は先行方向にさかのぼった順なので、反転して返す
    return (path[seen[n]:] + [n])[::-1]

戻り値は、最初と最後の要素が同じで、各要素が次の要素の直接の先行頂点になっているリストです。後で見るgraphlibの循環報告と同じ規則にそろえています。

Pythonのgraphlib.TopologicalSorter

Python 3.9から標準ライブラリにgraphlibモジュールがあり、トポロジカルソート用のTopologicalSorterクラスを提供しています。ノードはハッシュ可能でなければなりません。

最初につまずきやすいのは、グラフを渡す向きです。コンストラクタに渡す辞書は、キーがノード、値がそのノードの先行ノードです。よく使う「ノード → 後続ノード」の隣接リストとは逆なので、そのまま渡すと順序が反転します。

from graphlib import TopologicalSorter

# キー: タスク、値: そのタスクより先に終わっている必要があるタスク
deps = {
    "test": {"build"},
    "build": {"codegen", "fetch"},
    "codegen": {"fetch"},
}
print(list(TopologicalSorter(deps).static_order()))
# ['fetch', 'codegen', 'build', 'test']

このグラフはトポロジカル順序が一通りしかないので、出力は決まっています。互いに独立したノードがあると話が変わります。公式ドキュメントは、返される順序がグラフに要素を追加した順序に依存しうると明記しており、add()の呼び出し順だけを変えて結果が変わる例を示しています。テストでは特定の順序と比較するより、「すべての辺 u → v で u が前にあるか」を検査するほうが安全です。

並列実行向けのAPI

static_order()は順序を一度に受け取るときに使い、タスクを並列に動かすときはprepare()、get_ready()、done()を直接使います。

ts = TopologicalSorter(deps)
ts.prepare()
while ts.is_active():
    for task in ts.get_ready():
        run(task)      # 実際にはワークキューに入れる
        ts.done(task)  # done() を呼ぶまで後続タスクは解放されない

get_ready()は、今実行できるノードをすべて含むタプルを返します。最初は先行ノードを持たないノードすべてで、その後はdone()によって先行ノードがすべて処理済みになったノードが新たに返されます。同じ呼び出しで返されたノード同士は互いを待つ必要がないので、同時に実行できます。公式ドキュメントの例も、get_ready()の結果をタスクキューに入れ、ワーカーが終えたノードを受け取ってdone()を呼ぶ構成です。上のコードは流れを見せるために、その場でdone()を呼んでいるだけです。

is_active()は、まだ進められる場合にTrueを返します。ドキュメントの定義では、循環に妨げられておらず、かつget_ready()でまだ返されていない準備済みノードがあるか、done()済みのノード数がget_ready()で返されたノード数より少ない場合です。

CycleError

グラフに循環があると、prepare()はValueErrorのサブクラスであるgraphlib.CycleErrorを送出します。static_order()で走査する場合も同様です。

from graphlib import TopologicalSorter, CycleError

deps = {"a": {"b"}, "b": {"c"}, "c": {"a"}}
try:
    list(TopologicalSorter(deps).static_order())
except CycleError as e:
    print(e.args[1])

循環は、例外オブジェクトのargsの2番目の要素にノードのリストとして入っています。リスト内の各ノードは次のノードの直接の先行ノードで、循環であることがわかるよう最初と最後のノードが同じになっています。循環が複数ある場合は一つだけが報告され、どれが選ばれるかは定められていません。上の例でも、リストがどのノードから始まるかはドキュメントで保証されていないので、出力をテストに固定しないほうがよいでしょう。

prepare()がCycleErrorを送出した後もget_ready()は引き続き使え、循環に妨げられるまで処理できるノードをできるだけ多く取得できます。先に触れた「循環の後ろで止まっているノード」は、この方法でも最後まで出てきません。

バージョンによる違い

  • graphlibはPython 3.9で追加されました。
  • prepare()の後はグラフを変更できず、add()を呼ぶとValueErrorになります。
  • Python 3.14からは、ソートを開始する前であればprepare()を複数回呼べます。それ以前のバージョンでは2回目の呼び出しがValueErrorでした。static_order()やget_ready()でソートを開始した後にprepare()を呼ぶと、3.14以降でもValueErrorです。

実装するときに確認すること

  • 辺の向き：自前の実装では「u が v より先」、graphlibでは「キーの値＝先行ノード」です。二つの規則を混ぜると結果が逆になります。
  • 循環の判定：出力された頂点数が全体より少なければ循環があります。残った頂点の一覧をそのまま循環として報告すると、循環の後ろで止まっている頂点まで混ざります。
  • 出力順序：独立したノード間の順序は入力順に左右されうるので、テストでは特定の順序ではなく、制約を満たしているかを確認するべきです。

ブログへ