선택 심화: 세 파일 polyphase 병합의 분배와 역할 교대

외부 정렬의 다중 병합은 런 파일을 모두 한꺼번에 열었습니다. 하지만 한 번에 쓸 수 있는 파일이 몇 개로 정해져 있다면 어떻게 해야 할까요? 컴퓨터가 자료를 자기 테이프에 저장하던 시절에는 테이프 장치의 수가 곧 동시에 쓸 수 있는 파일의 수였고, 이 제약 아래에서 읽고 쓰는 양을 줄이는 여러 방법이 연구되었습니다. 1960년 R. L. Gilstad가 발표한 polyphase 병합은 그 가운데 하나로, 런을 여러 개씩 담은 작업 파일 세 개만으로 병합을 끝냅니다. 이 글에서는 polyphase 병합을 실제 파일로 실행해 보고, 런을 처음에 어떤 개수로 나누어 담아야 하는지, 왜 그렇게 하면 읽고 쓰는 양이 줄어드는지 설명합니다.

파일 셋으로 균형 있게 병합하면

작업 파일이 F0, F1, F2 셋이라고 합시다. 정렬할 입력 파일과 최종 출력 파일은 따로 있고, 셋은 런을 담아 두는 작업용입니다. 가장 쉬운 방법은 런을 F0과 F1에 번갈아 나누어 담고, 두 파일의 런을 하나씩 짝지어 병합해 F2에 쓰는 것입니다. 그러면 런의 수가 절반으로 줄지만, 다음 병합을 하려면 F2의 런들을 다시 F0과 F1로 나누어 옮겨야 합니다. 이 옮기는 일은 병합이 아니라 단순한 복사인데도 자료 전체를 한 번 더 읽고 써야 합니다.

polyphase 병합은 이 복사를 없앱니다. 한 단계에서는 런이 남아 있는 두 파일을 입력으로, 비어 있는 파일을 출력으로 씁니다. 두 입력 파일의 런을 하나씩 짝지어 병합하다가 한쪽 입력 파일의 런이 다 떨어지면 단계가 끝나고, 그 다 쓴 파일이 다음 단계의 출력이 됩니다. 다른 입력 파일에 남은 런과 방금 쓴 출력 파일의 런은 그대로 다음 단계의 입력이 됩니다. 이렇게 단계마다 세 파일의 역할이 돌아가며 바뀌므로, 런을 다시 나누어 옮길 필요가 없습니다.

런의 수를 피보나치 수로 나누어 담는다

이 방법이 끝까지 잘 돌아가려면 처음에 런을 담는 개수가 중요합니다. 단계마다 두 입력 파일 가운데 런이 적은 쪽이 다 떨어질 때까지 병합하므로, 런이 많은 파일에 L개, 적은 파일에 S개가 있다면 S쌍을 병합합니다. 그러면 많은 쪽에는 L - S개가 남고, 출력 파일에는 S개가 생겨서, 다음 단계의 두 입력은 S개와 L - S개가 됩니다. L과 S가 연속한 피보나치 수라면 L - S는 S 바로 앞의 피보나치 수이므로, 다음 단계도 연속한 피보나치 수가 됩니다. 이 관계가 1과 1까지 이어지다 마지막 병합에서 런 하나가 남습니다.

런이 8개라면 연속한 피보나치 수 5와 3으로 나누어 F0에 5개, F1에 3개를 담습니다. 세 파일의 런 수는 다음과 같이 바뀝니다.

5, 3, 0 → 2, 0, 3 → 0, 2, 1 → 1, 1, 0 → 0, 0, 1

첫 단계는 F0과 F1에서 3쌍을 병합해 F2에 3개를 쓰고, F1이 비었으므로 다음 출력이 됩니다. 둘째 단계는 F0의 남은 2개와 F2의 3개에서 2쌍을 병합해 F1에 씁니다. 이렇게 네 단계 만에 런 하나가 남습니다. 표의 0은 파일을 지웠다는 뜻이 아니라 그 파일에 아직 읽지 않은 런이 없다는 뜻입니다.

가상 런으로 모자란 자리를 채운다

실제 런의 수가 늘 두 피보나치 수의 합이 되지는 않습니다. 런이 7개라면 합이 7 이상인 가장 작은 연속 피보나치 수 5와 3을 고르고, 모자란 1개는 값이 하나도 없는 길이 0의 가상 런(dummy run)으로 채웁니다. 가상 런은 런 수를 맞추는 자리 하나로 계산에 참여하지만 값은 더하지도 빼지도 않습니다. 가상 런과 실제 런을 병합하면 실제 런이 그대로 나오고, 가상 런 둘을 병합하면 다시 길이 0인 런이 됩니다. 값이 0 하나인 런과 가상 런은 다른 것이라는 점에 주의해야 합니다.

