힙과 우선순위 큐 - 가장 급한 것부터

이 장에서 배우는 것

출력할 문서가 먼저 들어왔다고 해서 반드시 먼저 처리해야 하는 것은 아니다. 곧 마감되는 보고서가 뒤늦게 들어오면 그 문서를 먼저 처리할 수도 있다. 이때 필요한 것은 모든 작업을 순서대로 늘어놓는 기능보다, 지금 가장 급한 작업 하나를 빠르게 고르는 기능이다. 우선순위 큐(priority queue)는 이러한 규칙으로 항목을 넣고 꺼내는 자료구조다.

이 장에서는 우선순위 큐를 최소 힙(min heap)으로 직접 만든다. 힙은 부모와 자식 사이의 대소 관계를 유지하는 트리다. 앞 장에서 다룬 균형 트리가 여러 값의 탐색을 위해 높이를 관리했다면, 여기서는 맨 위의 작은 값을 빠르게 꺼내는 데 집중한다. 노드 객체를 서로 연결하지 않고, 고정 크기로 준비한 배열 안에서 트리의 위치를 계산한다.

• 완전 이진 트리의 모양을 배열 인덱스로 나타낸다.
• 삽입할 때 올리기, 삭제할 때 내리기로 힙의 조건을 회복한다.
• 기존 배열을 아래쪽부터 고쳐 힙을 만드는 이유와 비용을 확인한다.
• 같은 우선순위에서 접수 순서를 지키는 작업 스케줄러를 만든다.
• 직접 만든 힙과 표준 라이브러리 heapq의 결과 및 비교 횟수를 대조한다.

문제 상황

문서 처리 프로그램에 작업 다섯 개가 들어왔다고 하자. 작업마다 긴급도를 나타내는 정수가 붙어 있으며, 숫자가 작을수록 먼저 처리한다. 모든 작업은 처리 전에 대기열에 들어간다. 같은 긴급도라면 먼저 접수한 작업을 먼저 처리해야 한다.

숫자가 작을수록 먼저 처리하며, 같은 숫자는 접수 순서로 결정한다

접수 번호	작업	우선순위
0	보고서 변환	4
1	긴급 알림	1
2	미리보기 생성	3
3	마감 문서	1
4	첨부 검사	2

매번 배열 전체를 훑어 가장 작은 값을 찾을 수도 있다. 그러나 작업이 n개라면 한 번 선택하는 데 최대 n개를 살펴야 한다. 하나를 꺼낼 때마다 이 과정을 되풀이하면 전체 처리에 필요한 일이 빠르게 늘어난다. 처음에 한 번 정렬하는 방법도 있지만, 처리 도중 새 작업이 계속 들어오는 상황에서는 삽입과 선택을 함께 고려해야 한다.

여기서 만드는 스케줄러는 대기 중인 작업의 처리 순서만 정한다. 실제로 시간을 재거나 여러 작업을 동시에 실행하지 않는다. 숫자 하나를 비교하고 배열의 자리를 바꾸는 과정에 집중하기 위한 예다. 실행 결과는 현재 시각과 무관하며 언제 실행해도 같다.

배열 안에 완전 이진 트리 담기

각 노드가 자식을 최대 둘까지 가지는 트리를 이진 트리라고 한다. 완전 이진 트리(complete binary tree)는 마지막 층을 제외한 모든 층이 차 있고, 마지막 층은 왼쪽부터 빈틈없이 채워지는 이진 트리다. 전공 수업에서 말하는 ‘완전’은 값의 정렬 상태가 아니라 이 모양의 조건을 뜻한다.

이 모양을 위에서 아래로, 같은 층에서는 왼쪽에서 오른쪽으로 읽으면 배열의 연속된 자리에 담을 수 있다. 루트, 즉 부모가 없는 맨 위 노드를 인덱스 0에 놓는다. 인덱스가 i인 노드의 왼쪽 자식은 2 × i + 1, 오른쪽 자식은 2 × i + 2에 놓인다. 루트가 아닌 노드의 부모는 (i - 1) // 2다. //는 나눗셈의 몫을 구하는 파이썬 연산자다.

