multiprocessing — 프로세스 기반 병렬 처리

multiprocessing — 프로세스 기반 병렬 처리

threading 모듈과 유사한 API로 프로세스를 생성(spawn)하는 것을 지원하는 패키지예요. multiprocessing 패키지는 로컬 및 원격 동시성을 모두 제공하며, 스레드 대신 하위 프로세스를 사용함으로써 효과적으로 전역 인터프리터 잠금(GIL, Global Interpreter Lock)을 우회해요. 덕분에 이 모듈은 주어진 머신의 여러 프로세서를 완전히 활용할 수 있게 해주고, POSIX와 Windows 모두에서 실행돼요. 사용 가능한 환경: Android, iOS, WASI가 아님. 모바일 플랫폼이나 WebAssembly 플랫폼에서는 지원되지 않아요.

출처: Python 표준 라이브러리

본문

소개

multiprocessing 모듈은 여러 입력 값에 걸쳐 함수 실행을 병렬화하고 입력 데이터를 프로세스들에 분산시키는(데이터 병렬성) 편리한 수단을 제공하는 Pool 객체도 도입해요. 다음 예시는 자식 프로세스가 해당 모듈을 성공적으로 임포트할 수 있도록 그런 함수를 모듈에 정의하는 일반적인 관행을 보여줘요. Pool을 사용한 데이터 병렬성의 기본 예시,

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))

는 표준 출력에 다음을 출력해요.

[1, 4, 9]

multiprocessing 모듈은 threading 모듈에 대응물이 없는 API, 예를 들어 실행 중인 프로세스를 종료(terminate)·중단(interrupt)·죽이는(kill) 능력도 소개해요.

Process 클래스

multiprocessing에서 프로세스는 Process 객체를 만든 다음 start() 메서드를 호출해 생성돼요. Process는 threading.Thread의 API를 따릅니다. 멀티프로세스 프로그램의 간단한 예시는 다음과 같아요.

from multiprocessing import Process

def f(name):
    print('hello', name)

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

개별 프로세스 ID를 보여주는 확장 예시는 다음과 같아요.

from multiprocessing import Process
import os

def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())

def f(name):
    info('function f')
    print('hello', name)

if __name__ == '__main__':
    info('main line')
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

if __name__ == '__main__' 부분이 왜 필요한지에 대한 설명은 Programming guidelines를 참고하세요. Process의 인자들은 자식 프로세스에 전달될 수 있도록 보통 픽클(pickle) 가능해야 해요. 위 예시를 REPL에 직접 입력하면, 자식 프로세스가 __main__ 모듈에서 f 함수를 찾지 못해 AttributeError가 날 수 있어요.

컨텍스트와 시작 방법

플랫폼에 따라 multiprocessing은 프로세스를 시작하는 세 가지 방식을 지원해요.

• spawn — 부모 프로세스가 새 Python 인터프리터 프로세스를 시작해요. 자식 프로세스는 프로세스 객체의 run() 메서드를 실행하는 데 필요한 리소스만 상속받아요. 특히 부모 프로세스의 불필요한 파일 디스크립터와 핸들은 상속되지 않아요. 이 방법으로 프로세스를 시작하는 것은 fork나 forkserver보다 상당히 느려요. POSIX와 Windows에서 사용 가능하며, Windows와 macOS에서 기본값이에요.
• fork — 부모 프로세스가 os.fork()로 Python 인터프리터를 포크해요. 자식 프로세스는 시작할 때 사실상 부모와 동일해요. 부모의 모든 리소스가 자식에게 상속돼요. 멀티스레드 프로세스를 안전하게 포크하는 것은 문제가 있음을 기억하세요. POSIX 시스템에서 사용 가능.

  버전 3.14에서 변경: 어떤 플랫폼에서도 더 이상 기본 시작 방법이 아님. fork가 필요한 코드는 get_context()나 set_start_method()로 명시적으로 지정해야 해요. 버전 3.12에서 변경: Python이 프로세스에 여러 스레드가 있음을 감지하면, 이 시작 방법이 내부적으로 호출하는 os.fork()가 DeprecationWarning을 일으켜요.

• forkserver — 프로그램이 시작될 때 forkserver 시작 방법을 선택하면 서버 프로세스가 생성돼요. 이후 새 프로세스가 필요할 때마다 부모 프로세스가 서버에 연결해 새 프로세스를 포크하도록 요청해요. fork 서버 프로세스는 시스템 라이브러리나 미리 로드된 임포트가 부수 효과로 스레드를 생성하지 않는 한 단일 스레드이므로 os.fork()를 쓰기에 일반적으로 안전해요. 불필요한 리소스는 상속되지 않아요. Unix 파이프로 파일 디스크립터를 전달할 수 있는 Linux 같은 POSIX 플랫폼에서 사용 가능하며, 그런 곳에서 기본값이에요.

  버전 3.14에서 변경: POSIX 플랫폼에서 기본 시작 방법이 됨. 버전 3.4에서 변경: spawn이 모든 POSIX 플랫폼에 추가되고, forkserver가 일부 POSIX 플랫폼에 추가됨. 버전 3.8에서 변경: macOS에서 spawn이 기본이 됨. fork는 macOS 시스템 라이브러리가 스레드를 시작할 수 있어 하위 프로세스 충돌을 초래할 수 있으므로 안전하지 않은 것으로 간주해야 함. (bpo-33725) 버전 3.14에서 변경: POSIX 플랫폼에서 기본 시작 방법이 성능은 유지하면서 일반적인 멀티스레드 프로세스 호환성 문제를 피하려고 fork에서 forkserver로 변경됨. (gh-84559)

POSIX에서 spawn 또는 forkserver 시작 방법을 사용하면, 프로그램의 프로세스들이 만든 링크 해제된 이름 있는 시스템 리소스(이름 있는 세마포어나 SharedMemory 객체 같은)를 추적하는 리소스 트래커 프로세스도 시작돼요. 모든 프로세스가 종료되면 리소스 트래커는 남은 추적 객체를 링크 해제해요. 보통은 없어야 하지만, 프로세스가 시그널로 죽었다면 "누수된" 리소스가 있을 수 있어요. (누수된 세마포어나 공유 메모리 세그먼트는 다음 재부팅까지 자동으로 링크 해제되지 않아요.)

시작 방법을 선택하려면 set_start_method()를 메인 모듈의 if __name__ == '__main__' 절에서 사용해요. 예:

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    mp.set_start_method('spawn')
    q = mp.Queue()
    p = mp.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

set_start_method()는 프로그램에서 한 번보다 많이 사용하면 안 돼요. 대신 get_context()로 컨텍스트 객체를 얻을 수도 있어요. 컨텍스트 객체는 multiprocessing 모듈과 같은 API를 가지며, 같은 프로그램에서 여러 시작 방법을 사용할 수 있게 해줘요.

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    ctx = mp.get_context('spawn')
    q = ctx.Queue()
    p = ctx.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

한 컨텍스트와 관련된 객체는 다른 컨텍스트의 프로세스와 호환되지 않을 수 있음을 기억하세요. 특히 fork 컨텍스트로 만든 락은 spawn이나 forkserver 시작 방법으로 시작한 프로세스에 전달할 수 없어요. multiprocessing이나 ProcessPoolExecutor를 사용하는 라이브러리는 사용자가 자신의 multiprocessing 컨텍스트를 제공할 수 있게 설계되어야 해요. 라이브러리 안에서 특정 컨텍스트를 사용하면 라이브러리 사용자 애플리케이션의 나머지와 호환되지 않을 수 있어요. 라이브러리가 특정 시작 방법을 요구한다면 항상 문서화하세요.

경고: 'spawn'과 'forkserver' 시작 방법은 일반적으로 POSIX 시스템에서 "frozen" 실행 파일(즉 PyInstaller, cx_Freeze 같은 패키지가 만든 바이너리)에 사용할 수 없어요. 'fork' 시작 방법은 코드가 스레드를 사용하지 않으면 작동할 수 있어요.

프로세스 간 객체 교환

multiprocessing은 프로세스 간 통신 채널의 두 가지 유형을 지원해요.

Queues — Queue 클래스는 queue.Queue의 거의 복제품이에요. 예:

from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # prints "[42, None, 'hello']"
    p.join()

큐는 스레드와 프로세스 안전해요. multiprocessing 큐에 넣는 어떤 객체든 직렬화돼요.

Pipes — Pipe() 함수는 기본적으로 이중(양방향)인 파이프로 연결된 두 연결(connection) 객체 쌍을 반환해요. 예:

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    p.join()

Pipe()가 반환한 두 연결 객체는 파이프의 두 끝을 나타내요. 각 연결 객체에는 send()와 recv() 메서드가 있어요. 두 프로세스(또는 스레드)가 동시에 같은 파이프 끝에서 읽거나 쓰려고 하면 파이프의 데이터가 손상될 수 있음을 주의하세요. 물론 다른 끝을 동시에 사용하는 프로세스끼리는 손상 위험이 없어요. send()는 객체를 직렬화하고 recv()는 객체를 다시 만듭니다.

프로세스 간 동기화

multiprocessing은 threading의 모든 동기화 프리미티브의 대응물을 담고 있어요. 예를 들어 한 번에 하나의 프로세스만 표준 출력에 출력하도록 락을 쓸 수 있어요:

from multiprocessing import Process, Lock

def f(l, i):
    l.acquire()
    try:
        print('hello world', i)
    finally:
        l.release()

if __name__ == '__main__':
    lock = Lock()

    for num in range(10):
        Process(target=f, args=(lock, num)).start()

락 없이는 다른 프로세스들의 출력이 모두 섞이기 쉬워요.

프로세스 간 상태 공유

위에서 말했듯이, 동시성 프로그래밍을 할 때는 보통 공유 상태 사용을 가능한 한 피하는 게 가장 좋아요. 특히 여러 프로세스를 사용할 때 그렇죠. 그래도 정말 공유 데이터를 써야 한다면 multiprocessing은 몇 가지 방법을 제공해요.

Shared memory — 데이터는 Value나 Array를 사용해 공유 메모리 맵에 저장할 수 있어요. 예를 들어:

from multiprocessing import Process, Value, Array

def f(n, a):
    n.value = 3.1415927
    for i in range(len(a)):
        a[i] = -a[i]

if __name__ == '__main__':
    num = Value('d', 0.0)
    arr = Array('i', range(10))

    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()

    print(num.value)
    print(arr[:])

는 다음을 출력해요.

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

num과 arr을 만들 때 쓰인 'd'와 'i' 인자는 array 모듈이 쓰는 종류의 타입 코드(typecode)예요: 'd'는 배정밀도(float), 'i'는 부호 있는 정수를 나타내요. 이 공유 객체들은 프로세스와 스레드 안전해요. 공유 메모리 사용에 더 유연성이 필요하면, 공유 메모리에서 할당된 임의의 ctypes 객체 생성을 지원하는 multiprocessing.sharedctypes 모듈을 쓸 수 있어요.

Server process — Manager()가 반환하는 매니저 객체는 Python 객체를 보유하고 다른 프로세스가 프록시로 조작할 수 있게 하는 서버 프로세스를 제어해요. Manager()가 반환하는 매니저는 list, dict, set, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value, Array 타입을 지원해요. 예:

from multiprocessing import Process, Manager

def f(d, l, s):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.reverse()
    s.add('a')
    s.add('b')

if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()
        l = manager.list(range(10))
        s = manager.set()

        p = Process(target=f, args=(d, l, s))
        p.start()
        p.join()

        print(d)
        print(l)
        print(s)

는 다음을 출력해요.

{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]
{'a', 'b'}

서버 프로세스 매니저는 공유 메모리 객체보다 유연해요. 임의의 객체 타입을 지원하도록 만들 수 있기 때문이에요. 또한 단일 매니저를 네트워크를 통해 다른 컴퓨터에서 실행되는 프로세스들끼리 공유할 수 있어요. 다만 공유 메모리보다는 느려요.

작업자 풀 사용

Pool 클래스는 작업자 프로세스 풀을 나타내요. 작업을 몇 가지 다른 방식으로 작업자 프로세스에 넘길 수 있는 메서드가 있어요. 예:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

if __name__ == '__main__':
    # start 4 worker processes
    with Pool(processes=4) as pool:

        # print "[0, 1, 4,..., 81]"
        print(pool.map(f, range(10)))

        # print same numbers in arbitrary order
        for i in pool.imap_unordered(f, range(10)):
            print(i)

        # evaluate "f(20)" asynchronously
        res = pool.apply_async(f, (20,))      # runs in *only* one process
        print(res.get(timeout=1))             # prints "400"

        # evaluate "os.getpid()" asynchronously
        res = pool.apply_async(os.getpid, ()) # runs in *only* one process
        print(res.get(timeout=1))             # prints the PID of that process

        # launching multiple evaluations asynchronously *may* use more processes
        multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
        print([res.get(timeout=1) for res in multiple_results])

        # make a single worker sleep for 10 seconds
        res = pool.apply_async(time.sleep, (10,))
        try:
            print(res.get(timeout=1))
        except TimeoutError:
            print("We lacked patience and got a multiprocessing.TimeoutError")

        print("For the moment, the pool remains available for more work")

    # exiting the 'with'-block has stopped the pool
    print("Now the pool is closed and no longer available")

