Приоритетна опашка

Код на:JavaC#PythonJavaScript

Накратко: Приоритетната опашка винаги дава най-важния елемент, например най-спешната поръчка; върху heap и добавянето, и вземането са O(log n), а в Python това е модулът heapq.

Проблем

Кой тръгва пръв

Диспечерът на доставките разпределя поръчки към куриерите. Всяка поръчка има краен срок: след колко минути трябва да е при клиента. Свободният куриер винаги получава най-спешната поръчка.

Поръчките идват непрекъснато. В петък вечер в голям град чакат по 10 000 наведнъж, а за деня минават 100 000.

Първото решение е списък, подреден по срок. При всяка нова поръчка тя се добавя и списъкът се подрежда наново. Подреждането е O(n log n) (виж merge sort), и то за всяка поръчка. Дори най-умното подреждане, което забелязва, че списъкът почти е подреден, трябва поне да мине през него: n стъпки при всяка нова поръчка.

А на куриера не му трябва целият ред. Трябва му само първата.

Идеята

Опашка, която знае само кой е пръв

Структура с две действия: добави поръчка и вземи най-спешната. Това е приоритетна опашка (priority queue). Не обещава ред за всички, а само кой е пръв.

Точно това прави heap от урока за heap sort. Тук е обърнат: min-heap, в който всеки родител е най-много колкото децата си. Тогава най-ранният срок е винаги на място 0.

• Вземането вече го знаеш от heap sort. Върхът излиза, последният елемент отива на негово място и потъва надолу. Разликата е, че потъва към по-спешното дете.
• Добавянето е новото: изплуване (sift up). Новата поръчка влиза в края на масива. После се сравнява с родителя си на място (i - 1) / 2. Ако е по-спешна, сменят се местата, и тя продължава нагоре. Спира, когато родителят е по-спешен или когато стигне върха.

Виж го

Изплуване и потъване

Добавянето: в края и нагоре.

def push(self, order):
    self._items.append(order)
    i = len(self._items) - 1
    while i > 0:
        parent = (i - 1) // 2
        if not self._less(i, parent):
            return
        self._swap(i, parent)
        i = parent

Вземането: върхът излиза, последният застава на негово място и потъва.

def poll(self):
    items = self._items
    top = items[0]
    items[0] = items[-1]
    items.pop()
    if items:
        self._sift_down(0)
    return top

def _sift_down(self, i):
    n = len(self._items)
    while True:
        low = i
        left, right = 2 * i + 1, 2 * i + 2
        if left < n and self._less(left, low):
            low = left
        if right < n and self._less(right, low):
            low = right
        if low == i:
            return
        self._swap(i, low)
        i = low

__len__ е методът, който вика len(queue). Така опашката се ползва като всеки списък.

Шест поръчки чакат. Влиза нова с 18 минути, а куриерът взима най-спешната:

waiting = [
    Order("A-101", 10),
    Order("A-102", 15),
    Order("A-103", 30),
    Order("A-104", 20),
    Order("A-105", 25),
    Order("A-106", 40),
]
queue = DeadlineQueue()
for o in waiting:
    queue.push(o)
print(f"Heap: {queue.deadlines()}")
queue.push(Order("A-107", 18))
print(f"След A-107: {queue.deadlines()}")
order = queue.poll()
print(f"Куриерът взима {order.id} ({order.deadline} мин)")
print(f"Остават: {queue.deadlines()}")

Изход

Heap: [10, 15, 30, 20, 25, 40]
След A-107: [10, 15, 18, 20, 25, 40, 30]
Куриерът взима A-101 (10 мин)
Остават: [15, 20, 18, 30, 25, 40]

class DeadlineQueue:
    def __init__(self):
        self._items = []

    def push(self, order):
        self._items.append(order)
        i = len(self._items) - 1
        while i > 0:
            parent = (i - 1) // 2
            if not self._less(i, parent):
                return
            self._swap(i, parent)
            i = parent

    def poll(self):
        items = self._items
        top = items[0]
        items[0] = items[-1]
        items.pop()
        if items:
            self._sift_down(0)
        return top

    def _sift_down(self, i):
        n = len(self._items)
        while True:
            low = i
            left, right = 2 * i + 1, 2 * i + 2
            if left < n and self._less(left, low):
                low = left
            if right < n and self._less(right, low):
                low = right
            if low == i:
                return
            self._swap(i, low)
            i = low

    def _less(self, i, j):
        items = self._items
        return items[i].deadline < items[j].deadline

    def _swap(self, i, j):
        items = self._items
        items[i], items[j] = items[j], items[i]

    def __len__(self):
        return len(self._items)

    def deadlines(self):
        return [o.deadline for o in self._items]