런을 구분하는 파일 형식과 읽던 위치

파일 하나에 런이 여러 개 들어가므로, 어디서 한 런이 끝나고 다음 런이 시작하는지 알 수 있어야 합니다. 값이 줄어드는 곳을 런의 끝으로 볼 수도 있지만, 그러면 [3, 8] 다음에 [9]처럼 이어서 정렬된 두 런을 구분하지 못합니다. 그래서 이 글의 작업 파일은 각 런을 "길이 한 줄, 그다음 그 길이만큼의 값 줄"로 씁니다. [3, 8]은 2, 3, 8의 세 줄이고, 가상 런은 0 한 줄, 값이 0 하나인 런은 1과 0의 두 줄입니다.

또 한 단계에서 다 쓰지 못한 파일은 다음 단계에서 읽던 자리부터 이어 읽어야 합니다. 파일 객체의 tell()은 지금 읽고 있는 위치를 나타내는 값을 돌려주고, seek(위치)는 그 위치로 돌아갑니다. 단계가 끝날 때 두 입력 파일의 위치를 tell()로 저장해 두었다가, 다음에 그 파일을 입력으로 쓸 때 seek()로 이어서 읽습니다. 출력으로 쓰는 파일은 읽지 않은 런이 없는 파일뿐이므로, "w" 모드로 열어 처음부터 덮어써도 잃는 값이 없습니다. 텍스트 파일에서 tell()이 돌려주는 값은 Python이 위치를 기억하는 표식일 뿐이라, 몇 바이트째인지로 해석하면 안 됩니다.

손으로 따라가기

입력이 8, 1, 6, 2, 9, 3, 7이고 청크 크기가 1이면 실제 런은 값 하나짜리 7개입니다. F0의 맨 앞에 가상 런을 넣고 입력 순서대로 나머지를 채우면 F0은 ∅, [8], [1], [6], [2], F1은 [9], [3], [7]이 됩니다. ∅는 가상 런입니다.

첫 단계가 끝났을 때 F0에는 [6]과 [2]가 아직 읽히지 않은 채 남아 있고, 저장한 위치는 [6]의 길이 줄 바로 앞입니다. 둘째 단계는 F0을 그 위치부터 읽습니다. F1은 다 읽었으므로 둘째 단계의 출력으로 비워서 씁니다. 둘째 단계 뒤에도 F2의 [1, 7]은 읽지 않은 채 남아 셋째 단계의 입력이 되므로, 이때 F2를 비우면 안 됩니다.

코드로 옮기기

긴 코드는 역할별로 나누어 읽으면 됩니다. _chunks는 입력을 청크 크기씩 읽어 정렬한 런을 하나씩 내놓는 생성기이고, _write_run과 _read_integer는 길이 줄과 값 줄을 쓰고 읽습니다. _merge_one_run은 두 입력 파일에서 런 하나씩을 길이 줄에 적힌 만큼만 읽어 병합합니다. polyphase_sort_file은 런 수를 세어 피보나치 분배를 정하고, 단계마다 세 파일의 런 수와 읽던 위치를 관리하며, 병합한 값의 개수를 기록해 돌려줍니다.

from pathlib import Path
from tempfile import TemporaryDirectory


def _chunks(source, size):
    with source.open(encoding="utf-8") as stream:
        chunk = []
        for line in stream:
            chunk.append(int(line))
            if len(chunk) == size:
                yield sorted(chunk)
                chunk = []
        if chunk:
            yield sorted(chunk)


def _write_run(stream, values):
    stream.write(f"{len(values)}\n")
    for value in values:
        stream.write(f"{value}\n")


def _read_integer(stream):
    line = stream.readline()
    if not line:
        raise ValueError("런 파일이 중간에 끝났습니다")
    return int(line)


def _merge_one_run(left, right, output):
    left_count, right_count = _read_integer(left), _read_integer(right)
    length = left_count + right_count
    output.write(f"{length}\n")
    first = _read_integer(left) if left_count else None
    second = _read_integer(right) if right_count else None
    while first is not None or second is not None:
        if second is None or (first is not None and first <= second):
            output.write(f"{first}\n")
            left_count -= 1
            first = _read_integer(left) if left_count else None
        else:
            output.write(f"{second}\n")
            right_count -= 1
            second = _read_integer(right) if right_count else None
    return length


