파이썬 Multiprocessing

작성일 | 분류 , |
토요일 오후, 대형마트에는 장을 보러 온 500명의 고객이 계산을 기다리고 있다. 만약 계산대가 하나뿐이라면 한 명당 2분씩 걸린다고 해도 모든 고객이 계산을 마치는 데 무려 16시간이 넘게 걸릴 것이다. 하지만 2

토요일 오후, 대형마트에는 장을 보러 온 500명의 고객이 계산을 기다리고 있다. 만약 계산대가 하나뿐이라면 한 명당 2분씩 걸린다고 해도 모든 고객이 계산을 마치는 데 무려 16시간이 넘게 걸릴 것이다. 하지만 20개의 계산대가 동시에 운영되면 각 계산대마다 25명씩 나누어 처리하므로 약 50분이면 모든 고객이 계산을 마칠 수 있다.

특히 한 계산대에서 바코드 인식 오류나 카드 결제 문제가 발생하더라도 다른 19개의 계산대는 영향받지 않고 계속 고객을 처리할 수 있어, 전체적인 마트 운영이 안정적으로 유지된다. 이처럼 멀티프로세싱은 작업을 여러 개의 독립적인 처리 단위로 나누어 동시에 실행함으로써 전체 처리 시간을 획기적으로 단축시키고 시스템의 안정성도 높여준다.

파이썬의 multiprocessing 라이브러리는 바로 이러한 원리를 프로그래밍에 적용한 것으로, 전역 인터프리터 잠금(Global Interpreter Lock, GIL)을 우회하여 여러 CPU 코어를 동시에 활용하는 진정한 의미의 병렬 처리를 가능하게 하는 표준 라이브러리다. 각 프로세스는 독립적인 메모리 공간과 파이썬 인터프리터를 가지므로, 계산 집약적인(CPU-bound) 작업의 성능을 극대화할 수 있다.

1. 핵심 개념: 프로세스 기반 병렬 처리

multiprocessing은 threading과 비슷한 방식으로 사용할 수 있지만, 실제로는 완전히 독립된 프로세스들을 만들어서 작업을 처리한다. 스레드가 한 집 안의 여러 방에서 일하는 것이라면, 프로세스는 아예 다른 집에서 각자 일하는 것과 같다.

Process 클래스의 기본 사용법

Process 클래스는 병렬로 실행할 작업을 담는 그릇이다. target에는 실행할 함수를 지정하고, args에는 그 함수에 전달할 값들을 튜플로 넣는다. 예를 들어 Process(target=cook, args=(“파스타”, 2))라고 하면 cook 함수에 “파스타”와 2라는 인자를 전달해서 실행하겠다는 의미다.

프로세스의 생명주기

프로세스를 만들고 실행하는 과정은 세 단계로 이루어진다. 먼저 Process() 객체를 생성하는 것은 일꾼을 고용하는 것과 같다. 그 다음 start()를 호출하면 실제로 새로운 프로세스가 만들어지고 지정된 함수가 실행된다. 마지막으로 join()을 호출하면 해당 프로세스가 작업을 완료할 때까지 기다린다. 여러 프로세스를 동시에 실행시킨 후 모두 끝나기를 기다려야 할 때join()은 필수적이다.

if name == “main”: 보호 장치

멀티프로세싱 코드는 반드시 if name == “main”: 블록 안에 작성해야 한다. 이것은 매우 중요한 안전장치다. 새로운 프로세스가 생성될 때 파이썬은 원본 스크립트를 다시 불러오는데, 이 보호 장치가 없으면 각 자식 프로세스가 또 다른 프로세스를 만들고, 그 프로세스가 또 프로세스를 만드는 무한 반복에 빠진다. 마치 거울 속의 거울처럼 끝없이 프로세스가 생성되어 컴퓨터가 멈출 수 있다. 이 한 줄의 코드가 그런 재앙을 막아준다.

