Python의 multiprocessing으로 CPU 집약적 작업 병렬화하기

1. 개요

Python 코드가 이미지 처리, 체크섬 계산, 숫자 시뮬레이션 같은 계산이 많을 때, 스레드를 사용해도 거의 빨라지지 않는 것을 보게 됩니다. 원인은 GIL(Global Interpreter Lock)입니다. 이 글은 이 문제를 간단히 설명합니다. 그 후 multiprocessing 모듈과 ProcessPoolExecutor를 사용해 작업을 여러 프로세스로 나누는 방법을 안내합니다. 이 글을 읽고 나면, CPU 집약적 작업을 병렬화하는 방법, 이를 테스트하는 법, 그리고 흔한 실수를 피하는 방법을 알게 될 것입니다.

모든 예제는 Python 3.12 이상을 사용하며 표준 라이브러리만 필요합니다. 추가 설치는 필요 없습니다.

2. 스레딩이 도움이 되지 않는 이유

2.1. CPU 집약적 작업

순수 Python으로 작성한 소수 개수 세기 함수를 사용합니다. 이는 전형적인 CPU 집약적 작업입니다:

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

함수는 [start, end) 범위의 각 숫자를 순회하고 2부터 그 제곱근까지의 숫자로 나누어떨어지는지 확인합니다. 함수는 파일을 읽지 않고, 네트워크를 호출하지도 않습니다. 단순히 계산만 합니다.

2.2. 시간 측정: 순차 실행과 스레드

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} primes, {time.perf_counter() - t0:.2f}s")

RANGES는 [0, 1_000_000) 범위를 8개 부분으로 나눕니다. run_threads는 각 부분을 한 스레드에 할당합니다. 8코어 머신의 결과입니다(당신의 머신에서는 다를 수 있습니다):

run_sequential: 78498 primes, 6.41s
run_threads: 78498 primes, 6.58s

스레드를 사용한 버전이 오히려 더 느립니다. CPython 표준 빌드에서는 언제든지 한 번에 한 스레드만 Python 바이트코드를 실행할 수 있습니다. 이는 GIL을 유지해야 하기 때문입니다. 8개의 스레드는 단지 하나의 코어를 번갈아 사용할 뿐이며, 스레드 간 전환의 오버헤드까지 더해집니다.

참고: Python 3.13부터 free-threaded 빌드(python3.13t)가 있어 GIL을 제거합니다. 이 버전은 아직 실험 단계이며, 많은 C 확장 라이브러리가 아직 지원하지 않습니다. 대부분의 현재 프로젝트에서는 프로세스를 사용하는 것이 가장 확실한 방법입니다.

3. multiprocessing 기초

각 프로세스는 자신의 인터프리터와 자신의 GIL을 가집니다. 따라서 N개의 프로세스는 N개의 코어에서 실제로 병렬로 실행될 수 있습니다.

3.1. Process 수동 생성

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Total primes:", total)

각 부분을 설명합니다:

• Process(target=..., args=...)는 자식 프로세스를 설명합니다. start()를 호출할 때까지 실행되지 않습니다.
• 자식 프로세스는 부모 프로세스와 메모리를 공유하지 않습니다. 결과를 보내려면 통신 채널을 사용해야 합니다. 여기서는 Queue입니다.
• 프로그램이 join() 전에 results.get()을 호출합니다. 큐에 아직 읽지 않은 대용량 데이터가 있는데 join() 먼저 호출하면 프로그램이 데드락될 수 있습니다.

Total primes: 78498

이 방법은 유연하지만 코드가 깁니다. 실제로는 Process를 직접 관리하는 경우가 거의 없습니다.

3.2. Pool 사용

Pool은 미리 워커 그룹을 만들고 자동으로 작업을 분배합니다:

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Total:", sum(counts))

starmap은 RANGES의 각 튜플을 start, end 매개변수로 분해합니다. 결과는 입력 순서대로 반환됩니다:

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Total: 78498

processes를 지정하지 않으면, Pool은 os.cpu_count() 워커를 사용합니다. with 블록을 나가면 자동으로 워커들을 정리합니다.

4. ProcessPoolExecutor: 더 현대적인 API

concurrent.futures.ProcessPoolExecutor는 ThreadPoolExecutor와 같은 인터페이스를 가집니다. 덕분에 클래스 이름만 바꿔도 거의 스레드에서 프로세스로 전환할 수 있습니다:

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

시간이 약 6.4초에서 1.3초로 줄었습니다. 정확히 8배가 아닌 이유는 두 가지입니다. 첫째, 범위별 작업량이 불균등합니다: 큰 숫자 범위는 검사하는 데 더 오래 걸립니다. 둘째, 프로세스 시작 및 데이터 전송 오버헤드가 있습니다.

각 결과가 완료되자마자 처리하려면 submit을 as_completed와 함께 사용하세요:

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"범위 시작 {futures[future]}: {future.result()} primes")

워커가 예외를 던지면, 부모 프로세스의 future.result()가 그 예외를 다시 던집니다. 따라서 일반적인 try/except로 오류를 처리할 수 있습니다.

5. 작은 작업이 많은 경우: chunksize와 imap_unordered