배열의 모든 칸이 실제 작업인 것은 아니다. capacity는 준비한 칸 수, size는 현재 작업 수다. 유효한 작업은 인덱스 0부터 size - 1까지만 들어 있다. 자식 인덱스를 계산할 수 있어도 그 값이 size 이상이면 실제 자식은 없다. 그림에서 인덱스 5에 대응하는 자리는 계산 가능하지만 작업이 다섯 개라면 접근해서는 안 된다.

최소 힙은 여기에 값의 조건을 추가한다. 모든 부모는 자신의 자식보다 작거나 같아야 한다. 전공 수업에서는 이를 힙 순서 조건이라고 부른다. 부모를 따라 올라갈수록 값이 커지지 않으므로 루트에는 전체에서 가장 작은 값이 놓인다.

이 조건은 형제끼리의 순서를 요구하지 않는다. 왼쪽 자식이 오른쪽 자식보다 커도 괜찮다. 따라서 힙의 내부 배열을 그대로 읽으면 정렬된 결과가 나오지 않는다. 완성 코드의 힙 배열도 우선순위가 1, 1, 3, 4, 2인 순서로 놓인다. 마지막의 2가 앞의 3보다 작아도 부모보다 작지만 않으면 조건을 만족한다.

배열 위치 계산과 최소 힙의 조건을 구분한다

대상	식 또는 조건	주의할 점
부모 위치	(i - 1) // 2	i가 0보다 클 때만 사용한다
왼쪽 자식 위치	2 × i + 1	size보다 작아야 존재한다
오른쪽 자식 위치	2 × i + 2	왼쪽만 존재할 수도 있다
최소 힙 조건	부모의 키 ≤ 자식의 키	형제 사이의 순서는 정하지 않는다

비교에 사용하는 값을 키(key)라고 한다. 이번 키는 우선순위와 접수 번호를 묶은 튜플이다. 튜플은 여러 값을 순서대로 묶는 파이썬 값이며, 비교할 때 앞 원소부터 확인한다. 따라서 (1, 1)은 (1, 3)보다 작다. 작업 이름은 키에 넣지 않으므로 이름의 문자 순서가 처리 순서에 영향을 주지 않는다.

올리기와 내리기로 조건 회복하기

삽입은 끝에 놓고 위로 올린다

새 작업은 배열의 맨 끝 유효 위치에 놓는다. 이 위치를 택하면 완전 이진 트리의 모양을 유지할 수 있다. 기존 노드 사이의 관계는 그대로이므로 새 작업과 부모 사이만 먼저 확인하면 된다. 새 작업이 부모보다 작으면 두 자리를 바꾸고, 바뀐 위치에서 다시 부모와 비교한다.

이 과정을 올리기라고 하며, 전공 수업에서는 상향 조정(sift up)이라고도 부른다. 루트에 도착하거나 부모보다 작지 않으면 멈춘다. 멈춘 뒤에는 새 작업이 지나온 경로를 포함해 모든 부모와 자식 관계가 힙 조건을 만족한다.

아래 그림은 우선순위만 표시한다. 마지막에 들어온 1이 부모 4보다 작아 한 층 올라간다. 실제 프로그램은 접수 번호도 비교하므로, 이미 앞서 접수된 우선순위 1을 넘어 올라가지는 않는다. 같은 숫자라는 이유만으로 자리를 바꾸지 않는 것이다.

삭제는 마지막 값을 루트로 옮겨 아래로 내린다

최솟값을 꺼낼 때는 루트의 작업을 반환할 값으로 보관한다. 루트를 비운 채로 둘 수 없으므로 마지막 작업을 루트로 옮기고 size를 하나 줄인다. 마지막 칸을 제거하면 트리의 모양은 유지되지만, 새 루트가 자식보다 클 수 있다.

이때는 자식 둘 중 작은 자식을 고른다. 그 자식이 현재 값보다 작으면 두 자리를 바꾸고 계속 내려간다. 이를 내리기 또는 하향 조정(sift down)이라고 한다. 작은 자식을 골라야 새 부모가 양쪽 자식보다 작거나 같아진다. 왼쪽 자식만 있는 경우에는 그 자식 하나와 비교한다.

