상위 목록: 하위 목록: 작성 날짜: 읽는 데 17 분 소요

프로세스 기반 병렬 처리(Multi Processing)

프로세스 기반 병렬 처리(Multi Processing)란 여러 개의 프로세스를 생성해 작업을 동시에 처리하는 방식을 의미합니다.

Python의 multiprocessing 모듈은 스레딩(threading) 모듈과 비슷한 API로 프로세스 스포닝(Process Spawning)을 지원하는 표준 라이브러리입니다.

프로세스(Process)는 프로그램이 메모리 상에서 실행중인 작업을 의미합니다. 그러므로, 멀티 프로세싱은 하나 이상의 프로세스들을 동시에 처리하는 것을 의미합니다.

대용량 데이터를 처리하는 과정이나 데이터를 분배하여 동시에 처리하고자 할 때 주로 활용합니다.

일반적인 Python(CPython)은 전역 인터프리터 잠금(GIL, Global Interpreter Lock) 때문에 한 프로세스 안에서 여러 스레드가 동시에 Python 코드를 실행하지 못합니다. 그러므로 연산량이 많은(CPU-bound) 작업은 프로세스를 나눠 처리해야 여러 CPU 코어를 활용할 수 있습니다.

멀티 프로세싱은 각 프로세스가 독립된 메모리 영역을 가지므로 하나의 프로세스에서 발생한 문제가 다른 프로세스에 영향을 주지 않아 안전성이 높지만, 프로세스 생성과 데이터 전달에 비용이 들어 오버헤드(Overhead)가 발생할 수 있습니다.


제 31강의 스레드는 하나의 프로세스 안에서 메모리를 공유하며 여러 흐름을 실행합니다. 파일이나 네트워크 응답을 기다리는 동안 다른 일을 하기에는 적합하지만, GIL 때문에 계산 자체를 여러 코어로 나눠 빠르게 만들지는 못합니다.

멀티 프로세싱은 Python 인터프리터를 여러 개 실행하는 방식입니다. 각 프로세스는 자신만의 인터프리터와 GIL을 가지므로, CPU 코어 수만큼 계산을 동시에 실행할 수 있습니다. 대신 프로세스 사이에는 메모리가 공유되지 않으므로, 작업과 결과를 직렬화(pickle)해서 주고받는 비용이 생깁니다.

이번 강좌에서는 자식 프로세스를 만드는 시작 방식과 __main__ 가드, 프로세스를 직접 다루는 Process, 작업자를 재사용하는 Pool과 ProcessPoolExecutor, 스레드와의 성능 비교, 그리고 외부 패키지 joblib을 다룹니다.

  • Tip : 프로세스 스포닝(Process Spawning)이란 부모 프로세스(Parent Process)가 운영 체제에 요청해 새로운 자식 프로세스(Child Process)를 만들어내는 과정입니다.

  • Tip : 스레드(Thread)는 프로세스(Process) 안에서 실행되는 여러 흐름 단위를 의미합니다.

  • Tip : 이 강좌의 예제는 실행 시간과 PID(프로세스 번호)를 출력합니다. 이 값들은 실행할 때마다, 그리고 CPU 개수와 시스템 부하 등 실행 환경에 따라 달라지므로 결과의 숫자보다 경향을 확인합니다.



시작 방식(Start Method)

자식 프로세스를 만드는 방법은 운영 체제마다 다르며, multiprocessing 모듈은 이를 시작 방식으로 구분합니다.

시작 방식에 따라 자식 프로세스가 부모의 무엇을 물려받는지, 그리고 코드를 어떻게 작성해야 하는지가 달라지므로 멀티 프로세싱을 사용하기 전에 먼저 알아 두어야 합니다.


import multiprocessing

if __name__ == "__main__":
    print(multiprocessing.get_start_method())
    print(multiprocessing.get_all_start_methods())
결과
spawn
[‘spawn’]

multiprocessing 모듈은 자식 프로세스를 만드는 시작 방식(Start Method)으로 spawn, fork, forkserver 세 가지를 제공합니다.

get_start_method()는 현재 사용하는 시작 방식을 반환하며, get_all_start_methods()는 현재 플랫폼에서 사용할 수 있는 시작 방식을 반환합니다. 위 결과는 Windows에서 실행한 결과입니다.

Windows는 프로세스를 복제하는 fork 기능이 없으므로, 사용할 수 있는 방식이 spawn 하나뿐입니다.


