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

"""Three merge-work-file polyphase sorter; educational integer-line file format.

Original and destination files are additional files. Source is read twice;
only one sorted chunk, two merge front values and three stream offsets are kept.
Run header is an integer length; length zero is one dummy run slot.
"""
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:
                chunk.sort()
                yield chunk
                chunk = []
        if chunk:
            chunk.sort()
            yield 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("truncated run file")
    return int(line)


def _merge_one_run(left, right, output):
    left_count, right_count = _read_integer(left), _read_integer(right)
    if left_count < 0 or right_count < 0:
        raise ValueError("negative run length")
    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 polyphase_sort_file(source, destination, chunk_size=3, *, work_dir=None):
    """Sort integer lines with three temporary merge files; return phase counters.

    Source must remain unchanged during both streaming passes. Destination is a
    disposable result path. Exact int chunk size >= 1; bool/float are rejected.
    Header/offset counts refer to logical runs, including length-zero dummies.
    """
    source, destination = Path(source), Path(destination)
    if type(chunk_size) is not int or chunk_size < 1:
        raise ValueError("positive integer chunk size required")
    if source.resolve() == destination.resolve() or (source.exists() and destination.exists() and source.samefile(destination)):
        raise ValueError("distinct paths required")
    value_count = 0
    with source.open(encoding="utf-8") as stream:
        for line in stream:
            int(line)
            value_count += 1
    real_runs = (value_count + chunk_size - 1) // chunk_size
    if real_runs <= 1:
        counts = [real_runs, 0, 0]
    else:
        large, small = 1, 1
        while large + small < real_runs:
            large, small = large + small, large
        counts = [large, small, 0]
    dummy_runs = sum(counts) - real_runs
    initial = counts.copy()
    io = {"scanReadValues": value_count,
          "distributionReadValues": value_count,
          "distributionWrittenValues": value_count,
          "mergeReadValues": 0, "mergeWrittenValues": 0,
          "finalReadValues": 0, "finalWrittenValues": 0}
    phases = []
    with TemporaryDirectory(prefix="polyphase-", dir=work_dir) as temporary:
        paths = [Path(temporary) / f"file-{index}.txt" for index in range(3)]
        for path in paths:
            path.write_text("", encoding="utf-8")
        runs = _chunks(source, chunk_size)
        for index in (0, 1):
            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))
        # Finish the generator so its source stream closes and excess is noticed.
        try:
            next(runs)
        except StopIteration:
            pass
        else:
            raise ValueError("source changed between counting and run generation")
        offsets = [0, 0, 0]
        while sum(count > 0 for count in counts) > 1:
            inputs = [index for index, count in enumerate(counts) if count]
            output_index = next(index for index, count in enumerate(counts) if not count)
            first, second = inputs
            pairs = min(counts[first], counts[second])
            before = counts.copy()
            phase_values = 0
            # Readers close before a consumed file is reused as the next output.
            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):
                    phase_values += _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
            io["mergeReadValues"] += phase_values
            io["mergeWrittenValues"] += phase_values
            phases.append({"before": before, "inputs": inputs,
                           "output": output_index, "mergedSlots": pairs,
                           "after": counts.copy(), "readValues": phase_values,
                           "writtenValues": phase_values})
        final_index = next((index for index, count in enumerate(counts) if count), None)
        with destination.open("w", encoding="utf-8") as output:
            if final_index is not None:
                with paths[final_index].open(encoding="utf-8") as stream:
                    stream.seek(offsets[final_index])
                    length = _read_integer(stream)
                    if length != value_count:
                        raise ValueError("final run length does not preserve input count")
                    for _ in range(length):
                        output.write(f"{_read_integer(stream)}\n")
                    io["finalReadValues"] = io["finalWrittenValues"] = length
    return {"valueCount": value_count, "realRuns": real_runs,
            "dummyRuns": dummy_runs, "initial": initial, "phases": phases,
            "finalFile": final_index, "io": io}


if __name__ == "__main__":
    private_root = Path(__file__).resolve().parent / "private-temp"
    private_root.mkdir(exist_ok=True)
    with TemporaryDirectory(prefix="polyphase-example-", dir=private_root) as temporary:
        folder = Path(temporary)
        source, output = folder / "input.txt", folder / "output.txt"
        source.write_text("8\n1\n6\n2\n9\n3\n7\n", encoding="utf-8")
        before = source.read_text(encoding="utf-8")
        report = polyphase_sort_file(source, output, 1, work_dir=folder)
        print(report["initial"])
        for phase in report["phases"]:
            print(phase["after"], phase["readValues"], phase["writtenValues"])
        print([int(line) for line in output.read_text(encoding="utf-8").splitlines()])
        print(source.read_text(encoding="utf-8") == before)

[5, 3, 0]
[2, 0, 3] 5 5
[0, 2, 1] 5 5
[1, 1, 0] 4 4
[0, 0, 1] 7 7
[1, 2, 3, 6, 7, 8, 9]
True