Chọn hàng đợi phù hợp với thứ tự xử lý

Lan quản lý một quầy mượn sách. Ba bạn An, Bình và Chi đến lần lượt, nhưng một yêu cầu khẩn cấp cũng có thể xuất hiện. Với từng công việc, Lan cần biết ai được phục vụ tiếp theo và có phải kiểm tra lại cả hàng hay không.

Quy tắc lấy ra quyết định cấu trúc cần dùng

Một hàng đợi (queue) thông thường theo quy tắc vào trước ra trước (first in, first out, FIFO). Phần tử mới vào cuối, còn phần tử cũ nhất ra ở đầu. Điều này phù hợp khi thứ tự đến quyết định thứ tự phục vụ.

Một hàng đợi hai đầu (double-ended queue, deque) cho phép thêm và lấy ở cả hai đầu. Còn hàng đợi ưu tiên (priority queue) chọn theo độ ưu tiên, không nhất thiết theo thời gian đến. Hai yêu cầu cùng ưu tiên cần một quy tắc phụ nếu thứ tự của chúng quan trọng.

Ví dụ dưới đây nhận tên theo thứ tự đến và phục vụ hai bạn. Bạn có thể chạy để xem thứ tự còn lại.

from collections import deque
q = deque(["An", "Bình", "Chi"])
q.append("Dũng")
print(q.popleft())
print(q.popleft())
print(list(q))

An
Bình
['Chi', 'Dũng']

append thêm vào cuối; popleft lấy và xóa ở đầu. Hai bạn còn lại vẫn theo thứ tự đến. Gọi popleft() khi q rỗng gây IndexError, vì không có phần tử để lấy.

Phân biệt ba việc: thêm, xem và lấy. Thêm làm hàng dài hơn một; xem đầu giữ nguyên hàng; lấy đầu làm hàng ngắn đi một. Nếu hai bạn cùng tên, hàng vẫn có hai phần tử và cần hai lần lấy để xử lý cả hai.

Trong một chương trình mô phỏng, bạn nên quyết định trước điều gì xảy ra khi hàng rỗng. Đề có thể yêu cầu in một thông báo hoặc bỏ qua thao tác lấy. Không được tự in thông báo bổ sung nếu đề chỉ yêu cầu kết quả của thao tác xem.

deque thao tác ở hai đầu mà không dồn cả dãy

deque hỗ trợ append(x) và appendleft(x) để thêm; pop() và popleft() để lấy. Chỉ số q[0] và q[-1] xem hai đầu mà không xóa. Hãy kiểm tra deque khác rỗng trước khi xem hoặc lấy một đầu.

Chương trình sau cho một xe vào hai đầu đường một chiều, sau đó cho xe rời đi. Các lệnh in giúp phân biệt thao tác xem và thao tác xóa.

from collections import deque
lane = deque([4, 8])
lane.appendleft(2)
lane.append(10)
print(lane[0], lane[-1])
print(lane.pop())
print(lane.popleft())
print(list(lane))

2 10
10
2
[4, 8]

Sau hai lệnh lấy, xe đầu và xe cuối tương ứng đã rời đi. Đọc lane[0] không làm xe rời đi. Nếu chỉ cần thêm và lấy cuối, list cũng phù hợp như ngăn xếp ở Bài 18.

Với list, pop(0) phải dời các phần tử phía sau để lấp chỗ trống. Lấy hết n phần tử từ đầu một danh sách có thể mất tổng cộng O(n²). Với deque, thêm hoặc lấy tại hai đầu có chi phí xấp xỉ O(1) mỗi thao tác, nên n lần mất O(n).

Bảng sau đo trên máy chạy hiện tại. Hai cách xử lý cùng số phần tử; thời gian thực tế thay đổi theo máy và tải, nhưng phép đo cho thấy sự khác biệt khi kích thước tăng.

Đừng suy ra một tỉ lệ thời gian cố định từ một lần đo. Điểm cần nhớ là deque không cần dồn lại toàn bộ các phần tử ở mỗi lần lấy đầu. Truy cập ngẫu nhiên giữa deque không phải mục đích chính của nó; dùng list nếu cần truy cập chỉ số bất kỳ thường xuyên.

Hàng FIFO giúp duyệt theo từng lớp