시작 방식 설명
spawn 새로운 Python 인터프리터를 실행하고, 부모 프로세스의 주 모듈을 다시 가져와(import) 필요한 객체를 전달
fork 부모 프로세스를 그대로 복제(os.fork)해 자식 프로세스를 생성(POSIX 전용)
forkserver 별도의 서버 프로세스를 먼저 실행하고, 자식 프로세스가 필요할 때마다 서버 프로세스가 fork해 생성(POSIX 전용)


기본 시작 방식은 플랫폼마다 다릅니다. Windows와 macOS는 spawn을 사용합니다.

Python 3.14 이상에서는 Linux 등 POSIX 플랫폼(macOS 제외)의 기본 시작 방식이 fork에서 forkserver로 변경되었습니다.

fork 방식은 스레드를 사용하는 프로세스를 복제할 때 교착 상태 등의 문제가 발생할 수 있어, 더 안전한 방식으로 기본값이 바뀌었습니다.

시작 방식을 직접 지정하려면 multiprocessing.set_start_method("spawn") 또는 multiprocessing.get_context("spawn")을 사용합니다.

spawn은 매번 새 인터프리터를 시작하므로 자식 프로세스를 만드는 데 시간이 더 걸리지만, 부모의 상태를 복제하지 않아 예측하기 쉽습니다. fork는 빠르지만 부모의 잠금(lock)이나 스레드 상태까지 그대로 복제되어 문제가 생길 수 있습니다.


__main__ 가드가 필요한 이유

spawn과 forkserver 방식은 자식 프로세스에서 실행할 함수를 찾기 위해 부모 프로세스의 주 모듈(스크립트)을 다시 가져옵니다.

이때 프로세스를 생성하는 코드가 모듈의 최상위에 있다면, 자식 프로세스도 모듈을 가져오면서 다시 프로세스를 생성하려고 시도합니다.

그러므로 Windows에서 가드 없이 프로세스를 생성하면 다음과 같은 오류가 발생합니다.


RuntimeError:
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.


프로세스를 생성하는 코드는 반드시 if __name__ == "__main__": 블록 안에 작성해야 합니다.

자식 프로세스가 주 모듈을 가져올 때에는 __name__이 "__main__"이 아니므로, 가드 안쪽의 코드는 실행되지 않습니다.

Python 3.14부터 Linux의 기본 방식도 forkserver가 되었으므로, 운영체제에 관계없이 항상 가드를 작성하는 것이 안전합니다.

정리하면, 스크립트의 최상위 코드는 자식 프로세스마다 한 번씩 다시 실행된다고 생각해야 합니다. 함수와 클래스 정의, import 문은 최상위에 두고, 실제로 일을 시작하는 코드는 가드 안에 둡니다.

  • Tip : spawn과 forkserver 방식에서 자식 프로세스에 전달하는 함수와 인수는 pickle로 직렬화할 수 있어야 합니다. 그러므로 lambda나 함수 내부에 정의한 함수는 전달할 수 없으며, 모듈 최상위에 정의한 함수를 사용합니다.

  • Tip : 대화형 인터프리터나 일부 Jupyter 환경에서는 주 모듈을 다시 가져올 수 없어 spawn 방식의 예제가 동작하지 않을 수 있습니다. 예제는 .py 파일로 저장해 실행합니다.



프로세스(Process)

Process 클래스는 자식 프로세스 하나를 직접 만들고 관리하는 가장 기본적인 방법입니다.

사용법은 제 31강의 Thread와 거의 같습니다. 실행할 함수를 target, 인수를 args로 전달하고, start()로 시작한 뒤 join()으로 끝나기를 기다립니다.


import os
import time
from multiprocessing import Process, freeze_support


def task(idx, count):
    print(f"PID : {os.getpid()}", flush=True)
    logic = sum([i ** 2 for i in range(count)])
    return idx, logic


if __name__ == "__main__":
    freeze_support()

    job = [("첫 번째", 10 ** 7), ("두 번째", 10 ** 7), ("세 번째", 10 ** 7), ("네 번째", 10 ** 7)]

    start = time.time()
    
    process = []
    for idx, count in job:
        p = Process(target=task, args=(idx, count))
        p.start()
        process.append(p)

    for p in process:
        p.join()

    print(f"End Time : {time.time() - start}s")

    start = time.time()

    for idx, count in job:
        task(idx, count)

    print(f"End Time : {time.time() - start}s")
결과
PID : 10372
PID : 8840
PID : 27576
PID : 22060
End Time : 12.0833420753479s
PID : 29284
PID : 29284
PID : 29284
PID : 29284
End Time : 24.283005952835083s