풀의 메서드는 풀을 만든 프로세스만 사용해야 한다는 점을 기억하세요.

참고: 이 패키지의 기능은 __main__ 모듈이 자식들에 의해 임포트 가능해야 한다는 것을 요구해요. 이는 Programming guidelines에서 다루지만 여기서 짚고 넘어갈 가치가 있어요. 즉 multiprocessing.pool.Pool 예시 같은 일부 예시는 대화형 인터프리터에서 작동하지 않아요. 예:

>>> from multiprocessing import Pool
>>> p = Pool(5)
>>> def f(x):
...     return x*x
...
>>> with p:
...     p.map(f, [1,2,3])
Process PoolWorker-1:
Process PoolWorker-2:
Process PoolWorker-3:
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>

레퍼런스

multiprocessing 패키지는 대부분 threading 모듈의 API를 재현해요.

전역 시작 방법

Python은 프로세스를 만들고 초기화하는 여러 방식을 지원해요. 전역 시작 방법은 프로세스를 만드는 기본 메커니즘을 설정해요. 특정 객체를 인스턴스화할 수도 있는 여러 multiprocessing 함수와 메서드는, 아직 설정되지 않았다면 전역 시작 방법을 시스템 기본값으로 암묵적으로 설정해요. 전역 시작 방법은 한 번만 설정할 수 있어요. 시스템 기본값에서 시작 방법을 바꿔야 한다면, 함수나 메서드를 호출하거나 이 객체들을 만들기 전에 적극적으로 전역 시작 방법을 설정해야 해요.

Process와 예외

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None) — Process 객체는 별도 프로세스에서 실행되는 활동을 나타내요. Process 클래스는 threading.Thread의 모든 메서드의 대응물을 가져요. 생성자는 항상 키워드 인자로 호출해야 해요. group은 항상 None이어야 해요. threading.Thread와의 호환성만을 위해 존재해요. target은 run() 메서드가 호출할 callable 객체예요. 기본값은 None(아무것도 호출 안 함)이에요. name은 프로세스 이름이에요. args는 target 호출의 인자 튜플, kwargs는 target 호출의 키워드 인자 딕셔너리예요. 제공되면 키워드 전용 daemon 인자가 프로세스 데몬 플래그를 True/False로 설정해요. None(기본값)이면 이 플래그는 생성하는 프로세스에서 상속돼요. 기본적으로 target에 인자가 전달되지 않아요. args 인자(기본값 ())로 target에 전달할 인자의 리스트나 튜플을 지정할 수 있어요. 하위 클래스가 생성자를 오버라이드한다면, 프로세스에 다른 일을 하기 전에 기본 클래스 생성자(super().__init__())를 호출해야 해요.

참고: 일반적으로 Process의 모든 인자는 픽클 가능해야 해요. 로컬로 정의된 target 함수로 REPL에서 Process나 concurrent.futures.ProcessPoolExecutor를 만들려 할 때 자주 관찰돼요. 현재 REPL 세션에서 정의된 callable 객체를 넘기면, target은 언픽클링 중에 로드되도록 임포트 가능한 모듈 안에 정의되어야 하므로 자식 프로세스가 잡히지 않은 AttributeError로 죽어요. 자식에서의 이런 잡히지 않는 오류의 예:

>>> import multiprocessing as mp
>>> def knigit():
...     print("Ni!")
...
>>> process = mp.Process(target=knigit)
>>> process.start()
>>> Traceback (most recent call last):
  File ".../multiprocessing/spawn.py", line ..., in spawn_main
  File ".../multiprocessing/spawn.py", line ..., in _main
AttributeError: module '__main__' has no attribute 'knigit'
>>> process
<SpawnProcess name='SpawnProcess-1' pid=379473 parent=378707 stopped exitcode=1>

이 제한은 "fork" 시작 방법을 쓸 때는 사실이 아니지만, Python 3.14부터 어떤 플랫폼에서도 더 이상 기본값이 아니에요. (gh-132898) 버전 3.3에서 daemon 매개변수 추가.

• run() — 프로세스의 활동을 나타내는 메서드. 하위 클래스에서 오버라이드할 수 있어요. 표준 run() 메서드는 생성자에 target 인자로 전달된 callable 객체를, 있으면 args와 kwargs에서 가져온 순차 및 키워드 인자로 호출해요. 인자를 리스트나 튜플로 하는 Process에 전달된 args도 같은 효과를 얻어요. 예:

  >>> from multiprocessing import Process
  >>> p = Process(target=print, args=[1])
  >>> p.run()
  1
  >>> p = Process(target=print, args=(1,))
  >>> p.run()
  1
  

• start() — 프로세스의 활동을 시작해요. 프로세스 객체당 최대 한 번 호출해야 해요. 객체의 run() 메서드가 별도 프로세스에서 호출되도록 정리해요.

• join([timeout]) — 선택적 인자 timeout이 None(기본값)이면, 이 메서드는 join()이 호출된 프로세스가 종료될 때까지 블록해요. timeout이 양수면 최대 timeout 초만큼 블록해요. 프로세스가 종료되거나 메서드가 타임아웃되면 None을 반환함을 기억하세요. 프로세스가 종료됐는지 확인하려면 exitcode를 검사하세요. 프로세스는 여러 번 join될 수 있어요. 프로세스는 자신을 join할 수 없어요(교착 상태가 되기 때문). 시작 전에 join을 시도하는 것은 오류예요.

• name — 프로세스 이름. 식별 목적만을 위한 문자열로 의미론은 없어요. 여러 프로세스가 같은 이름을 가질 수 있어요. 초기 이름은 생성자가 설정해요. 명시적 이름이 주어지지 않으면 'Process-N1:N2:…:Nk' 형태의 이름이 만들어져요. 각 Nk는 부모의 N번째 자식이에요.

• is_alive() — 프로세스가 살아 있는지 반환. 대략 start() 메서드가 반환하는 순간부터 자식 프로세스가 종료될 때까지 프로세스 객체는 살아 있어요.

• daemon — 프로세스의 데몬 플래그, 부울 값. start()가 호출되기 전에 설정해야 해요. 초기 값은 생성하는 프로세스에서 상속돼요. 프로세스가 종료되면 모든 데몬성 자식 프로세스의 종료를 시도해요. 데몬성 프로세스는 자식 프로세스를 만들 수 없음을 주의하세요. 아니면 데몬성 프로세스가 부모가 종료될 때 종료되면 자식이 고아가 돼요. 또한 이들은 Unix 데몬이나 서비스가 아니라, 비데몬성 프로세스가 종료되면 종료(되고 join되지)되는 보통 프로세스예요.

threading.Thread API 외에 Process 객체는 다음 속성과 메서드도 지원해요.

• pid — 프로세스 ID 반환. 프로세스가 생성되기 전에는 None이에요.
• exitcode — 자식의 종료 코드. 프로세스가 아직 종료되지 않았으면 None. 자식의 run() 메서드가 정상 반환하면 종료 코드는 0, 정수 인자 N으로 sys.exit()로 종료하면 N. run() 안에서 잡히지 않은 예외로 종료하면 1, 시그널 N으로 종료되면 음수 -N.
• authkey — 프로세스의 인증 키(바이트 문자열). multiprocessing이 초기화될 때 메인 프로세스는 os.urandom()으로 임의 문자열을 할당받아요. Process 객체가 생성되면 부모의 인증 키를 상속하지만 다른 바이트 문자열로 설정해 바꿀 수 있어요.
• sentinel — 프로세스가 끝날 때 "준비"가 되는 시스템 객체의 숫자 핸들. 여러 이벤트를 한 번에 기다리려면 multiprocessing.connection.wait()에서 이 값을 쓸 수 있어요. 아니면 join()을 호출하는 게 더 단순해요. Windows에서는 WaitForSingleObject/WaitForMultipleObjects API 계열에 쓸 수 있는 OS 핸들이고, POSIX에서는 select 모듈의 프리미티브에 쓸 수 있는 파일 디스크립터예요. (버전 3.3 추가.)
• interrupt() — 프로세스를 중단해요. POSIX에서 SIGINT 시그널로 작동해요. Windows에서의 동작은 정의되지 않아요. 기본적으로 자식 프로세스에서 KeyboardInterrupt를 발생시켜 종료해요. 자식 프로세스의 signal.signal()로 SIGINT에 대한 적절한 시그널 핸들러를 설정해 이 동작을 바꿀 수 있어요. 참고: 자식이 KeyboardInterrupt를 잡아 버리면 프로세스는 종료되지 않아요. 기본 동작은 자식에서 잡히지 않은 예외가 발생한 것처럼 exitcode를 1로 설정하기도 해요. 다른 exitcode를 원하면 KeyboardInterrupt를 잡고 exit(your_code)를 호출하면 돼요. (버전 3.14 추가.)
• terminate() — 프로세스를 종료해요. POSIX에서는 SIGTERM 시그널로, Windows에서는 TerminateProcess()로 이루어져요. 종료 핸들러와 finally 절 등은 실행되지 않음을 기억하세요. 프로세스의 하위 프로세스는 종료되지 않고 그냥 고아가 돼요.

  경고: 이 메서드를 관련 프로세스가 파이프나 큐를 사용 중일 때 쓰면 파이프나 큐가 손상되어 다른 프로세스가 쓸 수 없게 될 수 있어요. 마찬가지로 프로세스가 락이나 세마포어를 획득한 상태에서 종료하면 다른 프로세스가 교착 상태에 빠질 수 있어요.

• kill() — terminate()와 같지만 POSIX에서 SIGKILL 시그널을 사용해요. (버전 3.7 추가.)
• close() — Process 객체를 닫아 관련된 모든 리소스를 해제해요. 기본 프로세스가 여전히 실행 중이면 ValueError가 발생해요. close()가 성공적으로 반환하면 Process 객체의 대부분의 다른 메서드와 속성은 ValueError를 일으켜요. (버전 3.7 추가.)

start(), join(), is_alive(), terminate(), exitcode 메서드는 프로세스 객체를 만든 프로세스만 호출해야 한다는 점을 주의하세요. Process의 일부 메서드의 사용 예시:

>>> import multiprocessing, time, signal
>>> mp_context = multiprocessing.get_context('spawn')
>>> p = mp_context.Process(target=time.sleep, args=(1000,))
>>> print(p, p.is_alive())
<...Process ... initial> False
>>> p.start()
>>> print(p, p.is_alive())
<...Process ... started> True
>>> p.terminate()
>>> time.sleep(0.1)
>>> print(p, p.is_alive())
<...Process ... stopped exitcode=-SIGTERM> False
>>> p.exitcode == -signal.SIGTERM
True

• exception multiprocessing.ProcessError — 모든 multiprocessing 예외의 기반 클래스.
• exception multiprocessing.BufferTooShort — Connection.recv_bytes_into()가 읽은 메시지에 대해 제공된 버퍼 객체가 너무 작을 때 발생. e가 BufferTooShort 인스턴스이면 e.args[0]이 메시지를 바이트 문자열로 줘요.
• exception multiprocessing.AuthenticationError — 인증 오류가 있을 때 발생.
• exception multiprocessing.TimeoutError — 타임아웃이 있는 메서드가 타임아웃이 만료될 때 발생.

파이프와 큐

여러 프로세스를 쓸 때는 보통 메시지 전달로 프로세스 간 통신을 하고 락 같은 동기화 프리미티브를 피해요. 메시지를 전달하려면 Pipe()(두 프로세스 간 연결)나 큐(여러 생산자와 소비자 허용)를 쓸 수 있어요. Queue, SimpleQueue, JoinableQueue 타입은 표준 라이브러리의 queue.Queue 클래스를 본뜬 다중 생산자·다중 소비자 FIFO 큐예요. 차이는 Queue가 Python 2.5의 queue.Queue 클래스에 도입된 task_done()과 join() 메서드가 없다는 점이에요. JoinableQueue를 쓰면 큐에서 꺼낸 각 작업에 대해 JoinableQueue.task_done()을 호출해야 해요. 아니면 미완료 작업 수를 세는 데 쓰는 세마포어가 결국 넘쳐서 예외를 일으킬 수 있어요.

다른 Python 큐 구현과 한 가지 다른 점은, multiprocessing 큐는 들어가는 모든 객체를 pickle로 직렬화한다는 것이에요. get 메서드가 반환하는 객체는 원본과 메모리를 공유하지 않는 다시 만들어진 객체예요. 매니저 객체로 공유 큐를 만들 수도 있다는 점을 기억하세요.

