−目次

最大流問題

最大流問題

問題設定

水道管のような、リンクを通じて何かが流れるネットワークを想像する

有向グラフ $G$ がある
ノード $s$ から $t$ に、水を流す
各リンクには容量 $cap$ があり、そのリンクを通じては $cap$ までの量しか流せない
$s, t$ 以外のノードでは流入量と流出量は一致する
最大流せる流量を求めよ

       ① -9→ ②
  10↗    ＼6    ＼8
  ／        ↘     ↘    ※リンク上の数値はcap
ｓ-4→ ③ -3→ ④ -4→ｔ

この場合、たとえば以下のように配分することで、12を流すことができる。

       ① -8→ ②
  10↗    ＼2    ＼8
  ／        ↘     ↘    ※リンク上の数値は流量
ｓ-2→ ③ -2→ ④ -4→ｔ

わかりやすい図とかは上記リンク先参照

解き方

「逆辺」という概念を利用する。逆辺とは、 $G$ の各辺を逆向きに繋ぐリンク。

答えとなる流量を $f$ とし、 $f=0$ で初期化する
$G$ に「逆辺」を追加する
各リンクに「残容量」を設定する
- オリジナルの辺の残容量の初期値は、そのリンクの容量 $cap$
- 逆辺の残容量の初期値は $0$
$s$ から $t$ まで流せる任意の経路を探す
- 残容量が0より大きい辺のみを辿る
- 経路が見つからなければ終了。その時点の $f$ が答え
- 見つかったら、その経路で流せる流量 $d$ を求める
  - $d=$ 経路中のリンク残容量の最小値
$G$ を更新する
- 経路の各リンクの残容量を $d$ ずつ減らす
- 経路の逆方向の各リンクの残容量を $d$ ずつ増やす
$f$ に $d$ を加え、4.に戻る

なぜ逆辺が必要なのか

グラフネットワーク〜フロー&カット〜 p.13～

いきなり逆辺なんて出てきて「何じゃこりゃ」と思うが、これを使うと貪欲法（見つかった経路に最大限流す、を繰り返す）が上手くいく。

逆辺の無い貪欲法で解くと、経路を発見する順番によっては最大を達成できなくなる場合がある。

逆辺の無いネットワーク
       ① -9→ ②        ｓ→ｔの経路をどれか1つ見つける際、
  10↗    ＼6    ＼8     ｓ-①-④-ｔの経路を最初に見つけてしまうと……
  ／        ↘     ↘    流量４を流した結果、④-ｔのリンクがそれ以上使えなくなる
ｓ-4→ ③ -3→ ④ -4→ｔ
                         （下図:ｓ-①-④-ｔに流量4を流した後の残容量）
       ① -9→ ②           
   6↗    ＼2    ＼8     最適解は、経路ｓ-①-②-ｔに多めに流すことで
  ／        ↘     ↘    経路ｓ-①-④-ｔに流す容量を少なくし、
ｓ-4→ ③ -3→ ④ -0→ｔ リンク④-ｔを経路ｓ-③-④-ｔのために開けておくのが良い

逆辺を張ったネットワーク（正方向残容量/逆方向残容量）
          9/0 
      ①ーー-→②        経路ｓ-①-④-ｔを最初に見つけて流した後の状態
 6/4↗  ＼2/4    ＼8/0   
  ／      ↘   0/4 ↘    同じく④-ｔのリンクは使えなくなっているが、
ｓー→③ー→④ーー-→ｔ  代わりにｓ③④①②ｔという経路が使えるようになっている
  4/0   3/0              リンク①-④の逆辺を利用し、流しすぎた容量を「押し戻す」感覚。

                         これにより、経路の発見順にかかわらず、全経路が網羅される

高速化

Ford-Fulkerson

$s$ から探索して $t$ までの経路を発見
$G$ を更新（経路の残容量を $d$ 減らし、逆を増やす）

という流れを繰り返す素朴な方法は、フォード・ファルカーソンのアルゴリズムという。