프로세스(Process) 클래스는 Process 객체를 생성한 후 start() 메서드를 호출해서 자식 프로세스를 스폰합니다.

이후 부모 프로세스는 join() 메서드를 통해 자식 프로세스가 종료될 때까지 대기합니다.

모든 프로세스를 먼저 시작한 다음, process 목록에 저장한 프로세스마다 join() 메서드를 호출해 모든 작업이 끝날 때까지 기다립니다.

처리해야하는 연산량이 많은 경우, 프로세스 클래스를 통해 병렬 처리를 진행할 수 있습니다.

프로세스(Process)는 각 작업마다 새로운 프로세스가 할당되어 작업을 처리합니다.


결과의 앞부분 네 줄은 서로 다른 PID이므로 네 작업이 각각 다른 프로세스에서 실행되었음을 알 수 있습니다. 뒷부분 네 줄은 부모 프로세스가 직접 순서대로 실행했으므로 PID가 모두 같습니다.

병렬 실행은 약 12초, 순차 실행은 약 24초가 걸렸습니다. 작업이 4개인데 4배가 아니라 약 2배만 빨라진 이유는 자식 프로세스를 시작하는 비용, CPU의 실제 코어 구성(논리 CPU와 물리 코어의 차이), 함께 실행 중인 다른 프로그램의 부하 등이 영향을 주기 때문입니다.

start()와 join()을 같은 반복문에서 번갈아 호출하면 하나가 끝나야 다음이 시작되므로 순차 실행과 다를 바가 없습니다. 예제처럼 모두 시작한 다음 모두 기다리는 순서가 중요합니다.

target 함수의 반환값은 부모 프로세스로 전달되지 않습니다. 결과를 돌려받으려면 Queue, Pipe 등을 사용하거나 아래의 Pool, ProcessPoolExecutor를 사용합니다.

  • Tip : freeze_support()는 Windows에서 스크립트를 PyInstaller 등으로 실행 파일로 만들 때(freeze) 필요한 함수입니다. 일반적인 .py 실행에서는 아무 동작도 하지 않으므로 생략해도 됩니다.

  • Tip : os.getpid()를 통해 프로세스마다 서로 다른 PID 값을 가진 프로세스가 실행되는 것을 확인할 수 있습니다. PID와 실행 시간은 실행할 때마다, CPU 개수와 시스템 부하 등 실행 환경에 따라 달라집니다. (이 강좌의 결과는 Windows, 논리 CPU 12개 환경에서 실행한 결과입니다.)

  • Tip : 자식 프로세스의 출력은 버퍼에 쌓였다가 나중에 출력되거나, 프로세스가 강제로 종료되면 사라질 수 있습니다. print(..., flush=True)로 즉시 출력하면 출력 결과를 확실하게 확인할 수 있습니다.


import os
import time
from multiprocessing import Process


def task(seconds):
    print(f"자식 PID : {os.getpid()}, 부모 PID : {os.getppid()}", flush=True)
    time.sleep(seconds)


if __name__ == "__main__":
    print(f"부모 PID : {os.getpid()}", flush=True)

    p = Process(target=task, args=(0.5,), name="worker-1")
    print(p.name, p.pid, p.is_alive(), flush=True)

    p.start()
    print(p.name, p.pid, p.is_alive(), flush=True)

    p.join()
    print(p.is_alive(), p.exitcode)
결과
부모 PID : 27056
worker-1 None False
worker-1 46940 True
자식 PID : 46940, 부모 PID : 27056
False 0

Process 객체가 생성 → 시작 → 종료되는 동안 상태가 어떻게 바뀌는지 확인하는 예제입니다.

  1. Process(...)로 객체를 만든 직후에는 아직 운영 체제의 프로세스가 없으므로 pid는 None, is_alive()는 False입니다. name은 직접 지정한 worker-1입니다.
  2. start()를 호출하면 운영 체제가 자식 프로세스를 만들고 PID(예제에서는 46940)가 부여되며, is_alive()가 True가 됩니다.
  3. 자식 프로세스는 os.getppid()로 자신을 만든 부모의 PID를 확인할 수 있으며, 부모가 출력한 PID와 같은 27056입니다.
  4. join()이 반환되면 자식 프로세스는 끝난 상태이므로 is_alive()는 False, 종료 코드 exitcode는 정상 종료를 뜻하는 0입니다.

exitcode는 자식 프로세스에서 처리되지 않은 예외가 발생하면 1, terminate()로 강제 종료되면 음수가 되므로 자식 프로세스가 정상적으로 끝났는지 확인하는 데 사용합니다.


from multiprocessing import Process

counter = 0