참고: multiprocessing은 타임아웃을 알리기 위해 평소의 queue.Empty와 queue.Full 예외를 사용해요. 이들은 multiprocessing 네임스페이스에는 없으니 queue에서 임포트해야 해요. 참고: 객체를 큐에 넣으면 그 객체가 피클되고 백그라운드 스레드가 나중에 피클 데이터를 기본 파이프로 플러시해요. 빈 큐에 객체를 넣은 후 empty() 메서드가 False를 반환하고 get_nowait()가 queue.Empty를 일으키지 않고 반환할 수 있을 때까지 무한히 작은 지연이 있을 수 있어요. 여러 프로세스가 객체를 큐에 넣으면 다른 쪽 끝에서 객체가 순서 밖으로 수신될 수 있어요. 하지만 같은 프로세스가 넣은 객체는 서로에 대해 항상 예상 순서로 돼요. 경고: 프로세스가 Queue를 쓰는 중 Process.terminate()나 os.kill()로 죽이면 큐의 데이터가 손상될 가능성이 있어요. 이로 인해 다른 프로세스가 나중에 큐를 쓸 때 예외를 얻을 수 있어요. 경고: 위에서 말했듯이, 자식 프로세스가 큐에 항목을 넣었으면(그리고 JoinableQueue.cancel_join_thread를 쓰지 않았다면) 버퍼된 모든 항목이 파이프에 플러시될 때까지 그 프로세스는 종료되지 않아요. 즉, 큐에 넣은 모든 항목이 소비됐다고 확신하지 않는 한 그 프로세스를 join하려 하면 교착 상태가 될 수 있어요. 마찬가지로 자식 프로세스가 비데몬성이라면 부모 프로세스는 모든 비데몬성 자식을 join하려 할 때 종료 시 멈출 수 있어요. 매니저로 만든 큐에는 이 문제가 없음을 기억하세요.

프로세스 간 통신을 위한 큐 사용 예시는 Examples를 보세요.

• multiprocessing.Pipe(duplex=True) — 파이프의 끝을 나타내는 Connection 객체 쌍 (conn1, conn2)을 반환해요. duplex가 True(기본값)면 파이프는 양방향이에요. False면 단방향으로, conn1은 수신에만, conn2는 송신에만 쓸 수 있어요. send()는 pickle로 객체를 직렬화하고 recv()는 객체를 다시 만들어요.
• class multiprocessing.Queue([maxsize]) — 파이프와 몇 개의 락/세마포어로 구현된 프로세스 공유 큐를 반환해요. 프로세스가 처음 항목을 큐에 넣으면 피더(feeder) 스레드가 시작되어 객체를 버퍼에서 파이프로 옮겨요. 이 클래스를 인스턴스화하면 전역 시작 방법을 설정할 수 있어요. 표준 라이브러리 queue 모듈의 평소 queue.Empty/queue.Full 예외로 타임아웃을 알려요. Queue는 task_done(), join(), shutdown()을 제외한 queue.Queue의 모든 메서드를 구현해요.
  • qsize() — 큐의 대략적 크기 반환. 멀티스레딩/멀티프로세싱 의미론 때문에 이 숫자는 신뢰할 수 없어요. macOS처럼 sem_getvalue()가 구현되지 않은 플랫폼에서는 NotImplementedError를 일으킬 수 있어요.
  • empty() — 큐가 비어 있으면 True, 아니면 False. 신뢰할 수 없어요. 닫힌 큐에서는 OSError를 일으킬 수 있어요(보장되지 않음).
  • full() — 큐가 가득 차 있으면 True, 아니면 False. 신뢰할 수 없어요.
  • put(obj[, block[, timeout]]) — obj를 큐에 넣어요. 선택적 인자 block이 True(기본값)이고 timeout이 None(기본값)이면 필요할 때까지 빈 슬롯이 생길 때까지 블록해요. timeout이 양수면 최대 timeout 초 블록하고 그 시간 안에 빈 슬롯이 없으면 queue.Full을 일으켜요. 그 외(block이 False)에는 빈 슬롯이 즉시 있으면 항목을 넣고, 아니면 queue.Full을 일으켜요(timeout은 무시됨). (버전 3.8: 큐가 닫히면 AssertionError 대신 ValueError 발생.)
  • put_nowait(obj) — put(obj, False)와 동등.
  • get([block[, timeout]]) — 큐에서 항목을 제거하고 반환해요. (put과 대칭적인 의미론, 타임아웃 시 queue.Empty 발생.) (버전 3.8: 닫힌 큐에서 OSError 대신 ValueError 발생.)
  • get_nowait() — get(False)와 동등.

multiprocessing.Queue에는 queue.Queue에는 없는 추가 메서드가 몇 개 있어요. 이 메서드들은 대부분의 코드에서 보통 불필요해요.

• close() — 큐를 닫아 내부 리소스를 해제해요. 닫힌 후에는 큐를 더 사용하면 안 돼요. 예를 들어 get(), put(), empty() 메서드를 더 호출하면 안 돼요. 백그라운드 스레드는 버퍼된 모든 데이터를 파이프에 플러시하면 종료돼요. 큐가 가비지 컬렉션될 때 자동으로 호출돼요.
• join_thread() — 백그라운드 스레드를 join해요. close() 호출 후에만 쓸 수 있어요. 백그라운드 스레드가 종료될 때까지 블록하며 모든 버퍼 데이터가 파이프에 플러시됐음을 보장해요. 기본적으로 프로세스가 큐의 생성자가 아니면 종료 시 큐의 백그라운드 스레드를 join하려 해요. 프로세스는 cancel_join_thread()를 호출해 join_thread()가 아무것도 안 하게 할 수 있어요.
• cancel_join_thread() — join_thread()가 블록하지 않게 해요. 특히 프로세스 종료 시 백그라운드 스레드가 자동으로 join되지 않게 해요. 더 나은 이름은 allow_exit_without_flush()일 거예요. 큐에 넣은 데이터가 유실될 가능성이 높고, 거의 쓸 일이 없을 거예요. 현재 프로세스가 큐에 넣은 데이터를 기본 파이프에 플러시하는 걸 기다리지 않고 즉시 종료해야 하고, 데이터 유실을 신경 쓰지 않을 때만 필요해요.

참고: 이 클래스의 기능은 호스트 OS에 작동하는 공유 세마포어 구현이 있어야 해요. 없으면 이 클래스의 기능이 비활성화되고 Queue를 인스턴스화하려는 시도는 ImportError를 일으켜요. (bpo-3770) 아래의 전문화된 큐 타입에도 동일하게 적용돼요.

• class multiprocessing.SimpleQueue — 단순화된 Queue 타입으로 잠긴 Pipe에 매우 가까워요.
  • close() — 큐를 닫아 리소스 해제. 닫힌 후에는 사용하면 안 됨. (버전 3.9 추가.)
  • empty() — 큐가 비면 True, 아니면 False. SimpleQueue가 닫히면 항상 OSError 발생.
  • get() — 큐에서 항목 제거·반환.
  • put(item) — 항목을 큐에 넣음.
• class multiprocessing.JoinableQueue([maxsize]) — Queue의 하위 클래스로, 추가로 task_done()과 join() 메서드가 있는 큐예요.
  • task_done() — 이전에 큐에 넣은 작업이 완료됐음을 나타냄. 큐 소비자가 사용해요. 작업을 가져오는 각 get()에 대해 후속 task_done() 호출이 작업 처리가 완료됐음을 큐에 알려요. join()이 현재 블록 중이면 모든 항목이 처리됐을 때(큐에 put()된 모든 항목에 대해 task_done() 호출을 받았다는 뜻) 재개돼요. 큐에 넣은 항목보다 더 많이 호출하면 ValueError 발생.
  • join() — 큐의 모든 항목이 꺼내지고 처리될 때까지 블록. 미완료 작업 수는 항목이 큐에 추가될 때마다 올라가고, 소비자가 task_done()을 호출하면 내려가요. 미완료 작업 수가 0이 되면 join()이 풀려요.

기타

• multiprocessing.active_children() — 현재 프로세스의 모든 살아 있는 자식의 리스트 반환. 호출하면 이미 끝난 프로세스를 "joining"하는 부수 효과가 있어요.

• multiprocessing.cpu_count() — 시스템의 CPU 수 반환. 이 숫자는 현재 프로세스가 사용할 수 있는 CPU 수와 동등하지 않아요. 사용 가능한 CPU 수는 os.process_cpu_count()(또는 len(os.sched_getaffinity(0)))로 얻을 수 있어요. CPU 수를 결정할 수 없으면 NotImplementedError 발생. (os.cpu_count(), os.process_cpu_count() 참고) 버전 3.13: 반환값을 -X cpu_count 플래그나 PYTHON_CPU_COUNT로 덮어쓸 수도 있음(os cpu count API의 래퍼이므로).

• multiprocessing.current_process() — 현재 프로세스에 해당하는 Process 객체 반환. threading.current_thread()의 대응물.

• multiprocessing.parent_process() — current_process()의 부모 프로세스에 해당하는 Process 객체 반환. 메인 프로세스의 경우 parent_process는 None. (버전 3.8 추가.)

• multiprocessing.freeze_support() — multiprocessing을 사용하는 프로그램이 얼려져 실행 파일이 됐을 때 지원을 추가해요. (py2exe, PyInstaller, cx_Freeze로 테스트됨.) 메인 모듈의 if __name__ == '__main__' 줄 바로 뒤에 이 함수를 호출해야 해요. 예:

  from multiprocessing import Process, freeze_support
  
  def f():
      print('hello world!')
  
  if __name__ == '__main__':
      freeze_support()
      Process(target=f).start()
  

  freeze_support() 줄을 빼먹으면 frozen 실행 파일을 실행하려 할 때 RuntimeError가 발생해요. 시작 방법이 spawn이 아니면 freeze_support()는 효과가 없어요. 또한 모듈이 Python 인터프리터로 정상 실행되면(프로그램이 frozen이 아니면) 효과가 없어요.

• multiprocessing.get_all_start_methods() — 지원되는 시작 방법의 리스트 반환. 첫 번째가 기본값이에요. 가능한 시작 방법은 'fork', 'spawn', 'forkserver'예요. 모든 플랫폼이 모든 방법을 지원하지는 않아요. (버전 3.4 추가.)

• multiprocessing.get_context(method=None) — multiprocessing 모듈과 같은 속성을 가진 컨텍스트 객체 반환. method가 None이면 기본 컨텍스트를 반환해요. 전역 시작 방법이 설정되지 않았다면 시스템 기본값으로 설정함을 주의하세요. 아니면 method는 'fork', 'spawn', 'forkserver'여야 해요. 지정한 시작 방법을 쓸 수 없으면 ValueError 발생. (버전 3.4 추가.)

• multiprocessing.get_start_method(allow_none=False) — 프로세스 시작에 쓰는 시작 방법의 이름 반환. 전역 시작 방법이 설정되지 않았고 allow_none이 False면 전역 시작 방법을 기본값으로 설정하고 그 이름을 반환해요. 반환값은 'fork', 'spawn', 'forkserver' 또는 None이에요. (버전 3.4 추가.) 버전 3.8: macOS에서 spawn이 기본이 됨. (bpo-33725)

• multiprocessing.set_executable(executable) — 자식 프로세스를 시작할 때 쓸 Python 인터프리터의 경로를 설정해요. (기본적으로 sys.executable이 사용돼요.) 임베더는 아마 다음과 같이 해야 해요:

  set_executable(os.path.join(sys.exec_prefix, 'pythonw.exe'))
  

  (버전 3.4: POSIX에서 'spawn' 시작 방법 사용 시 지원됨. 버전 3.11: 경로류 객체를 받음.)

• multiprocessing.set_forkserver_preload(module_names) — forkserver 메인 프로세스가 임포트하려 시도할 모듈 이름 목록을 설정해, 이미 임포트된 상태가 포크된 프로세스에 상속되게 해요. 임포트 중 어떤 ImportError든 조용히 무시돼요. 매 프로세스에서 반복 작업을 피하는 성능 향상으로 쓸 수 있어요. 작동하려면 forkserver 프로세스가 시작되기 전에(Pool을 만들거나 Process를 시작하기 전에) 호출해야 해요. 'forkserver' 시작 방법을 쓸 때만 의미가 있어요. (버전 3.4 추가.)

• multiprocessing.set_start_method(method, force=False) — 자식 프로세스를 시작하는 데 쓸 방법을 설정해요. method 인자는 'fork', 'spawn', 'forkserver'일 수 있어요. 시작 방법이 이미 설정되었고 force가 True가 아니면 RuntimeError 발생. method가 None이고 force가 True면 시작 방법을 None으로 설정. method가 None이고 force가 False면 컨텍스트를 기본 컨텍스트로 설정. 최대 한 번 호출해야 하고, 메인 모듈의 if __name__ == '__main__' 절 안에서 보호해야 해요. (버전 3.4 추가.)

  참고: multiprocessing에는 threading.active_count(), threading.enumerate(), threading.settrace(), threading.setprofile(), threading.Timer, threading.local의 대응물이 없어요.

Connection 객체

Connection 객체는 픽클 가능한 객체나 문자열의 송수신을 허용해요. 메시지 지향적 연결 소켓이라고 생각할 수 있어요. Connection 객체는 보통 Pipe로 만들며 — Listener와 Client도 참고하세요.

class multiprocessing.connection.Connection

• send(obj) — 연결의 다른 끝으로 객체를 보내는데 recv()로 읽어야 해요. 객체는 픽클 가능해야 해요. 매우 큰 픽클(대략 32 MiB+, OS에 따라 다름)은 ValueError를 일으킬 수 있어요.