2. 프로세스 생성 및 관리 방법

multiprocessing은 작업을 병렬로 처리하는 두 가지 주요 방법을 제공한다. 상황에 따라 적절한 방법을 선택하면 효율적인 병렬 처리가 가능하다.

Process 클래스 직접 사용

가장 기본적이고 직관적인 방법이다. 마치 직원을 한 명씩 직접 채용해서 특정 업무를 맡기는 것과 같다. 각 프로세스를 개별적으로 생성하고, 시작하고, 종료를 기다리는 과정을 직접 관리한다. 예를 들어 3개의 파일을 각각 다른 방식으로 처리해야 한다면, 3개의 Process 객체를 만들어 각각 다른 함수를 실행시킬 수 있다. 이 방식은 적은 수의 작업을 처리하거나, 각 프로세스가 서로 다른 일을 해야 할 때 특히 유용하다. 프로세스 하나하나를 세밀하게 제어할 수 있다는 장점이 있지만, 수십 개 이상의 프로세스를 다뤄야 한다면 코드가 복잡해질 수 있다.

import multiprocessing
import os
import time

def worker(task_id):
    """자식 프로세스에서 실행될 작업"""
    print(f"작업 {task_id}: 프로세스 ID {os.getpid()} 시작")
    time.sleep(2)
    print(f"작업 {task_id}: 종료")

if __name__ == "__main__":
    processes = []
    for i in range(3):
        # target에 함수, args에 인자를 전달하여 프로세스 생성
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start() # 프로세스 시작

    print("메인 프로세스: 모든 자식 프로세스가 끝날 때까지 대기")
    for p in processes:
        p.join() # 각 프로세스가 종료될 때까지 대기

    print("메인 프로세스: 모든 작업 완료")

이 코드는 3개의 독립적인 프로세스를 만들어서 동시에 작업을 처리하는 방법을 보여준다. 각 프로세스는 자신만의 고유한 프로세스 ID를 가지고 독립적으로 실행된다.

worker 함수는 각 자식 프로세스에서 실행될 작업을 정의한다. 함수가 시작되면 자신의 작업 번호와 프로세스 ID를 출력하고, 2초 동안 대기한 후 종료 메시지를 출력한다. os.getpid()를 통해 각 프로세스가 서로 다른 ID를 가진 독립적인 프로세스임을 확인할 수 있다.

메인 코드에서는 빈 리스트를 만들어 생성된 프로세스들을 관리한다. for 루프를 통해 3개의 프로세스를 순차적으로 생성하는데, 각 프로세스는 worker 함수를 실행하도록 설정되고 고유한 작업 번호(0, 1, 2)를 인자로 받는다. p.start()를 호출하는 순간 새로운 프로세스가 생성되어 독립적으로 worker 함수를 실행하기 시작한다.

모든 프로세스를 시작한 후, 메인 프로세스는 “모든 자식 프로세스가 끝날 때까지 대기”라는 메시지를 출력한다. 그 다음 각 프로세스에 대해 join()을 호출하여 해당 프로세스가 완료될 때까지 기다린다. 3개의 프로세스가 동시에 실행되므로 전체 실행 시간은 약 2초가 된다. 만약 순차적으로 실행했다면 6초가 걸렸을 작업을 2초 만에 완료하는 것이다. 모든 프로세스가 종료되면 “모든 작업 완료” 메시지가 출력되고 프로그램이 종료된다.

Pool을 사용한 고수준 병렬 처리

Pool은 프로세스들을 미리 만들어두고 작업을 효율적으로 분배하는 똑똑한 관리자와 같다. 식당에서 주문이 들어올 때마다 요리사를 새로 고용하는 것이 아니라, 미리 여러 명의 요리사를 대기시켜두고 주문을 분배하는 것과 비슷한 원리다.

Pool의 작동 방식