def fibonacci_counts(real_runs):
    if real_runs <= 1:
        return [real_runs, 0, 0]
    large, small = 1, 1
    while large + small < real_runs:
        large, small = large + small, large
    return [large, small, 0]


def polyphase_sort_file(source, destination, chunk_size):
    source, destination = Path(source), Path(destination)
    if type(chunk_size) is not int or chunk_size < 1:
        raise ValueError("청크 크기는 1 이상인 정수여야 합니다")
    if source.resolve() == destination.resolve():
        raise ValueError("원본을 덮어쓰지 않도록 출력 경로를 다르게 정해야 합니다")
    with source.open(encoding="utf-8") as stream:
        value_count = sum(1 for _ in stream)
    real_runs = (value_count + chunk_size - 1) // chunk_size
    counts = fibonacci_counts(real_runs)
    dummy_runs = sum(counts) - real_runs
    report = {"initial": counts.copy(), "dummy": dummy_runs, "phases": []}
    with TemporaryDirectory() as folder:
        paths = [Path(folder) / f"F{index}.txt" for index in range(3)]
        runs = _chunks(source, chunk_size)
        for index in range(3):
            with paths[index].open("w", encoding="utf-8") as stream:
                for slot in range(counts[index]):
                    if index == 0 and slot < dummy_runs:
                        _write_run(stream, [])
                    else:
                        _write_run(stream, next(runs))
        leftover = next(runs, None)
        runs.close()
        if leftover is not None:
            raise ValueError("두 번 읽는 사이에 원본 파일이 바뀌었습니다")
        offsets = [0, 0, 0]
        while sum(count > 0 for count in counts) > 1:
            first, second = [index for index, count in enumerate(counts) if count > 0]
            output_index = counts.index(0)
            pairs = min(counts[first], counts[second])
            moved = 0
            with paths[first].open(encoding="utf-8") as left, \
                    paths[second].open(encoding="utf-8") as right, \
                    paths[output_index].open("w", encoding="utf-8") as output:
                left.seek(offsets[first])
                right.seek(offsets[second])
                for _ in range(pairs):
                    moved += _merge_one_run(left, right, output)
                offsets[first], offsets[second] = left.tell(), right.tell()
            offsets[output_index] = 0
            counts[first] -= pairs
            counts[second] -= pairs
            counts[output_index] = pairs
            report["phases"].append((counts.copy(), moved))
        with destination.open("w", encoding="utf-8") as output:
            if real_runs > 0:
                final_index = next(index for index, count in enumerate(counts) if count > 0)
                with paths[final_index].open(encoding="utf-8") as stream:
                    stream.seek(offsets[final_index])
                    for _ in range(_read_integer(stream)):
                        output.write(f"{_read_integer(stream)}\n")
    return report


with TemporaryDirectory() as folder:
    source = Path(folder) / "input.txt"
    destination = Path(folder) / "output.txt"
    source.write_text("8\n1\n6\n2\n9\n3\n7\n", encoding="utf-8")
    report = polyphase_sort_file(source, destination, chunk_size=1)
    print(report["initial"], "가상 런", report["dummy"])
    for counts, moved in report["phases"]:
        print(counts, moved)
    print([int(line) for line in destination.read_text(encoding="utf-8").split()])

[5, 3, 0] 가상 런 1
[2, 0, 3] 5
[0, 2, 1] 5
[1, 1, 0] 4
[0, 0, 1] 7
[1, 2, 3, 6, 7, 8, 9]

함수는 입력을 두 번 읽습니다. 처음에는 값의 개수만 세어 런의 수와 분배를 정하고, 두 번째로 _chunks가 청크를 하나씩 정렬해 작업 파일에 씁니다. 그래서 메모리에는 청크 하나만 올라옵니다. 분배가 끝나면 next(runs, None)으로 남은 런이 없는지 확인합니다. 남은 런이 없으면 이 호출에서 생성기가 끝까지 실행되어 with 블록을 벗어나므로 입력 파일이 이때 닫히고, 런이 남았다면 이어지는 runs.close()가 멈춰 있던 생성기를 끝내 입력 파일을 닫습니다. 두 번 읽는 사이에 입력 파일이 바뀌어 런이 남았다면 오류를 냅니다.

병합 반복의 counts.index(0)은 읽지 않은 런이 없는 파일, 곧 이번 단계의 출력 파일을 찾습니다. 마지막에 런이 남는 파일은 런의 수에 따라 다르므로 F2로 정해 두지 않고, 런이 남은 파일을 찾아 그 런을 출력 파일로 옮깁니다. 출력의 각 줄은 단계가 끝난 뒤 F0, F1, F2의 런 수와 그 단계에서 병합한 값의 개수입니다.