def work():
    global counter
    counter += 100
    print("자식 프로세스 :", counter, flush=True)


if __name__ == "__main__":
    p = Process(target=work)
    p.start()
    p.join()
    print("부모 프로세스 :", counter)
결과
자식 프로세스 : 100
부모 프로세스 : 0

멀티 프로세싱에서 가장 많이 하는 실수는 전역 변수로 결과를 모으려는 것입니다.

자식 프로세스는 부모와 별개의 메모리를 가지므로, 자식이 전역 변수 counter를 100으로 바꿔도 그것은 자식 프로세스 안의 counter일 뿐입니다. 부모 프로세스의 counter는 여전히 0입니다.

스레드라면 같은 메모리를 공유하므로 부모에서도 100이 보였을 것입니다. 이 차이 때문에 프로세스 사이에서는 데이터를 명시적으로 주고받는 도구가 필요합니다.


from multiprocessing import Process, Queue


def task(idx, count, queue):
    queue.put((idx, sum(i * i for i in range(count))))


if __name__ == "__main__":
    queue = Queue()
    job = [("첫 번째", 10), ("두 번째", 100), ("세 번째", 1000)]

    process = [Process(target=task, args=(idx, count, queue)) for idx, count in job]
    for p in process:
        p.start()

    result = [queue.get() for _ in job]

    for p in process:
        p.join()

    print(sorted(result, key=lambda r: r[1]))
결과
[(‘첫 번째’, 285), (‘두 번째’, 328350), (‘세 번째’, 332833500)]

multiprocessing.Queue는 여러 프로세스가 함께 사용할 수 있는 선입선출(FIFO) 대기열입니다. 자식 프로세스는 put()으로 결과를 넣고, 부모 프로세스는 get()으로 꺼냅니다.

put()에 넣은 객체는 pickle로 직렬화되어 부모 프로세스로 전달되고, get()에서 다시 객체로 복원됩니다.

결과는 먼저 끝난 프로세스부터 들어오므로 작업 순서와 다를 수 있습니다. 예제는 결과를 계산 값 기준으로 정렬해 출력했습니다.

join()보다 get()을 먼저 호출한 이유는, 대기열에 꺼내지 않은 데이터가 남아 있으면 자식 프로세스가 종료되지 못하고 기다릴 수 있기 때문입니다. 결과를 모두 꺼낸 뒤 join()을 호출합니다.



풀(Pool)

작업이 수백, 수천 개라면 작업마다 프로세스를 만드는 것은 비효율적입니다. 프로세스를 만드는 비용이 작업 자체보다 커질 수 있고, 코어 수보다 많은 프로세스는 서로 CPU를 나눠 쓸 뿐입니다.

Pool은 정해진 개수의 작업자(worker) 프로세스를 미리 만들어 두고, 작업을 나눠 맡기며 재사용합니다.


import os
import time
from multiprocessing import Pool


def task(pairs):
    print(f"PID : {os.getpid()}", flush=True)
    idx, count = pairs
    logic = sum([i ** 2 for i in range(count)])
    return idx, logic


if __name__ == "__main__":
    job = [("첫 번째", 10 ** 7), ("두 번째", 10 ** 7), ("세 번째", 10 ** 7), ("네 번째", 10 ** 7)]

    start = time.time()

    with Pool(processes=2) as p:
        result = p.map(task, job)

    print(result)
    print(f"End Time : {time.time() - start}s")

    start = time.time()

    result = [task(j) for j in job]

    print(result)
    print(f"End Time : {time.time() - start}s")
결과
PID : 26692
PID : 17944
PID : 17944
PID : 26692
[(‘첫 번째’, 333333283333335000000), (‘두 번째’, 333333283333335000000), (‘세 번째’, 333333283333335000000), (‘네 번째’, 333333283333335000000)]
End Time : 30.840702772140503s
PID : 27512
PID : 27512
PID : 27512
PID : 27512
[(‘첫 번째’, 333333283333335000000), (‘두 번째’, 333333283333335000000), (‘세 번째’, 333333283333335000000), (‘네 번째’, 333333283333335000000)]
End Time : 41.745346546173096s

풀(Pool) 객체는 여러 입력 값에 걸쳐 함수의 실행을 병렬 처리하고 입력 데이터를 프로세스에 분산시킵니다.

풀의 인스턴스를 생성하고 2개의 작업자를 생성합니다.

map() 메서드를 통해 실행하려는 함수와 반복 가능한 객체를 입력하여 각 프로세스에 매핑합니다. 결과는 입력 순서대로 반환됩니다.