Стъпка по стъпка, с думи

1. Шест чакащи поръчки, сроковете са в минути. Min-heap: всеки родител е най-много колкото децата си, затова най-спешната, 10, е отгоре.
2. push: нова поръчка с 18 минути влиза в края на масива, място 6. В дървото е дете на 30.
3. Изплуване: родителят на 6 е (6 - 1) / 2 = 2. 18 е по-спешно от 30, разменят се.
4. Родителят на 2 е 0, с 10 минути. 10 е по-спешно от 18: поръчката спира. Една размяна, най-много по една на ниво.
5. poll: куриерът взима върха, поръчката с 10 минути. На място 0 остава дупка.
6. Последният, 30, отива на върха и масивът се свива до шест. Сега _sift_down го слага на мястото му.
7. Потъване, както в heap sort, но към по-спешното дете: 30 срещу 15 и 18 се разменя с 15, после срещу 20 и 25 се разменя с 20.
8. Място 3 няма деца: край. Отгоре е 15, следващата по спешност. И двете движения струват O(log n): височината на дървото.

Масивът не е подреден: 18 стои преди 20. Подредено е само това, което трябва: най-ранният срок е отгоре.

Пробвай

Предвиди изхода

Пет поръчки влизат в празна опашка. Какво има в масива след това?

queue = DeadlineQueue()
queue.push(Order("B-1", 40))
queue.push(Order("B-2", 25))
queue.push(Order("B-3", 30))
queue.push(Order("B-4", 10))
queue.push(Order("B-5", 35))
print(queue.deadlines())

Отговор

[10, 25, 30, 35, 40]

[40, 25, 30, 10, 35]

[10, 25, 30, 40, 35]

Провери

Пробвай

Попълни

Изплуването сравнява елемента на място i с родителя му. Къде е родителят?

parent = 

Провери

Идеята

Колко струва

• Добавяне: най-много едно сравнение на ниво. Дървото с n елемента има около log₂ n нива: O(log n).
• Вземане: две сравнения на ниво при потъването: също O(log n).
• Кой е пръв: върхът е на място 0, без нито едно сравнение: O(1).

Пусни диспечера с двата варианта. Поръчките идват една по една, след всяка втора куриер взима най-спешната, а накрая куриерите взимат останалите. Броят се сравненията между два срока:

• Подреден списък: подреждане след всяка поръчка O(n²)
• Min-heap: изплуване и потъване O(n log n)

Алгоритъм	n = 10	n = 1000	n = 10 000	n = 100 000
Подреден списък: подреждане след всяка поръчка	11	63 430	6 250 042	≈ 625 004 200
Min-heap: изплуване и потъване	28	17 397	256 807	3 386 161

Кодът се пуска в твоя браузър на JavaScript. Броят операции е същият на всеки език, защото алгоритъмът е същият. Времето зависи от машината, затова е само за сравнение. „≈“ е оценка по кривата: толкова голямо n не пускаме, за да не замръзне страницата.

class DeadlineQueue:
    def __init__(self):
        self._items = []

    def push(self, order):
        self._items.append(order)
        i = len(self._items) - 1
        while i > 0:
            parent = (i - 1) // 2
            if not self._less(i, parent):
                return
            self._swap(i, parent)
            i = parent

    def poll(self):
        items = self._items
        top = items[0]
        items[0] = items[-1]
        items.pop()
        if items:
            self._sift_down(0)
        return top

    def _sift_down(self, i):
        n = len(self._items)
        while True:
            low = i
            left, right = 2 * i + 1, 2 * i + 2
            if left < n and self._less(left, low):
                low = left
            if right < n and self._less(right, low):
                low = right
            if low == i:
                return
            self._swap(i, low)
            i = low

    def _less(self, i, j):
        items = self._items
        return items[i].deadline < items[j].deadline

    def _swap(self, i, j):
        items = self._items
        items[i], items[j] = items[j], items[i]

    def __len__(self):
        return len(self._items)

    def deadlines(self):
        return [o.deadline for o in self._items]

При 10 поръчки подреденият списък прави 11 сравнения, а heap-ът 28: при малко поръчки списъкът е напълно добре. При 10 000 са 6 250 042 срещу 256 807. При 100 000 heap-ът прави 3 386 161, а списъкът около 625 милиона. Него Bench не го пуска, а го оценява по кривата.