• recv() — send()로 보낸 연결의 다른 끝에서 보낸 객체를 반환해요. 받을 것이 있을 때까지 블록. 받을 것이 남지 않고 다른 끝이 닫히면 EOFError 발생.

• fileno() — 연결이 쓰는 파일 디스크립터 또는 핸들 반환.

• close() — 연결을 닫아요. 연결이 가비지 컬렉션될 때 자동으로 호출돼요.

• poll([timeout]) — 읽을 수 있는 데이터가 있는지 여부 반환. timeout을 지정하지 않으면 즉시 반환해요. timeout이 숫자면 블록할 최대 시간(초). None이면 무한 타임아웃. 여러 연결 객체를 한 번에 폴링하려면 multiprocessing.connection.wait()를 쓰세요.

• send_bytes(buf[, offset[, size]]) — bytes-like 객체에서 완전한 메시지로 바이트 데이터를 보내요. offset이 주어지면 buf에서 그 위치부터 데이터를 읽어요. size가 주어지면 그만큼의 바이트를 buf에서 읽어요. 매우 큰 버퍼(대략 32 MiB+, OS에 따라 다름)는 ValueError를 일으킬 수 있어요.

• recv_bytes([maxlength]) — 연결의 다른 끝에서 보낸 바이트 데이터의 완전한 메시지를 문자열로 반환해요. 받을 것이 있을 때까지 블록. 받을 것이 남지 않고 다른 끝이 닫히면 EOFError. maxlength가 지정되고 메시지가 그보다 길면 OSError 발생 후 연결은 더 이상 읽을 수 없어요. (버전 3.3: IOError를 일으키던 것이 OSError로, 지금은 그 별명.)

• recv_bytes_into(buf[, offset]) — 연결의 다른 끝에서 보낸 바이트 데이터의 완전한 메시지를 buf로 읽고 메시지의 바이트 수를 반환해요. buf는 쓰기 가능한 bytes-like 객체여야 해요. offset이 주어지면 그 위치부터 메시지를 써요. 버퍼가 너무 짧으면 BufferTooShort 예외가 발생하고 완전한 메시지는 e.args[0]로 쓸 수 있어요(e는 예외 인스턴스).

  버전 3.3에서 변경: Connection 객체 자체가 이제 Connection.send()와 Connection.recv()로 프로세스 간에 전송될 수 있음. Connection 객체는 이제 컨텍스트 관리 프로토콜도 지원함. __enter__()는 연결 객체를 반환하고 __exit__()는 close()를 호출해요. 예:

  >>> from multiprocessing import Pipe
  >>> a, b = Pipe()
  >>> a.send([1, 'hello', None])
  >>> b.recv()
  [1, 'hello', None]
  >>> b.send_bytes(b'thank you')
  >>> a.recv_bytes()
  b'thank you'
  >>> import array
  >>> arr1 = array.array('i', range(5))
  >>> arr2 = array.array('i', [0] * 10)
  >>> a.send_bytes(arr1)
  >>> count = b.recv_bytes_into(arr2)
  >>> assert count == len(arr1) * arr1.itemsize
  >>> arr2
  array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])
  

  경고: Connection.recv() 메서드는 받은 데이터를 자동으로 언픽클하는데, 메시지를 보낸 프로세스를 신뢰할 수 없으면 보안 위험이 될 수 있어요. 따라서 연결 객체가 Pipe()로 만들어진 게 아니면, 인증을 수행한 후에만 recv()와 send()를 써야 해요. 경고: 프로세스가 파이프에 읽거나 쓰는 중에 죽으면, 메시지 경계가 어디인지 확실히 알 수 없게 될 수 있기 때문에 파이프의 데이터가 손상될 가능성이 있어요.

동기화 프리미티브

일반적으로 동기화 프리미티브는 멀티스레드 프로그램만큼 멀티프로세스 프로그램에서 필요하지 않아요. threading 모듈 문서를 참고하세요. 매니저 객체로 동기화 프리미티브를 만들 수도 있음을 기억하세요.

• class multiprocessing.Barrier(parties[, action[, timeout]]) — threading.Barrier의 복제품인 배리어 객체. (버전 3.3 추가.)
• class multiprocessing.BoundedSemaphore([value]) — threading.BoundedSemaphore의 밀접한 대응물인 바운디드 세마포어 객체. 밀접한 대응물과 한 가지 차이: acquire 메서드의 첫 인자 이름이 Lock.acquire()와 일관되게 block이다.
  • locked() — 이 객체가 지금 잠겨 있는지 나타내는 부울 반환. (버전 3.14 추가.) 참고: macOS에서는 sem_getvalue()가 구현되지 않아 Semaphore와 구분할 수 없어요.
• class multiprocessing.Condition([lock]) — threading.Condition의 별명인 조건 변수. lock이 지정되면 multiprocessing의 Lock 또는 RLock 객체여야 해요. (버전 3.3: wait_for() 메서드 추가.)
• class multiprocessing.Event — threading.Event의 복제품.
• class multiprocessing.Lock — 재진입하지 않는 락 객체: threading.Lock의 밀접한 대응물. 프로세스나 스레드가 락을 획득하면, 이후 어떤 프로세스/스레드가 획득하려 해도 해제될 때까지 블록하고, 어떤 프로세스/스레드든 해제할 수 있어요. threading.Lock이 스레드에 적용되는 개념과 동작이 여기 multiprocessing.Lock에서 프로세스나 스레드에 적용된다고 복제돼요(기록된 바와 같은 예외 제외). Lock은 사실 기본 컨텍스트로 초기화된 multiprocessing.synchronize.Lock 인스턴스를 반환하는 팩토리 함수라는 점을 주의하세요. Lock은 컨텍스트 관리자 프로토콜을 지원해 with 문에 쓸 수 있어요.
  • acquire(block=True, timeout=None) — 락을 획득, 블록 또는 비블록. block이 True(기본값)이면 락이 잠금 해제 상태가 될 때까지 블록한 다음 잠금 상태로 설정하고 True 반환. block이 False면 블록하지 않아요. 락이 잠금 상태면 False 반환, 아니면 락을 잠금 상태로 설정하고 True 반환. timeout에 양의 부동소수점 값을 주면, 락을 획득하지 못하는 동안 최대 timeout 초 블록. timeout 음수는 0과 동등. None(기본값)은 무한. 락을 획득하면 True, 타임아웃이 경과하면 False.
  • release() — 락을 해제해요. 원래 획득한 프로세스/스레드뿐 아니라 어느 프로세스/스레드에서든 호출할 수 있어요. 잠금되지 않은 락에서 호출하면 ValueError가 발생한다는 점을 제외하고 threading.Lock.release()와 동일해요.
  • locked() — 이 객체가 지금 잠겨 있는지 나타내는 부울 반환. (버전 3.14 추가.)
• class multiprocessing.RLock — 재귀 락 객체: threading.RLock의 밀접한 대응물. 재귀 락은 그것을 획득한 프로세스/스레드가 해제해야 해요. 프로세스/스레드가 재귀 락을 획득하면, 같은 프로세스/스레드는 블록 없이 다시 획득할 수 있고, 획득한 횟수만큼 해제해야 해요. RLock은 사실 기본 컨텍스트로 초기화된 multiprocessing.synchronize.RLock을 반환하는 팩토리 함수예요. 컨텍스트 관리자 프로토콜 지원.
  • acquire(block=True, timeout=None) — 락을 획득, 블록 또는 비블록. block이 True면 락이 이미 현재 프로세스/스레드가 소유하지 않는 한, 잠금 해제(어느 프로세스/스레드도 소유하지 않음) 상태가 될 때까지 블록. 현재 프로세스/스레드가 소유권을 취하고(이미 없으면) 락 내부의 재귀 레벨이 1 증가해 True 반환. block이 False면 블록하지 않아요. 다른 프로세스/스레드가 이미 획득(소유)했으면 소유권을 취하지 않고 재귀 레벨도 바뀌지 않아 False 반환. 잠금 해제 상태면 소유권을 취하고 재귀 레벨이 증가해 True 반환. timeout 사용과 동작은 Lock.acquire()와 같아요.
  • release() — 재귀 레벨을 감소시키며 락을 해제해요. 감소 후 재귀 레벨이 0이면 락을 잠금 해제로 재설정하고(어느 프로세스/스레드도 소유하지 않음) 락이 풀리기를 기다리며 블록 중인 다른 프로세스/스레드가 있으면 정확히 하나가 진행하도록 허용해요. 감소 후에도 0이 아니면 락은 잠기고 호출 프로세스/스레드가 소유한 상태로 남아요. 호출 프로세스/스레드가 락을 소유할 때만 호출하세요. 소유자가 아닌 프로세스/스레드가 호출하거나 락이 잠금 해제 상태면 AssertionError 발생.
  • locked() — 이 객체가 지금 잠겨 있는지 부울 반환. (버전 3.14 추가.)
• class multiprocessing.Semaphore([value]) — threading.Semaphore의 밀접한 대응물인 세마포어 객체. acquire 첫 인자 이름이 block인 한 가지 차이.
  • get_value() — 세마포어의 현재 값 반환. macOS처럼 sem_getvalue()가 구현되지 않은 플랫폼에서는 NotImplementedError 발생할 수 있어요.
  • locked() — 지금 잠겨 있는지 부울 반환. (버전 3.14 추가.) 참고: macOS는 sem_timedwait를 지원하지 않아, 타임아웃으로 acquire()를 호출하면 잠자는 루프로 그 동작을 흉내 내요. 참고: 이 패키지의 일부 기능은 호스트 OS에 작동하는 공유 세마포어 구현이 필요해요. 없으면 multiprocessing.synchronize 모듈이 비활성화되고 임포트 시도는 ImportError를 일으켜요. (bpo-3770)

공유 ctypes 객체

자식 프로세스가 상속할 수 있는 공유 메모리를 사용해 공유 객체를 만들 수 있어요.

• multiprocessing.Value(typecode_or_type, *args, lock=True) — 공유 메모리에서 할당된 ctypes 객체 반환. 기본적으로 반환값은 사실 객체의 동기화된 래퍼예요. 객체 자체는 Value의 value 속성으로 접근할 수 있어요. typecode_or_type가 반환 객체의 타입을 정해요: ctypes 타입 또는 array 모듈이 쓰는 한 문자 타입 코드. *args는 타입의 생성자에 전달돼요. lock이 True(기본값)면 값에 대한 접근을 동기화할 새 재귀 락 객체가 만들어져요. lock이 Lock/RLock 객체면 그걸로 동기화해요. False면 접근이 자동으로 보호되지 않아 반드시 "process-safe"하지 않아요. +=처럼 읽기와 쓰기를 수반하는 연산은 원자적이지 않아요. 예를 들어 공유 값을 원자적으로 증가시키려면 다음만으로 충분하지 않아요:

  counter.value += 1
  

  관련 락이 재귀적(기본값)이라고 가정하면 대신 이렇게 할 수 있어요:

  with counter.get_lock():
      counter.value += 1
  

  lock은 키워드 전용 인자임을 주의하세요.

• multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True) — 공유 메모리에서 할당된 ctypes 배열 반환. 기본적으로 반환값은 사실 배열의 동기화된 래퍼예요. typecode_or_type가 배열 요소의 타입을 정해요: ctypes 타입 또는 array 모듈이 쓰는 한 문자 타입 코드('w'는 제외). 또한 'c' 타입 코드는 ctypes.c_char의 별명이에요. size_or_initializer가 정수면 배열의 길이를 정하고 배열은 처음에 0으로 채워져요. 아니면 시퀀스로 배열을 초기화하고 그 길이가 배열의 길이를 정해요. lock 동작은 Value와 유사해요. ctypes.c_char 배열은 value와 raw 속성을 가지며 둘 다 바이트 문자열 저장·검색에 쓸 수 있어요. raw가 배열 전체 크기의 bytes 객체와 상호작용할 수 있는 반면, value 읽기는 대부분의 프로그래밍 언어가 문자열을 다루듯 null 바이트에서 멈춰요. lock은 키워드 전용 인자임을 주의하세요.

multiprocessing.sharedctypes 모듈

multiprocessing.sharedctypes 모듈은 자식 프로세스가 상속할 수 있는 공유 메모리에서 ctypes 객체를 할당하는 함수를 제공해요.

참고: 포인터를 공유 메모리에 저장하는 것이 가능하지만, 이것은 특정 프로세스의 주소 공간의 위치를 가리키게 될 것임을 기억하세요. 하지만 그 포인터는 두 번째 프로세스의 문맥에서는 매우 유효하지 않을 가능성이 높고, 두 번째 프로세스에서 역참조하려 하면 충돌할 수 있어요.