풀(Pool)은 사전에 프로세스(processes)의 개수를 설정하고, 작업자 프로세스를 재사용하며 작업을 처리합니다.

Pool.map()은 내장 함수 map()처럼 인수를 하나만 전달하므로, 예제의 task는 (이름, 개수) 튜플 하나를 받아 함수 안에서 나눕니다.

with 구문을 사용하면 블록을 벗어날 때 terminate()가 호출되어 작업자 프로세스가 종료됩니다. map()은 모든 결과를 받은 뒤 반환하므로, 블록 안에서 결과를 받아 두면 안전합니다.

결과를 보면 PID가 26692와 17944 두 종류만 나타나며, 각 작업자가 두 개씩 작업을 처리했습니다. 작업자가 2개이므로 4개의 작업을 두 번에 나눠 처리하게 되어, 작업자가 4개인 앞의 Process 예제보다 시간이 더 걸렸습니다.

  • Tip : 프로세스의 개수가 2개라면, 4개의 작업을 처리해도 PID는 2종류만 나타납니다. 어떤 작업이 어느 프로세스에 배정될지는 실행할 때마다 달라질 수 있습니다.

  • Tip : processes를 생략하면 사용할 수 있는 CPU 개수(os.process_cpu_count(), Python 3.13 이상)만큼 작업자를 생성합니다.


from multiprocessing import Pool


def square(x):
    return x * x


if __name__ == "__main__":
    with Pool(processes=2) as pool:
        print(pool.map(square, range(5)))
        print(pool.starmap(pow, [(2, 3), (3, 2), (10, 2)]))
        print(list(pool.imap(square, range(5))))

        r = pool.apply_async(square, (12,))
        print(r.get(timeout=10))
결과
[0, 1, 4, 9, 16]
[8, 9, 100]
[0, 1, 4, 9, 16]
144

Pool은 작업을 나눠 주는 여러 메서드를 제공합니다.

starmap()은 각 요소를 언패킹해서 여러 인수로 전달합니다. (2, 3)은 pow(2, 3)으로 호출되어 8이 됩니다. 앞의 예제처럼 함수 안에서 튜플을 나눌 필요가 없습니다.

imap()은 결과를 리스트로 한꺼번에 반환하지 않고, 결과가 준비되는 대로 하나씩 꺼낼 수 있는 반복자를 반환합니다. 작업이 많을 때 앞의 결과부터 처리를 시작할 수 있습니다.

apply_async()는 작업 하나를 제출하고 즉시 AsyncResult 객체를 반환하며, get()으로 결과를 기다려 받습니다. timeout을 지정하면 그 시간 안에 결과가 없을 때 예외가 발생합니다.


메서드 인수 전달 반환
map(함수, 반복 가능한 객체) 요소 하나 입력 순서대로 결과 리스트
starmap(함수, 반복 가능한 객체) 요소를 언패킹 입력 순서대로 결과 리스트
imap(함수, 반복 가능한 객체) 요소 하나 입력 순서대로 결과를 내는 반복자
imap_unordered(함수, 반복 가능한 객체) 요소 하나 끝난 순서대로 결과를 내는 반복자
apply_async(함수, 인수 튜플) 작업 하나 AsyncResult 


import time
from multiprocessing import Pool


def square(x):
    return x * x


if __name__ == "__main__":
    data = range(10000)

    start = time.perf_counter()
    result1 = [square(x) for x in data]
    print(f"단일 프로세스 : {time.perf_counter() - start:.4f}s")

    start = time.perf_counter()
    with Pool(processes=4) as pool:
        result2 = pool.map(square, data)
    print(f"Pool(4) : {time.perf_counter() - start:.4f}s")

    print(result1 == result2)
결과
단일 프로세스 : 0.0015s
Pool(4) : 0.3058s
True

멀티 프로세싱이 항상 빠른 것은 아닙니다. 곱셈 한 번처럼 작업 하나가 매우 가벼우면 오히려 수백 배 느려질 수 있습니다.

Pool(4)의 시간에는 작업자 프로세스 4개를 시작하는 시간, 1만 개의 입력과 결과를 pickle로 직렬화해 주고받는 시간이 모두 포함됩니다. 실제 계산 시간은 이에 비하면 아주 작습니다.

그러므로 멀티 프로세싱은 작업 하나하나가 충분히 무거울 때, 그리고 주고받는 데이터가 계산량에 비해 작을 때 효과가 있습니다.



프로세스 풀 실행기(ProcessPoolExecutor)

