heapq — Algoritmo de fila heap

Código-fonte: Lib/heapq.py

Este módulo fornece uma implementação do algoritmo de fila heap, também conhecido como algoritmo de fila de prioridade.

Heaps são árvores binárias para as quais cada nó pai tem um valor menor ou igual a qualquer um de seus filhos. Chamamos essa condição de invariante de heap.

Esta implementação usa arrays para os quais heap[k] <= heap[2*k+1] e heap[k] <= heap[2*k+2] para todos k, contando elementos de zero. Para efeito de comparação, os elementos inexistentes são considerados infinitos. A propriedade interessante de um heap é que seu menor elemento é sempre a raiz, heap[0].

A API abaixo difere dos algoritmos de heap de livros didáticos em dois aspectos: (a) Usamos indexação baseada em zero. Isso torna o relacionamento entre o índice de um nó e os índices de seus filhos um pouco menos óbvio, mas é mais adequado, pois o Python usa indexação baseada em zero. (b) Nosso método pop retorna o menor item, não o maior (chamado de “min heap” em livros didáticos; um “max heap” é mais comum em textos devido à sua adequação para classificação no local).

Esses dois tornam possível visualizar o heap como uma lista regular do Python sem surpresas: heap[0] é o menor item, e heap.sort() mantém o invariante de heap!

Para criar um heap, use uma lista inicializada com [], ou você pode transformar uma lista preenchida em um heap através da função heapify().

As seguintes funções são fornecidas:

heapq.heappush(heap, item)

Coloca o valor item no heap, mantendo o invariante de heap.

heapq.heappop(heap)

Retira e retorna o menor item do heap, mantendo o invariante de heap. Se o heap estiver vazio, a exceção IndexError será levantada. Para acessar o menor item sem retirá-lo, use heap[0].

heapq.heappushpop(heap, item)

Coloca item no heap, depois retira e retorna o menor item do heap. A ação combinada é executada com mais eficiência do que heappush() seguida por uma chamada separada para heappop().

heapq.heapify(x)

Transforma a lista x em um heap, no local, em tempo linear.

heapq.heapreplace(heap, item)

Abre e retorna o menor item da heap e também coloca o novo item. O tamanho do heap não muda. Se o heap estiver vazio, a exceção IndexError será levantada.

Esta operação de uma etapa é mais eficiente que heappop() seguida por heappush() e pode ser mais apropriada ao usar um heap de tamanho fixo. A combinação de retirar/colocar sempre retorna um elemento do heap e o substitui por item.

O valor retornado pode ser maior que o item adicionado. Se isso não for desejado, considere usar heappushpop(). Sua combinação de retirar/colocar retorna o menor dos dois valores, deixando o valor maior no heap.

O módulo também oferece três funções de propósito geral baseadas em heaps.

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

Mescla diversas entradas classificadas em uma única saída classificada (por exemplo, mescla entradas com registro de data e hora de vários arquivos de log). Retorna um iterador sobre os valores classificados.

Semelhante a sorted(itertools.chain(*iterables)) mas retorna um iterável, não puxa os dados para a memória todos de uma vez e assume que cada um dos fluxos de entrada já está classificado (do menor para o maior).

Possui dois argumentos opcionais que devem ser especificados como argumentos nomeados.

key especifica uma função chave de um argumento que é usado para extrair uma chave de comparação de cada elemento de entrada. O valor padrão é None (compare os elementos diretamente).

reverse é um valor booleano. Se definido como True, então os elementos de entrada serão mesclados como se cada comparação fosse invertida. Para obter um comportamento semelhante a sorted(itertools.chain(*iterables), reverse=True), todos os iteráveis devem ser classificados do maior para o menor.

Alterado na versão 3.5: Adicionados os parâmetros opcionais key e reverse.

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

Retorna uma lista com os n maiores elementos do conjunto de dados definido por iterable. key, se fornecido, especifica uma função de um argumento que é usado para extrair uma chave de comparação de cada elemento em iterable (por exemplo, key=str.lower). Equivalente a: sorted(iterable, key=key, reverse=True)[:n].

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

Retorna uma lista com os n menores elementos do conjunto de dados definido por iterable. key, se fornecido, especifica uma função de um argumento que é usado para extrair uma chave de comparação de cada elemento em iterable (por exemplo, key=str.lower). Equivalente a: sorted(iterable, key=key)[:n].

As duas últimas funções têm melhor desempenho para valores menores de n. Para valores maiores, é mais eficiente usar a função sorted(). Além disso, quando n==1, é mais eficiente usar as funções embutidas min() e max(). Se for necessário o uso repetido dessas funções, considere transformar o iterável em um heap real.

Exemplos básicos

Um heapsort pode ser implementado colocando todos os valores em um heap e, em seguida, retirando os menores valores, um de cada 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]

Isto é semelhante a sorted(iterable), mas diferente de sorted(), esta implementação não é estável.

Os elementos de heap podem ser tuplas. Isto é útil para atribuir valores de comparação (como prioridades de tarefas) juntamente com o registro principal que está sendo rastreado:

>>> 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 implementação da fila de prioridade