• multiprocessing.sharedctypes.RawArray(typecode_or_type, size_or_initializer) — 공유 메모리에서 할당된 ctypes 배열 반환. 요소 설정·조회가 잠재적으로 비원자적이므로, 접근이 락으로 자동 동기화되도록 하려면 Array()를 대신 쓰세요.
• multiprocessing.sharedctypes.RawValue(typecode_or_type, *args) — 공유 메모리에서 할당된 ctypes 객체 반환. 값 설정·조회는 잠재적으로 비원자적이므로 Value()를 쓰세요.
• multiprocessing.sharedctypes.Array(typecode_or_type, size_or_initializer, *, lock=True, ctx=None) — RawArray()와 같지만 lock 값에 따라 프로세스 안전한 동기화 래퍼가 반환될 수 있어요. lock은 True면 새 락 생성, Lock/RLock 객체면 그걸로 동기화, False면 자동 보호 안 됨. ctx는 컨텍스트 객체 또는 None(현재 컨텍스트 사용). lock과 ctx는 키워드 전용 매개변수.
• multiprocessing.sharedctypes.Value(typecode_or_type, *args, lock=True, ctx=None) — RawValue()와 같지만 lock 값에 따라 프로세스 안전한 동기화 래퍼가 반환될 수 있어요.
• multiprocessing.sharedctypes.copy(obj) — ctypes 객체 obj의 복사본인 공유 메모리에서 할당된 ctypes 객체 반환.
• multiprocessing.sharedctypes.synchronized(obj, lock=None, ctx=None) — lock으로 접근을 동기화하는 ctypes 객체의 프로세스 안전한 래퍼 객체 반환. lock이 None(기본값)이면 multiprocessing.RLock 객체가 자동으로 생성돼요. 동기화된 래퍼는 감싼 객체의 메서드 외에 두 메서드를 가져요: get_obj()는 감싼 객체를, get_lock()은 동기화에 쓰는 락 객체를 반환해요. 래퍼를 통한 ctypes 객체 접근은 원시 ctypes 객체 접근보다 훨씬 느릴 수 있어요. (버전 3.5: 동기화된 객체가 컨텍스트 관리자 프로토콜 지원.)

다음 표는 공유 메모리에서 공유 ctypes 객체를 만드는 문법을 일반 ctypes 문법과 비교해요. (표에서 MyStruct는 ctypes.Structure의 하위 클래스.)

ctypes	sharedctypes using type	sharedctypes using typecode
c_double(2.4)	RawValue(c_double, 2.4)	RawValue('d', 2.4)
MyStruct(4, 6)	RawValue(MyStruct, 4, 6)
(c_short * 7)()	RawArray(c_short, 7)	RawArray('h', 7)
(c_int * 3)(9, 2, 8)	RawArray(c_int, (9, 2, 8))	RawArray('i', (9, 2, 8))

아래는 여러 ctypes 객체가 자식 프로세스에 의해 수정되는 예시예요.

from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Value, Array
from ctypes import Structure, c_double

class Point(Structure):
    _fields_ = [('x', c_double), ('y', c_double)]

def modify(n, x, s, A):
    n.value **= 2
    x.value **= 2
    s.value = s.value.upper()
    for a in A:
        a.x **= 2
        a.y **= 2

if __name__ == '__main__':
    lock = Lock()

    n = Value('i', 7)
    x = Value(c_double, 1.0/3.0, lock=False)
    s = Array('c', b'hello world', lock=lock)
    A = Array(Point, [(1.875,-6.25), (-5.75,2.0), (2.375,9.5)], lock=lock)

    p = Process(target=modify, args=(n, x, s, A))
    p.start()
    p.join()

    print(n.value)
    print(x.value)
    print(s.value)
    print([(a.x, a.y) for a in A])

출력되는 결과는 다음과 같아요.

49
0.1111111111111111
HELLO WORLD
[(3.515625, 39.0625), (33.0625, 4.0), (5.640625, 90.25)]

매니저

매니저는 서로 다른 프로세스 간에 공유할 수 있는 데이터를 만드는 방법을 제공하며, 다른 머신에서 실행 중인 프로세스 간 네트워크를 통한 공유를 포함해요. 매니저 객체는 공유 객체를 관리하는 서버 프로세스를 제어해요. 다른 프로세스는 프록시를 사용해 공유 객체에 접근할 수 있어요.

• multiprocessing.Manager() — 프로세스 간 객체 공유에 쓸 수 있는 시작된 SyncManager 객체 반환. 반환된 매니저 객체는 생성된 자식 프로세스에 해당하며, 공유 객체를 만들고 해당 프록시를 반환할 메서드를 가져요. 매니저 프로세스는 가비지 컬렉션되거나 부모 프로세스가 종료되는 즉시 종료돼요.

• class multiprocessing.managers.BaseManager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0) — BaseManager 객체를 만들어요. 만든 후에는 start()나 get_server().serve_forever()를 호출해 매니저 객체가 시작된 매니저 프로세스를 가리키도록 해야 해요. address는 매니저 프로세스가 새 연결을 수신하는 주소. None이면 임의로 선택. authkey는 서버 프로세스에 대한 수신 연결의 유효성을 검사하는 데 쓰일 인증 키. None이면 current_process().authkey 사용. 아니면 바이트 문자열이어야 함. serializer는 'pickle'(pickle 직렬화) 또는 'xmlrpclib'(xmlrpc.client 직렬화)여야 해요. ctx는 컨텍스트 객체 또는 None. shutdown_timeout은 shutdown() 메서드에서 매니저가 쓰는 프로세스가 완료될 때까지 기다리는 타임아웃(초)이에요. 종료가 타임아웃되면 프로세스가 종료되고, 그것도 타임아웃되면 죽여요. (버전 3.11: shutdown_timeout 매개변수 추가.)

  • start([initializer[, initargs]]) — 매니저를 시작할 하위 프로세스를 시작해요. initializer가 None이 아니면 하위 프로세스가 시작할 때 initializer(*initargs)를 호출해요.
  • get_server() — Manager의 제어 아래 실제 서버를 나타내는 Server 객체 반환. Server 객체는 serve_forever() 메서드를 지원해요:

    >>> from multiprocessing.managers import BaseManager
    >>> manager = BaseManager(address=('', 50000), authkey=b'abc')
    >>> server = manager.get_server()
    >>> server.serve_forever()
    

    Server는 address 속성도 가져요.
  • connect() — 로컬 매니저 객체를 원격 매니저 프로세스에 연결해요:

    >>> from multiprocessing.managers import BaseManager
    >>> m = BaseManager(address=('127.0.0.1', 50000), authkey=b'abc')
    >>> m.connect()
    

  • shutdown() — 매니저가 쓰는 프로세스를 중지해요. start()가 서버 프로세스를 시작하는 데 쓰였을 때만 사용 가능해요. 여러 번 호출할 수 있어요.
  • register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]]) — 매니저 클래스에 타입이나 callable을 등록하는 데 쓸 수 있는 클래스 메서드. typeid는 특정 유형의 공유 객체를 식별하는 "타입 식별자"로, 문자열이어야 해요. callable은 이 타입 식별자의 객체를 만드는 데 쓰는 callable. connect()로 서버에 연결할 매니저 인스턴스이거나 create_method가 False면 None으로 남길 수 있어요. proxytype은 이 typeid의 공유 객체용 프록시를 만드는 데 쓰는 BaseProxy의 하위 클래스. None이면 프록시 클래스가 자동 생성돼요. exposed는 이 typeid의 프록시가 BaseProxy._callmethod()로 접근할 수 있어야 하는 메서드 이름 시퀀스를 지정. (exposed가 None이면 존재한다면 proxytype._exposed_가 대신 사용.) exposed 목록이 지정되지 않으면 공유 객체의 모든 "공개 메서드"에 접근 가능해요. (여기서 "공개 메서드"는 __call__() 메서드를 가지며 이름이 '_'로 시작하지 않는 어떤 속성이든 의미.) method_to_typeid는 프록시를 반환해야 하는 노출된 메서드들의 반환 타입을 지정하는 매핑으로, 메서드 이름을 typeid 문자열로 매핑해요. 메서드 이름이 이 매핑의 키가 아니거나 매핑이 None이면 메서드가 반환한 객체는 값으로 복사돼요. create_method는 서버 프로세스에 새 공유 객체를 만들고 그것에 대한 프록시를 반환하라고 지시하는 데 쓸 수 있는, 이름이 typeid인 메서드를 만들지 여부를 정해요. 기본적으로 True. BaseManager 인스턴스는 읽기 전용 속성도 하나 가져요.
  • address — 매니저가 쓰는 주소.

  버전 3.3에서 변경: Manager 객체는 컨텍스트 관리 프로토콜 지원. __enter__()는 서버 프로세스를 시작하고(아직 시작하지 않았다면) 매니저 객체를 반환. __exit__()는 shutdown() 호출.

• class multiprocessing.managers.SyncManager — 프로세스 동기화에 쓸 수 있는 BaseManager의 하위 클래스. 이 타입의 객체는 multiprocessing.Manager()가 반환해요. 그 메서드들은 프로세스 간 동기화할 여러 흔한 데이터 타입에 대한 프록시 객체를 만들고 반환해요. 특히 공유 리스트와 딕셔너리를 포함해요.

  • Barrier(parties[, action[, timeout]]) — 공유 threading.Barrier 객체를 만들고 프록시 반환. (버전 3.3 추가.)
  • BoundedSemaphore([value]) — 공유 threading.BoundedSemaphore 객체를 만들고 프록시 반환.
  • Condition([lock]) — 공유 threading.Condition 객체를 만들고 프록시 반환. lock이 주어지면 threading.Lock/threading.RLock 객체의 프록시여야 해요. (버전 3.3: wait_for() 추가.)
  • Event() — 공유 threading.Event 객체를 만들고 프록시 반환.
  • Lock() — 공유 threading.Lock 객체를 만들고 프록시 반환.
  • Namespace() — 공유 Namespace 객체를 만들고 프록시 반환.
  • Queue([maxsize]) — 공유 queue.Queue 객체를 만들고 프록시 반환.
  • RLock() — 공유 threading.RLock 객체를 만들고 프록시 반환.
  • Semaphore([value]) — 공유 threading.Semaphore 객체를 만들고 프록시 반환.
  • Array(typecode, sequence) — 배열을 만들고 프록시 반환.
  • Value(typecode, value) — 쓰기 가능한 value 속성을 가진 객체를 만들고 프록시 반환.
  • dict(), dict(mapping), dict(sequence) — 공유 dict 객체를 만들고 프록시 반환.
  • list(), list(sequence) — 공유 list 객체를 만들고 프록시 반환.
  • set(), set(sequence), set(mapping) — 공유 set 객체를 만들고 프록시 반환. (버전 3.14: set 지원 추가.)

  버전 3.6에서 변경: 공유 객체는 중첩될 수 있음. 예를 들어 공유 리스트 같은 공유 컨테이너 객체는 모두 SyncManager가 관리·동기화할 다른 공유 객체를 담을 수 있어요.

• class multiprocessing.managers.Namespace — SyncManager에 등록할 수 있는 타입. 네임스페이스 객체는 공개 메서드가 없지만 쓰기 가능한 속성은 있어요. 그 표현(representation)은 속성 값을 보여줘요. 그러나 네임스페이스 객체의 프록시를 쓸 때는 '_'로 시작하는 속성이 referent의 속성이 아니라 프록시의 속성이에요:

  >>> mp_context = multiprocessing.get_context('spawn')
  >>> manager = mp_context.Manager()
  >>> Global = manager.Namespace()
  >>> Global.x = 10
  >>> Global.y = 'hello'
  >>> Global._z = 12.3    # this is an attribute of the proxy
  >>> print(Global)
  Namespace(x=10, y='hello')
  

커스터마이즈된 매니저 — 자신만의 매니저를 만들려면 BaseManager의 하위 클래스를 만들고 register() 클래스 메서드로 새 타입이나 callable을 매니저 클래스에 등록해요. 예:

from multiprocessing.managers import BaseManager

class MathsClass:
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # prints 7
        print(maths.mul(7, 8))         # prints 56

원격 매니저 사용 — 한 머신에서 매니저 서버를 실행하고 다른 머신의 클라이언트가 사용하게 하는 것이 가능해요(관련 방화벽이 허용한다고 가정). 다음 명령을 실행하면 원격 클라이언트가 접근할 수 있는 단일 공유 큐를 위한 서버를 만들어요:

>>> from multiprocessing.managers import BaseManager
>>> from queue import Queue
>>> queue = Queue()
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue', callable=lambda:queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

한 클라이언트는 다음과 같이 서버에 접근할 수 있어요:

>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.put('hello')

다른 클라이언트도 사용할 수 있어요:

>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.get()
'hello'

프록시 객체

프록시는 (아마도) 다른 프로세스에 살고 있는 공유 객체를 가리키는 객체예요. 공유 객체를 프록시의 referent라고 해요. 여러 프록시 객체가 같은 referent를 가질 수 있어요. 프록시 객체는 referent의 해당 메서드를 호출하는 메서드가 있어요(단 referent의 모든 메서드가 프록시를 통해 반드시 사용 가능한 건 아니에요). 이런 식으로 프록시는 referent처럼 쓸 수 있어요:

>>> mp_context = multiprocessing.get_context('spawn')
>>> manager = mp_context.Manager()
>>> l = manager.list([i*i for i in range(10)])
>>> print(l)
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
>>> print(repr(l))
<ListProxy object, typeid 'list' at 0x...>
>>> l[4]
16
>>> l[2:5]
[4, 9, 16]