concurrent.futures 모듈은 스레드 풀과 프로세스 풀을 같은 사용법으로 다룰 수 있게 만든 고수준 API입니다.

Pool보다 기능은 단순하지만, 예외 처리와 결과 수집이 일관되고 ThreadPoolExecutor로 쉽게 바꿀 수 있어 새로 작성하는 코드에서 많이 사용됩니다.


import os
import time
from concurrent.futures import ProcessPoolExecutor


def task(idx, count):
    print(f"PID : {os.getpid()}", flush=True)
    logic = sum([i ** 2 for i in range(count)])
    return idx, logic


if __name__ == "__main__":
    job = [("첫 번째", 10 ** 7), ("두 번째", 10 ** 7), ("세 번째", 10 ** 7), ("네 번째", 10 ** 7)]
    idxs = [idx for idx, _ in job]
    counts = [count for _, count in job]

    start = time.time()

    with ProcessPoolExecutor(max_workers=4) as executor:
        result = list(executor.map(task, idxs, counts))

    print(result)
    print(f"End Time : {time.time() - start}s")
결과
PID : 31560
PID : 19636
PID : 4212
PID : 13268
[(‘첫 번째’, 333333283333335000000), (‘두 번째’, 333333283333335000000), (‘세 번째’, 333333283333335000000), (‘네 번째’, 333333283333335000000)]
End Time : 11.72168493270874s

concurrent.futures 모듈의 ProcessPoolExecutor는 프로세스 풀을 더 간단하게 사용할 수 있는 표준 라이브러리의 고수준 API입니다.

ProcessPoolExecutor(max_workers=작업자 수)로 실행기를 생성하며, with 구문을 벗어날 때 모든 작업이 끝날 때까지 기다린 뒤 프로세스를 정리합니다.

executor.map(함수, 반복 가능한 객체1, 반복 가능한 객체2, ...)은 내장 함수 map()처럼 여러 인수를 전달할 수 있으며, 결과를 입력 순서대로 반환합니다.

Pool의 with 구문은 블록을 벗어날 때 작업자를 강제로 종료하지만, ProcessPoolExecutor의 with 구문은 제출된 작업이 모두 끝날 때까지 기다린 뒤 종료한다는 차이가 있습니다.

작업자 4개가 4개의 작업을 동시에 처리했으므로, 앞의 Process 예제와 비슷한 약 12초가 걸렸습니다.


import os
from concurrent.futures import ProcessPoolExecutor, as_completed


def task(idx, count):
    logic = sum([i ** 2 for i in range(count)])
    return idx, os.getpid(), logic


if __name__ == "__main__":
    job = [("첫 번째", 10 ** 7), ("두 번째", 10 ** 6), ("세 번째", 10 ** 5), ("네 번째", 10 ** 4)]

    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(task, idx, count): idx for idx, count in job}

        for future in as_completed(futures):
            print(futures[future], future.result())
결과
네 번째 (‘네 번째’, 32788, 333283335000)
세 번째 (‘세 번째’, 8600, 333328333350000)
두 번째 (‘두 번째’, 17180, 333332833333500000)
첫 번째 (‘첫 번째’, 33152, 333333283333335000000)

submit(함수, 인수...) 메서드는 작업 하나를 제출하고, 작업의 결과를 나중에 받을 수 있는 Future 객체를 즉시 반환합니다.

future.result()는 작업이 끝날 때까지 기다린 후 반환값을 반환하며, 작업 중 예외가 발생했다면 그 예외를 다시 발생시킵니다.

as_completed()는 제출한 순서와 관계없이 먼저 끝난 작업부터 Future 객체를 반환합니다. 그러므로 연산량이 적은 작업의 결과가 먼저 출력될 수 있으며, 출력 순서는 실행할 때마다 달라질 수 있습니다.

예제는 Future 객체를 키로, 작업 이름을 값으로 하는 사전을 만들어 두었으므로 futures[future]로 어떤 작업의 결과인지 알 수 있습니다. 연산량이 가장 적은 네 번째가 가장 먼저 출력되었습니다.

  • Tip : 스레드 기반의 ThreadPoolExecutor도 같은 사용법을 가지므로, 작업의 성격(CPU 연산 또는 입출력 대기)에 따라 실행기만 바꿔서 사용할 수 있습니다.


from concurrent.futures import ProcessPoolExecutor


def divide(x):
    return 10 / x


if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(divide, x) for x in (5, 0, 2)]

        for f in futures:
            try:
                print(f.result())
            except ZeroDivisionError as e:
                print(type(e).__name__, ":", e)
결과
2.0
ZeroDivisionError : division by zero
5.0

