[python] multiprocessing 사용법

파이썬으로 작업을 하다보면 multiprocessing을 사용할 일이 생기는데, 사용법과 주의할 점을 정리하고자 한다.

 

1. Multiprocessing 사용하기

1) Process, Queue

기본적인 형태는 Process를 선언하여 새로운 프로세스를 만들 때 실행할 function과 arguement들을 넘겨주는 방식이다. 그리고 Queue를 사용하여 프로세스끼리 데이터를 주고 받을 수 있다. 주의할 점은 새로운 프로세스를 실행할 때 넘겨줄 function과 arguement들은 fork가 가능한 객체들이어야 한다. 대부분은 문제 없지만 GPU 자원을 활용하는 모델이나 threading 기반으로 돌아가는 객체들은 안될 때가 있다.

아래 예제는 consumer & producer 프로세스를 생성한 예시이다.

import multiprocessing as mp
import time


def create_work(workload=0, work_queue):
    for _ in range(100):
        workload += 1
        work_queue.put(workload)
    
def finish_work(work_queue):
	while not work_queue.empty():
		workload = work_queue.get(False)  # non-blocking을 위함
    	workload -= 1


if __name__ == "__main__":  
	# Process를 선언하는 code가 main에 있어야 함!
    workloads = [100, 200]
    work_queue = mp.Queue()
	procs = []
    
	# producer process 선언
    for work in workloads:
        producer_proc = mp.Process(target=create_work, args=(work, work_queue))
        procs.append(producer_proc)
		producer_proc.start()


    # producer process 선언
    consumer_proc = mp.Process(target=finish_work, args=(work_queue,))
    procs.append(consumer_proc)
    consumer_proc.start()
    
    for proc in procs:
    	proc.join()

 

2) List, Dict

프로세스끼리 데이터를 주고 받을 때 사용가능한 mp.Manager().list(), mp.Manager().dict()도 있다!

import multiprocessing as mp
import time

    
def other_work(work_list, work_dict):
	for work in range(100):
	    work_list.append(work)
        work_dict["a"] = work_dict["a"] + work


if __name__ == "__main__":  
	# Process를 선언하는 code가 main에 있어야 함!
    workloads = [100, 200]
    work_list = mp.Manager().list()
    work_dict = mp.Manager().dict()
    work_dict["a"] = 0
	
    proc = mp.Process(target=other_work, args=(work_list, work_dict))
    
	proc.start()
    proc.join()

 

2. 주의할 사항, 참고할 사항

1) .get(), .get_nowait() 차이

Queue가 비어있는 상황에서 .get()으로 데이터를 불러오면, 새로운 데이터가 들어올 때 까지 기다리기 때문에 block이 발생한다. 이를 우회하기 위해서 queue.empty()를 사용하거나, get_nowait() 또는 .get(False)와 같이 사용하면 된다.

 

2) List, Dict등을 사용할 때 

mp.Manager().dict(), mp.Manager().list()를 사용해서 생성한 shared variable들은 python native한 연산을 사용하면 안된다. +=, list+[1]를 사용하면 안되고 아래와 같이 사용해야 한다.

import multiprocessing as mp
import time

    
def other_work(work_list, work_dict):
	for work in range(100):
    	# 올바른 방법
	    work_list.append(work)
        work_dict["a"] = work_dict["a"] + work
        
        # 안되는 방법
		work_list += [work]
        work_dict["a"] -= work

if __name__ == "__main__":  
	# Process를 선언하는 code가 main에 있어야 함!
    workloads = [100, 200]
    work_list = mp.Manager().list()
    work_dict = mp.Manager().dict()
    work_dict["a"] = 0
	
    proc = mp.Process(target=other_work, args=(work_list, work_dict))
    
	proc.start()
    proc.join()

 

 

 

 

[참고자료]

• https://www.digitalocean.com/community/tutorials/python-multiprocessing-example
• https://stackoverflow.com/questions/56364119/managed-dict-of-list-not-updated-in-multiprocessing-when-using-operator