프록시에 str()을 적용하면 referent의 표현을 반환하고, repr()을 적용하면 프록시의 표현을 반환함을 주목하세요. 프록시 객체의 중요한 특징은 픽클 가능해서 프로세스 간에 전달될 수 있다는 것이에요. 따라서 referent가 Proxy 객체를 담을 수 있어요. 이로써 관리되는 리스트·딕셔너리·다른 Proxy 객체의 중첩이 허용돼요:

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # referent of a now contains referent of b
>>> print(a, b)
[<ListProxy object, typeid 'list' at ...>] []
>>> b.append('hello')
>>> print(a[0], b)
['hello'] ['hello']

마찬가지로 dict와 list 프록시를 서로 안에 중첩할 수 있어요. 참고 자료에서 표준(비프록시) list나 dict 객체가 referent에 담겨 있으면, 프록시는 내부 값이 언제 수정되는지 알 방법이 없어 그 가변 값의 수정이 매니저를 통해 전파되지 않아요. 하지만 컨테이너 프록시에 값을 저장하면(프록시 객체에서 __setitem__을 촉발) 매니저를 통해 전파돼요. 예:

# create a list proxy and append a mutable object (a dictionary)
lproxy = manager.list()
lproxy.append({})
# now mutate the dictionary
d = lproxy[0]
d['a'] = 1
d['b'] = 2
# at this point, the changes to d are not yet synced, but by
# updating the dictionary, the proxy is notified of the change
lproxy[0] = d

참고: multiprocessing의 프록시 타입은 값에 의한 비교를 지원하는 데 아무것도 하지 않아요. 그래서 예를 들어 manager.list([1,2,3]) == [1,2,3]는 False예요. 비교할 때는 referent의 복사본을 사용해야 해요.

• class multiprocessing.managers.BaseProxy — 프록시 객체는 BaseProxy의 하위 클래스의 인스턴스예요.
  • _callmethod(methodname[, args[, kwds]]) — 프록시 referent의 메서드를 호출하고 결과를 반환해요. proxy가 referent obj의 프록시라면, proxy._callmethod(methodname, args, kwds)는 매니저의 프로세스에서 getattr(obj, methodname)(*args, **kwds)를 평가해요. 반환값은 호출 결과의 복사본 또는 새 공유 객체의 프록시예요. 호출이 예외를 일으키면 _callmethod()가 다시 일으켜요. 매니저 프로세스에서 다른 예외가 발생하면 RemoteError로 변환되어 _callmethod()가 일으켜요. 특히 methodname이 노출되지 않았으면 예외가 발생. 예:

    >>> l = manager.list(range(10))
    >>> l._callmethod('__len__')
    10
    >>> l._callmethod('__getitem__', (slice(2, 7),)) # equivalent to l[2:7]
    [2, 3, 4, 5, 6]
    >>> l._callmethod('__getitem__', (20,))          # equivalent to l[20]
    Traceback (most recent call last):
    ...
    IndexError: list index out of range
    

  • _getvalue() — referent의 복사본 반환. referent가 언픽클 가능하지 않으면 예외 발생.
  • __repr__() — 프록시 객체의 표현 반환.
  • __str__() — referent의 표현 반환.

정리(Cleanup) — 프록시 객체는 weakref 콜백을 사용해서, 가비지 컬렉션될 때 그 referent를 소유한 매니저에서 자신을 등록 해제해요. 공유 객체는 더 이상 그것을 가리키는 프록시가 없으면 매니저 프로세스에서 삭제돼요.

프로세스 풀

Pool 클래스로 제출된 작업을 수행할 프로세스 풀을 만들 수 있어요.

• class multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]]) — 작업을 제출할 수 있는 작업자 프로세스 풀을 제어하는 프로세스 풀 객체. 타임아웃과 콜백이 있는 비동기 결과와 병렬 map 구현을 지원해요. processes는 사용할 작업자 프로세스 수. None이면 os.process_cpu_count()가 반환한 수를 사용해요. initializer가 None이 아니면 각 작업자 프로세스가 시작할 때 initializer(*initargs)를 호출해요. maxtasksperchild는 작업자 프로세스가 사용되지 않는 리소스를 해제하도록 종료되어 새 작업자로 교체되기 전에 완료할 수 있는 작업 수. 기본 maxtasksperchild는 None으로, 작업자 프로세스가 풀만큼 오래 산다는 뜻. context는 작업자 프로세스 시작에 쓸 컨텍스트를 지정. 보통 풀은 multiprocessing.Pool() 함수나 컨텍스트 객체의 Pool() 메서드로 만들어져요. 두 경우 모두 context는 적절히 설정돼요. None이면 이 함수를 호출하면 아직 설정되지 않았다면 현재 전역 시작 방법을 설정하는 부수 효과가 있어요. 풀 객체의 메서드는 풀을 만든 프로세스만 호출해야 한다는 점을 주의하세요.

  경고: multiprocessing.pool 객체에는 (다른 리소스처럼) 제대로 관리해야 하는 내부 리소스가 있어요 — 풀을 컨텍스트 매니저로 사용하거나 close()와 terminate()를 수동으로 호출하세요. 그렇게 하지 않으면 프로세스가 마무리(finalization) 시 멈출 수 있어요. CPython이 풀의 파이널라이저 호출을 보장하지 않으므로(자세한 내용은 object.__del__() 참고) 가비지 컬렉터에 의존하는 것은 올바르지 않아요. 버전 3.2: maxtasksperchild 추가. 버전 3.4: context 추가. 버전 3.13: processes가 os.cpu_count() 대신 기본적으로 os.process_cpu_count() 사용. 참고: Pool 내 작업자 프로세스는 보통 Pool 작업 큐의 전체 기간 동안 산다. 다른 시스템(Apache, mod_wsgi 등)에서 흔한, 작업자가 보유한 리소스를 해제하는 패턴은, 풀의 작업자가 일정량의 작업만 완료하고 종료·정리되며 새 프로세스가 교체되도록 허용하는 것이다. Pool의 maxtasksperchild 인자가 이 능력을 최종 사용자에게 노출해요.

  • apply(func[, args[, kwds]]) — args와 키워드 인자 kwds로 func를 호출해요. 결과가 준비될 때까지 블록. 블록하므로 apply_async()가 병렬 작업에 더 적합해요. 또한 func는 풀의 작업자 중 정확히 하나에서만 실행돼요.
  • apply_async(func[, args[, kwds[, callback[, error_callback]]]]) — AsyncResult 객체를 반환하는 apply()의 변형. callback이 지정되면 단일 인자를 받는 callable이어야 해요. 결과가 준비되면 적용되는데, 호출이 실패하면 error_callback이 대신 적용돼요. error_callback이 지정되면 target 함수가 실패할 때 예외 인스턴스로 호출돼요. 콜백은 즉시 완료되어야 해요. 아니면 결과를 처리하는 스레드가 블록될 수 있으니까요.
  • map(func, iterable[, chunksize]) — map() 내장 함수의 병렬 대응물(단 하나의 iterable 인자만 지원해요. 여러 iterable은 starmap() 참고). 결과가 준비될 때까지 블록. 이 메서드는 iterable을 여러 청크로 나눠 별도 작업으로 프로세스 풀에 제출해요. 이 청크들의 (대략적) 크기는 chunksize를 양의 정수로 설정해 지정할 수 있어요. 매우 긴 iterable에서는 높은 메모리 사용을 일으킬 수 있음을 주의하세요. 더 효율적으로 하려면 명시적 chunksize 옵션으로 imap()이나 imap_unordered()를 사용하는 것을 고려하세요.
  • map_async(func, iterable[, chunksize[, callback[, error_callback]]]) — AsyncResult 객체를 반환하는 map()의 변형.
  • imap(func, iterable[, chunksize]) — map()의 이터레이터 기반 버전. chunksize는 map() 메서드와 같아요. 매우 긴 iterable에서 chunksize에 큰 값을 쓰면 기본값 1보다 훨씬 빨리 작업을 끝낼 수 있어요. chunksize가 1이면 imap()이 반환한 이터레이터의 next() 메서드가 선택적 timeout 매개변수를 가져요: next(timeout)은 결과가 timeout 초 안에 반환될 수 없으면 multiprocessing.TimeoutError를 일으켜요.
  • imap_unordered(func, iterable[, chunksize]) — imap()과 같지만 반환된 이터레이터의 결과 순서는 임의로 간주해야 해요. (작업자 프로세스가 하나뿐일 때만 순서가 "올바른" 것이 보장돼요.)
  • starmap(func, iterable[, chunksize]) — iterable의 요소가 인자로 풀리는 iterable일 것으로 예상된다는 점을 제외하면 map()과 같아요. 따라서 [(1,2), (3, 4)]의 iterable은 [func(1,2), func(3,4)]가 돼요. (버전 3.3 추가.)
  • starmap_async(func, iterable[, chunksize[, callback[, error_callback]]]) — iterable의 iterable을 순회하며 iterable을 풀어 func를 호출하는 starmap()과 map_async()의 결합. 결과 객체 반환. (버전 3.3 추가.)
  • close() — 더 이상 작업이 풀에 제출되지 않게 해요. 모든 작업이 완료되면 작업자 프로세스가 종료돼요.
  • terminate() — 미완료 작업을 완료하지 않고 작업자 프로세스를 즉시 중지해요. 풀 객체가 가비지 컬렉션되면 terminate()가 즉시 호출돼요.
  • join() — 작업자 프로세스가 종료될 때까지 기다려요. join()을 쓰기 전에 close()나 terminate()를 호출해야 해요.

  버전 3.3에서 변경: Pool 객체는 컨텍스트 관리 프로토콜 지원. __enter__()는 풀 객체 반환, __exit__()는 terminate() 호출.

• class multiprocessing.pool.AsyncResult — Pool.apply_async()와 Pool.map_async()가 반환하는 결과의 클래스.
  • get([timeout]) — 결과가 도착하면 반환. timeout이 None이 아니고 결과가 timeout 초 안에 도착하지 않으면 multiprocessing.TimeoutError 발생. 원격 호출이 예외를 일으켰으면 그 예외를 get()이 다시 일으켜요.
  • wait([timeout]) — 결과가 사용 가능하거나 timeout 초가 지날 때까지 대기.
  • ready() — 호출이 완료됐는지 여부 반환.
  • successful() — 호출이 예외를 일으키지 않고 완료됐는지 여부 반환. 결과가 준비되지 않았으면 ValueError 발생. (버전 3.7: 준비되지 않으면 AssertionError 대신 ValueError 발생.)

다음 예시는 풀의 사용을 보여줘요.

from multiprocessing import Pool
import time

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(processes=4) as pool:         # start 4 worker processes
        result = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously in a single process
        print(result.get(timeout=1))        # prints "100" unless your computer is *very* slow

        print(pool.map(f, range(10)))       # prints "[0, 1, 4,..., 81]"

        it = pool.imap(f, range(10))
        print(next(it))                     # prints "0"
        print(next(it))                     # prints "1"
        print(it.next(timeout=1))           # prints "4" unless your computer is *very* slow

        result = pool.apply_async(time.sleep, (10,))
        print(result.get(timeout=1))        # raises multiprocessing.TimeoutError

리스너와 클라이언트

보통 프로세스 간 메시지 전달은 큐나 Pipe()가 반환한 Connection 객체로 이루어져요. 하지만 multiprocessing.connection 모듈은 추가 유연성을 허용해요. 소켓이나 Windows 명명된 파이프를 다루기 위한 고수준 메시지 지향 API를 기본적으로 제공해요. hmac 모듈을 쓴 다이제스트 인증과 여러 연결을 동시에 폴링하는 것도 지원해요.