자식 프로세스에서 발생한 예외는 사라지지 않고 Future 객체에 저장되었다가, result()를 호출하는 시점에 부모 프로세스에서 다시 발생합니다.

그러므로 예외 처리 코드는 작업 함수 안이 아니라 result()를 호출하는 곳에 작성할 수 있으며, 하나의 작업이 실패해도 나머지 작업의 결과(2.0, 5.0)는 정상적으로 받을 수 있습니다.

Process를 직접 사용했다면 자식 프로세스의 예외는 오류 메시지를 출력하고 exitcode만 1로 남기므로, 이 점에서 ProcessPoolExecutor가 다루기 편합니다.


from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        future = executor.submit(lambda x: x * 2, 10)
        try:
            future.result()
        except Exception as e:
            print(type(e).__name__, ":", e)
결과
PicklingError : Can’t pickle <function <lambda> at 0x000001C0CC56B690>: it’s not found as __main__.<lambda>

앞에서 설명한 pickle 제약을 실제로 확인하는 예제입니다.

작업을 자식 프로세스로 보내려면 함수를 직렬화해야 하는데, pickle은 함수를 모듈 이름과 함수 이름으로 기록합니다. lambda는 이름이 없어 자식 프로세스에서 다시 찾을 수 없으므로 PicklingError가 발생합니다.

lambda 대신 모듈 최상위에 def로 정의한 함수를 사용하면 해결됩니다. 오류 메시지의 메모리 주소는 실행할 때마다 달라집니다.



스레드와 프로세스 비교

같은 계산 작업을 순차 실행, 스레드, 프로세스로 각각 실행해 GIL의 영향을 직접 확인합니다.


import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def task(count):
    return sum(i * i for i in range(count))


if __name__ == "__main__":
    job = [2 * 10 ** 6] * 4

    start = time.perf_counter()
    [task(c) for c in job]
    print(f"순차 실행 : {time.perf_counter() - start:.2f}s")

    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=4) as ex:
        list(ex.map(task, job))
    print(f"스레드 4개 : {time.perf_counter() - start:.2f}s")

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=4) as ex:
        list(ex.map(task, job))
    print(f"프로세스 4개 : {time.perf_counter() - start:.2f}s")
결과
순차 실행 : 1.55s
스레드 4개 : 1.40s
프로세스 4개 : 0.85s

스레드 4개는 순차 실행과 거의 차이가 없습니다. GIL 때문에 한 번에 하나의 스레드만 Python 코드를 실행할 수 있어, 4개의 스레드가 번갈아 가며 실행되기 때문입니다.

프로세스 4개는 각 프로세스가 별도의 GIL을 가지므로 실제로 동시에 계산하여 시간이 줄었습니다. 작업이 작아서 프로세스 시작 비용이 차지하는 비율이 크기 때문에 4배까지 빨라지지는 않았습니다.

반대로 파일 다운로드나 데이터베이스 응답 대기처럼 기다리는 시간이 대부분인(I/O-bound) 작업은 기다리는 동안 GIL이 풀리므로 스레드로도 충분히 빨라지며, 프로세스보다 가볍습니다.


구분 스레드(Thread) 프로세스(Process)
메모리 같은 프로세스의 메모리를 공유 프로세스마다 독립된 메모리
GIL의 영향 한 번에 하나의 스레드만 Python 코드를 실행 프로세스마다 별도의 GIL
적합한 작업 입출력 대기(I/O-bound) 계산 위주(CPU-bound)
생성 비용 작음 큼(인터프리터 시작)
데이터 전달 변수를 직접 공유(동기화 필요) pickle로 직렬화해 전달


  • Tip : Python 3.13 이상에서는 GIL을 사용하지 않는 free-threaded 빌드를 별도로 설치할 수 있습니다. 이 강좌의 결과는 GIL을 사용하는 일반 빌드에서 실행한 결과이며, sys._is_gil_enabled()로 현재 인터프리터가 GIL을 사용하는지 확인할 수 있습니다.



병렬(Parallel)

joblib은 과학 계산 분야에서 많이 사용하는 병렬 처리 패키지로, 반복문을 거의 그대로 둔 채 병렬화할 수 있는 간결한 문법을 제공합니다.

joblib은 표준 라이브러리가 아닌 외부 패키지입니다. 앞의 표준 라이브러리로 충분하다면 설치하지 않아도 되며, 필요한 경우 다음과 같이 설치합니다.

pip install joblib


import os
import time
from joblib import Parallel, delayed