매번 작업을 워커에 보낼 때마다 Python은 매개변수를 pickle, 파이프를 통해 전송, 그리고 결과를 unpickle해야 합니다. 100,000개의 아주 작은 작업이 있다면, 이 오버헤드가 실제 계산보다 클 수 있습니다. chunksize로 작업을 배치로 묶으면 해결됩니다:

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000은 1개 대신 10,000개씩 보냅니다. 프로세스 간 통신 횟수가 크게 줄어듭니다.
• imap_unordered는 lazy 방식으로 결과를 반환하고 순서를 유지하지 않습니다. 합계 계산처럼 합칠 때만 필요하고 메모리를 절약하고 싶을 때 적합합니다.
• ProcessPoolExecutor.map도 chunksize 매개변수를 받습니다. 기본값은 1이므로 작은 작업이 많을 때 조정하세요.

6. 각 워커마다 한 번만 리소스 초기화

각 작업이 시간이 많이 걸리는 룩업 테이블을 필요하다고 가정합니다. 매번 함수 호출할 때마다 테이블을 재작성하면 매우 낭비입니다. initializer 매개변수를 사용하면 워커 시작 시 정확히 한 번 함수를 실행할 수 있습니다:

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

_price_table 변수는 각 워커 내에서 전역입니다. 부모 프로세스는 그 변경사항을 보지 못합니다. 이는 각 워커가 무거운 리소스를 가질 수 있으면서도 매 작업마다 테이블을 보낼 필요가 없는 안전한 방법입니다.

7. 병렬 코드 테스트 작성

pytest를 사용해 일반 코드처럼 프로세스를 사용하는 코드를 테스트할 수 있습니다. 조건은 워커 함수가 모듈 수준에서 정의되어야 한다는 것입니다:

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 passed in 0.62s

가장 중요한 테스트는 test_parallel_matches_sequential입니다. 범위 분할이 요소를 놓치거나 중복으로 세지 않는지 확인합니다. 이는 split_range를 직접 작성할 때 매우 흔한 오류입니다.

8. 흔한 오류와 피해야 할 경우들

if __name__ == "__main__": 잊기. spawn start method(Windows와 macOS의 기본값) 또는 forkserver(Python 3.14 이후 Linux의 기본값)에서 자식 프로세스는 메인 모듈을 다시 import합니다. 풀 생성 코드가 모듈 수준에 있으면 자식 프로세스마다 새로운 풀이 생깁니다. 그러면 RuntimeError가 나거나 프로세스가 계속 생깁니다.

람다나 중첩 함수 전달. 워커로 전송되는 모든 것은 pickle 가능해야 합니다:

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

모듈 수준에서 def로 함수를 정의하세요. 일부 매개변수를 고정해야 하면 functools.partial을 사용하세요.

너무 큰 데이터 전달. 수백만 요소의 리스트를 각 작업에 전달하면 여러 번 pickle됩니다. 복사 시간이 계산 시간을 초과할 수 있습니다. 몇 가지 해결 방법이 있습니다: 인덱스나 파일 경로만 보내기, initializer 사용, 또는 multiprocessing.shared_memory와 배열 형식의 데이터 사용.

fork 동작에 의존. Linux에서 fork로 작동하는 코드(자식 프로세스가 전역 변수를 상속)가 macOS, Windows 또는 Python 3.14에서 실패할 수 있습니다. 일관된 동작을 원하면 start method를 명시적으로 설정하세요:

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

참고: HTTP 호출이나 데이터베이스 조회 같은 I/O 바운드 작업에는 multiprocessing을 사용하면 안 됩니다. I/O를 기다리는 동안 스레드도 GIL을 해제하므로 ThreadPoolExecutor나 asyncio가 훨씬 가볍습니다. 각 작업이 몇 마이크로초에 불과하고 배치로 묶지 않는 경우, 또는 계산이 이미 C 라이브러리에서 GIL을 해제하는 경우(NumPy, 대용량 데이터 해시 함수…)에도 multiprocessing을 사용하면 안 됩니다. 이러한 경우 스레드 또는 벡터화가 충분합니다.

너무 많은 워커 만들기. 워커 개수를 CPU 코어 개수보다 크게 설정해도 CPU 바운드 코드를 더 빠르게 하지 못합니다. 단지 RAM을 낭비할 뿐입니다. 각 프로세스가 전체 인터프리터이기 때문입니다. os.cpu_count()로 시작하는 것이 합리적입니다.

9. 결론

GIL은 스레드가 순수 Python의 무거운 CPU 작업을 가속화하지 못하게 합니다. 각 프로세스가 자신의 GIL을 가지므로, 작업을 여러 프로세스로 나누는 것이 모든 CPU 코어를 사용하는 간단한 방법입니다. 대부분의 경우 ProcessPoolExecutor가 가장 깔끔한 선택입니다. Pool은 imap_unordered나 starmap이 필요할 때 유용합니다. Process를 수동으로 다루는 것은 특수한 경우에만 해당합니다.

multiprocessing을 사용할 때 기억해야 할 몇 가지:

• 항상 if __name__ == "__main__":을 사용하세요.
• pickle 가능한 함수와 데이터만 전달하세요.
• 작은 작업이 많을 때 chunksize를 증가시키세요.
• 무거운 리소스에는 initializer를 사용하세요.
• 병렬 결과를 순차 결과와 비교하는 테스트를 작성해 범위 분할이 올바른지 확인하세요.