Xét mạng đường đi không hướng: A nối B và C, B nối D, C nối E. Từ A, các đỉnh B và C cách một cạnh; D và E cách hai cạnh. Duyệt theo chiều rộng (breadth-first search, BFS) xử lý các đỉnh theo từng lớp bằng hàng FIFO.

Mỗi đỉnh được đánh dấu ngay khi đưa vào hàng. Nhờ vậy hai người hàng xóm không thể đưa cùng một đỉnh vào hàng hai lần. Ví dụ tự khai báo cả graph và start trước khi duyệt.

from collections import deque
graph = {"A": ["B", "C"], "B": ["A", "D"],
         "C": ["A", "E"], "D": ["B"], "E": ["C"]}
start = "A"
distance = {start: 0}
q = deque([start])
while q:
    u = q.popleft()
    for v in graph[u]:
        if v not in distance:
            distance[v] = distance[u] + 1
            q.append(v)
print(distance)

{'A': 0, 'B': 1, 'C': 1, 'D': 2, 'E': 2}

Khi lấy một đỉnh cách d cạnh, mọi đỉnh được phát hiện từ nó có đường đi dài d + 1. FIFO xử lý hết lớp d trước lớp d + 1, nên lần đầu phát hiện cho số cạnh ít nhất. Kết luận này áp dụng khi mỗi cạnh có cùng chi phí; bài có trọng số cần phương pháp khác.

Với V đỉnh và E cạnh, danh sách kề và đánh dấu khi thêm giúp duyệt O(V + E) và dùng O(V + E) bộ nhớ kể cả đồ thị. Chẳng hạn 10.000 đỉnh và 30.000 cạnh chỉ cần xét cỡ 70.000 lượt kề trong đồ thị vô hướng. Không cần học thêm thuật toán đồ thị để hiểu hợp đồng FIFO của ví dụ này.

Nếu một đỉnh không có đường đi từ A, ví dụ này không đưa đỉnh ấy vào distance. Không nên suy ra nó có khoảng cách bằng không. Một chương trình đọc đồ thị lớn có thể khởi tạo khoảng cách chưa biết bằng -1, rồi đặt lại khi phát hiện từng đỉnh.

Mạng có chu trình vẫn duyệt được nhờ kiểm tra v not in distance. Trong đồ thị vô hướng, mỗi cạnh xuất hiện ở danh sách kề của hai đầu. Dù vậy, mỗi đỉnh vẫn chỉ vào hàng một lần vì đã được đánh dấu từ lúc thêm.

Deque đơn điệu giữ các ứng viên lớn nhất của cửa sổ

Cho dãy nhiệt độ [4, 2, 2, 5, 1, 3] và cửa sổ dài 3. Cần tìm nhiệt độ lớn nhất trong từng đoạn liên tiếp đủ ba ngày. Quét lại cả cửa sổ tốn O(nk), tức gần 500 triệu lượt so sánh khi n = 50000, k = 10000.

Ta dùng deque đơn điệu (monotonic deque) chứa chỉ số của những ứng viên còn hữu ích. Chỉ số tăng từ đầu đến cuối, còn giá trị tương ứng giảm nghiêm ngặt. Đầu deque là ứng viên lớn nhất đang còn trong cửa sổ.

Ở ngày thứ ba (chỉ số 2), giá trị 2 mới loại ứng viên 2 của ngày thứ hai. Giá trị mới bằng nó nhưng hết hạn muộn hơn, nên ứng viên cũ không thể tốt hơn trong cửa sổ tương lai. Khi 5 đến, mọi ứng viên nhỏ hơn ở cuối đều bị loại vì cùng lý do.

Tại chỉ số i, cửa sổ có biên trái i - k + 1. Mọi chỉ số <= i - k đã hết hạn và phải bỏ khỏi đầu. Sau đó bỏ từ cuối mọi chỉ số có giá trị <= a[i], rồi thêm i.

from collections import deque
a = [4, 2, 2, 5, 1, 3]
k = 3
q = deque()
answers = []
for i, value in enumerate(a):
    while q and q[0] <= i - k:
        q.popleft()
    while q and a[q[-1]] <= value:
        q.pop()
    q.append(i)
    if i >= k - 1:
        answers.append(a[q[0]])