Pool(processes=4)를 실행하면 4개의 작업자 프로세스가 미리 생성되어 대기 상태에 들어간다. 프로세스 수를 지정하지 않으면 컴퓨터의 CPU 코어 수만큼 자동으로 생성된다. 이렇게 미리 만들어둔 프로세스들은 작업이 들어오면 즉시 처리를 시작할 수 있어서, 매번 새로운 프로세스를 생성하는 것보다 훨씬 빠르다.

map 메서드의 강력함

pool.map(func, iterable)은 Pool의 가장 핵심적인 기능이다. 예를 들어 pool.map(square, [1, 2, 3, 4, 5])를 실행하면, 각 숫자에 대해 square 함수를 병렬로 실행한다. 일반적인 파이썬의 map 함수와 사용법은 동일하지만, 여러 프로세스가 동시에 작업을 처리한다는 점이 다르다.

map 메서드는 모든 작업이 완료될 때까지 기다린 후, 결과를 원래 순서대로 정렬된 리스트로 반환한다. 100개의 이미지를 처리한다면, Pool이 알아서 4개의 프로세스에 25개씩 분배하여 동시에 처리하고, 최종적으로 100개의 결과를 순서대로 돌려준다. 이런 방식으로 대량의 데이터를 처리할 때 코드는 간단하면서도 실행 속도는 크게 향상된다.

import multiprocessing
import time

def power(x, n):
    return x ** n

if __name__ == "__main__":
    tasks = [(2, 8), (3, 5), (4, 4), (5, 3)]

    # with 구문을 사용하면 풀이 자동으로 닫히고 정리됩니다.
    with multiprocessing.Pool(processes=4) as pool:
        # starmap은 튜플 형태의 인자들을 각각 풀어서 함수에 전달합니다.
        results = pool.starmap(power, tasks)

    print(f"결과: {results}") # 출력: [256, 243, 256, 125]

이 코드는 여러 개의 거듭제곱 연산을 동시에 처리하는 방법을 보여준다. power 함수는 두 개의 인자(x와 n)를 받아서 x의 n제곱을 계산하는 간단한 함수다. tasks 리스트에는 (2, 8), (3, 5) 같은 튜플들이 들어있는데, 각 튜플은 밑수와 지수의 쌍을 나타낸다.

여기서 starmap의 특별한 기능이 드러난다. 일반적인 map은 하나의 인자만 전달할 수 있지만, starmap은 각 튜플을 자동으로 풀어서 함수에 전달한다. 즉, (2, 8)은 power(2, 8)로, (3, 5)는 power(3, 5)로 실행되는 것이다. 이렇게 여러 개의 인자를 가진 함수도 쉽게 병렬 처리할 수 있다.

with multiprocessing.Pool(processes=4) as pool: 구문을 사용하면 Pool 관리가 매우 편해진다. 4개의 프로세스를 가진 Pool을 생성하고, with 블록을 벗어나면 자동으로 Pool이 닫히고 모든 프로세스가 정리된다. 파일을 열고 닫는 것처럼 Pool 자원을 깔끔하게 관리할 수 있는 파이썬의 편리한 기능이다.

실행 결과를 보면 4개의 프로세스가 동시에 각각의 거듭제곱을 계산한다. 2의 8제곱은 256, 3의 5제곱은 243, 4의 4제곱은 256, 5의 3제곱은 125가 된다. Pool은 작업을 분배하고 결과를 수집하여 원래 순서대로 [256, 243, 256, 125]라는 리스트로 반환한다. 이 연산들이 동시에 처리되어 순차적으로 처리하는 것보다 전체 실행 시간이 크게 단축된다.

3. 프로세스 간 데이터 교환 (IPC)

프로세스들은 각자 독립된 메모리 공간을 사용하기 때문에 서로 데이터를 주고받으려면 특별한 방법이 필요하다. 이런 프로세스 간 통신 방법을 IPC(Inter-Process Communication)라고 부른다.

Queue