내리기의 시작 위치 아래에 있는 두 하위 트리가 이미 힙이라는 전제가 중요하다. 하위 트리는 어떤 노드와 그 아래에 연결된 노드 전체를 뜻한다. 삭제 직후에는 루트로 옮긴 값만 문제이므로 이 전제가 성립한다. 자리를 바꿀 때마다 문제 가능성이 있는 위치가 한 층 아래로 이동한다.

연산 횟수로 높이의 의미를 확인한다

노드에서 가장 아래까지 내려가는 간선 수를 그 노드의 높이라고 한다. 간선은 부모와 자식을 잇는 연결이다. 완전 이진 트리는 층이 하나 늘 때 담을 수 있는 노드 수가 대략 두 배가 된다. 따라서 노드가 n개일 때 높이는 log₂ n에 비례한다. log₂ n은 2를 몇 번 곱해야 n에 도달하는지를 나타내는 값이다.

올리기는 한 층에서 키를 한 번 비교한다. 내리기는 자식 선택에 최대 한 번, 부모와 선택한 자식의 비교에 한 번을 쓴다. 따라서 둘 다 최악의 경우 O(log n)이다. O 표기는 입력이 커질 때 필요한 일이 어떤 비율로 늘어나는지 나타낸다. 이번에는 전체 실행 시간을 재는 대신 키 비교 횟수를 직접 센다.

키 비교 한 번을 단위 연산으로 삼는다. 튜플 내부의 정수 비교까지 따로 세지는 않는다. 키의 길이가 항상 둘이므로 이 측정으로도 크기에 따른 증가를 관찰할 수 있다. 배열 접근, 대입, 인덱스 계산은 비교 횟수에 포함하지 않지만 한 층마다 일정 횟수만 수행한다.

배열 전체를 힙으로 만들기

이미 작업이 모여 있다면 빈 힙에 하나씩 넣을 수도 있다. 이 방법의 최악 비용은 O(n log n)이다. 다른 방법은 입력을 배열에 그대로 복사한 뒤, 마지막 부모부터 루트까지 차례로 내리기를 수행하는 것이다. 이 작업을 힙 만들기(heapify)라고 한다.

자식이 없는 노드를 잎이라고 한다. 잎은 비교할 자식이 없으므로 그 자체로 힙이다. 마지막 부모의 인덱스는 size // 2 - 1이다. 여기서 시작해 인덱스를 줄이면, 어떤 부모를 처리할 때 그 자식 쪽 하위 트리는 이미 힙으로 정리되어 있다. 따라서 내리기의 전제를 만족한다.

모든 부모에서 내리기를 수행한다고 해서 비용이 O(n log n)이 되는 것은 아니다. 아래쪽에는 노드가 많지만 내려갈 거리가 짧다. 위쪽은 내려갈 거리가 길지만 노드가 적다. 대략 절반은 잎이라 내려가지 않고, 약 4분의 1은 한 층, 약 8분의 1은 두 층 정도 내려간다. 이 일을 합친 크기는 n에 비례하므로 전체 비용은 O(n)이다.

완성 코드에는 작은 작업 예뿐 아니라 크기를 8, 16, 32로 늘리는 실험도 넣는다. 오름차순 값을 힙으로 만든 뒤 가장 작은 값을 삽입하고 곧바로 꺼낸다. 힙 만들기의 비교 횟수는 각각 7, 15, 31이고, 삽입은 3, 4, 5다. 입력이 두 배가 될 때 한쪽은 거의 두 배로, 다른 쪽은 한 번씩 늘어나는 모습을 확인한다. 이 표본은 일반적인 비용 설명을 눈으로 확인하는 실험이며, 모든 입력에 대한 증명을 대신하지는 않는다.

내부 저장에는 [None] * capacity로 한 번 준비한 리스트만 사용한다. append, pop, 정렬 기능으로 직접 구현을 대신하지 않는다. 용량을 넘으면 예외를 발생시키는 고정 용량 설계다. 마지막 비교 구간에서만 heapq가 관리하는 리스트와 표준 연산을 사용한다. heapq의 배열이 자동으로 커지는 비용과 여기의 고정 용량 설계를 같은 것으로 보아서는 안 된다.

완성 코드

