2.38. 우선순위 큐

우선순위 큐는 전체를 정렬하지 않고도 항목들 중 가장 작은(또는 가장 큰) 것이 무엇인지 항상 알고 있는 컨테이너입니다. heapq 는 저장소로 일반 list 를 사용하고 가장 작은 항목을 인덱스 0에 유지하기 위해 최소 힙(min-heap) 이라는 트리 기반 알고리즘을 사용하여 MicroPython에서 이를 구현합니다.

2.38.1. 개념 모델

최소 힙은 모든 인덱스의 항목이 인덱스 2*i + 1 및 2*i + 2 의 항목보다 작거나 같도록 배열된 리스트입니다. 이 배열은 세 가지 연산을 저렴하게 만듭니다:

  • 가장 작은 항목 찾기 – heap[0] 에 있습니다. 스캔 없이 단일 인덱스 조회만 하면 됩니다.

  • 항목 추가 – 항목이 트리를 따라 위로 올라가며 레벨당 하나의 부모와 비교됩니다. 천 개의 항목은 약 열 번의 비교, 백만 개는 약 스무 번의 비교로 끝납니다.

  • 가장 작은 항목 제거 – 같은 과정을 역방향으로 진행하며 같은 정도의 적은 비교만 필요합니다.

리스트는 정렬되지 않습니다. 이를 반복하면 임의의 순서로 항목이 나옵니다. heap[0] 만 최솟값임이 보장됩니다.

2.38.2. MicroPython의 세 가지 함수

MicroPython의 heapq 는 정확히 세 가지 함수를 제공합니다:

  • heapq.heappush() – 힙 불변식을 그대로 유지하면서 항목을 삽입합니다.

  • heapq.heappop() – 가장 작은 항목을 제거하고 반환합니다.

  • heapq.heapify() – 기존 리스트를 제자리에서 힙으로 재배열합니다.

이것이 전부입니다. CPython의 heappushpop, heapreplace, nlargest, nsmallest, merge 는 존재하지 않습니다. 아래 패턴들은 일반적인 것들을 직접 만드는 방법을 보여줍니다.

새로 시작하기:

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

기존 데이터에서 시작하기:

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

heapify는 각 항목을 한 번씩만 건드리며 같은 항목들을 하나씩 푸시하는 것보다 눈에 띄게 빠릅니다.

2.38.3. 스트림에서 상위 N개 구하기

우선순위 큐의 자연스러운 활용: 한 번만 읽을 수 있는 스트림에서 가장 큰 N 개의 항목을 추적합니다.

요령은 크기가 N 인 최소 힙을 유지하는 것입니다. 새 값이 들어올 때마다 푸시하고, 힙이 N 을 넘어 커지면 가장 작은 것을 버립니다. 가장 작은 것은 항상 heap[0] 에 있으므로, 단 한 번의 비교로 들어오는 각 항목을 유지할지 버릴지 결정할 수 있습니다:

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)

힙은 N 을 넘어 커지지 않으므로, 스트림이 아무리 길어도 들어오는 각 값은 동일한 정도의 적은 비교 비용만 듭니다. 마지막 sorted 만이 그 비용이 직접적으로 N 에 의존하는 유일한 단계입니다.

2.38.4. 예약된 이벤트

(deadline, payload) 튜플의 우선순위 큐는 이벤트 스케줄러의 표준 형태입니다. 튜플은 사전식으로 비교되므로, 가장 작은 deadline을 가진 튜플이 자동으로 맨 위로 떠오릅니다:

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)

두 이벤트가 동일한 deadline을 공유하면 튜플 비교는 두 번째 요소로 넘어갑니다. 자연스럽게 비교 가능하지 않은 payload의 경우, 안정적인 순서를 강제하기 위해 튜플의 두 번째 요소로 카운터를 추가합니다:

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

이제 가장 작은 deadline이 먼저 처리되는 동작은 그대로 유지되고, 동점은 카운터로 결정되어 먼저 예약된 작업이 먼저 실행되며, task 는 비교 가능할 필요가 없습니다.

2.38.5. heapq 가 제공하지 않는 것

동작할 것처럼 보이지만 그렇지 않은 몇 가지 패턴:

  • for x in heap 으로 반복해도 정렬된 출력이 나오지 않습니다 – 리스트의 기본 순서는 힙의 트리 레이아웃입니다. 힙을 순서대로 소비하려면 비워질 때까지 heappop 하세요.

  • 이미 힙에 있는 항목의 우선순위를 변경 할 방법은 없습니다. 표준적인 우회 방법은 새 우선순위를 새 항목으로 푸시하고 기존 것을 취소된 것으로 표시(팝될 때 건너뛰기)하는 것입니다.

  • 힙은 항목의 복사본이 아니라 참조를 저장합니다. 항목을 푸시한 후 변경하면 힙 불변식이 깨질 수 있습니다. 비교 키로는 불변 값(숫자, 튜플)을 사용하세요.

일반 리스트 위의 최소 힙은 소형 임베디드 장치에서 “가장 작은 것은 무엇인가” 또는 “가장 긴급한 것은 무엇인가”라는 모든 질문에 적합한 도구입니다. 세 개의 함수, 하나의 리스트면 나머지는 힙이 알아서 처리합니다.