heapq — Heap-Warteschlange-Algorithmus

Quellcode: Lib/heapq.py

Dieses Modul bietet eine Implementierung des Heap-Warteschlange-Algorithmus, auch bekannt als Prioritätswarteschlangen-Algorithmus.

Min-Heaps sind binäre Bäume, bei denen jeder Elternknoten einen Wert hat, der kleiner oder gleich jedem seiner Kinder ist. Wir bezeichnen diese Bedingung als Heap-Invariant.

Für Min-Heaps verwendet diese Implementierung Listen, für die heap[k] <= heap[2*k+1] und heap[k] <= heap[2*k+2] für alle k gilt, für die die verglichenen Elemente existieren. Die Elemente werden ab Null gezählt. Die interessante Eigenschaft eines Min-Heaps ist, dass sein kleinster Wert immer die Wurzel, heap[0], ist.

Max-Heaps erfüllen die umgekehrte Invariante: Jeder Elternknoten hat einen Wert, der *größer* als jeder seiner Kinder ist. Diese werden als Listen implementiert, für die maxheap[2*k+1] <= maxheap[k] und maxheap[2*k+2] <= maxheap[k] für alle k gilt, für die die verglichenen Elemente existieren. Die Wurzel, maxheap[0], enthält das *größte* Element; heap.sort(reverse=True) erhält die Max-Heap-Invariant.

Die heapq API unterscheidet sich in zwei Aspekten von Lehrbuch-Heap-Algorithmen: (a) Wir verwenden nullbasierte Indizierung. Dies macht die Beziehung zwischen dem Index eines Knotens und den Indizes seiner Kinder etwas weniger offensichtlich, ist aber besser geeignet, da Python nullbasierte Indizierung verwendet. (b) Lehrbücher konzentrieren sich oft auf Max-Heaps, da diese für das In-Place-Sortieren geeignet sind. Unsere Implementierung bevorzugt Min-Heaps, da sie besser zu den Python Listen passen.

Diese beiden Aspekte ermöglichen es, den Heap als normale Python-Liste ohne Überraschungen zu betrachten: heap[0] ist das kleinste Element, und heap.sort() erhält die Heap-Invariant!

Wie list.sort() verwendet diese Implementierung nur den < Operator für Vergleiche, sowohl für Min-Heaps als auch für Max-Heaps.

In der folgenden API und in dieser Dokumentation bezieht sich der unqualifizierte Begriff Heap im Allgemeinen auf einen Min-Heap. Die API für Max-Heaps ist mit dem Suffix _max benannt.

Um einen Heap zu erstellen, verwenden Sie eine als [] initialisierte Liste oder transformieren Sie eine vorhandene Liste mithilfe der Funktionen heapify() bzw. heapify_max() in einen Min-Heap oder Max-Heap.

Die folgenden Funktionen werden für Min-Heaps bereitgestellt

heapq.heappush(heap, item)

Fügt den Wert item in den heap ein und erhält dabei die Min-Heap-Invariant.

heapq.heappop(heap)

Entnimmt und gibt das kleinste Element aus dem heap zurück und erhält dabei die Min-Heap-Invariant. Wenn der Heap leer ist, wird IndexError ausgelöst. Um das kleinste Element abzurufen, ohne es zu entnehmen, verwenden Sie heap[0].

heapq.heappushpop(heap, item)

Fügt item in den Heap ein und entnimmt dann das kleinste Element aus dem heap und gibt es zurück. Die kombinierte Aktion ist effizienter als ein Aufruf von heappush() gefolgt von einem separaten Aufruf von heappop().

heapq.heapify(x)

Transformiert die Liste x in-place in einen Min-Heap mit linearer Zeit.

heapq.heapreplace(heap, item)

Entnimmt und gibt das kleinste Element aus dem heap zurück und fügt gleichzeitig das neue item ein. Die Heap-Größe ändert sich nicht. Wenn der Heap leer ist, wird IndexError ausgelöst.

Diese Einzelschrittoperation ist effizienter als ein heappop() gefolgt von einem heappush() und kann besser geeignet sein, wenn ein Heap mit fester Größe verwendet wird. Die Pop/Push-Kombination gibt immer ein Element aus dem Heap zurück und ersetzt es durch item.

Der zurückgegebene Wert kann größer sein als das hinzugefügte item. Wenn dies nicht erwünscht ist, sollten Sie stattdessen heappushpop() verwenden. Seine Push/Pop-Kombination gibt den kleineren der beiden Werte zurück und lässt den größeren Wert im Heap.