다음 내용을 main.py로 저장한다. 표준 라이브러리만 사용하며 입력이나 난수를 요구하지 않는다. 비교 계수기는 __lt__가 호출된 횟수를 센다. __lt__는 두 객체를 <로 비교할 때 실행되는 특별한 메서드다. 직접 만든 힙과 heapq 모두 같은 작업 객체를 비교하므로 같은 단위로 횟수를 측정한다.

import heapq


class Counter:
    def __init__(self):
        self.value = 0


class Task:
    def __init__(self, priority, order, name, counter):
        self.priority = priority
        self.order = order
        self.name = name
        self.counter = counter

    def __lt__(self, other):
        self.counter.value += 1
        return (self.priority, self.order) < (
            other.priority, other.order
        )


class MinHeap:
    def __init__(self, capacity):
        if capacity < 0:
            raise ValueError("용량은 음수일 수 없다")
        self.data = [None] * capacity
        self.size = 0

    def _up(self, index):
        while index > 0:
            parent = (index - 1) // 2
            if not self.data[index] < self.data[parent]:
                break
            self.data[index], self.data[parent] = (
                self.data[parent], self.data[index]
            )
            index = parent

    def _down(self, index):
        while 2 * index + 1 < self.size:
            left = 2 * index + 1
            right = left + 1
            small = left
            if right < self.size:
                if self.data[right] < self.data[left]:
                    small = right
            if not self.data[small] < self.data[index]:
                break
            self.data[index], self.data[small] = (
                self.data[small], self.data[index]
            )
            index = small

    def push(self, task):
        if self.size == len(self.data):
            raise OverflowError("힙 용량 초과")
        self.data[self.size] = task
        self.size += 1
        self._up(self.size - 1)

    def peek(self):
        if self.size == 0:
            raise IndexError("빈 힙")
        return self.data[0]

    def pop(self):
        if self.size == 0:
            raise IndexError("빈 힙")
        result = self.data[0]
        self.size -= 1
        last = self.data[self.size]
        self.data[self.size] = None
        if self.size > 0:
            self.data[0] = last
            self._down(0)
        return result

    def build(self, items):
        if len(items) > len(self.data):
            raise OverflowError("힙 용량 초과")
        for index in range(self.size):
            self.data[index] = None
        self.size = len(items)
        for index in range(self.size):
            self.data[index] = items[index]
        for index in range(self.size // 2 - 1, -1, -1):
            self._down(index)


RECORDS = (
    (4, "보고서 변환"),
    (1, "긴급 알림"),
    (3, "미리보기 생성"),
    (1, "마감 문서"),
    (2, "첨부 검사"),
)


def make_tasks(counter):
    return tuple(
        Task(priority, order, name, counter)
        for order, (priority, name) in enumerate(RECORDS)
    )


def show_layout(heap):
    text = ""
    for index in range(heap.size):
        if index > 0:
            text += " | "
        task = heap.data[index]
        text += f"{task.priority}:{task.order}"
    return text


def check_scaling():
    for count in (8, 16, 32):
        counter = Counter()
        heap = MinHeap(count + 1)
        items = tuple(
            Task(value, value, "", counter)
            for value in range(count)
        )
        heap.build(items)
        build_count = counter.value

        counter.value = 0
        heap.push(Task(-1, count, "", counter))
        push_count = counter.value

        counter.value = 0
        assert heap.pop().priority == -1
        pop_count = counter.value
        print(
            f"크기 {count}: 만들기={build_count}, "
            f"삽입={push_count}, 삭제={pop_count}"
        )


def check_boundaries():
    heap = MinHeap(1)
    errors = 0
    try:
        heap.pop()
    except IndexError:
        errors += 1

    counter = Counter()
    heap.push(Task(1, 0, "하나", counter))
    try:
        heap.push(Task(2, 1, "둘", counter))
    except OverflowError:
        errors += 1

    assert heap.pop().name == "하나"
    assert heap.size == 0
    assert heap.data[0] is None
    assert errors == 2
    print("빈 힙·용량 초과 검사: 통과")


def main():
    counter = Counter()
    heap = MinHeap(8)
    for task in make_tasks(counter):
        before = counter.value
        heap.push(task)
        print(
            f"삽입 {task.order}: "
            f"비교={counter.value - before}"
        )
    print(f"반복 삽입 비교 합계: {counter.value}")

    counter.value = 0
    heap.build(make_tasks(counter))
    print(f"힙 배열: {show_layout(heap)}")
    print(f"직접 만들기 비교: {counter.value}")

    before = counter.value
    first = heap.peek()
    print(
        f"최소 조회: {first.name}, "
        f"비교={counter.value - before}"
    )

    standard_counter = Counter()
    standard = list(make_tasks(standard_counter))
    heapq.heapify(standard)
    print(f"heapq 만들기 비교: {standard_counter.value}")

    counter.value = 0
    standard_counter.value = 0
    while heap.size > 0:
        before = counter.value
        task = heap.pop()
        expected = heapq.heappop(standard)
        assert (task.priority, task.order, task.name) == (
            expected.priority, expected.order, expected.name
        )
        print(
            f"처리 {task.order}: {task.name}, "
            f"비교={counter.value - before}"
        )

    assert len(standard) == 0
    print(
        f"삭제 비교 합계: 직접={counter.value}, "
        f"heapq={standard_counter.value}"
    )
    print("처리 순서 비교: 일치")
    check_scaling()
    check_boundaries()


if __name__ == "__main__":
    main()

줄별 해설

비교 기준과 계수기를 준비한다

Counter의 value는 처음에 0이다. Task 하나마다 별도 숫자를 두는 대신, 같은 힙에 속한 모든 작업이 같은 계수기 객체를 참조한다. 어느 작업의 __lt__가 실행되든 같은 숫자가 증가한다. 직접 만든 힙용 계수기와 heapq용 계수기는 분리하여 서로의 측정에 영향을 주지 않는다.

Task.__lt__의 첫 줄은 비교 횟수를 증가시킨다. 다음 줄은 우선순위와 접수 번호를 차례로 비교한다. 이 예에서는 enumerate가 접수 번호를 0부터 부여하며, 같은 대기열 안에서 번호가 겹치지 않는다. 실행 도중 새 작업을 받는 프로그램이라면 꺼낸 작업의 번호를 재사용하지 않고 다음 번호를 계속 증가시켜야 한다.

유효 구간 안에서만 이동한다

MinHeap.__init__은 용량만큼 칸을 확보한다. None은 아직 작업이 없다는 표시다. size가 0이므로 이 시점에는 어떤 칸도 힙의 원소가 아니다. _up과 _down의 이름 앞 밑줄은 클래스 내부에서 사용하는 보조 메서드라는 관례를 나타낸다.

_up의 while은 루트에서 부모를 계산하지 않도록 막는다. 비교 결과가 거짓이면 현재 작업은 부모보다 작지 않으므로 break로 반복을 끝낸다. 교환 뒤 index를 parent로 바꾸는 줄이 있어야 같은 작업을 계속 따라간다. 이 갱신이 빠지면 한 번 교환한 뒤 필요한 위쪽 관계를 확인하지 못한다.

_down의 while은 왼쪽 자식의 존재를 검사한다. 완전 이진 트리에서는 왼쪽 없이 오른쪽만 있는 경우가 없으므로 이 조건 하나로 자식 유무를 알 수 있다. 오른쪽은 별도로 범위를 확인한 후 비교한다. small에는 자식의 값이 아니라 선택한 자식의 인덱스를 저장한다.

push는 먼저 용량을 검사한다. 작업을 기록한 다음 size를 늘리므로 새 작업까지 유효 구간에 포함된다. 이후 마지막 인덱스에서 올리기를 시작한다. peek는 원소가 있는지만 검사하고 루트를 반환하므로 키 비교는 0회이며 시간은 O(1)이다.

꺼낸 자리와 다시 만드는 범위를 정리한다

pop은 루트를 result에 보관한 다음 size를 줄인다. 줄어든 size는 이전 마지막 작업의 인덱스이므로 last를 읽는 데 쓸 수 있다. 그 칸에는 None을 넣어 제거한 작업의 참조를 남기지 않는다. 원소가 하나뿐이었다면 이 과정으로 바로 빈 힙이 되며, 루트 대입과 내리기를 수행하지 않는다.

build는 기존 유효 구간을 비운 뒤 새 입력을 복사한다. 호출자의 입력 자체를 재배치하지 않는다. 마지막 range의 끝값 -1은 포함되지 않으므로 인덱스 0까지 처리한다. 원소가 없거나 하나뿐이면 시작값이 -1이어서 반복이 실행되지 않는다. 기존 원소가 m개인 힙을 다시 만드는 경우, 이전 구간을 지우는 비용까지 포함하면 O(m + n)이다. 빈 힙에서 n개로 처음 만들 때는 O(n)이다.

측정 구간을 분리한다

main의 before는 연산 직전 누적 횟수를 보관한다. 연산 뒤 값에서 이를 빼면 해당 연산만의 횟수가 나온다. 반복 삽입 실험을 마친 뒤 계수기를 0으로 돌리고 build를 실행하므로 두 방법의 횟수가 섞이지 않는다.

heapq에는 같은 원본 기록으로 만든 별도 작업 객체를 넣는다. 삭제할 때마다 우선순위, 접수 번호, 작업 이름을 assert로 대조한다. assert는 조건이 거짓이면 실행을 멈춰 검사 실패를 알리는 문장이다. 두 힙의 내부 배열 모양이 같을 필요는 없지만 꺼낸 결과는 같아야 한다.

check_scaling은 크기 변화와 비교 횟수의 관계를 보여 준다. check_boundaries는 빈 힙 삭제, 용량 초과, 마지막 원소 삭제 뒤의 정리를 확인한다. 프로그램 끝의 조건문은 이 파일을 직접 실행했을 때 main을 호출한다. 전체 검사와 출력은 한 번 실행한 뒤 종료된다.

실행 결과

터미널에서 파일이 있는 디렉터리로 이동한 뒤 실행한다. 다음은 Python 3.12의 heapq 비교 절차를 기준으로 한 예상 출력이다. 직접 구현의 횟수는 코드가 정하며, heapq의 세부 비교 횟수는 구현이 달라지면 바뀔 수 있다. 처리 순서의 일치는 그러한 내부 절차와 별개로 검사한다.

python3 main.py

삽입 0: 비교=0
삽입 1: 비교=1
삽입 2: 비교=1
삽입 3: 비교=2
삽입 4: 비교=1
반복 삽입 비교 합계: 5
힙 배열: 1:1 | 1:3 | 3:2 | 4:0 | 2:4
직접 만들기 비교: 6
최소 조회: 긴급 알림, 비교=0
heapq 만들기 비교: 5
처리 1: 긴급 알림, 비교=3
처리 3: 마감 문서, 비교=2
처리 4: 첨부 검사, 비교=1
처리 2: 미리보기 생성, 비교=0
처리 0: 보고서 변환, 비교=0
삭제 비교 합계: 직접=6, heapq=7
처리 순서 비교: 일치
크기 8: 만들기=7, 삽입=3, 삭제=5
크기 16: 만들기=15, 삽입=4, 삭제=7
크기 32: 만들기=31, 삽입=5, 삭제=9
빈 힙·용량 초과 검사: 통과

작은 작업 예에서는 반복 삽입이 5회, 아래에서부터 만들기가 6회다. O(n)인 방법이 작은 모든 입력에서도 비교 횟수가 적다는 뜻은 아니다. 시간 복잡도는 입력이 커질 때의 증가 경향과 상한을 설명한다. 입력 순서와 중간에 멈추는 위치에 따라 실제 횟수는 달라진다.

heapq와 직접 구현의 삭제 횟수가 다른 이유도 결과의 오류가 아니다. 여기서는 매 층에서 현재 값이 멈춰도 되는지 검사한다. 다른 구현은 작은 자식을 먼저 위로 옮기며 내려간 뒤 마지막 값을 위로 조정할 수 있다. 같은 힙 조건을 지켜도 비교 순서와 횟수는 달라질 수 있다. 이 측정만으로 실제 실행 속도의 우열을 판단할 수는 없다.

실무에서 자주 틀리는 것

오른쪽 자식이 항상 있다고 가정한다

마지막 부모에게 왼쪽 자식만 있을 수 있다. 다음 코드는 유효 범위를 벗어난 None을 작업과 비교할 수 있다.

small = left
if self.data[right] < self.data[left]:
    small = right

오른쪽 자식의 존재를 확인한 뒤 비교한다. 파이썬의 and는 앞 조건이 거짓이면 뒤 식을 실행하지 않는다.

small = left
if right < self.size and self.data[right] < self.data[left]:
    small = right

같은 우선순위의 처리 순서를 맡겨 둔다

우선순위만 비교하면 같은 값의 작업이 접수 순서로 나올 근거가 없다. 힙은 같은 키의 입력 순서를 자동으로 보존하는 구조가 아니다.

return self.priority < other.priority

고유한 접수 번호를 두 번째 비교 기준으로 넣는다. 앞에서 접수한 작은 번호가 먼저 나온다.

return (self.priority, self.order) < (
    other.priority, other.order
)

대기 중인 작업의 우선순위를 그대로 바꾼다

필드를 바꾸는 것만으로는 배열 위치가 갱신되지 않는다. 다음 변경 뒤에는 작은 값이 루트 아래에 남을 수 있다.

heap.data[heap.size - 1].priority = 0

이번처럼 원소 수가 작고 변경이 드물다면 남은 항목을 복사하여 다시 힙으로 만드는 방법이 단순하다. 다음 코드는 main 안에서 힙이 비어 있지 않을 때 실행할 수 있다. 다시 만드는 비용은 O(n)이다.

items = tuple(heap.data[i] for i in range(heap.size))
items[-1].priority = 0
heap.build(items)

이 방법은 항목이 힙에 있는 동안 키를 바꾸어도 된다는 뜻이 아니다. 변경 직후 다시 만들기를 마칠 때까지 조회나 삭제를 끼워 넣지 않아야 한다. 빈번한 우선순위 변경에는 항목 위치를 추적하는 별도 설계가 필요하다.

힙 조건을 검사하면서 측정 횟수까지 바꾼다

디버깅을 위해 모든 부모와 자식을 비교한 뒤 그 값을 연산 횟수로 출력하면 검증 비용이 섞인다. 다음 코드는 Task.__lt__를 호출하므로 계수기를 증가시킨다.

for i in range(1, heap.size):
    parent = (i - 1) // 2
    assert not heap.data[i] < heap.data[parent]
print(counter.value)

측정값을 먼저 저장하고 검증에는 작업 객체 대신 키 튜플을 사용한다. 이 검증 자체는 O(n)이지만 측정한 힙 연산의 비교 횟수에는 포함되지 않는다.

measured = counter.value
for i in range(1, heap.size):
    child = heap.data[i]
    parent = heap.data[(i - 1) // 2]
    assert (parent.priority, parent.order) <= (
        child.priority, child.order
    )
print(measured)

한눈에 보기

고정 용량 최소 힙의 비용과 이번 코드의 측정 기준

연산	시간 복잡도	키 비교의 범위	구현의 핵심
최소 조회	O(1)	0회	루트를 읽는다
삽입	최악 O(log n)	올라간 경로에서 층당 최대 1회	끝에 넣고 올린다
최솟값 삭제	최악 O(log n)	내려간 경로에서 층당 최대 2회	마지막 값을 루트로 옮긴다
빈 힙에서 만들기	O(n)	전체에서 n에 비례한다	마지막 부모부터 내린다

용량 C만큼 배열을 미리 준비하므로 저장 공간은 O(C)다. 생성 시 배열을 채우는 비용도 O(C)이며, 이는 삽입 한 번의 비용과 구분한다. 올리기와 내리기는 반복문으로 구현하므로 이동 중에 추가 배열을 만들지 않는다. 반면 예제의 입력 튜플과 비교용 heapq는 검증을 위해 별도 공간을 사용한다.

우선순위 큐는 ‘어떤 항목을 먼저 꺼내는가’라는 사용 규칙이고, 힙은 그 규칙을 구현하는 방법이다. 최소 힙의 강점은 전체 정렬 상태를 유지하지 않으면서 가장 작은 항목을 빠르게 제공한다는 데 있다. 다음 장에서는 작은 값의 순서 대신 문자열의 앞부분을 함께 저장하여 찾는 구조를 다룬다.

연습 문제

1. check_scaling의 크기 목록에 64를 추가하라. 만들기, 삽입, 삭제 비교 횟수를 실행 전에 예상하고 실제 출력과 대조하라.
2. 완성 코드의 삽입 실험을 우선순위 1, 2, 3, 4, 5 순서로 바꾸었을 때 총 비교 횟수를 예상하라. 접수 번호는 0부터 증가시킨다. 내림차순 5, 4, 3, 2, 1과도 비교하라.
3. pop으로 모든 작업을 꺼낸 뒤에도 고정 크기 배열의 길이가 유지되는지 확인하라. 모든 칸이 None인지 검사하는 코드도 작성하라.
4. 내리기에서 작은 자식 대신 왼쪽 자식만 고르면 왜 틀리는지 설명하라. 우선순위 1, 4, 2, 5, 6, 3을 접수 순서대로 build한 뒤 첫 pop을 실행하여 확인하라.

정답과 해설

1. 크기 64에서는 만들기 63회, 삽입 6회, 삭제 11회다. 오름차순 입력은 처음부터 힙 조건을 만족한다. 만들기는 각 부모에서 한 번씩 자식 관계를 확인하고 멈추며, 합계는 이 입력에서 n - 1이다. 삽입한 -1은 여섯 층을 올라간다. 삭제에서는 다섯 층에서 두 번씩, 마지막 왼쪽 자식과 한 번 비교한다.

   for count in (8, 16, 32, 64):
       # check_scaling의 기존 반복문 본문을 그대로 둔다.
       ...
   

   위의 ... 자리는 새로운 구현을 뜻하지 않는다. 완성 코드에 있는 for문의 크기 튜플만 바꾸면 다음 줄이 추가된다.

   크기 64: 만들기=63, 삽입=6, 삭제=11
   

2. 오름차순은 첫 삽입을 제외하면 매번 부모와 한 번 비교하고 멈추므로 총 4회다. 내림차순은 각 새 값이 루트까지 올라가며 0, 1, 1, 2, 2회, 총 6회를 비교한다. 다음 함수를 main에서 호출하면 확인할 수 있다.

   def compare_orders():
       for values in ((1, 2, 3, 4, 5), (5, 4, 3, 2, 1)):
           counter = Counter()
           heap = MinHeap(5)
           for order, priority in enumerate(values):
               heap.push(Task(priority, order, "", counter))
           print(counter.value)
   

   출력은 차례로 4와 6이다. 같은 크기라도 입력 순서에 따라 실제 횟수는 달라진다.

3. main에서 처리 순서 비교를 마친 뒤 다음 검사를 추가한다. 배열의 길이는 8이고 size는 0이다. 작업 수가 줄어도 확보한 칸 수는 유지된다.

   assert len(heap.data) == 8
   assert heap.size == 0
   for value in heap.data:
       assert value is None
   print("배열 정리 검사: 통과")
   

   이 검사는 각 삭제가 마지막 유효 칸을 None으로 정리했는지도 함께 확인한다. 배열 길이와 원소 수를 같은 뜻으로 사용하면 이 차이를 놓치기 쉽다.

4. 주어진 배열은 이미 최소 힙이다. 루트 1을 꺼내면 마지막 값 3이 루트로 온다. 왼쪽 자식 4만 보면 3이 더 작으므로 멈추지만, 오른쪽 자식 2는 3보다 작다. 다음 코드는 올바른 구현이 첫 삭제 뒤 2를 루트에 두는지 확인한다.

   def check_right_child():
       counter = Counter()
       values = (1, 4, 2, 5, 6, 3)
       items = tuple(
           Task(value, order, "", counter)
           for order, value in enumerate(values)
       )
       heap = MinHeap(6)
       heap.build(items)
       assert heap.pop().priority == 1
       assert heap.peek().priority == 2
       print("오른쪽 자식 선택 검사: 통과")
   

   작은 자식을 고르는 규칙은 비교 횟수를 줄이기 위한 선택 사항이 아니다. 교환 뒤 부모가 양쪽 자식보다 작거나 같아야 한다는 힙 조건을 지키기 위해 필요하다.