읽고 쓰는 값의 개수

단계마다 병합한 값은 5, 5, 4, 7개로, 모두 21개를 읽고 21개를 썼습니다. 앞의 세 단계는 자료 전체가 아니라 일부만 다루었다는 점이 중요합니다. 같은 일곱 런을 파일 셋으로 균형 있게 병합했다면, 병합 세 번과 그 사이에 런을 다시 두 파일로 나누어 옮기는 두 번이 필요해서, 매번 값 7개씩 모두 35개를 읽고 썼을 것입니다. polyphase 병합은 다시 나누어 옮기는 일을 없애 이 차이를 만듭니다.

이 밖에 값의 개수를 세는 첫 읽기 7개, 처음 분배의 읽기 7개와 쓰기 7개, 마지막에 결과를 출력 파일로 옮기는 읽기 7개와 쓰기 7개가 있어, 전체로는 값을 42번 읽고 35번 씁니다. 여기서 센 것은 값 줄의 개수이고, 길이 줄이나 실제로 디스크에 접근한 횟수와는 다릅니다. 파일 객체는 버퍼에 모아서 읽고 쓰기 때문입니다.

왜 맞고 왜 끝나는가

처음에 쓴 실제 런은 모두 정렬되어 있고, 가상 런은 값이 없는 정렬된 런으로 볼 수 있습니다. _merge_one_run은 두 런의 맨 앞 값 가운데 작은 것을 쓰므로 병합 정렬의 병합과 같은 이유로 정렬된 런을 만들고, 두 길이의 합을 길이 줄에 쓴 뒤 값도 정확히 그만큼 쓰므로 값을 잃거나 더하지 않습니다.

단계가 끝날 때 다 쓰지 못한 파일은 읽던 위치를 저장해 두고, 다 쓴 파일만 다음 출력으로 덮어쓰므로 아직 읽지 않은 런을 지우는 일이 없습니다. 단계마다 두 입력에서 S쌍을 병합해 S개를 만드므로 전체 런의 수가 S만큼 줄고, 앞에서 본 것처럼 런 수가 연속한 피보나치 수의 관계를 유지하다가 마지막에 하나만 남아 끝납니다. 그 하나는 모든 값을 담은 정렬된 런입니다.

비용과 한계

청크 크기가 M이고 값이 N개이면, 처음에 청크를 정렬하는 데 O(N log M)이 듭니다. 런이 R개일 때 피보나치 수는 단계마다 약 1.618배씩 줄어들므로 단계는 O(log R)번이고, 한 단계에서 병합하는 값은 많아야 N개이므로 병합에서 옮기는 값은 O(N log R) 이하입니다. 메모리에는 청크 하나, 병합할 때의 맨 앞 값 두 개, 세 파일의 읽던 위치만 있으면 됩니다.

이 구현은 원리를 보여 주기 위한 것이라 정수만 다루며, 같은 키의 서로 다른 기록이 순서를 지키는지는 다루지 않습니다. 실행 도중에 디스크가 가득 차거나 프로그램이 멈추면 출력 파일이 일부만 쓰인 채 남을 수 있으므로, 실제로 쓴다면 임시 이름으로 끝까지 쓴 뒤 바꾸어 놓는 식의 처리가 더 필요합니다. 또 파일을 여러 개 동시에 열 수 있는 오늘날의 환경에서는 외부 정렬의 다중 병합이 더 간단하고 빠른 경우가 많습니다.

연습 문제

1. 가상 런과 병합한 값의 개수

첫 단계에서 가상 런과 [9]를 병합할 때 읽고 쓰는 값은 몇 개이고, 출력 런의 길이 줄에는 무엇이 쓰이나요? 이 병합을 한 쌍으로 세는 이유는 무엇인가요?

2. 둘째 단계 뒤에 F2를 비워도 될까

둘째 단계가 끝난 뒤 F2를 비워도 될까요? 다음 단계에서 세 파일의 역할이 무엇인지로 설명하세요.

3. 런이 12개일 때의 분배

실제 런이 12개라면 처음에 세 파일에 런을 몇 개씩 담고, 가상 런은 몇 개가 필요하며, 병합은 몇 단계를 거치나요?

런의 분배와 가상 런의 배경은 Gordon College의 외부 정렬 강의에서, 버퍼와 파일 입출력의 비용 모델은 UC Berkeley CS186의 정렬 노트에서 더 읽을 수 있습니다. 다음 장의 종합: 지도 요구에서 경로와 검사까지에서는 지금까지 배운 알고리즘을 하나의 문제에 함께 적용합니다.