print(answers)

[4, 5, 5, 5]

Sau khi xóa các chỉ số hết hạn, mọi ứng viên còn lại đều thuộc cửa sổ. Một ứng viên bị loại ở cuối có giá trị không lớn hơn phần tử mới và rời cửa sổ sớm hơn. Do đó nó không thể là lựa chọn cần thiết về sau; đầu deque cho giá trị lớn nhất.

Mỗi chỉ số được thêm một lần và bị loại tối đa một lần ở một trong hai đầu. Tổng số thao tác deque là O(n), dù có vòng while lồng trong vòng for. Bộ nhớ là O(k) cho deque và O(n) nếu lưu toàn bộ câu trả lời.

Bảng sau sinh trực tiếp từ cùng dãy, nhưng hiển thị ứng viên dưới dạng cặp chỉ số và giá trị. Dấu gạch ngang trong cột kết quả nghĩa là cửa sổ chưa đủ phần tử. Hãy quan sát riêng bước chỉ số 3: ứng viên 0 hết hạn và giá trị 5 loại các ứng viên còn nhỏ hơn.

Một cách dùng bộ nhớ ít hơn là xử lý từng đáp án khi cửa sổ đủ thay vì giữ danh sách answers. Tuy nhiên, nếu cần in trên một dòng, hãy tuân theo định dạng của đề. Số lượng kết quả chính xác là n-k+1 khi 1 <= k <= n.

Để tìm giá trị nhỏ nhất, giữ các giá trị tăng dần. Khi thêm giá trị mới, loại ở cuối các ứng viên có giá trị lớn hơn hoặc bằng nó. Quy tắc hết hạn và cách tính số cửa sổ vẫn giữ nguyên.

Trường hợp k = 1 trả lại chính dãy ban đầu. Với k = n, chỉ có một cửa sổ. Các số âm hoặc bằng nhau không đổi lập luận vì chỉ cần so sánh giá trị và hạn của chỉ số.

Nếu chỉ cần giá trị lớn nhất, có thể thay <= bằng < và giữ cả các giá trị bằng nhau. Khi đó các giá trị trong deque không còn giảm nghiêm ngặt, nhưng việc xóa chỉ số hết hạn vẫn giữ nguyên. Nếu đề hỏi vị trí đạt cực đại, cách xử lý bằng nhau quyết định chọn vị trí cũ hay mới.

Heap lấy khóa nhỏ nhất mà không sắp xếp lại toàn bộ

Một hệ thống có các công việc chờ, mỗi việc mang số phút xử lý dự kiến. Giả sử quy tắc phục vụ là việc có số phút nhỏ nhất được lấy trước. Sau khi lấy một việc, việc mới vẫn có thể đến.

Đống nhị phân (binary heap) duy trì quan hệ giữa mỗi nút cha và các nút con. heapq của Python dùng min-heap: phần tử nhỏ nhất nằm ở h[0]. Danh sách biểu diễn heap không nhất thiết được sắp xếp toàn bộ.

import heapq
jobs = [7, 3, 6]
heapq.heapify(jobs)
print(heapq.heappop(jobs))
heapq.heappush(jobs, 2)
print(heapq.heappop(jobs))
print(jobs[0])

3
2
6

heapify biến danh sách hiện có thành heap tại chỗ trong O(n). heappush và heappop mất O(log n) mỗi lần; đọc h[0] mất O(1) khi heap không rỗng. Lấy từ heap rỗng gây IndexError, nên kiểm tra if h trước.

Với 100.000 công việc và 100.000 lần lấy, xây heap một lần rồi lấy từng việc cần O(n + q log n). Sắp xếp lại toàn bộ sau từng lần thay đổi sẽ tốn công hơn nhiều. Heap không hỗ trợ tìm kiếm hoặc xóa một phần tử tùy ý trong O(log n) chỉ bằng các thao tác trên.

Python so sánh tuple từ trái sang phải. Đặt (độ_ưu_tiên, thứ_tự_đến, tên) giúp chọn số ưu tiên nhỏ hơn trước, rồi chọn người đến trước nếu bằng nhau. Bộ đếm thứ tự cũng tránh phải so sánh các đối tượng tên có kiểu không tương thích.

