Sắp xếp công việc theo thứ tự phụ thuộc với DAG

Lan chuẩn bị bốn công việc cho hội chợ. Bạn phải thiết kế bảng trước khi in bảng, và phải chuẩn bị nội dung trước khi in. Việc trang trí cần bảng đã in. Lan muốn biết có thể làm các việc theo thứ tự nào mà không bỏ sót điều kiện.

Cạnh có hướng ghi lại việc nào phải làm trước

Gọi bốn công việc là 1: thiết kế, 2: chuẩn bị nội dung, 3: in và 4: trang trí. Cạnh 1 -> 3 nghĩa là phải hoàn thành việc 1 trước việc 3. Cạnh 2 -> 3 có nghĩa tương tự. Cạnh 3 -> 4 nối việc in với trang trí.

Một thứ tự topo (topological order) là cách xếp tất cả đỉnh thành một hàng. Với mỗi cạnh u -> v, đỉnh u phải nằm trước v. Dãy 1, 2, 3, 4 hợp lệ. Dãy 2, 1, 3, 4 cũng hợp lệ vì không có điều kiện nào giữa hai việc đầu.

Hàm sau kiểm tra một thứ tự cụ thể. Mảng pos lưu vị trí của từng đỉnh, nên mỗi cạnh chỉ cần một phép so sánh.

edges = [(1, 3), (2, 3), (3, 4)]
for order in ([1, 2, 3, 4], [1, 3, 2, 4]):
    pos = {v: i for i, v in enumerate(order)}
    print(all(pos[u] < pos[v] for u, v in edges))

True
False

Phương án thứ hai đặt 3 trước 2 nên vi phạm cạnh 2 -> 3. Việc kiểm tra trên giả sử mỗi đỉnh xuất hiện đúng một lần. Nếu dữ liệu là một phương án bất kỳ, bạn còn phải kiểm tra nó là hoán vị của tất cả các đỉnh.

Chẳng hạn, dãy 1, 2, 3, 3, 4 đặt đúng chiều mọi cạnh nhưng lặp lại đỉnh 3. Nó không phải thứ tự topo vì chứa thừa một lần xuất hiện. Để kiểm tra đầy đủ, so độ dài với n và so tập các phần tử với tập đỉnh trước khi kiểm tra các cạnh.

Chu trình có hướng (directed cycle) là một đường đi theo chiều cạnh và trở lại đỉnh ban đầu. Nếu có 1 -> 2 -> 3 -> 1, mỗi đỉnh phải đứng trước đỉnh tiếp theo. Suy ra 1 phải đứng trước chính nó, điều không thể xảy ra.

Đồ thị có hướng không chu trình (directed acyclic graph), viết tắt DAG, là đồ thị có hướng không chứa chu trình như vậy. Một đồ thị có thứ tự topo khi và chỉ khi nó là DAG. Một đỉnh đứng riêng cũng thuộc đồ thị và vẫn phải xuất hiện trong thứ tự.

edges = [(1, 1)]
order = [1]
pos = {v: i for i, v in enumerate(order)}
print(all(pos[u] < pos[v] for u, v in edges))

False

Ở đây không có vị trí nào nhỏ hơn chính nó. Nếu đồ thị có nhiều thành phần tách rời, mỗi thành phần vẫn có thể đóng góp đỉnh vào thứ tự chung. Không được bắt đầu duyệt từ một đỉnh rồi bỏ qua phần còn lại.

Bậc vào cho biết còn bao nhiêu điều kiện chưa được gỡ

Bậc vào (indegree) của một đỉnh là số cạnh đi vào đỉnh đó. Trong đồ thị còn lại, bậc vào bằng 0 nghĩa là không có cạnh nào từ một đỉnh chưa xử lý đi vào nó. Vì vậy đỉnh này có thể được đưa tiếp vào thứ tự.

n = 4
edges = [(1, 3), (2, 3), (3, 4)]
indegree = [0] * (n + 1)
for u, v in edges:
    indegree[v] += 1