Für Max-Heaps werden die folgenden Funktionen bereitgestellt

heapq.heapify_max(x)

Transformiert die Liste x in-place in einen Max-Heap mit linearer Zeit.

Hinzugefügt in Version 3.14.

heapq.heappush_max(heap, item)

Fügt den Wert item in den Max-Heap heap ein und erhält dabei die Max-Heap-Invariant.

Hinzugefügt in Version 3.14.

heapq.heappop_max(heap)

Entnimmt und gibt das größte Element aus dem Max-Heap heap zurück und erhält dabei die Max-Heap-Invariant. Wenn der Max-Heap leer ist, wird IndexError ausgelöst. Um das größte Element abzurufen, ohne es zu entnehmen, verwenden Sie maxheap[0].

Hinzugefügt in Version 3.14.

heapq.heappushpop_max(heap, item)

Fügt item in den Max-Heap heap ein und entnimmt dann das größte Element aus heap und gibt es zurück. Die kombinierte Aktion ist effizienter als ein Aufruf von heappush_max() gefolgt von einem separaten Aufruf von heappop_max().

Hinzugefügt in Version 3.14.

heapq.heapreplace_max(heap, item)

Entnimmt und gibt das größte Element aus dem Max-Heap heap zurück und fügt gleichzeitig das neue item ein. Die Max-Heap-Größe ändert sich nicht. Wenn der Max-Heap leer ist, wird IndexError ausgelöst.

Der zurückgegebene Wert kann kleiner sein als das hinzugefügte item. Siehe die analoge Funktion heapreplace() für detaillierte Anwendungshinweise.

Hinzugefügt in Version 3.14.

Das Modul bietet außerdem drei allgemeine Funktionen, die auf Heaps basieren.

heapq.merge(*iterables, key=None, reverse=False)

Führt mehrere sortierte Eingaben zu einer einzigen sortierten Ausgabe zusammen (z. B. Zusammenführen von zeitgestempelten Einträgen aus mehreren Logdateien). Gibt einen Iterator über die sortierten Werte zurück.

Ähnlich wie sorted(itertools.chain(*iterables)), gibt aber ein Iterable zurück, zieht die Daten nicht auf einmal in den Speicher und geht davon aus, dass jeder der Eingabeströme bereits sortiert ist (vom kleinsten zum größten).

Hat zwei optionale Argumente, die als Schlüsselwortargumente angegeben werden müssen.

key gibt eine Schlüsselfunktion für ein Argument an, die verwendet wird, um einen Vergleichsschlüssel aus jedem Eingabeelement zu extrahieren. Der Standardwert ist None (vergleicht die Elemente direkt).

reverse ist ein boolescher Wert. Wenn auf True gesetzt, werden die Eingabeelemente so zusammengeführt, als ob jeder Vergleich umgekehrt wäre. Um ein Verhalten ähnlich wie sorted(itertools.chain(*iterables), reverse=True) zu erzielen, müssen alle Iterables von groß nach klein sortiert sein.

Geändert in Version 3.5: Die optionalen Parameter key und reverse wurden hinzugefügt.

heapq.nlargest(n, iterable, key=None)

Gibt eine Liste mit den n größten Elementen aus dem durch iterable definierten Datensatz zurück. key gibt, falls angegeben, eine Funktion für ein Argument an, die verwendet wird, um einen Vergleichsschlüssel aus jedem Element in iterable zu extrahieren (z. B. key=str.lower). Entspricht: sorted(iterable, key=key, reverse=True)[:n].

heapq.nsmallest(n, iterable, key=None)

Gibt eine Liste mit den n kleinsten Elementen aus dem durch iterable definierten Datensatz zurück. key gibt, falls angegeben, eine Funktion für ein Argument an, die verwendet wird, um einen Vergleichsschlüssel aus jedem Element in iterable zu extrahieren (z. B. key=str.lower). Entspricht: sorted(iterable, key=key)[:n].

Die beiden letzteren Funktionen sind für kleinere Werte von n am besten geeignet. Für größere Werte ist es effizienter, die Funktion sorted() zu verwenden. Wenn n==1 ist, ist es außerdem effizienter, die integrierten Funktionen min() und max() zu verwenden. Wenn diese Funktionen wiederholt verwendet werden müssen, sollten Sie das Iterable in einen tatsächlichen Heap umwandeln.

