Python multiprocessing

개요

파이썬은 기본적으로 멀티스레딩을 지원하지만, Global Interpreter Lock(GIL)의 영향으로 CPU-bound 작업에서는 실제로 병렬 실행을 보장하지 못함

이에 대한 대안 중 하나가 multiprocessing

multiprocessing 모듈은 파이썬의 표준 라이브러리에 포함되어 있으며, 여러 프로세스를 사용하여 작업을 병렬로 실행할 수 있게 해 줌

multiprocessing은 별도의 프로세스를 생성하며, 각 프로세스는 별도의 메모리 공간을 가지므로 GIL의 영향을 받지 않고  병렬 처리를 지원함

이를 통해 CPU-bound 작업의 성능을 향상할 수 있음

기본 사용법

import multiprocessing

def worker(num):
    """작업을 수행하는 함수"""
    print(f'Worker: {num}')

if __name__ == '__main__':
    # 부모 프로세스 (메인 프로세스)
    # 프로세스 스포닝 (자식 프로세스 생성)
    processes = []
    for i in range(5):
        p = multiprocessing.Process(name="SubProcess", target=worker, args=(i,))
        processes.append(p)
        p.start()

    # 모든 프로세스 종료 대기
    for p in processes:
        p.join()

multiprocessing 모듈에서 프로세스 스포닝은 Process 클래스의 인스턴스를 생성한 후 start( ) 메서드를 호출하면 됨

Process 클래스의 인스턴스를 생성할 때 생성될 자식 프로세스의 이름(name)과 위함 하고자 하는 일(함수, target), 인자(args)를 전달

※ 스포닝 : 부모 프로세스가 운영체제에 요청하여 자식 프로세스를 새로 만들어내는 과정

프로세스 풀 활용

multiprocessing.Pool을 사용하면 풀 내에서 여러 작업을 병렬로 실행 가능

multiprocessing.Pool 객체를 생성할 때 processes 매개변수를 사용하여 동시에 실행할 프로세스의 수를 지정 가능 ( 만약 지정하지 않는다면 CPU 코어 수에 따라 자동으로 설정됨 )

이를 통해 동시에 몇 개의 작업을 처리할지를 유연하게 관리할 수 있음

풀이 가득 차있는 상황에서 새로운 작업이 추가되면 작업은 작업 큐에 대기하며, 만약 작업 큐에 공간이 있으면 풀 내의 프로세스가 새로운 작업을 가져와서 처리함

from multiprocessing import Pool

def worker(num):
    """작업을 수행하는 함수"""
    return f'Worker: {num}'

if __name__ == '__main__':
    with Pool(processes=3) as pool:
        results = pool.map(worker, range(5))
    print(results)

데몬 활용

일반적으로 생성한 경우 부모 프로세스는 자식 프로세스가 종료되지 않으면 종료되지 않음

하지만 multiprocessing 모듈에서 daemon=True 옵션을 사용하면 부모 프로세스가 종료될 때 자식 프로세스도 함께 종료됨

 

데몬 프로세스를 사용하여 메인 프로세스에 영향을 주지 않으면서 동작시킬 작업들을 구현할 수 있음

• daemon=True 옵션을 주면 유용한 예시
  • 로그 감시 데몬
    시스템 로그 파일을 지속적으로 모니터링하고 변경사항을 감지하여 중요한 이벤트를 기록하는 데몬
    이 데몬은 백그라운드에서 지속적으로 실행되며, 부모 프로세스가 종료되면 함께 종료되는 것이 좋음
  • 자동 백업 프로세스
    주기적으로 파일 시스템을 스캔하여 중요한 데이터를 백업하는 데몬
    이러한 데몬은 사용자가 시스템에 로그인할 필요 없이 백그라운드에서 실행되며, 부모 프로세스가 종료되면 함께 종료되는 것이 좋음
    

# Daemon Process
# 메인 프로세스가 종료시 서브 프로세스도 종료됨
import multiprocessing as mp 
import time


def work():
    print("Sub Process start")
    time.sleep(5) 
    print("Sub Process end")


if __name__ == "__main__":
    print("Main Process start")
    proc = mp.Process(name="Sub Process", target=work, daemon=True)
    proc.start()
    print("Main Process end")

서브 프로세스 상태 확인 및 종료

import multiprocessing as mp 
import time

def work():
    while True:
        print("sub process is running") 
        time.sleep(1)


if __name__ == "__main__":
    p = mp.Process(target=work, name="SubProcess")
    print("Status: ", p.is_alive())

    p.start()
    print("Status: ", p.is_alive())

    time.sleep(5)   # 메인 프로세스 3초 대기
    p.kill()        # 서브 프로세스 종료
    print("Status: ", p.is_alive())

    p.join()        # 메인 프로세스는 서브 프로세스가 종료될 때까지 블록됨
    print("Status: ", p.is_alive())

 

결과

Status:  False
Status:  True
sub process is running
sub process is running
sub process is running
sub process is running
sub process is running
Status:  True
Status:  False

 

multiprocessing 사용 시 주의사항

• 메모리 사용량이 증가할 수 있으므로 대용량 데이터 처리 시 주의 필요
• 각 프로세스가 독립적인 메모리 공간을 가지므로 데이터를 공유하려면 명시적인 메커니즘(예: 공유 메모리, 큐 등)을 사용해야 함

import multiprocessing as mp

def worker(q):
    q.put(1) 
    q.put(2) 
    q.put(3)

if __name__ == "__main__":
    # multiprocessing.Manager().Queue()로 Queue 생성
    manager = mp.Manager()
    q = manager.Queue() 
    
    # 프로세스 생성
    processes = []
    for _ in range(3):
        p = mp.Process(target=worker, args=(q,))
        processes.append(p)
        p.start()

    # 모든 프로세스가 종료될 때까지 대기
    for p in processes:
        p.join()

    # 멀티프로세스 간에 데이터를 공유하기 위해 Queue를 사용하여 데이터를 가져옴
    results = []
    while not q.empty():
        results.append(q.get())

    print(results)

 

결과

[1, 2, 3, 1, 2, 3, 1, 2, 3]