Queue는 여러 프로세스가 데이터를 안전하게 주고받을 수 있는 통로다. 편의점의 택배 보관함을 생각하면 이해하기 쉽다. 한 사람(생산자 프로세스)이 택배를 보관함에 넣으면(put), 다른 사람(소비자 프로세스)이 그것을 꺼내간다(get). 여러 사람이 동시에 접근해도 내부적으로 잠금장치가 있어서 택배가 사라지거나 손상되는 일이 없다. 특히 여러 프로세스가 동시에 데이터를 넣고 빼는 상황에서도 안전하게 작동한다는 것이 큰 장점이다.

from multiprocessing import Process, Queue

def producer(queue):
    for i in range(5):
        queue.put(f"데이터 {i}")
    queue.put(None) # 작업 종료 신호

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"소비: {item}")

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

이 코드는 한 프로세스가 데이터를 생성하고, 다른 프로세스가 그 데이터를 소비하는 전형적인 패턴을 보여준다. 택배 기사가 물건을 배송함에 넣고, 고객이 그것을 꺼내가는 것과 비슷한 방식이다.

생산자(producer) 프로세스는 “데이터 0”부터 “데이터 4”까지 5개의 메시지를 큐에 넣는다. 모든 데이터를 넣은 후에는 None을 넣어서 작업이 끝났다는 신호를 보낸다. 이것은 마치 택배 기사가 마지막에 “오늘 배송 끝” 이라는 메모를 남기는 것과 같다.

소비자(consumer) 프로세스는 큐에서 계속 데이터를 꺼내서 화면에 출력한다. while True 루프를 돌면서 새로운 데이터가 있는지 확인하고, None을 받으면 더 이상 올 데이터가 없다는 뜻이므로 작업을 종료한다.

메인 코드에서는 하나의 Queue를 만들고, 생산자와 소비자 프로세스를 각각 생성한다. 두 프로세스는 동시에 실행되는데, 생산자가 데이터를 넣는 속도와 소비자가 꺼내는 속도가 달라도 Queue가 알아서 데이터를 안전하게 보관해준다. 마지막에 join()으로 두 프로세스가 모두 끝날 때까지 기다린 후 프로그램이 종료된다. Queue덕분에 별도의 Lock 없이도 데이터가 손실되거나 꼬이는 일 없이 안전하게 전달된다.

Pipe

Pipe는 두 프로세스를 직접 연결하는 전화선 같은 것이다. multiprocessing.Pipe()를 호출하면 양쪽 끝을 나타내는 두 개의 연결 객체를 받는다. 한쪽에서 send()로 메시지를 보내면, 다른 쪽에서 recv()로 받을 수 있다. 전화통화처럼 양방향 통신이 가능하지만, 주로 두 프로세스 간의 1대1 통신에 사용된다. Queue보다 빠르지만 여러 프로세스가 동시에 사용하기에는 적합하지 않다.

from multiprocessing import Process, Pipe

def sender(conn):
    conn.send("Hello from sender")
    conn.close()

def receiver(conn):
    msg = conn.recv()
    print(f"수신: {msg}")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=sender, args=(child_conn,))
    p.start()
    receiver(parent_conn)
    p.join()

이 코드는 Pipe를 사용해서 두 프로세스가 전화통화하듯이 직접 메시지를 주고받는 방법을 보여준다. Pipe는 양쪽 끝이 있는 통신 선로를 만들어서 한쪽에서 보낸 데이터를 다른 쪽에서 받을 수 있게 해준다.

Pipe() 함수를 호출하면 두 개의 연결 객체가 반환된다. parent_conn은 부모 프로세스가 사용할 끝단이고, child_conn은 자식 프로세스가 사용할 끝단이다. 마치 전화기 두 대가 선으로 연결된 것과 같다.

sender 함수는 자식 프로세스에서 실행되며, conn.send()를 통해 “Hello from sender”라는 메시지를 파이프로 보낸다. 메시지를 보낸 후에는 conn.close()로 연결을 닫는다. 이는 전화를 끊는 것과 같은 동작이다.