Chẳng hạn, (1, 4, "Mai") đứng trước (2, 0, "An") vì tiêu chí đầu tiên đã khác. Còn (1, 4, "Mai") đứng trước (1, 7, "Bình") vì tiêu chí đầu bằng nhau và 4 < 7. Đổi thứ tự hai trường đầu sẽ đổi cả quy tắc phục vụ.

Khi nhiều phần tử cùng giá trị, heapq vẫn giữ đủ số bản sao. Một lần heappop chỉ xóa một bản sao. Nếu cần hủy hoặc sửa ưu tiên của một việc bất kỳ, phải thiết kế thêm cách quản lý; không nên xóa trực tiếp giữa danh sách rồi tưởng heap vẫn hợp lệ.

import heapq
waiting = []
for order, (priority, name) in enumerate([(2, "An"), (1, "Bình"), (1, "Chi")]):
    heapq.heappush(waiting, (priority, order, name))
while waiting:
    print(heapq.heappop(waiting)[2])

Bình
Chi
An

Để lấy số lớn nhất trước, có thể đưa số đối của mỗi giá trị vào min-heap. Lấy xong, đổi dấu lần nữa để nhận giá trị gốc. Cách này cần điều chỉnh cẩn thận nếu khóa có nhiều tiêu chí.

import heapq
scores = [5, -2, 9, 9]
h = [-x for x in scores]
heapq.heapify(h)
print(-heapq.heappop(h))
print(-heapq.heappop(h))
print(-heapq.heappop(h))

9
9
5

Hai điểm bằng nhau vẫn là hai phần tử riêng. Giá trị âm trong dữ liệu gốc cũng được phục hồi đúng sau hai lần đổi dấu. Đây là hàng ưu tiên theo giá trị, không bảo đảm FIFO cho các phần tử bằng nhau nếu không thêm khóa phụ.

Kết hợp heap với thời gian đến để phân công công việc

Giả sử các công việc đã có mặt, được chọn theo thời gian xử lý ngắn nhất. Sau khi xử lý một việc, ta ghi thời điểm hoàn tất của nó. Tổng thời gian chờ hoàn tất có thể được tính bằng cách cộng các thời điểm đó.

Với ba thời lượng 6, 2, 4, chương trình dùng heap để lấy lần lượt thời lượng nhỏ nhất. Đây là một quy tắc phục vụ đã cho; nếu công việc có thời điểm đến khác nhau, cần thêm điều kiện chỉ đưa việc đã đến vào heap.

import heapq
durations = [6, 2, 4]
heapq.heapify(durations)
elapsed = 0
total = 0
while durations:
    elapsed += heapq.heappop(durations)
    total += elapsed
print(total)

20

Ở mỗi bước, elapsed là tổng thời lượng các việc đã lấy. total là tổng thời điểm hoàn tất đã ghi. Bất biến này đúng ban đầu với hai biến bằng không và được duy trì khi cộng thời lượng mới.

Xây heap mất O(n), còn n lần lấy mất O(n log n); bộ nhớ là O(n). Nếu dữ liệu cố định và chỉ cần thứ tự một lần, sắp xếp cũng là lựa chọn tốt. Heap đặc biệt hữu ích khi có thêm việc giữa các lần lấy.

Với thời lượng bằng nhau, kết quả tổng không thay đổi khi đổi chỗ hai việc. Nếu đề còn yêu cầu tên việc, cần nêu rõ cách phá hòa và đưa tiêu chí đó vào khóa. Nếu không có việc nào, cả elapsed và total vẫn bằng không và vòng lặp không chạy.

Lỗi thường gặp

Tóm tắt

• FIFO lấy phần tử đến trước; deque cho phép thao tác hiệu quả ở cả hai đầu.
• Deque đơn điệu lưu chỉ số còn hạn và không bị ứng viên mới lấn át; tổng thời gian O(n).
• heapq là min-heap: lấy khóa nhỏ nhất trong O(log n) và có thể dùng tuple để phá hòa.
• Chọn cấu trúc theo quy tắc lấy phần tử, giới hạn đầu vào và trường hợp rỗng.

Kiểm tra nhanh

Đọc thêm

• Tài liệu chính thức về collections.deque.
• Tài liệu chính thức về heapq.