計算量は $O(Ef)$ 、ただしE:リンク数、f:最大フロー

Dinic法

1つの経路 $R$ が見つかって $G$ を更新した後、次の経路を求める際に、影響があるのは $R$ と辺を共有する経路だけである。他の経路は、途中までの探索結果を利用できるはずだ。

それをアルゴリズムとして確立したのが、Dinic法となる。

$s$ からBFSを行い、各頂点の $s$ からの距離 $level_v$ を記憶しておく（距離＝たどるリンク数）
$s$ からDFSを行い、 $t$ までの経路を1つ見つける
- BFSで求めた $level_v$ が増加する方向にのみ移動する
- 各ノードにつき、「探索した結果、 $t$ にたどり着けなかった辺」を覚えておく
- 経路が見つかったら、 $G$ を更新する
改めて $s$ からDFSを行う
- やり方は同様
- 前のDFSで、 $t$ にたどり着けないとわかった辺は使わない
- 見つかったら、 $G$ を更新し、3.に戻る
- 見つからなくなったら、探索した辺をリセットし、1.に戻る

※BFSもDFSも、残容量が残っているリンクのみを探索する。

BFSの時点で $t$ までの経路が見つからない状態になれば、それが最大流。

And more.

もっと高速化は進んでいるらしいのだが、理解が追いつかないため、無視。

事前処理に時間をかけることで、クエリの時間を短縮しているアルゴリズムもある。

実装

Dinic法 - tkw’s diaryを参考に。

        
          Dinic法 - Python3
          
              
              from collections import deque
 
 
class Dinic:
    def __init__(self, n):
        self.n = n
        self.links = [[] for _ in range(n)]
        self.depth = None
        self.progress = None
 
    def add_link(self, _from, to, cap):
        self.links[_from].append([cap, to, len(self.links[to])])
        self.links[to].append([0, _from, len(self.links[_from]) - 1])
 
    def bfs(self, s):
        depth = [-1] * self.n
        depth[s] = 0
        q = deque([s])
        while q:
            v = q.popleft()
            for cap, to, rev in self.links[v]:
                if cap > 0 and depth[to] < 0:
                    depth[to] = depth[v] + 1
                    q.append(to)
        self.depth = depth
 
    def dfs(self, v, t, flow):
        if v == t:
            return flow
        links_v = self.links[v]
        for i in range(self.progress[v], len(links_v)):
            self.progress[v] = i
            cap, to, rev = link = links_v[i]
            if cap == 0 or self.depth[v] >= self.depth[to]:
                continue
            d = self.dfs(to, t, min(flow, cap))
            if d == 0:
                continue
            link[0] -= d
            self.links[to][rev][0] += d
            return d
        return 0
 
    def max_flow(self, s, t):
        flow = 0
        while True:
            self.bfs(s)
            if self.depth[t] < 0:
                return flow
            self.progress = [0] * self.n
            current_flow = self.dfs(s, t, float('inf'))
            while current_flow > 0:
                flow += current_flow
                current_flow = self.dfs(s, t, float('inf'))
 
# 使い方
mf = Dinic(6)
mf.add_link(0, 1, 10)
mf.add_link(0, 3, 4)
mf.add_link(1, 2, 9)
mf.add_link(1, 4, 6)
mf.add_link(2, 5, 8)
mf.add_link(3, 4, 3)
mf.add_link(4, 5, 4)
print(mf.max_flow(0, 5))

            

        
      

最小カット問題

最大流問題の答えは、最小カット問題の答えと一致する。

問題設定

有向グラフ $G$ がある
各リンクには非負のコストが与えられている
ある2点 $s, t$ について、 $s→t$ まで行けなくなるように、リンクを取り除く
取り除くリンクのコストをできるだけ少なくしたい場合、いくつになる？

最大フロー最小カット定理

最小流量制約

使用シーン
- 最低これだけ流さないといけない、というのが各辺（または一部の辺）にあって、その上で最大流を求めたい
- 最大流が不要でも「このリンクに最低これだけ流した上で、 $s→t$ まで破綻せず流せる？」というのを調べたい