• multiprocessing.connection.deliver_challenge(connection, authkey) — 연결의 다른 끝에 임의로 생성된 메시지를 보내고 답을 기다려요. 답이 authkey를 키로 한 메시지의 다이제스트와 일치하면 연결의 다른 끝에 환영 메시지를 보내요. 아니면 AuthenticationError 발생.
• multiprocessing.connection.answer_challenge(connection, authkey) — 메시지를 받고 authkey를 키로 메시지의 다이제스트를 계산해 다이제스트를 보내요. 환영 메시지를 받지 못하면 AuthenticationError 발생.
• multiprocessing.connection.Client(address[, family[, authkey]]) — address 주소를 사용하는 리스너에 연결을 설정하려 시도해 Connection을 반환해요. 연결 타입은 family 인자로 결정되지만, 보통 address 형식에서 추론할 수 있어 일반적으로 생략할 수 있어요. authkey가 주어지고 None이 아니면 바이트 문자열이어야 하며 HMAC 기반 인증 챌린지의 비밀 키로 쓰여요. None이면 인증이 수행되지 않아요. 인증이 실패하면 AuthenticationError 발생.
• class multiprocessing.connection.Listener([address[, family[, backlog[, authkey]]]]) — 연결을 '듣고' 있는 바인딩된 소켓 또는 Windows 명명된 파이프의 래퍼. address는 리스너 객체의 바인딩된 소켓 또는 명명된 파이프가 쓰는 주소.

  참고: '0.0.0.0' 주소를 쓰면 그 주소는 Windows에서 연결 가능한 끝점이 아니에요. 연결 가능한 끝점이 필요하면 '127.0.0.1'을 쓰세요. family는 쓸 소켓(또는 명명된 파이프)의 타입. 'AF_INET'(TCP 소켓), 'AF_UNIX'(Unix 도메인 소켓), 'AF_PIPE'(Windows 명명된 파이프) 중 하나일 수 있어요. 이 중 첫 번째만 사용 가능함이 보장돼요. family가 None이면 address 형식에서 추론돼요. address도 None이면 기본값이 선택되는데, 가장 빠르다고 가정되는 family예요. family가 'AF_UNIX'이고 address가 None이면 tempfile.mkstemp()로 만든 개인 임시 디렉터리에 소켓이 생성돼요. 리스너 객체가 소켓을 쓰면 backlog(기본 1)가 소켓을 바인딩한 후 listen() 메서드에 전달돼요. authkey가 주어지고 None이 아니면 HMAC 기반 인증 챌린지의 비밀 키로 쓰여요. 인증 실패 시 AuthenticationError.

  • accept() — 리스너 객체의 바인딩된 소켓 또는 명명된 파이프에서 연결을 수락하고 Connection 객체 반환. 인증이 시도되어 실패하면 AuthenticationError 발생.
  • close() — 리스너 객체의 바인딩된 소켓 또는 명명된 파이프를 닫아요. 리스너가 가비지 컬렉션될 때 자동으로 호출돼요. 명시적으로 호출하는 게 좋아요. Listener 객체는 읽기 전용 속성을 가져요.
  • address — Listener 객체가 쓰는 주소.
  • last_accepted — 마지막 수락된 연결이 온 주소. 사용할 수 없으면 None.

  버전 3.3: Listener 객체가 컨텍스트 관리 프로토콜 지원.

• multiprocessing.connection.wait(object_list, timeout=None) — object_list의 객체가 준비될 때까지 대기. object_list에서 준비된 객체의 리스트를 반환해요. timeout이 부동소수점이면 최대 그 초만큼 블록. None이면 무한 블록. 음수 timeout은 0과 동등. POSIX와 Windows 모두에서, object_list에 나타날 수 있는 객체는: 읽을 수 있는 Connection 객체, 연결되고 읽을 수 있는 socket.socket 객체, 또는 Process 객체의 sentinel 속성. 연결이나 소켓 객체는 읽을 수 있는 데이터가 있거나 다른 끝이 닫혔을 때 준비돼요.
  • POSIX: wait(object_list, timeout)은 select.select(object_list, [], [], timeout)과 거의 동등. 차이는 select.select()가 시그널에 중단되면 EINTR 오류 번호로 OSError를 일으킬 수 있는데, wait()는 그렇지 않다는 점.
  • Windows: object_list의 항목은 Win32 WaitForMultipleObjects() 문서 정의에 따라 대기 가능한 정수 핸들이거나, 소켓 핸들이나 파이프 핸들을 반환하는 fileno() 메서드가 있는 객체여야 해요. (파이프 핸들과 소켓 핸들은 대기 가능한 핸들이 아님을 주의하세요.)

  버전 3.3 추가.

예시: 다음 서버 코드는 'secret password'를 인증 키로 쓰는 리스너를 만들고, 연결을 기다린 다음 클라이언트에 데이터를 보내요.

from multiprocessing.connection import Listener
from array import array

address = ('localhost', 6000)     # family is deduced to be 'AF_INET'

with Listener(address, authkey=b'secret password') as listener:
    with listener.accept() as conn:
        print('connection accepted from', listener.last_accepted)

        conn.send([2.25, None, 'junk', float])

        conn.send_bytes(b'hello')

        conn.send_bytes(array('i', [42, 1729]))

다음 코드는 서버에 연결하고 서버에서 데이터를 받아요.

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)

with Client(address, authkey=b'secret password') as conn:
    print(conn.recv())                  # => [2.25, None, 'junk', float]

    print(conn.recv_bytes())            # => 'hello'

    arr = array('i', [0, 0, 0, 0, 0])
    print(conn.recv_bytes_into(arr))    # => 8
    print(arr)                          # => array('i', [42, 1729, 0, 0, 0])

다음 코드는 wait()를 사용해 여러 프로세스에서 한 번에 메시지를 기다려요.

from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait

def foo(w):
    for i in range(10):
        w.send((i, current_process().name))
    w.close()

if __name__ == '__main__':
    readers = []

    for i in range(4):
        r, w = Pipe(duplex=False)
        readers.append(r)
        p = Process(target=foo, args=(w,))
        p.start()
        # We close the writable end of the pipe now to be sure that
        # p is the only process which owns a handle for it.  This
        # ensures that when p closes its handle for the writable end,
        # wait() will promptly report the readable end as being ready.
        w.close()

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
            except EOFError:
                readers.remove(r)
            else:
                print(msg)

주소 형식 — 'AF_INET' 주소는 (hostname, port) 형태의 튜플로, hostname은 문자열이고 port는 정수예요. 'AF_UNIX' 주소는 파일시스템의 파일 이름을 나타내는 문자열. 'AF_PIPE' 주소는 r'\\.\pipe\PipeName' 형태의 문자열. Client()로 ServerName이라는 원격 컴퓨터의 명명된 파이프에 연결하려면 r'\\ServerName\pipe\PipeName' 형태의 주소를 써야 해요. 두 개의 백슬래시로 시작하는 문자열은 기본적으로 'AF_UNIX' 주소가 아니라 'AF_PIPE' 주소로 간주된다는 점을 주의하세요.

인증 키

Connection.recv를 쓸 때 받은 데이터는 자동으로 언픽클돼요. 불행히도 신뢰할 수 없는 소스의 데이터를 언픽클하는 것은 보안 위험이에요. 그래서 Listener와 Client()는 다이제스트 인증을 제공하기 위해 hmac 모듈을 사용해요. 인증 키는 비밀번호로 생각할 수 있는 바이트 문자열이에요: 연결이 설정되면 양쪽 끝이 상대방이 인증 키를 안다는 증거를 요구해요. (양쪽이 같은 키를 쓴다는 것을 보여주는 데 키를 연결을 통해 보내는 것은 수반되지 않아요.) 인증이 요청되지만 인증 키가 지정되지 않으면 current_process().authkey의 반환값이 사용돼요. 이 값은 현재 프로세스가 만드는 어떤 Process 객체에도 자동으로 상속돼요. 즉 (기본적으로) 멀티프로세스 프로그램의 모든 프로세스는 서로 간 연결 설정에 쓸 수 있는 단일 인증 키를 공유해요. 적합한 인증 키는 os.urandom()으로도 생성할 수 있어요. 이 인증은 주소로 도달할 수 있는 Listener와 Client() 연결을 보호해요. Pipe()가 만들거나 Queue가 내부적으로 쓰는 익명 파이프에는 적용되지 않아요. multiprocessing은 같은 사용자로 실행되는 모든 로컬 프로세스를 신뢰된 것으로 취급해요. 대부분의 OS에서 그런 프로세스들은 서로의 파이프 파일 디스크립터에 어떻게든 접근할 수 있거든요. 같은 사용자의 프로세스 간 격리가 필요한 애플리케이션은 OS 레벨에서 처리해야 해요 — 예를 들어 다른 사용자 계정이나 샌드박스에서 작업자를 실행하는 식으로.

로깅

약간의 로깅 지원이 가능해요. 다만 logging 패키지는 프로세스 공유 락을 사용하지 않으므로(핸들러 타입에 따라) 다른 프로세스의 메시지가 섞일 수 있어요.

• multiprocessing.get_logger() — multiprocessing이 쓰는 로거 반환. 필요하면 새로 만들어져요. 처음 만들어질 때 로거는 레벨 logging.NOTSET이고 기본 핸들러가 없어요. 이 로거에 보낸 메시지는 기본적으로 루트 로거로 전파되지 않아요. Windows에서 자식 프로세스는 부모 프로세스 로거의 레벨만 상속하고, 로거의 다른 커스터마이징은 상속되지 않는다는 점을 주의하세요.
• multiprocessing.log_to_stderr(level=None) — 이 함수는 get_logger()를 호출하지만, get_logger가 만든 로거를 반환하는 것 외에 '[%(levelname)s/%(processName)s] %(message)s' 형식으로 sys.stderr에 출력을 보내는 핸들러를 추가해요. level 인자를 넘겨 로거의 levelname을 수정할 수 있어요. 다음은 로깅이 켜진 예시 세션:

  >>> import multiprocessing, logging
  >>> logger = multiprocessing.log_to_stderr()
  >>> logger.setLevel(logging.INFO)
  >>> logger.warning('doomed')
  [WARNING/MainProcess] doomed
  >>> m = multiprocessing.Manager()
  [INFO/SyncManager-...] child process calling self.run()
  [INFO/SyncManager-...] created temp directory /.../pymp-...
  [INFO/SyncManager-...] manager serving at '/.../listener-...'
  >>> del m
  [INFO/MainProcess] sending shutdown message to manager
  [INFO/SyncManager-...] manager exiting with exitcode 0
  

  로깅 레벨의 전체 표는 logging 모듈을 참고하세요.

multiprocessing.dummy 모듈

multiprocessing.dummy는 multiprocessing의 API를 복제하지만 threading 모듈의 래퍼에 지나지 않아요. 특히 multiprocessing.dummy가 제공하는 Pool 함수는, 모든 같은 메서드 호출을 지원하지만 작업자 프로세스 대신 작업자 스레드 풀을 사용하는 Pool의 하위 클래스인 ThreadPool의 인스턴스를 반환해요.

• class multiprocessing.pool.ThreadPool([processes[, initializer[, initargs]]]) — 작업을 제출할 수 있는 작업자 스레드 풀을 제어하는 스레드 풀 객체. ThreadPool 인스턴스는 Pool 인스턴스와 완전히 인터페이스 호환되며, 리소스도 제대로 관리해야 해요(컨텍스트 매니저로 풀을 쓰거나 close()와 terminate()를 수동 호출). processes는 쓸 작업자 스레드 수. None이면 os.process_cpu_count()를 사용. initializer가 None이 아니면 각 작업자 프로세스가 시작할 때 호출돼요. Pool과 달리 maxtasksperchild와 context는 제공할 수 없어요.

  참고: ThreadPool은 Pool과 같은 인터페이스를 공유하는데, Pool은 프로세스 풀을 중심으로 설계됐고 concurrent.futures 모듈 도입보다 앞서요. 그래서 스레드 기반 풀에 말이 안 되는 일부 연산을 상속하고, 비동기 작업 상태를 나타내는 자신만의 타입 AsyncResult를 가지는데 이는 다른 라이브러리가 이해하지 못해요. 사용자는 일반적으로 처음부터 스레드 중심으로 설계된 더 단순한 인터페이스이고 asyncio를 포함한 많은 라이브러리와 호환되는 concurrent.futures.Future 인스턴스를 반환하는 concurrent.futures.ThreadPoolExecutor를 쓰는 게 좋아요.

프로그래밍 지침

multiprocessing을 사용할 때 지켜야 할 특정 지침과 관용법이 있어요.

모든 시작 방법

다음은 모든 시작 방법에 적용돼요.

• 공유 상태 피하기 — 가능한 한 프로세스 간에 많은 양의 데이터를 옮기는 것을 피해야 해요. 프로세스 간 통신에는 낮은 수준의 동기화 프리미티브를 쓰기보다 큐나 파이프를 쓰는 게 아마 가장 좋아요.
• 픽클 가능성 — 프록시 메서드의 인자가 픽클 가능한지 확인하세요.
• 프록시의 스레드 안전성 — 락으로 보호하지 않는 한 여러 스레드에서 프록시 객체를 사용하지 마세요. (다른 프로세스가 같은 프록시를 쓰는 데는 문제가 없어요.)
• 좀비 프로세스 join하기 — POSIX에서 프로세스가 끝났지만 join되지 않으면 좀비가 돼요. 새 프로세스가 시작될 때마다(또는 active_children()를 호출할 때마다) 아직 join되지 않은 완료 프로세스들이 모두 join되므로 아주 많이 쌓이지 않아요. 또한 끝난 프로세스의 Process.is_alive를 호출해도 그 프로세스를 join해요. 그래도 시작한 모든 프로세스를 명시적으로 join하는 것이 좋은 관행이에요.
• 피클/언피클보다 상속이 낫다 — spawn이나 forkserver 시작 방법을 쓸 때는 많은 multiprocessing 타입이 자식 프로세스가 쓸 수 있도록 픽클 가능해야 해요. 하지만 일반적으로 파이프나 큐로 공유 객체를 다른 프로세스에 보내는 것은 피해야 해요. 대신 다른 곳에서 만들어진 공유 리소스가 필요한 프로세스가 조상 프로세스에서 상속받을 수 있도록 프로그램을 설계하세요.
• 프로세스 종료 피하기 — Process.terminate 메서드로 프로세스를 멈추면, 프로세스가 현재 사용 중인 공유 리소스(락, 세마포어, 파이프, 큐 같은)가 고장나거나 다른 프로세스가 사용할 수 없게 될 가능성이 있어요. 그래서 공유 리소스를 전혀 쓰지 않는 프로세스에만 terminate를 쓰는 게 아마 가장 좋아요.
• 큐를 쓰는 프로세스 join하기 — 큐에 항목을 넣은 프로세스는 버퍼된 모든 항목이 "피더" 스레드에 의해 기본 파이프에 공급될 때까지 종료를 기다릴 것임을 명심하세요. (자식 프로세스는 큐의 Queue.cancel_join_thread 메서드를 호출해 이 동작을 피할 수 있어요.) 즉 큐를 쓸 때마다, 큐에 넣은 모든 항목이 프로세스가 join되기 전에 결국 제거되도록 해야 해요. 아니면 큐에 항목을 넣은 프로세스가 종료될 것이라고 확신할 수 없어요. 또한 비데몬성 프로세스는 자동으로 join된다는 것도 기억하세요. 교착 상태가 되는 예시:

  from multiprocessing import Process, Queue
  
  def f(q):
      q.put('X' * 1000000)
  
  if __name__ == '__main__':
      queue = Queue()
      p = Process(target=f, args=(queue,))
      p.start()
      p.join()                    # this deadlocks
      obj = queue.get()
  

  수정은 마지막 두 줄을 바꾸면 돼요(또는 그냥 p.join() 줄을 제거).