Подреденият списък в Bench е в най-добрия си случай: новата поръчка само се плъзга до мястото си, без да се подрежда нищо друго. Пак е O(n) на поръчка, защото минава средно през половината списък. Heap-ът е O(log n) на поръчка.

По-точно, за любопитните

Кое подреждане забелязва, че списъкът почти е подреден? TimSort, който ползват sort в Python, List.sort и Collections.sort в Java и Array.prototype.sort в JavaScript (виж вградения sort). Той намира подредената част с около n сравнения и вкарва новия елемент в нея: O(n) на поръчка. Bench брои още по-евтин вариант, плъзгането от края, и пак излиза O(n). C# List.Sort е introsort и не търси подредени части: при всяка поръчка прави около n log n сравнения, още повече.

Защо изплуването е бързо и на практика? Новата поръчка обикновено е някъде по средата по спешност. Половината елементи на heap са листа, четвъртината са точно над тях, и т.н. Затова средно случайна поръчка спира след едно-две нива. Най-лошият случай (новата е най-спешната от всички) пак е само log₂ n.

Пробвай

Намери грешката

Колега е написал потъването, но куриерите понякога тръгват с по-късна поръчка преди по-ранна:

def _sift_down(self, i):
    n = len(self._items)
    while True:
        low = i
        left, right = 2 * i + 1, 2 * i + 2
        if left < n and self._less(left, low):
            low = left
        if right < n and self._less(right, i):
            low = right
        if low == i:
            return
        self._swap(i, low)
        i = low

queue = DeadlineQueue()
queue.push(Order("C-1", 5))
queue.push(Order("C-2", 10))
queue.push(Order("C-3", 20))
queue.push(Order("C-4", 30))
queue.push(Order("C-5", 40))
queue.push(Order("C-6", 50))
taken = []
while len(queue) > 0:
    taken.append(queue.poll().deadline)
print(taken)

Изход

[5, 20, 10, 30, 40, 50]

На кой ред е грешката?

Избери реда с грешката.

1def _sift_down(self, i):2 n = len(self._items)3 while True:4 low = i5 left, right = 2 * i + 1, 2 * i + 26 if left < n and self._less(left, low):7 low = left8 if right < n and self._less(right, i):9 low = right10 if low == i:11 return12 self._swap(i, low)13 i = low

Провери

Виж го

Готовите опашки на всеки език

В проект не пишеш heap сам. Почти всеки език има приоритетна опашка, и всяка си има особеност.

В Python heapq е min-heap върху обикновен списък: heappush изплува, heappop вади върха, а queue[0] е върхът. Поръчката се слага като кортеж (срок, номер): кортежите се сравняват отляво надясно, затова първо решава срокът.

Особеността: списъкът е heap, не подреден списък. По ред вади само heappop.

import heapq

queue = []
heapq.heappush(queue, (30, "A-101"))
heapq.heappush(queue, (10, "A-102"))
heapq.heappush(queue, (20, "A-103"))
heapq.heappush(queue, (25, "A-104"))
print(f"Най-спешна: {queue[0][1]}")
print("Отвътре:", *[deadline for deadline, _ in queue])
while queue:
    deadline, order_id = heapq.heappop(queue)
    print(f"{order_id} ({deadline} мин)")

Изход

Най-спешна: A-102
Отвътре: 10 25 20 30
A-102 (10 мин)
A-103 (20 мин)
A-104 (25 мин)
A-101 (30 мин)

Втората особеност е равенството. При еднакъв срок Python сравнява следващото в кортежа. Ако там стои речник или обект без <, опашката пада:

import heapq

queue = []
heapq.heappush(queue, (20, {"id": "A-101"}))
try:
    heapq.heappush(queue, (20, {"id": "A-102"}))
except TypeError as e:
    print("TypeError:", e)

Изход

TypeError: '<' not supported between instances of 'dict' and 'dict'

Поправката е пореден номер между срока и поръчката. Номерата никога не са равни, затова речникът не се сравнява, а при равен срок първа е по-ранната поръчка:

import heapq
from itertools import count

queue = []
seq = count()
heapq.heappush(queue, (20, next(seq), {"id": "A-101"}))
heapq.heappush(queue, (20, next(seq), {"id": "A-102"}))
heapq.heappush(queue, (10, next(seq), {"id": "A-103"}))
while queue:
    deadline, _, order = heapq.heappop(queue)
    print(order["id"], deadline)

Изход