def task(idx, count):
    print(f"PID : {os.getpid()}", flush=True)
    logic = sum([i ** 2 for i in range(count)])
    return idx, logic


if __name__ == "__main__":
    job = [("첫 번째", 10 ** 7), ("두 번째", 10 ** 7), ("세 번째", 10 ** 7), ("네 번째", 10 ** 7)]

    start = time.time()

    result = Parallel(n_jobs=4)(delayed(task)(idx, count) for idx, count in job)

    print(result)
    print(f"End Time : {time.time() - start}s")

    start = time.time()

    result = [task(*j) for j in job]

    print(result)
    print(f"End Time : {time.time() - start}s")
결과
PID : 31540
PID : 28564
PID : 24744
PID : 7120
[(‘첫 번째’, 333333283333335000000), (‘두 번째’, 333333283333335000000), (‘세 번째’, 333333283333335000000), (‘네 번째’, 333333283333335000000)]
End Time : 13.654357433319092s
PID : 32504
PID : 32504
PID : 32504
PID : 32504
[(‘첫 번째’, 333333283333335000000), (‘두 번째’, 333333283333335000000), (‘세 번째’, 333333283333335000000), (‘네 번째’, 333333283333335000000)]
End Time : 51.84334325790405s

joblib 라이브러리는 multiprocessing 모듈과 비슷한 병렬 처리 기능을 제공합니다.

주요한 차이점으로는 매개 변수를 조금 더 쉽게 전달할 수 있으며, 대규모 Numpy 기반 데이터 구조에 대해 작업자 프로세스와 공유 메모리를 효율적으로 사용할 수 있습니다.

Parallel(n_jobs=프로세스 개수)(delayed(함수)(인수))의 구조로 사용할 수 있습니다.

Parallel 클래스는 병렬 매핑을 위한 클래스입니다. 백엔드(backend)를 설정하거나, 배치 크기(batch_size) 등을 추가로 설정할 수 있습니다.

delayed 함수는 함수와 인수를 바로 실행하지 않고 나중에 실행할 수 있도록 캡처하는 데 사용됩니다.


예제의 한 줄을 나눠서 읽으면 다음과 같습니다.

  1. delayed(task)(idx, count)는 task(idx, count)를 호출하지 않고, 호출할 함수와 인수를 묶은 작업 정보를 만듭니다.
  2. 바깥의 (... for idx, count in job)은 제 42강의 생성자 표현식으로, 작업 정보를 하나씩 만들어 냅니다.
  3. Parallel(n_jobs=4)가 만든 실행기에 생성자 표현식을 전달하면, 작업 정보를 작업자 4개에 나눠 실행하고 결과를 입력 순서대로 리스트로 반환합니다.

즉, 일반 반복문 [task(idx, count) for idx, count in job]에서 task를 delayed(task)로 감싸고 앞에 Parallel(...)을 붙이는 것만으로 병렬화할 수 있습니다.

대규모 Numpy 배열을 다루는 작업에서는 joblib를 활용하는 것도 효율적인 방법입니다.

  • Tip : 멀티 프로세싱은 여러 프로세스를 생성하는 데 시간이 소요되므로, 비교적 작업량이 적은 경우 단일 프로세스가 더 빠를 수 있습니다.

  • Tip : n_jobs=-1로 지정하면 사용할 수 있는 모든 CPU를 사용합니다.



정리

이름 설명 반환값/특징
Process(target, args) 자식 프로세스 하나를 생성 start(), join(), exitcode
Queue() 프로세스 사이에 데이터를 전달 put(), get()
Pool(processes) 작업자 프로세스를 재사용하는 풀 map, starmap, imap, apply_async
ProcessPoolExecutor(max_workers) 고수준 프로세스 풀 submit()은 Future, map()은 반복자
as_completed(futures) 끝난 순서대로 Future 반환 반복자
get_start_method() 현재 시작 방식 spawn / fork / forkserver
Parallel(n_jobs)(delayed(함수)(인수)) joblib의 병렬 실행 입력 순서대로 결과 리스트


  • 프로세스를 만드는 코드는 항상 if __name__ == "__main__": 안에 작성하고, 작업 함수는 모듈 최상위에 정의합니다.
  • 프로세스는 메모리를 공유하지 않으므로, 결과는 반환값(Pool, Future)이나 Queue로 받습니다.
  • 계산 위주의 작업은 프로세스, 기다리는 시간이 많은 작업은 스레드가 적합합니다.
  • 작업 하나가 가볍다면 프로세스 생성과 데이터 전달 비용 때문에 오히려 느려질 수 있습니다.

댓글 남기기

함께 보면 좋은 글