heapq — Algoritmo de colas montículos (heap)

Código fuente: Lib/heapq.py

Este módulo proporciona una implementación del algoritmo de montículos, también conocido como algoritmo de cola con prioridad.

Los montículos son árboles binarios para los cuales cada nodo padre tiene un valor menor o igual que cualquiera de sus hijos. Esta implementación utiliza matrices para las cuales heap[k] <= heap[2*k+1] y heap[k] <= heap[2*k+2] para todo k, contando los elementos desde cero. Para poder comparar, los elementos inexistentes se consideran infinitos. La propiedad interesante de un montículo es que su elemento más pequeño es siempre la raíz, heap[0].

El API que se presenta a continuación difiere de los algoritmos de los libros de texto en dos aspectos: (a) Utilizamos la indexación basada en cero. Esto hace que la relación entre el índice de un nodo y los índices de sus hijos sea un poco menos evidente, pero es más adecuado ya que Python utiliza la indexación basada en cero. (b) Nuestro método «pop» retorna el elemento más pequeño, no el más grande (llamado «min heap» o montículo por mínimo en los libros de texto; un «max heap» o montículo por máximos es más común en los textos debido a su idoneidad para la clasificación in situ).

Estos dos permiten ver el montículo como una lista Python normal sin sorpresas: heap[0] es el ítem más pequeño, y heap.sort() mantiene el montículo invariable!

Para crear un montículo, usa una lista inicializada como [], o puedes transformar una lista poblada en un montículo a través de la función heapify().

Las siguientes funciones están provistas:

heapq.heappush(heap, item)

Empujar el valor item en el heap, manteniendo el montículo invariable.

heapq.heappop(heap)

Desapila o pop y retorna el elemento más pequeño del heap, manteniendo el montículo invariable. Si el montículo está vacío, IndexError se lanza. Para acceder al elemento más pequeño sin necesidad de desapilar, usa``heap[0]``.

heapq.heappushpop(heap, item)

Apila el elemento o iem en el montículo, y luego desapila y retorna el elemento más pequeño del montículo. La acción combinada se ejecuta más eficientemente que heappush() seguido de una llamada separada a heappop().

heapq.heapify(x)

Transformar la lista x en un montículo, en el lugar, en tiempo lineal.

heapq.heapreplace(heap, item)

Desapila y retorna el elemento más pequeño del heap, y también apile el nuevo item. El tamaño del montículo no cambia. Si el montículo está vacío, IndexError se lanza.

Esta operación de un solo paso es más eficiente que un heappop() seguido por heappush() y puede ser más apropiada cuando se utiliza un montículo de tamaño fijo. La combinación pop/push siempre retorna un elemento del montículo y lo reemplaza con item.

El valor retornado puede ser mayor que el item añadido. Si no se desea eso, considere usar heappushpop() en su lugar. Su combinación push/pop retorna el menor de los dos valores, dejando el mayor valor en el montículo.

El módulo también ofrece tres funciones de propósito general basadas en los montículos.

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

Fusionar varias entradas ordenadas en una sola salida ordenada (por ejemplo, fusionar entradas con marca de tiempo de varios archivos de registro). Retorna un iterator sobre los valores ordenados.

Similar a sorted(itertools.chain(*iterables)) pero retorna un iterable, no hala los datos a la memoria de una sola vez, y asume que cada uno de los flujos de entrada ya están ordenado (de menor a mayor).

Tiene dos argumentos opcionales que deben ser especificados como argumentos de palabras clave.

key especifica una key function de un argumento que se utiliza para extraer una clave de comparación de cada elemento de entrada. El valor por defecto es None (compara los elementos directamente).

reverse es un valor booleano. Si se establece en True, entonces los elementos de entrada se fusionan como si cada comparación se invirtiera. Para lograr un comportamiento similar a sorted(itertools.chain(*iterables), reverse=True), todos los iterables deben ser ordenados de mayor a menor.

Distinto en la versión 3.5: Añadió los parámetros opcionales de key y reverse.

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