A-103 10
A-101 20
A-102 20

По-точно, за любопитните

Нито една от тези опашки не е стабилна: две поръчки с еднакъв срок може да излязат в какъвто и да е ред (виж стабилността). Ако редът на идване е важен, сложи пореден номер, както в Python отгоре.

Смяна на срока на поръчка, която вече чака, няма в нито една от трите. remove в Java търси елемента линейно: O(n). Обичайното решение е да добавиш поръчката пак с новия срок, а старото копие да прескочиш, когато излезе. Така работи и Dijkstra в модула за графи (algo-graphs), който предстои.

Как е в другите езици: Java има PriorityQueue, C# PriorityQueue<TElement, TPriority> (от .NET 6), Python heapq (функции върху списък), а JavaScript няма вградена приоритетна опашка.

Виж го

Още две задачи за heap

Първите k. Кои 10 продукта се продават най-много от милион? Min-heap, който никога няма повече от 10 елемента: всеки нов се сравнява с върха. Това е в урока за heap, там е и Bench.

Сливане на k подредени списъка. Три склада пращат поръчките си, подредени по срок. Трябва един общ маршрут, пак по срок. В heap-а стои само началото на всеки списък. Вадиш най-ранното, а на негово място слагаш следващото от същия склад:

import heapq

lists = [[5, 20, 45], [10, 15, 50], [12, 30]]
names = ["Север", "Юг", "Център"]
heads = [(lst[0], k, 0) for k, lst in enumerate(lists)]
heapq.heapify(heads)
route = []
while heads:
    deadline, k, i = heapq.heappop(heads)
    route.append(f"{deadline} {names[k]}")
    if i + 1 < len(lists[k]):
        heapq.heappush(heads, (lists[k][i + 1], k, i + 1))
print("Маршрут:", ", ".join(route))

Изход

Маршрут: 5 Север, 10 Юг, 12 Център, 15 Юг, 20 Север, 30 Център, 45 Север, 50 Юг

Heap-ът никога няма повече от k елемента (тук 3). Затова N поръчки струват O(N log k), а не O(N log N) за подреждане на всичко. Така се сливат и подредени части от файл, който не се събира в паметта.

По-точно, за любопитните

За това в Python има готово: heapq.merge(*lists) прави точно същото и връща сливането поред, без да го събира в списък.

Merge sort също слива подредени части, но по две (merge sort). С heap се сливат k наведнъж.

Упражнение

Честна опашка

Реши задачата в редактора си.

Еднакъв срок, първа е по-ранната. Направи опашка FairQueue, в която при еднакъв срок първа излиза поръчката, която е дошла по-рано. Пази пореден номер с всяка поръчка и сравнявай първо по срок, после по номер. Пусни я с поръчките A-1 (30), A-2 (20), A-3 (30), A-4 (20), A-5 (10), A-6 (20), в този ред:

Изход

Ред: A-5, A-2, A-4, A-6, A-1, A-3

Още задачи

• Max-heap. Обърни сравнението и направи опашка, която дава най-скъпата поръчка първа.
• Брой сравненията. Добави брояч в изплуването и потъването и пусни 1000 случайни поръчки. Колко сравнения се падат на поръчка?
• Първите три. Как да вземеш трите най-спешни поръчки, без да вадиш другите от опашката?

Тест

Провери се

1. Какво обещава приоритетната опашка?

Вади елементите в реда, в който са влезлиПази елементите подредени, за да ги обхождаш по редВади най-спешния елемент и приема нови, без да подрежда всичко

2. Новата поръчка влиза в heap с 1 000 000 поръчки. Колко сравнения прави изплуването най-много?

Около 1000Около 20Около 1 000 000

3. В Python пазиш поръчките като (срок, речник) в heapq. Две поръчки имат еднакъв срок. Какво става?

heapq ги слива в еднаВзима се тази, която е влязла първаTypeError: кортежите се сравняват по втория елемент, а речници не се сравняват

4. Три склада пращат подредени по срок списъци, общо N поръчки. Как ги сливаш в един маршрут?

Взимаш поред по една от всеки складСливаш ги в един списък и го подреждашHeap с началото на всеки списък: вадиш най-ранното и слагаш следващото от същия склад

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: Най-добър, среден и най-лош случай · Merge sort и „разделяй и владей“ · Heap и heap sort · Стабилно сортиране и вграденият sort
• Продължава в: Dijkstra · Минимално покриващо дърво · Хеш таблица като брояч · Навигация в града
• Задачи по темата: Структури от данни