• 자식 프로세스에 리소스를 명시적으로 전달 — POSIX에서 fork 시작 방법을 쓰면 자식 프로세스는 전역 리소스로 부모 프로세스에서 만든 공유 리소스를 쓸 수 있어요. 하지만 객체를 자식 프로세스의 생성자에 인자로 전달하는 것이 더 좋아요. 코드를 (잠재적으로) Windows와 다른 시작 방법과 호환되게 하는 것 외에도, 자식 프로세스가 살아 있는 동안 객체가 부모 프로세스에서 가비지 컬렉션되지 않음을 보장하기 때문이에요. 예를 들어

  from multiprocessing import Process, Lock
  
  def f():
      ... do something using "lock" ...
  
  if __name__ == '__main__':
      lock = Lock()
      for i in range(10):
          Process(target=f).start()
  

  는 다음과 같이 다시 써야 해요.

  from multiprocessing import Process, Lock
  
  def f(l):
      ... do something using "l" ...
  
  if __name__ == '__main__':
      lock = Lock()
      for i in range(10):
          Process(target=f, args=(lock,)).start()
  

• sys.stdin을 "파일류 객체"로 교체할 때 주의 — multiprocessing은 원래 multiprocessing.Process._bootstrap() 메서드에서 무조건적으로 os.close(sys.stdin.fileno())를 호출했어요 — 이것이 프로세스 속 프로세스 문제를 일으켰죠. 이제는 다음과 같이 바뀌었어요:

  sys.stdin.close()
  sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)
  

  이는 프로세스들이 서로 충돌해 잘못된 파일 디스크립터 오류가 나는 근본 문제를 해결하지만, 출력 버퍼링이 있는 "파일류 객체"로 sys.stdin()을 교체하는 애플리케이션에는 잠재적 위험을 도입해요. 그 위험은 여러 프로세스가 이 파일류 객체에 close()를 호출하면 같은 데이터가 객체에 여러 번 플러시되어 손상될 수 있다는 것이에요. 파일류 객체를 작성하고 자신만의 캐싱을 구현한다면, 캐시에 추가할 때마다 pid를 저장하고 pid가 바뀌면 캐시를 버려 fork에 안전하게 만들 수 있어요. 예:

  @property
  def cache(self):
      pid = os.getpid()
      if pid != self._pid:
          self._pid = pid
          self._cache = []
      return self._cache
  

  자세한 내용은 bpo-5155, bpo-5313, bpo-5331 참고.

spawn과 forkserver 시작 방법

fork 시작 방법에 적용되지 않는 몇 가지 추가 제약이 있어요.

• 더 많은 픽클 가능성 — Process에 대한 모든 인자가 픽클 가능한지 확인하세요. 또한 Process.__init__을 하위 분류한다면 Process.start 메서드가 호출될 때 인스턴스가 픽클 가능해야 합니다.
• 전역 변수 — 자식 프로세스에서 실행되는 코드가 전역 변수에 접근하려 하면, 그것이 보는 값(있으면)은 Process.start가 호출될 때 부모 프로세스의 값과 같지 않을 수 있음을 명심하세요. 하지만 그냥 모듈 레벨 상수인 전역 변수는 문제를 일으키지 않아요.
• 메인 모듈의 안전한 임포트 — 메인 모듈이 (새 프로세스를 시작하는 것 같은) 의도하지 않은 부수 효과 없이 새 Python 인터프리터에 의해 안전하게 임포트될 수 있는지 확인하세요. 예를 들어 spawn이나 forkserver 시작 방법을 쓰면서 다음 모듈을 실행하면 RuntimeError로 실패해요:

  from multiprocessing import Process
  
  def foo():
      print('hello')
  
  p = Process(target=foo)
  p.start()
  

  대신 프로그램의 "진입점"을 이렇게 if __name__ == '__main__':으로 보호해야 해요:

  from multiprocessing import Process, freeze_support, set_start_method
  
  def foo():
      print('hello')
  
  if __name__ == '__main__':
      freeze_support()
      set_start_method('spawn')
      p = Process(target=foo)
      p.start()
  

  (freeze_support() 줄은 프로그램이 frozen이 아니라 정상 실행된다면 생략할 수 있어요.) 이로써 새로 생성된 Python 인터프리터가 모듈을 안전하게 임포트한 다음 모듈의 foo() 함수를 실행할 수 있어요. 풀이나 매니저가 메인 모듈에서 만들어진 경우에도 유사한 제약이 적용돼요.

예시

커스터마이즈된 매니저와 프록시를 만들고 사용하는 방법의 데모:

from multiprocessing import freeze_support
from multiprocessing.managers import BaseManager, BaseProxy
import operator

##

class Foo:
    def f(self):
        print('you called Foo.f()')
    def g(self):
        print('you called Foo.g()')
    def _h(self):
        print('you called Foo._h()')

# A simple generator function
def baz():
    for i in range(10):
        yield i*i

# Proxy type for generator objects
class GeneratorProxy(BaseProxy):
    _exposed_ = ['__next__']
    def __iter__(self):
        return self
    def __next__(self):
        return self._callmethod('__next__')

# Function to return the operator module
def get_operator_module():
    return operator

##

class MyManager(BaseManager):
    pass

# register the Foo class; make `f()` and `g()` accessible via proxy
MyManager.register('Foo1', Foo)

# register the Foo class; make `g()` and `_h()` accessible via proxy
MyManager.register('Foo2', Foo, exposed=('g', '_h'))

# register the generator function baz; use `GeneratorProxy` to make proxies
MyManager.register('baz', baz, proxytype=GeneratorProxy)

# register get_operator_module(); make public functions accessible via proxy
MyManager.register('operator', get_operator_module)

##

def test():
    manager = MyManager()
    manager.start()

    print('-' * 20)

    f1 = manager.Foo1()
    f1.f()
    f1.g()
    assert not hasattr(f1, '_h')
    assert sorted(f1._exposed_) == sorted(['f', 'g'])

    print('-' * 20)

    f2 = manager.Foo2()
    f2.g()
    f2._h()
    assert not hasattr(f2, 'f')
    assert sorted(f2._exposed_) == sorted(['g', '_h'])

    print('-' * 20)

    it = manager.baz()
    for i in it:
        print('<%d>' % i, end=' ')
    print()

    print('-' * 20)

    op = manager.operator()
    print('op.add(23, 45) =', op.add(23, 45))
    print('op.pow(2, 94) =', op.pow(2, 94))
    print('op._exposed_ =', op._exposed_)

##

if __name__ == '__main__':
    freeze_support()
    test()

Pool 사용:

import multiprocessing
import time
import random
import sys

#
# Functions used by test code
#

def calculate(func, args):
    result = func(*args)
    return '%s says that %s%s = %s' % (
        multiprocessing.current_process().name,
        func.__name__, args, result
        )

def calculatestar(args):
    return calculate(*args)

def mul(a, b):
    time.sleep(0.5 * random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5 * random.random())
    return a + b

def f(x):
    return 1.0 / (x - 5.0)

def pow3(x):
    return x ** 3

def noop(x):
    pass

#
# Test code
#

def test():
    PROCESSES = 4
    print('Creating pool with %d processes\n' % PROCESSES)

    with multiprocessing.Pool(PROCESSES) as pool:
        #
        # Tests
        #

        TASKS = [(mul, (i, 7)) for i in range(10)] + \
                [(plus, (i, 8)) for i in range(10)]

        results = [pool.apply_async(calculate, t) for t in TASKS]
        imap_it = pool.imap(calculatestar, TASKS)
        imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)

        print('Ordered results using pool.apply_async():')
        for r in results:
            print('\t', r.get())
        print()

        print('Ordered results using pool.imap():')
        for x in imap_it:
            print('\t', x)
        print()

        print('Unordered results using pool.imap_unordered():')
        for x in imap_unordered_it:
            print('\t', x)
        print()

        print('Ordered results using pool.map() --- will block till complete:')
        for x in pool.map(calculatestar, TASKS):
            print('\t', x)
        print()

        #
        # Test error handling
        #

        print('Testing error handling:')

        try:
            print(pool.apply(f, (5,)))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from pool.apply()')
        else:
            raise AssertionError('expected ZeroDivisionError')

        try:
            print(pool.map(f, list(range(10))))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from pool.map()')
        else:
            raise AssertionError('expected ZeroDivisionError')

        try:
            print(list(pool.imap(f, list(range(10)))))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from list(pool.imap())')
        else:
            raise AssertionError('expected ZeroDivisionError')

        it = pool.imap(f, list(range(10)))
        for i in range(10):
            try:
                x = next(it)
            except ZeroDivisionError:
                if i == 5:
                    pass
            except StopIteration:
                break
            else:
                if i == 5:
                    raise AssertionError('expected ZeroDivisionError')

        assert i == 9
        print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
        print()

        #
        # Testing timeouts
        #

        print('Testing ApplyResult.get() with timeout:', end=' ')
        res = pool.apply_async(calculate, TASKS[0])
        while 1:
            sys.stdout.flush()
            try:
                sys.stdout.write('\n\t%s' % res.get(0.02))
                break
            except multiprocessing.TimeoutError:
                sys.stdout.write('.')
        print()
        print()

        print('Testing IMapIterator.next() with timeout:', end=' ')
        it = pool.imap(calculatestar, TASKS)
        while 1:
            sys.stdout.flush()
            try:
                sys.stdout.write('\n\t%s' % it.next(0.02))
            except StopIteration:
                break
            except multiprocessing.TimeoutError:
                sys.stdout.write('.')
        print()
        print()


if __name__ == '__main__':
    multiprocessing.freeze_support()
    test()

큐를 사용해 작업자 프로세스 모음에 작업을 공급하고 결과를 모으는 방법을 보여주는 예시:

import time
import random

from multiprocessing import Process, Queue, current_process, freeze_support

#
# Function run by worker processes
#

def worker(input, output):
    for func, args in iter(input.get, 'STOP'):
        result = calculate(func, args)
        output.put(result)

#
# Function used to calculate result
#

def calculate(func, args):
    result = func(*args)
    return '%s says that %s%s = %s' % \
        (current_process().name, func.__name__, args, result)

#
# Functions referenced by tasks
#

def mul(a, b):
    time.sleep(0.5*random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5*random.random())
    return a + b

#
#
#

def test():
    NUMBER_OF_PROCESSES = 4
    TASKS1 = [(mul, (i, 7)) for i in range(20)]
    TASKS2 = [(plus, (i, 8)) for i in range(10)]

    # Create queues
    task_queue = Queue()
    done_queue = Queue()

    # Submit tasks
    for task in TASKS1:
        task_queue.put(task)

    # Start worker processes
    for i in range(NUMBER_OF_PROCESSES):
        Process(target=worker, args=(task_queue, done_queue)).start()

    # Get and print results
    print('Unordered results:')
    for i in range(len(TASKS1)):
        print('\t', done_queue.get())

    # Add more tasks using `put()`
    for task in TASKS2:
        task_queue.put(task)

    # Get and print some more results
    for i in range(len(TASKS2)):
        print('\t', done_queue.get())

    # Tell child processes to stop
    for i in range(NUMBER_OF_PROCESSES):
        task_queue.put('STOP')


if __name__ == '__main__':
    freeze_support()
    test()

더 알아보기

• concurrent.futures.ProcessPoolExecutor — 호출 프로세스의 실행을 막지 않고 백그라운드 프로세스에 작업을 밀어넣는 더 높은 수준의 인터페이스를 제공해요. Pool 인터페이스를 직접 쓰는 것과 비교해, concurrent.futures API는 기본 프로세스 풀에 작업을 제출하는 것과 결과를 기다리는 것을 더 쉽게 분리할 수 있게 해줘요.