いま、

最低 $a$ 流さないといけなくて、容量が $b$ の辺 $(u,v)$
最低 $c$ 流さないといけなくて、容量が $d$ の辺 $(w,x)$

があったとすると、

ⓢ -..-> ⓤ --[a,b]--> ⓥ -..-> ⓦ --[c,d]--> ⓧ -..-> ⓣ

最小流量制約を処理するための便宜的な起終点 $S,T$ を用意して、以下のように辺を張りなおす。
（ここでの辺は、容量のみが決められた、通常の最大流の辺）

                        ,----a---- S ----c----,
                       ↓                     ↓
ⓢ -..-> ⓤ --(b-a)--> ⓥ -..-> ⓦ --(d-c)--> ⓧ -..-> ⓣ
         ｜                     ｜
          `---a----> T <----c---'

こうしてフローを流したとき、 $S$ から出る辺、 $T$ に入る辺の容量が最小流量制約を表現している。
最低流量分はとりあえず辺の上流側頂点から $T$ に分離させて、改めて下流側頂点に $S$ から供給する。
こうすることで、最低流量分を流したかどうかが、全体に紛れずに、単独で確認できるようになる。

つまり、 $S,T$ につながる辺の容量は全て使い切らないといけない。
使い切った上で、「 $s→t$ への最大流 + 各辺の最小流量制約」が、最大流となる。

「 $S$ から出る辺、 $T$ に入る辺」を説明上、「必須辺」と呼ぶことにする。

必須辺の容量を使い切れない場合、条件を満たすフローは存在しないことになる。

しかし、単純に流すだけでは、「上手いことやれば必須辺の容量が全て使えるが、探索順の関係で $t$ への辺などに優先的に流れてしまい、 $T$ への辺の容量を全て使えず、『破綻』と判定されてしまう」などということが発生しうる。

以下の順序で流すと、必須辺を優先的に使うことが保証される。

$S→T$
$S→t$
$s→T$
ここまでで、 $S$ から出る辺、 $T$ に入る辺に流れた量を調べて、可能不可能を判定
$s→t$ で最大流を得る

また、可能不可能の判定だけでよいなら、 $t→s$ に容量無限の辺を張った上で、 $S→T$ へ最大流を流し、必須辺の容量使い切りチェックすればフロー1回でいける。

Python3

        
          
          
              
              from collections import deque
from typing import Tuple, List, Optional
 
 
class DinicEdge:
    __slots__ = ['to', 'cap', 'is_orig', 'rev']
 
    def __init__(self, to: int, cap: int, is_orig: bool, rev: Optional['DinicEdge'] = None):
        self.to = to
        self.cap = cap
        self.is_orig = is_orig
        self.rev = rev
 
class DinicWithLowerBound:
    """
    最小流量制約付き最大流
 
    使い方:
      インスタンス生成
       mf = DinicWithLowerBound(n, S, T)
       ※ nは、最小流量を表現する超頂点S,Tを含めた頂点数。 S,T の頂点番号を与える
 
       辺追加
       基本的にS,Tに関係する辺はクラスが自動的に追加するため、元のグラフの頂点間の辺のみ与えれば良い
       mf.add_link(from, to, capacity)
       mf.add_bounded_link(from, to, lower_bound, capacity)
 
       流す
       mf.max_flow(s, t)
       または最小流量が実現可能かチェックする（高速）
       mf.exists_fulfill_flow(s, t)
 
       流した後、カット辺を列挙する
       mf.cut_edges(s)
    """
 
    def __init__(self, n: int, S: int, T: int):
        self.n = n
        self.S = S
        self.T = T
        self.links: List[List[DinicEdge]] = [[] for _ in range(n)]
        self.reserved = 0
 
    def add_link(self, from_: int, to: int, capacity: int) -> None:
        fwd = DinicEdge(to, capacity, True)
        bwd = DinicEdge(from_, 0, False)
        fwd.rev = bwd
        bwd.rev = fwd
        self.links[from_].append(fwd)
        self.links[to].append(bwd)
 
    def add_bounded_link(self, from_: int, to: int, lower_bound: int, capacity: int) -> None:
        assert capacity >= lower_bound
        if lower_bound > 0:
            self.add_link(self.S, to, lower_bound)
            self.add_link(from_, self.T, lower_bound)
            self.reserved += lower_bound
        if capacity > lower_bound:
            self.add_link(from_, to, capacity - lower_bound)
 
    def bfs(self, s: int, t: int) -> bool:
        links = self.links
        INF = 1 << 60
        self.level = level = [INF] * self.n
        level[s] = 0
        q = deque([s])
        goal = False
        while q:
            u = q.popleft()
            nd = level[u] + 1
            for e in links[u]:
                if e.cap > 0 and level[e.to] == INF:
                    level[e.to] = nd
                    q.append(e.to)
                    if e.to == t:
                        goal = True
                        break
            if goal:
                break
 
        return level[t] != INF
 
    def dfs(self, s: int, t: int) -> int:
        links = self.links
        level = self.level
        progress = self.progress
        link_counts = self.link_counts
        # 逆から進めた方が速くなることが多いらしい
        stack = [t]
 
        while stack:
            u = stack[-1]
            if u == s:
                break
            for i in range(progress[u], link_counts[u]):
                progress[u] = i
                e = links[u][i]
                if level[u] <= level[e.to] or progress[e.to] >= link_counts[e.to]:
                    continue
                rev = e.rev
                if rev.cap == 0:
                    continue
                stack.append(e.to)
                break
            else:
                progress[u] += 1
                stack.pop()
        else:
            return 0
 
        f = 1 << 60
        fwd_links = []
        bwd_links = []
        stack.pop()
        for u in stack:
            rev = links[u][progress[u]]
            link = rev.rev
            f = min(f, link.cap)
            fwd_links.append(link)
            bwd_links.append(rev)
 
        for link in fwd_links:
            link.cap -= f
 
        for link in bwd_links:
            link.cap += f
 
        return f
 
    def _max_flow(self, s: int, t: int) -> int:
        flow = 0
        while self.bfs(s, t):
            self.progress = [0] * self.n
            current_flow = 1  # Anything > 0 is ok
            while current_flow > 0:
                current_flow = self.dfs(s, t)
                flow += current_flow
        return flow
 
    def max_flow(self, s: int, t: int) -> int:
        self.link_counts = list(map(len, self.links))
        self._max_flow(self.S, self.T)
        self._max_flow(self.S, t)
        self._max_flow(s, self.T)
        if not self.check_lower_bound():
            return -1
        result = self._max_flow(s, t)
        return result + self.reserved
 
    def exists_fulfill_flow(self, s: int, t: int) -> bool:
        self.add_link(t, s, 1 << 60)
        self.link_counts = list(map(len, self.links))
        self._max_flow(self.S, self.T)
        return self.check_lower_bound()
 
    def check_lower_bound(self) -> bool:
        for e in self.links[self.S]:
            if e.is_orig and e.cap > 0:
                return False
        for e in self.links[self.T]:
            if e.is_orig:
                continue
            if e.rev.cap > 0:
                return False
        return True
 
    def cut_edges(self, s: int) -> List[Tuple[int, int]]:
        """ max_flowしたあと、最小カットにおいてカットすべき辺を復元する """
        q = [s]
        reachable = [0] * self.n
        reachable[s] = 1
        while q:
            v = q.pop()
            for e in self.links[v]:
                if e.cap == 0 or reachable[e.to]:
                    continue
                reachable[e.to] = 1
                q.append(e.to)
        edges = []
        for v in range(self.n):
            if reachable[v] == 0:
                continue
            for e in self.links[v]:
                if e.is_orig and reachable[e.to] == 0:
                    edges.append((v, e.to))
        return edges

            

        
      