Grundlegende Beispiele

Ein Heapsort kann implementiert werden, indem alle Werte in einen Heap eingefügt und dann die kleinsten Werte nacheinander entnommen werden.

>>> def heapsort(iterable):
...     h = []
...     for value in iterable:
...         heappush(h, value)
...     return [heappop(h) for i in range(len(h))]
...
>>> heapsort([1, 3, 5, 7, 9, 2, 4, 6, 8, 0])
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Dies ähnelt sorted(iterable), aber im Gegensatz zu sorted() ist diese Implementierung nicht stabil.

Heap-Elemente können Tupel sein. Dies ist nützlich, um Vergleichswerte (wie z. B. Aufgabenebenen) neben dem Hauptdatensatz zuzuweisen, der verfolgt wird.

>>> h = []
>>> heappush(h, (5, 'write code'))
>>> heappush(h, (7, 'release product'))
>>> heappush(h, (1, 'write spec'))
>>> heappush(h, (3, 'create tests'))
>>> heappop(h)
(1, 'write spec')

Hinweise zur Implementierung von Prioritätswarteschlangen

Eine Prioritätswarteschlange ist eine häufige Verwendung für einen Heap und birgt mehrere Implementierungsherausforderungen.

  • Stabilität der Sortierung: Wie erhalten Sie zwei Aufgaben mit gleichen Prioritäten, damit sie in der Reihenfolge zurückgegeben werden, in der sie ursprünglich hinzugefügt wurden?

  • Die Tupelvergleichung schlägt bei (Priorität, Aufgabe)-Paaren fehl, wenn die Prioritäten gleich sind und die Aufgaben keine standardmäßige Vergleichsreihenfolge haben.

  • Wenn sich die Priorität einer Aufgabe ändert, wie verschieben Sie sie an eine neue Position im Heap?

  • Oder wenn eine ausstehende Aufgabe gelöscht werden muss, wie finden Sie sie und entfernen sie aus der Warteschlange?

Eine Lösung für die ersten beiden Herausforderungen besteht darin, Einträge als 3-Elemente-Liste zu speichern, die die Priorität, eine Eintragsanzahl und die Aufgabe enthält. Die Eintragsanzahl dient als Tie-Breaker, sodass zwei Aufgaben mit der gleichen Priorität in der Reihenfolge zurückgegeben werden, in der sie hinzugefügt wurden. Da keine zwei Eintragszählungen gleich sind, wird der Tupelvergleich niemals versuchen, zwei Aufgaben direkt zu vergleichen.

Eine weitere Lösung für das Problem nicht vergleichbarer Aufgaben ist die Erstellung einer Wrapper-Klasse, die das Aufgabenelement ignoriert und nur das Prioritätsfeld vergleicht.

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

Die verbleibenden Herausforderungen drehen sich um das Auffinden einer ausstehenden Aufgabe und das Ändern ihrer Priorität oder das vollständige Entfernen. Das Auffinden einer Aufgabe kann mit einem Wörterbuch erfolgen, das auf einen Eintrag in der Warteschlange verweist.

Das Entfernen des Eintrags oder das Ändern seiner Priorität ist schwieriger, da dies die Heap-Struktur-Invarianten brechen würde. Eine mögliche Lösung ist daher, den Eintrag als entfernt zu markieren und einen neuen Eintrag mit der überarbeiteten Priorität hinzuzufügen.

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
counter = itertools.count()     # unique sequence count