print(*indegree[1:])

0 0 2 1

Hai việc đầu có bậc vào 0. Việc 3 có bậc vào 2 vì cần cả hai việc đó. Việc 4 có bậc vào 1. Khi bỏ đỉnh 1, ta giảm bậc vào của 3 xuống 1; việc 3 chưa thể bắt đầu.

Hình sau mô tả một lượt xử lý theo hàng đợi đầu vào trước, đầu ra trước. Mỗi dòng ghi trạng thái sau khi lấy một đỉnh; mảng bậc vào theo thứ tự 1, 2, 3, 4.

Khi xử lý một cạnh u -> v, ta giảm indegree[v] đúng một lần. Chỉ khi bậc vào vừa về 0 mới đưa v vào hàng đợi. Nếu đầu vào cho phép cạnh trùng, mỗi bản sao đều được đếm và đều được giảm.

Ví dụ, nếu dữ liệu liệt kê 1 -> 3 hai lần, bậc vào của 3 tăng hai. Danh sách kề của 1 cũng chứa hai lần số 3. Khi xử lý 1, hai lượt giảm cân bằng với hai lượt tăng. Nếu chỉ loại cạnh trùng ở một trong hai nơi, bậc vào không còn phản ánh số cạnh chưa gỡ.

Thuật toán Kahn lần lượt lấy các đỉnh sẵn sàng