receiver 함수는 부모 프로세스에서 실행되며, conn.recv()로 파이프를 통해 전달된 메시지를 받아서 화면에 출력한다. 여기서 주목할 점은 receiver가 별도의 프로세스로 실행되지 않고 메인 프로세스에서 직접 호출된다는 것이다.

실행 흐름을 보면, 먼저 자식 프로세스를 시작해서 sender 함수를 실행시킨다. 그 다음 메인 프로세스에서 receiver 함수를 호출해서 메시지를 받는다. Pipe의 특징은 한쪽에서 보낸 데이터를 다른 쪽에서 받을 때까지 기다린다는 점이다. 따라서 sender가 메시지를 보내기 전까지 receiver는 대기 상태에 있게 된다. 이런 방식으로 두 프로세스 간에 동기화된 통신이 가능하다.

공유 메모리 (Value, Array)

여러 프로세스가 같은 메모리 공간을 직접 사용할 수 있게 해주는 도구다. 마치 가족들이 냉장고를 공유해서 사용하는 것처럼, 모든 프로세스가 동일한 데이터에 접근할 수 있다.

Value와 Array의 차이

Value는 하나의 값(숫자, 문자 등)을 공유할 때 사용한다. 예를 들어 현재 처리된 작업의 개수를 세는 카운터 같은 것이다. Array는 여러 개의 값을 담는 배열을 공유할 때 사용한다. 각 프로세스의 작업 진행률을 기록하는 리스트 같은 것을 만들 수 있다.

장점과 주의사항

공유 메모리의 가장 큰 장점은 속도다. 프로세스 간에 데이터를 복사하거나 전송할 필요 없이 직접 읽고 쓸 수 있어서 매우 빠르다. 하지만 여러 프로세스가 동시에 같은 데이터를 수정하려고 하면 문제가 발생한다. 두 사람이 동시에 냉장고에서 마지막 남은 우유를 꺼내려고 하는 것처럼, 데이터가 엉망이 될 수 있다. 이런 문제를 ‘경쟁 상태(Race Condition)’라고 부른다.

따라서 공유 메모리를 사용할 때는 반드시 Lock을 함께 사용해야 한다. Lock은 한 번에 한 프로세스만 데이터에 접근하도록 보장해주는 안전장치 역할을 한다.

from multiprocessing import Process, Value, Lock

def increment(shared_value, lock):
    for _ in range(10000):
        with lock: # 락을 획득하고 블록을 빠져나갈 때 자동으로 해제
            shared_value.value += 1

if __name__ == "__main__":
    # 'i'는 타입을 나타내는 타입 코드 (integer)
    counter = Value('i', 0)
    lock = Lock()
    
    p1 = Process(target=increment, args=(counter, lock))
    p2 = Process(target=increment, args=(counter, lock))

    p1.start()
    p2.start()
    p1.join()
    p2.join()

    print(f"최종 값: {counter.value}") # 출력: 20000

이 코드는 두 개의 프로세스가 하나의 숫자를 함께 증가시키면서도 정확한 결과를 얻는 방법을 보여준다. 먼저 Value(‘i’, 0)으로 모든 프로세스가 접근할 수 있는 정수형 공유 변수를 만들고 초기값을 0으로 설정한다. 동시에 Lock()으로 잠금장치를 만드는데, 이것은 여러 프로세스가 동시에 공유 변수를 건드리지 못하게 막아주는 역할을 한다.

increment 함수는 공유 변수를 10,000번 증가시키는 간단한 작업을 수행한다. 핵심은 with lock: 구문인데, 이 덕분에 한 번에 하나의 프로세스만 카운터를 증가시킬 수 있다. 마치 화장실에 들어가면 자동으로 문이 잠기고, 나올 때 자동으로 열리는 것과 같은 원리다.

