2.38. Prioritätswarteschlangen

Eine Prioritätswarteschlange ist ein Container, der immer weiß, welches seiner Elemente das kleinste (oder das größte) ist, ohne das Ganze zu sortieren. heapq implementiert eine solche in MicroPython, wobei eine einfache list als Speicher und ein baumbasierter Algorithmus namens Min-Heap verwendet wird, um das kleinste Element an Index 0 zu halten.

2.38.1. Das gedankliche Modell

Ein Min-Heap ist eine Liste, die so angeordnet ist, dass das Element an jedem Index kleiner-oder-gleich den Elementen an den Indizes 2*i + 1 und 2*i + 2 ist. Diese Anordnung macht drei Operationen kostengünstig:

• Das kleinste Element finden – es befindet sich in heap[0]. Ein einzelner Index-Zugriff, kein Durchsuchen.

• Ein Element hinzufügen – es steigt durch den Baum auf und vergleicht sich pro Ebene mit einem Elternknoten. Tausend Elemente sind etwa zehn Vergleiche; eine Million sind etwa zwanzig.

• Das kleinste Element entfernen – derselbe Durchlauf umgekehrt, mit derselben Handvoll Vergleiche.

Die Liste ist nicht sortiert. Beim Iterieren erhält man die Elemente in beliebiger Reihenfolge. Nur heap[0] ist garantiert das Minimum.

2.38.2. Die drei Funktionen in MicroPython

Das heapq von MicroPython stellt genau drei Funktionen bereit:

• heapq.heappush() – fügt ein Element ein und hält dabei die Heap-Invariante intakt.

• heapq.heappop() – entfernt das kleinste Element und gibt es zurück.

• heapq.heapify() – ordnet eine bestehende Liste an Ort und Stelle in einen Heap um.

Das ist der gesamte Funktionsumfang. heappushpop, heapreplace, nlargest, nsmallest und merge aus CPython existieren nicht; die folgenden Muster zeigen, wie man die gebräuchlichen davon von Hand nachbaut.

Von Grund auf neu beginnen:

>>> import heapq
>>> heap = []
>>> heapq.heappush(heap, 5)
>>> heapq.heappush(heap, 1)
>>> heapq.heappush(heap, 3)
>>> heap
[1, 5, 3]
>>> heapq.heappop(heap)
1
>>> heapq.heappop(heap)
3
>>> heapq.heappop(heap)
5

Von bestehenden Daten ausgehen:

>>> samples = [5, 1, 3, 2, 4]
>>> heapq.heapify(samples)
>>> samples
[1, 2, 3, 5, 4]
>>> heapq.heappop(samples)
1

Heapify berührt jedes Element einmal und ist spürbar schneller, als dieselben Elemente einzeln nacheinander einzufügen.

2.38.3. Top-N über einen Stream

Eine natürliche Anwendung einer Prioritätswarteschlange: die N größten Elemente in einem Stream verfolgen, den man nur einmal lesen kann.

Der Trick besteht darin, einen Min-Heap der Größe N zu führen. Jeder neue Wert wird eingefügt; wächst der Heap über N hinaus, wird das kleinste Element entfernt. Das kleinste befindet sich stets in heap[0], sodass ein einzelner Vergleich entscheidet, ob jedes eingehende Element behalten oder verworfen wird:

def top_n_brightest(readings, n):
    heap = []
    for r in readings:
        if len(heap) < n:
            heapq.heappush(heap, r)
        elif r > heap[0]:
            heapq.heappop(heap)
            heapq.heappush(heap, r)
    return sorted(heap, reverse=True)

Der Heap wächst nie über N hinaus, sodass jeder eingehende Wert dieselbe Handvoll Vergleiche kostet, egal wie lang der Stream ist. Das abschließende sorted ist der einzige Schritt, dessen Kosten direkt von N abhängen.

2.38.4. Geplante Ereignisse

Eine Prioritätswarteschlange aus (deadline, payload)-Tupeln ist die Standardform für einen Ereignisplaner. Tupel werden lexikografisch verglichen, sodass das Tupel mit der kleinsten Deadline automatisch nach oben steigt:

import heapq
import time

queue = []
heapq.heappush(queue, (time.time() + 5, 'check sensors'))
heapq.heappush(queue, (time.time() + 1, 'blink LED'))
heapq.heappush(queue, (time.time() + 3, 'send heartbeat'))

while queue:
    deadline, task = queue[0]
    now = time.time()
    if now < deadline:
        time.sleep(deadline - now)
    heapq.heappop(queue)
    run(task)

Wenn sich zwei Ereignisse eine Deadline teilen, fällt der Tupelvergleich auf das zweite Element zurück. Für Payloads, die nicht von Natur aus vergleichbar sind, fügt man als zweites Element des Tupels einen Zähler hinzu, um eine stabile Reihenfolge zu erzwingen:

counter = 0
def schedule(deadline, task):
    global counter
    counter += 1
    heapq.heappush(queue, (deadline, counter, task))

Nun ist das Verhalten „kleinste Deadline zuerst“ intakt, Gleichstände werden über den Zähler aufgelöst, sodass früher geplante Aufgaben zuerst laufen, und task muss nicht vergleichbar sein.

2.38.5. Was heapq nicht bietet

Einige Muster, die aussehen, als sollten sie funktionieren, tun es aber nicht:

• Das Iterieren mit for x in heap liefert keine sortierte Ausgabe – die zugrunde liegende Reihenfolge der Liste ist die Baumstruktur des Heaps. Um den Heap in Reihenfolge zu verarbeiten, ruft man heappop auf, bis er leer ist.

• Es gibt keine Möglichkeit, die Priorität eines bereits im Heap befindlichen Elements zu ändern. Die übliche Behelfslösung besteht darin, die neue Priorität als neuen Eintrag einzufügen und den alten als abgebrochen zu markieren (ihn überspringen, wenn er herausgepoppt wird).

• Der Heap speichert Referenzen auf seine Elemente, keine Kopien. Das Verändern eines Elements, nachdem es eingefügt wurde, kann die Heap-Invariante verletzen. Verwenden Sie unveränderliche Werte (Zahlen, Tupel) als Vergleichsschlüssel.

Ein Min-Heap auf einer einfachen Liste ist das richtige Werkzeug für jede Frage nach „dem kleinsten“ oder „dem dringendsten“ Element auf einem kleinen eingebetteten Gerät. Drei Funktionen, eine Liste, und der Heap kümmert sich um den Rest.