Retorna una lista con los n elementos más grandes del conjunto de datos definidos por iterable. key, si se proporciona, especifica una función de un argumento que se utiliza para extraer una clave de comparación de cada elemento en iterable (por ejemplo, key=str.lower). Equivalente a: sorted(iterable, key=clave, reverse=True)[:n].

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

Retorna una lista con los n elementos más pequeños del conjunto de datos definidos por iterable. key, si se proporciona, especifica una función de un argumento que se utiliza para extraer una clave de comparación de cada elemento en iterable (por ejemplo, key=str.lower). Equivalente a: sorted(iterable, key=clave)[:n].

Las dos últimas funciones funcionan mejor para valores más pequeños de n. Para valores más grandes, es más eficiente usar la función sorted(). Además, cuando n==1, es más eficiente usar las funciones incorporadas min() y max`(). Si se requiere el uso repetido de estas funciones, considere convertir lo iterable en un verdadero montículo.

Ejemplos Básicos

Un heapsort» puede ser implementado empujando todos los valores en un montículo y luego desapilando los valores más pequeños uno a la vez:

>>> 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]

Esto es similar a sorted(iterable), pero a diferencia de sorted(), esta implementación no es estable.

Los elementos del montículo pueden ser tuplas. Esto es útil para asignar valores de comparación (como las prioridades de las tareas) junto con el registro principal que se está rastreando:

>>> 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')

Notas de Aplicación de la Cola de Prioridades

Una cola de prioridad es de uso común para un montículo, y presenta varios desafíos de implementación:

  • Estabilidad de la clasificación: ¿cómo se consigue que dos tareas con iguales prioridades sean retornadas en el orden en que fueron añadidas originalmente?

  • Interrupciones de comparación en tupla para pares (prioridad, tarea) si las prioridades son iguales y las tareas no tienen un orden de comparación por defecto.

  • ¿Si la prioridad de una tarea cambia, cómo la mueves a una nueva posición en el montículo?

  • ¿O si una tarea pendiente necesita ser borrada, cómo la encuentras y la eliminas de la cola?

Una solución a los dos primeros desafíos es almacenar las entradas como una lista de 3 elementos que incluya la prioridad, un recuento de entradas y la tarea. El recuento de entradas sirve como un desempate para que dos tareas con la misma prioridad sean retornadas en el orden en que fueron añadidas. Y como no hay dos recuentos de entradas iguales, la comparación tupla nunca intentará comparar directamente dos tareas.

Otra solución al problema de las tareas no comparables es crear una clase envolvente que ignore el elemento de la tarea y sólo compare el campo de prioridad:

from dataclasses import dataclass, field
from typing import Any

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

Los desafíos restantes giran en torno a encontrar una tarea pendiente y hacer cambios en su prioridad o eliminarla por completo. Encontrar una tarea se puede hacer con un diccionario que apunta a una entrada en la cola.

Eliminar la entrada o cambiar su prioridad es más difícil porque rompería las invariantes de la estructura del montículo. Por lo tanto, una posible solución es marcar la entrada como eliminada y añadir una nueva entrada con la prioridad revisada:

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')

Teoría

Los montículos son conjuntos para los cuales``a[k] <= a[2*k+1]`` y «a[k] <= a[2*k+2]`` para todos los k, contando los elementos desde 0. Para comparar, los elementos no existentes se consideran infinitos. La interesante propiedad de un montículo es que a[0] es siempre su elemento más pequeño.

La extraña invariante de arriba intenta ser una representación eficiente de la memoria para un torneo. Los números de abajo son k, no``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

En el árbol de arriba, cada celda k está coronada por 2*k+1 y 2*k+2. En un torneo binario habitual que vemos en los deportes, cada celda es el ganador sobre las dos celdas que supera, y podemos rastrear al ganador hasta el árbol para ver todos los oponentes que tuvo. Sin embargo, en muchas aplicaciones informáticas de tales torneos, no necesitamos rastrear la historia de un ganador. Para ser más eficientes en la memoria, cuando un ganador es ascendido, tratamos de reemplazarlo por algo más en un nivel inferior, y la regla se convierte en que una celda y las dos celdas que supera contienen tres elementos diferentes, pero la celda superior «gana» sobre las dos celdas superiores.

Si esta invariante del montículo está protegida en todo momento, el índice 0 es claramente el ganador general. La forma algorítmica más simple de eliminarlo y encontrar el «próximo» ganador es mover algún perdedor (digamos la celda 30 en el diagrama de arriba) a la posición 0, y luego filtrar este nuevo 0 por el árbol, intercambiando valores, hasta que la invariante se reestablezca. Esto es claramente logarítmico en el número total de elementos del árbol. Al iterar sobre todos los elementos, se obtiene una clasificación O(n log n).

Una buena característica de este tipo es que puedes insertar nuevos elementos de manera eficiente mientras se realiza la clasificación, siempre y cuando los elementos insertados no sean «mejores» que el último 0’th elemento que has extraído. Esto es especialmente útil en contextos de simulación, donde el árbol contiene todos los eventos entrantes, y la condición de «ganar» significa el menor tiempo programado. Cuando un evento programa otros eventos para su ejecución, se programan en el futuro, para que puedan ir fácilmente al montículo. Por lo tanto, un montículo es una buena estructura para implementar planificadores o schedulers (esto es lo que usé para mi secuenciador MIDI :-).

Se han estudiado extensamente varias estructuras para implementar los planificadores, y los montículos son buenos para ello, ya que son razonablemente rápidos, la velocidad es casi constante, y el peor de los casos no es muy diferente del caso promedio. Sin embargo, hay otras representaciones que son más eficientes en general, aunque los peores casos podrían ser terribles.

Los montículos también son muy útiles en las grandes ordenaciones de elementos en discos de memoria. Lo más probable es que todos sepan que un tipo grande implica la producción de «ejecuciones» (que son secuencias preclasificadas, cuyo tamaño suele estar relacionado con la cantidad de memoria de la CPU), seguidas de una fusión de pases para estas ejecuciones, cuya fusión suele estar muy inteligentemente organizada 1. Es muy importante que la clasificación inicial produzca las ejecuciones posibles más largas. Los torneos son una buena manera de lograrlo. Si, utilizando toda la memoria disponible para celebrar un torneo, sustituyes y filtras los elementos que encajan en la carrera actual, producirás carreras que tienen el doble del tamaño de la memoria para la entrada aleatoria, y mucho mejor para la entrada ordenada de forma difusa.

Además, si se da salida al 0’th item en el disco y se obtiene una entrada que no puede caber en el torneo actual (porque el valor «gana» sobre el último valor de salida), no puede caber en el montículo, por lo que el tamaño del montículo disminuye. La memoria liberada podría ser ingeniosamente reutilizada inmediatamente para construir progresivamente un segundo montículo, que crece exactamente al mismo ritmo que el primer montículo se está fundiendo. Cuando el primer montículo se desvanece completamente, se cambia de montículo y se inicia una nueva carrera. ¡Ingenioso y muy efectivo!

En una palabra, los montículos son estructuras de memoria útiles a conocer. Las uso en algunas aplicaciones, y creo que es bueno tener un módulo “heap” alrededor. :-)

Notas al pie de página

1

Los algoritmos de balanceo de discos que están vigentes hoy en día, son más molestos que inteligentes, y esto es una consecuencia de las capacidades de búsqueda de los discos. En los dispositivos que no pueden buscar, como las grandes unidades de cinta, la historia era muy diferente, y había que ser muy inteligente para asegurarse (con mucha antelación) de que cada movimiento de la cinta fuera el más efectivo (es decir, que participara mejor en el «progreso» de la fusión). Algunas cintas eran incluso capaces de leer al revés, y esto también se utilizó para evitar el tiempo rebobinado. Créanme, ¡la ordenación de elementos en cinta realmente buenos fueron espectaculares de ver! ¡Desde todos los tiempos, la ordenación de elementos siempre ha sido un Gran Arte! :-)