Thuật toán Kahn (Kahn's algorithm) giữ một hàng đợi các đỉnh có bậc vào 0. Ban đầu, ta tính bậc vào rồi đưa tất cả đỉnh sẵn sàng vào hàng. Sau đó ta lấy một đỉnh, ghi vào kết quả và gỡ các cạnh đi ra từ nó.

Chương trình dưới đọc n m, sau đó đọc m cạnh. Nó in thứ tự tìm được hoặc -1 khi không thể xếp đủ. Các đỉnh đánh số từ 1 đến n.

4 3
1 3
2 3
3 4

from collections import deque

n, m = map(int, input().split())
adj = [[] for _ in range(n + 1)]
indegree = [0] * (n + 1)
for _ in range(m):
    u, v = map(int, input().split())
    adj[u].append(v)
    indegree[v] += 1

queue = deque(v for v in range(1, n + 1) if indegree[v] == 0)
order = []
while queue:
    u = queue.popleft()
    order.append(u)
    for v in adj[u]:
        indegree[v] -= 1
        if indegree[v] == 0:
            queue.append(v)

if len(order) != n:
    print(-1)
else:
    print(*order)

1 2 3 4

Trong ví dụ này, 1 và 2 có thể đổi chỗ mà vẫn đúng. Kết quả phụ thuộc vào thứ tự ban đầu và thứ tự thêm các đỉnh kề. Hàng đợi thường không hứa đưa ra thứ tự nhỏ nhất.

Lý do đúng là một bất biến (invariant): bậc vào đang lưu đúng số cạnh từ các đỉnh chưa xử lý. Mỗi khi lấy u, ta gỡ đúng các cạnh đi ra từ u. Vì u có bậc vào 0, không còn điều kiện nào bắt buộc một đỉnh chưa xử lý đứng trước u.

Một DAG hữu hạn còn đỉnh thì có ít nhất một đỉnh bậc vào 0. Nếu không, đi ngược một cạnh từ mỗi đỉnh sẽ phải gặp lại một đỉnh, tạo chu trình. Bởi vậy Kahn tiếp tục lấy được đỉnh cho tới khi DAG hết đỉnh.

Khi các môn đã sẵn sàng có thể học cùng một học kỳ, ta xử lý Kahn theo từng lớp. Trước mỗi lượt, lưu số phần tử đang có trong hàng đợi. Các môn vừa được mở sẽ chờ đến lượt sau, tương ứng học kỳ sau.

from collections import deque
adj = [[], [3], [3], [4, 5], [], []]
deg = [0, 0, 0, 2, 1, 1]
q = deque([1, 2])
semesters = 0
done = 0
while q:
    semesters += 1
    for _ in range(len(q)):
        u = q.popleft()
        done += 1
        for v in adj[u]:
            deg[v] -= 1
            if deg[v] == 0:
                q.append(v)
print(semesters if done == 5 else -1)

3

Hai môn 1 và 2 học ở kỳ đầu, môn 3 ở kỳ hai, còn 4 và 5 ở kỳ ba. Mỗi môn chỉ được học sau mọi môn tiên quyết, nên không thể xếp sớm hơn. Cách học tất cả môn sẵn sàng đạt đúng cận đó. Các đỉnh và cạnh vẫn chỉ được duyệt một lần, nên chi phí là O(n + m).

Kahn cũng kiểm tra được thứ tự topo có duy nhất hay không. Nếu tại một lượt có hai đỉnh sẵn sàng, có thể chọn một trong hai trước và tiếp tục hoàn thành DAG. Hai lựa chọn tạo hai thứ tự khác nhau. Nếu mỗi lượt chỉ có một đỉnh sẵn sàng, mọi vị trí đều bị xác định, nên thứ tự duy nhất.

Điều kiện tương đương là mỗi cặp đỉnh liên tiếp trong thứ tự có một cạnh trực tiếp từ đỉnh trước tới đỉnh sau. Nếu thiếu cạnh đó, hai đỉnh liên tiếp có thể đổi chỗ mà vẫn giữ mọi điều kiện. Khi đồ thị có chu trình, không có thứ tự topo; phải kiểm tra điều này trước khi kết luận duy nhất.

Mỗi đỉnh vào và ra khỏi hàng nhiều nhất một lần. Mỗi cạnh được duyệt một lần. Thời gian là O(n + m) và bộ nhớ là O(n + m). Với n = 100000, m = 200000, phần duyệt gồm khoảng 300000 lượt đỉnh và cạnh, chưa tính chi phí đọc.

Số đỉnh đã lấy giúp phát hiện chu trình

Nếu hàng đợi rỗng trước khi lấy đủ n đỉnh, các đỉnh còn lại đều có ít nhất một cạnh đi vào từ phần còn lại. Đi ngược các cạnh ấy sẽ gặp lại một đỉnh. Vì vậy đồ thị chứa chu trình.

Ngược lại, một chu trình không bao giờ được gỡ hết. Mỗi đỉnh trên chu trình vẫn còn cạnh từ đỉnh trước nó. Do đó len(order) < n tương đương với việc có chu trình.

Đoạn sau có một đỉnh độc lập và một chu trình giữa hai đỉnh khác. Nó cho thấy thuật toán vẫn xử lý được phần độc lập trước khi dừng.

from collections import deque
n = 3
edges = [(1, 2), (2, 1)]
adj = [[] for _ in range(n + 1)]
deg = [0] * (n + 1)
for u, v in edges:
    adj[u].append(v)
    deg[v] += 1
q = deque(v for v in range(1, n + 1) if deg[v] == 0)
order = []
while q:
    u = q.popleft()
    order.append(u)
    for v in adj[u]:
        deg[v] -= 1
        if deg[v] == 0:
            q.append(v)
print(order)
print(len(order) == n)

[3]
False

Chỉ có 3 được lấy, nên kết quả kiểm tra là False. Một đỉnh còn lại sau Kahn không nhất thiết nằm trên chu trình. Chẳng hạn, một đỉnh ở sau chu trình cũng có thể bị chặn bởi cạnh từ chu trình đó.

Nếu không có cạnh nào, mọi đỉnh đều sẵn sàng từ đầu. Một đỉnh duy nhất không có cạnh là DAG. Một cạnh tự quay về tạo chu trình, dù đồ thị chỉ có một đỉnh. Các trường hợp này thường xuất hiện trong bộ kiểm.

Hàng đợi ưu tiên tạo thứ tự nhỏ nhất theo từ điển

Một đề có thể yêu cầu thứ tự nhỏ nhất theo từ điển (lexicographically smallest). So hai thứ tự từ trái sang phải; tại vị trí đầu tiên khác nhau, thứ tự có số nhỏ hơn là thứ tự nhỏ hơn.

Bạn thay hàng đợi thường bằng hàng đợi ưu tiên (priority queue) dạng min-heap. Mỗi lượt lấy đỉnh nhỏ nhất đang sẵn sàng. Một đỉnh mới sẵn sàng phải được đưa vào heap ngay trước lượt chọn kế tiếp.

import heapq
n = 4
edges = [(1, 2), (1, 3)]
adj = [[] for _ in range(n + 1)]
deg = [0] * (n + 1)
for u, v in edges:
    adj[u].append(v)
    deg[v] += 1
ready = [v for v in range(1, n + 1) if deg[v] == 0]
heapq.heapify(ready)
order = []
while ready:
    u = heapq.heappop(ready)
    order.append(u)
    for v in adj[u]:
        deg[v] -= 1
        if deg[v] == 0:
            heapq.heappush(ready, v)
print(*order)

1 2 3 4

Ban đầu heap chứa 1 và 4. Sau khi lấy 1, hai đỉnh 2 và 3 cũng sẵn sàng. Lượt kế tiếp chọn 2, không phải 4. Một hàng đợi thường đã có 4 ở trước 2 nên có thể cho thứ tự lớn hơn.

Tại mỗi vị trí, mọi thứ tự hợp lệ tiếp theo phải chọn một đỉnh đang sẵn sàng. Chọn đỉnh nhỏ nhất trong số đó tạo tiền tố nhỏ nhất. Sau khi bỏ đỉnh này, cùng lập luận áp dụng cho phần còn lại.

Mỗi đỉnh được đưa vào và lấy ra khỏi heap tối đa một lần. Thời gian là O(m + n log n), bộ nhớ là O(n + m). Với 100000 đỉnh, log₂ n nhỏ hơn 17; khoảng vài triệu thao tác heap là một ước lượng thô, không phải thời gian đo thực tế.

Thứ tự topo cho phép tính thời điểm hoàn thành sớm nhất

Giả sử mỗi việc có thời lượng và có thể chạy song song với việc khác. Một việc chỉ được bắt đầu khi mọi việc trước trực tiếp của nó hoàn thành. Ta cần thời điểm hoàn thành sớm nhất của từng việc.

Đặt finish[v] là thời điểm hoàn thành sớm nhất của việc v. Việc không có điều kiện trước hoàn thành tại duration[v]. Với cạnh u -> v, việc v không thể bắt đầu trước finish[u].

Do đó, với mọi đỉnh trước v, ta lấy thời điểm hoàn thành lớn nhất. Công thức là finish[v] = duration[v] + max(finish[u]) trên các cạnh u -> v. Nếu không có cạnh đi vào, phần lớn nhất được coi là 0.

Cho bốn việc có thời lượng lần lượt 3, 2, 4, 1. Các cạnh vẫn là 1 -> 3, 2 -> 3, 3 -> 4. Việc 3 chờ cả 1 và 2, nên bắt đầu ở thời điểm 3, không phải 2 hay 5.

from collections import deque
n = 4
duration = [0, 3, 2, 4, 1]
edges = [(1, 3), (2, 3), (3, 4)]
adj = [[] for _ in range(n + 1)]
deg = [0] * (n + 1)
for u, v in edges:
    adj[u].append(v)
    deg[v] += 1
q = deque(v for v in range(1, n + 1) if deg[v] == 0)
finish = duration[:]
order = []
while q:
    u = q.popleft()
    order.append(u)
    for v in adj[u]:
        finish[v] = max(finish[v], finish[u] + duration[v])
        deg[v] -= 1
        if deg[v] == 0:
            q.append(v)
if len(order) == n:
    print(*finish[1:])
else:
    print(-1)

3 2 7 8

Khi v được lấy ra, mọi đỉnh trước trực tiếp của nó đã được xử lý. Vì vậy mọi ứng viên đã được chuyển tới finish[v]. Áp dụng lập luận này theo thứ tự topo chứng minh từng kết quả đúng.

Đây là quy hoạch động (dynamic programming) trên DAG: tính trạng thái sau từ những trạng thái trước. Độ phức tạp vẫn là O(n + m), vì mỗi cạnh chỉ cập nhật một lần. Không dùng các giá trị chưa được xác nhận nếu đồ thị có chu trình.

Giả thiết được chạy song song rất quan trọng. Nếu chỉ có một người làm và không được làm đồng thời, tổng thời gian không còn được mô tả bởi công thức trên. Đề phải nói rõ điều kiện sử dụng tài nguyên.

Khi đề hỏi thời gian hoàn thành cả dự án, bạn lấy giá trị lớn nhất trong finish sau khi kiểm tra chu trình. Các thành phần không có đường nối vẫn được tính vì tất cả công việc phải hoàn thành. Một công việc độc lập có thời lượng lớn có thể quyết định thời gian chung.

Đường đi trên DAG dùng cùng thứ tự nhưng khác trạng thái

Cùng một thứ tự topo còn giúp đếm đường đi. Giả sử cần đếm số đường đi từ đỉnh 1, tính cả đường đi độ dài 0 từ 1 tới chính nó. Mỗi cạnh u -> v nối mọi đường tới u với v.

order = [1, 2, 3, 4]
adj = [[], [2, 3], [4], [4], []]
ways = [0] * 5
ways[1] = 1
for u in order:
    for v in adj[u]:
        ways[v] += ways[u]
print(*ways[1:])

1 1 1 2

Đỉnh 4 có hai đường: 1 -> 2 -> 4 và 1 -> 3 -> 4. Các đỉnh không đến được từ 1 giữ giá trị 0. Nếu đề yêu cầu chia dư, bạn lấy phần dư sau mỗi lần cộng.

Trạng thái này khác finish: ở đây các đóng góp được cộng, còn lịch sớm nhất lấy giá trị lớn nhất. Cả hai đều đúng vì khi xử lý một đỉnh, mọi cạnh đi vào nó đã được xét. Duyệt theo số hiệu đỉnh không đảm bảo điều đó.

Ví dụ sau có cạnh đi từ số lớn sang số nhỏ. Số hiệu của đỉnh không phải là thứ tự topo.

edges = [(3, 1), (1, 2)]
order = [3, 1, 2]
pos = {v: i for i, v in enumerate(order)}
print(all(pos[u] < pos[v] for u, v in edges))

True

Khi tính số đường đi, mỗi cạnh đóng góp một lần nên chi phí là O(n + m) sau khi tìm topo. Nếu có cạnh trùng và đề xem chúng là hai lựa chọn khác nhau, chúng tạo hai đóng góp. Nếu đề định nghĩa đường đi theo dãy đỉnh, bạn phải xử lý cạnh trùng theo quy ước đề.

Lỗi thường gặp

Tóm tắt

• Cạnh u -> v yêu cầu u đứng trước v; thứ tự topo tồn tại đúng khi đồ thị là DAG.
• Kahn lấy các đỉnh bậc vào 0, gỡ cạnh đi ra và phát hiện chu trình bằng số đỉnh đã lấy.
• Min-heap chọn thứ tự nhỏ nhất theo từ điển. Hàng đợi thường chạy trong O(n + m).
• Thứ tự topo giúp chuyển trạng thái quy hoạch động từ các việc trước sang việc sau.

Kiểm tra nhanh

Đọc thêm

• Python: graphlib và thứ tự topo.
• CP-Algorithms: Topological Sorting.