def add_task(task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED.  Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

Theorie

Heaps sind Arrays, für die a[k] <= a[2*k+1] und a[k] <= a[2*k+2] für alle k gilt, wobei die Elemente ab 0 gezählt werden. Zum Vergleich werden nicht vorhandene Elemente als unendlich betrachtet. Die interessante Eigenschaft eines Heaps ist, dass a[0] immer sein kleinster Wert ist.

Die seltsame obige Invariante soll eine effiziente Speicherrepräsentation für ein Turnier sein. Die Zahlen unten sind k, nicht a[k]

                               0

              1                                 2

      3               4                5               6

  7       8       9       10      11      12      13      14

15 16   17 18   19 20   21 22   23 24   25 26   27 28   29 30

Im obigen Baum überragt jede Zelle k die von ihr überragten Zellen 2*k+1 und 2*k+2. In einem üblichen binären Turnier, wie wir es im Sport sehen, ist jede Zelle der Gewinner über die beiden Zellen, die sie überragt, und wir können den Gewinner den Baum hinunterverfolgen, um alle Gegner zu sehen, gegen die er angetreten ist. In vielen Computeranwendungen solcher Turniere müssen wir jedoch nicht die Geschichte eines Gewinners verfolgen. Um den Speicher effizienter zu nutzen, ersetzen wir beim Aufstieg eines Gewinners ihn durch etwas anderes auf einer niedrigeren Ebene, und die Regel lautet, dass eine Zelle und die beiden von ihr überragten Zellen drei verschiedene Elemente enthalten, aber die obere Zelle "gewinnt" über die beiden überragten Zellen.

Wenn diese Heap-Invariant jederzeit geschützt ist, ist Index 0 eindeutig der Gesamtsieger. Der einfachste algorithmische Weg, ihn zu entfernen und den "nächsten" Gewinner zu finden, besteht darin, einen Verlierer (sagen wir Zelle 30 im obigen Diagramm) an die Position 0 zu verschieben und diesen neuen 0er dann den Baum hinunter zu befördern, Werte auszutauschen, bis die Invariant wiederhergestellt ist. Dies ist eindeutig logarithmisch zur Gesamtzahl der Elemente im Baum. Durch Iteration über alle Elemente erhalten Sie eine O(n log n) Sortierung.

Ein schöner Nebeneffekt dieser Sortierung ist, dass Sie neue Elemente effizient einfügen können, während die Sortierung läuft, vorausgesetzt, dass die eingefügten Elemente nicht "besser" sind als das letzte extrahierte 0er-Element. Dies ist besonders nützlich in Simulationskontexten, in denen der Baum alle eingehenden Ereignisse enthält und die "Gewinnbedingung" die kleinste geplante Zeit bedeutet. Wenn ein Ereignis andere Ereignisse zur Ausführung plant, werden diese für die Zukunft geplant, sodass sie leicht in den Heap passen. Ein Heap ist also eine gute Struktur zur Implementierung von Planern (damit habe ich meinen MIDI-Sequenzer gemacht :-)).

Verschiedene Strukturen zur Implementierung von Planern wurden ausgiebig untersucht, und Heaps sind dafür gut geeignet, da sie relativ schnell sind, die Geschwindigkeit fast konstant ist und der Worst Case nicht viel vom Durchschnittsfall abweicht. Es gibt jedoch andere Darstellungen, die insgesamt effizienter sind, aber die Worst Cases können schrecklich sein.

Heaps sind auch sehr nützlich bei großen Festplattensortierungen. Sie wissen wahrscheinlich alle, dass eine große Sortierung die Erzeugung von "Läufen" (vorab sortierte Sequenzen, deren Größe normalerweise mit der CPU-Speichergröße zusammenhängt) und anschließende Zusammenführungsläufe dieser Läufe impliziert, die oft sehr clever organisiert ist [1]. Es ist sehr wichtig, dass die anfängliche Sortierung die längsten möglichen Läufe erzeugt. Turniere sind eine gute Möglichkeit, dies zu erreichen. Wenn Sie mit dem gesamten verfügbaren Speicher einen Tournament erstellen und Elemente ersetzen und aufsteigen lassen, die in den aktuellen Lauf passen, produzieren Sie Läufe, die bei zufälligen Eingaben doppelt so groß sind wie der Speicher, und für leicht geordnete Eingaben viel besser.

Darüber hinaus, wenn Sie das 0er-Element auf die Festplatte ausgeben und eine Eingabe erhalten, die möglicherweise nicht in das aktuelle Turnier passt (da der Wert über den zuletzt ausgegebenen Wert "gewinnt"), kann er nicht in den Heap passen, sodass die Größe des Heaps abnimmt. Der freigewordene Speicher kann sofort geschickt wiederverwendet werden, um progressiv einen zweiten Heap aufzubauen, der genau im gleichen Tempo wächst, wie der erste Heap schmilzt. Wenn der erste Heap vollständig verschwindet, wechseln Sie die Heaps und starten einen neuen Lauf. Clever und ziemlich effektiv!

Kurz gesagt, Heaps sind nützliche Speicherstrukturen, die man kennen sollte. Ich verwende sie in einigen Anwendungen, und ich denke, es ist gut, ein „heap“-Modul bereitzuhalten. :-)

Fußnoten