Uma fila de prioridade é de uso comum para um heap e apresenta vários desafios de implementação:

  • Estabilidade de classificação: como fazer com que duas tarefas com prioridades iguais sejam retornadas na ordem em que foram adicionadas originalmente?

  • A comparação de tuplas quebra para pares (prioridade, tarefa) se as prioridades forem iguais e as tarefas não tiverem uma ordem de comparação padrão.

  • Se a prioridade de uma tarefa mudar, como movê-la para uma nova posição no heap?

  • Ou se uma tarefa pendente precisar ser excluída, como encontrá-la e removê-la da fila?

Uma solução para os dois primeiros desafios é armazenar as entradas como uma lista de 3 elementos, incluindo a prioridade, uma contagem de entradas e a tarefa. A contagem de entradas serve de desempate para que duas tarefas com a mesma prioridade sejam retornadas na ordem em que foram adicionadas. E como não há duas contagens de entradas iguais, a comparação de tuplas nunca tentará comparar diretamente duas tarefas.

Outra solução para o problema de tarefas não comparáveis é criar uma classe wrapper que ignore o item da tarefa e compare apenas o campo de prioridade:

from dataclasses import dataclass, field
from typing import Any

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

Os desafios restantes giram em torno de encontrar uma tarefa pendente e fazer alterações em sua prioridade ou removê-la totalmente. Encontrar uma tarefa pode ser feito com um dicionário apontando para uma entrada na fila.

Remover a entrada ou alterar sua prioridade é mais difícil porque quebraria os invariantes da estrutura de heap. Assim, uma possível solução é marcar a entrada como removida e adicionar uma nova entrada com a prioridade 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')

Teoria

Heaps são arrays para os quais a[k] <= a[2*k+1] e a[k] <= a[2*k+2] para todos k, contando elementos de 0. Para fins de comparação, os elementos inexistentes são considerados infinitos. A propriedade interessante de um heap é que a[0] é sempre seu menor elemento.

O estranho invariante acima pretende ser uma representação de memória eficiente para um torneio. Os números abaixo são k, não 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

Na árvore acima, cada célula k está no topo de 2*k+1 e 2*k+2. Num torneio binário normal que vemos nos desportos, cada célula é a vencedora das duas células que está no topo, e podemos rastrear o vencedor na árvore para ver todos os adversários que teve. Contudo, em muitas aplicações informáticas de tais torneios, não precisamos de traçar a história de um vencedor. Para sermos mais eficientes em termos de memória, quando um vencedor é promovido, tentamos substituí-lo por algo de nível inferior, e a regra passa a ser que uma célula e as duas células que ela cobre contêm três itens diferentes, mas a célula de cima “ganha” sobre as duas células superiores.

Se esse invariante de heap estiver protegido o tempo todo, o índice 0 é claramente o vencedor geral. A maneira algorítmica mais simples de removê-lo e encontrar o “próximo” vencedor é mover algum perdedor (digamos a célula 30 no diagrama acima) para a posição 0 e, em seguida, filtrar esse novo 0 pela árvore, trocando valores, até que o invariante é restabelecido. Isto é claramente logarítmico do número total de itens na árvore. Ao iterar todos os itens, você obtém uma classificação O(n log n).

Um recurso interessante desse tipo é que você pode inserir novos itens com eficiência enquanto a classificação está em andamento, desde que os itens inseridos não sejam “melhores” que o último 0º elemento extraído. Isto é especialmente útil em contextos de simulação, onde a árvore contém todos os eventos recebidos e a condição “vitória” significa o menor tempo programado. Quando um evento agenda outros eventos para execução, eles são agendados para o futuro, para que possam entrar facilmente no heap. Portanto, um heap é uma boa estrutura para implementar escalonadores (foi isso que usei no meu sequenciador MIDI :-).

Várias estruturas para implementação de escalonadores foram extensivamente estudadas, e os heaps são bons para isso, pois são razoavelmente rápidos, a velocidade é quase constante e o pior caso não é muito diferente do caso médio. No entanto, existem outras representações que são globalmente mais eficientes, embora os piores casos possam ser terríveis.

Heaps também são muito úteis em classificações de discos grandes. Provavelmente todos vocês sabem que uma classificação grande implica a produção de “execuções” (que são sequências pré-classificadas, cujo tamanho geralmente está relacionado à quantidade de memória da CPU), seguidas de passagens de fusão para essas execuções, cuja fusão geralmente é muito inteligente. organizado [1]. É muito importante que a classificação inicial produza as execuções mais longas possíveis. Os torneios são uma boa maneira de conseguir isso. Se, usando toda a memória disponível para realizar um torneio, você substituir e filtrar itens que se encaixem na corrida atual, você produzirá corridas que têm o dobro do tamanho da memória para entradas aleatórias e muito melhores para entradas ordenadas de maneira imprecisa.

Além disso, se você gerar o 0º item no disco e obter uma entrada que pode não caber no torneio atual (porque o valor “ganha” sobre o último valor de saída), ele não poderá caber no heap, então o tamanho do heap diminui. A memória liberada poderia ser reutilizada de maneira inteligente e imediata para a construção progressiva de um segundo heap, que cresce exatamente na mesma proporção que o primeiro heap está derretendo. Quando o primeiro heap desaparece completamente, você troca os heaps e inicia uma nova execução. Inteligente e bastante eficaz!

Em uma palavra, heaps são estruturas de memória úteis para conhecer. Eu os uso em alguns aplicativos e acho bom manter um módulo “heap” por perto. :-)

Notas de rodapé