실행 과정을 보면, 두 개의 프로세스를 만들어서 각각 increment 함수를 실행하도록 설정한 후 동시에 시작시킨다. 두 프로세스가 모두 작업을 마칠 때까지 기다린 다음 최종 결과를 출력하면 정확히 20,000이 나온다. 만약 Lock을 사용하지 않았다면 두 프로세스가 동시에 같은 값을 읽고 수정하려다가 일부 증가 연산이 무시되어 20,000보다 작은 값이 나왔을 것이다. Lock이 있기 때문에 각 프로세스가 차례대로 안전하게 값을 증가시켜 항상 정확한 결과를 보장받을 수 있다.

4. 프로세스 동기화

여러 프로세스가 공유 자원에 접근할 때 충돌을 방지하고 데이터의 정확성을 보장하기 위한 도구들이다. Lock은 가장 기본적인 동기화 도구다. 화장실 문에 달린 잠금장치처럼, 한 번에 하나의 프로세스만 중요한 코드 영역(임계 영역)에 들어갈 수 있도록 한다. 한 프로세스가 Lock을 획득하면 다른 프로세스들은 그 Lock이 해제될 때까지 기다려야 한다.

Semaphore는 정해진 수만큼의 프로세스가 동시에 자원을 사용할 수 있게 하는 고급 잠금장치다. 주차장에 3개의 빈 자리가 있다면, 동시에 최대 3대의 차만 주차할 수 있는 것과 같은 원리다. 자원의 개수를 관리하여 과도한 접근을 제한한다.

Event는 프로세스 간에 신호를 주고받는 간단한 도구다. 육상 경기의 출발 신호탄처럼, 한 프로세스가 event.set()으로 “시작” 신호를 보내면, event.wait()로 기다리던 다른 프로세스들이 작업을 시작한다.

Condition은 Lock과 Event의 기능을 합친 복잡한 동기화 도구다. 식당에서 주문한 음식이 나올 때까지 기다리다가, 준비가 완료되면 알림을 받는 것과 비슷하다. 특정 조건이 충족될 때까지 여러 프로세스를 대기시켰다가, 조건이 만족되면 한꺼번에 깨워서 작업을 진행시킨다. 주로 생산자-소비자 패턴 같은 복잡한 상황에서 사용한다.

파이썬의 multiprocessing은 GIL의 한계를 넘어서 진정한 병렬 처리를 가능하게 하는 강력한 도구다. 대형마트의 계산대처럼 여러 작업을 동시에 처리함으로써 전체 처리 시간을 획기적으로 단축시킬 수 있다. Process 클래스로 개별 작업을 세밀하게 제어하거나, Pool로 대량의 작업을 효율적으로 분배하는 등 상황에 맞는 방법을 선택할 수 있다.

프로세스 간 통신을 위한 Queue와 Pipe, 데이터 공유를 위한 Value와 Array, 그리고 안전한 동기화를 위한 Lock, Semaphore, Event, Condition 등 다양한 도구들을 제공한다. 이러한 도구들을 적절히 활용하면 복잡한 병렬 처리 작업도 안전하고 효율적으로 구현할 수 있다.

특히 CPU 집약적인 작업에서 multiprocessing의 진가가 드러난다. 대용량 데이터 처리, 이미지 변환, 복잡한 계산 작업 등에서 순차 처리 대비 수배에서 수십 배의 성능 향상을 기대할 수 있다. 다만 프로세스 생성과 통신에는 오버헤드가 있으므로, 작업의 특성을 고려하여 적절한 병렬화 전략을 선택하는 것이 중요하다.

multiprocessing을 마스터하면 파이썬으로도 고성능 병렬 처리 프로그램을 작성할 수 있다. 현대의 멀티코어 CPU 환경에서 하드웨어 자원을 최대한 활용하여, 더 빠르고 효율적인 프로그램을 만들어보라.