파이썬 멀티프로세싱 풀

지난 이야기

지난 시간에는 파이썬 멀티프로세싱 중 프로세스(Process)¹를 이야기했다.
이번에는 멀티프로세싱의 또다른 방법인 풀(Pool)을 설명하겠다.

Pool

• 풀은 스타크래프트의 스포닝 풀을 생각하면 된다.
  • 산란못을 지으면 부화장에서 저글링을 생성할 수 있다. 부화장을 많이 지을수록 더 많은 저글링이 생성된다.
  • 여기서
    • 산란못을 풀 객체
    • 저글링 생성을 프로세스(함수)
    • 부화장을 프로세서
    • 저글링을 반환값
    • 이라 보면 된다.
  • 산란못을 지은 뒤 부화장들에게 저글링을 계속 생성하라고 명령을 내린다.
  • 부화장들은 명령을 받아 저글링을 생산할 것이다. 저글링 생성이 끝난 부화장은 명령을 받아 또 저글링을 생성할 것이다.
  • 산란못만 있는 군락지에서는 저글링을 생산할 수 있지만, 히드라는 생성할 수 없다. 히드라를 생성하려면 히드라리스크 굴이라는 또 다른 객체를 만들어 줘야 한다.
• 즉 하나의 풀 객체에서는 하나의 함수만 실행시킬 수 있다. 멀티프로세싱 프로세스가 각 프로세스 객체마다 다른 함수를 배치할 수 있는 것과 차이점이다.
• 그럼 풀은 어쩔 때 유리하나. 눈치가 빠르면 알겠지만 같은 함수를 인자만 다르게 하여 여러 번 실행해야 할 때 프로세스보다 간단하게 멀티프로세싱을 사용할 수 있다.
  • 풀 객체를 선언할 때 프로세서 개수를 설정하면 객체에 함수와 인자들을 할당했을 때 풀 객체는 각 프로세서들에게 인자 하나씩을 할당해주고, 프로세서가 그 인자를 받아 함수를 끝내면 남아있는 인자들 중 또 알아서 프로세서에게 할당해주는 원리이다.
  • 프로세스는 부화장 하나마다 일일이 저글링 생성 명령을 내린다 보면 되고, 풀은 부화장 전부를 부대지정 한 후-그럴 수는 없지만-저글링 생성 명령을 한번에 하는 것으로 볼 수 있다.

사용법과 예제

기초 사용법

• 백문이 불여일견. 우선 간단한 사용법을 알려주자면

from multiprocessing import Pool
(변수명) = Pool( (워커 개수) )
(변수명).map( (실행 함수 이름) , (함수 인자 리스트) )
(변수명).close()
(변수명).join()

• 먼저 Pool() 객체를 선언해야 한다. (워커 개수)는 풀 객체에서 사용할 프로세서 개수를 의미한다.
  그냥 괄호 안에 입력해도 되고, processes=(개수) 이렇게 명시적으로 입력해도 된다. 입력하지 않으면 os.cpu_count() 값으로 자동 처리된다.
• .map()에서 (함수 인자 리스트)는 리스트 형태로 받아야 한다. 함수에 들어갈 인자들을 리스트 형태로 만들어서 넣어주자.
  • 만일 함수가 두 개 이상의 인자를 받는 경우에는(즉 def func(a,b) 이런 형태) .map() 대신 .starmap()을 사용해야 하고, 각 인자는 투플 형태로 구성되어야 한다. (즉 풀에 넣는 인자 리스트는 [(1,2),(3,4),…] 이런 식)
    이에 관해서는 아래 실전 예제에서 설명하겠다.
  • .map() 대신 위 .starmap(), .imap() 등을 사용할 수도 있다. 이것도 마찬가지로 실전 예제에서 다룬다.
• pool.close()는 이 작업이 끝나면 풀 객체를 닫는다는 말이다. 이후 또다른 풀 객체를 생성했을 때 꼬일 수 있기 때문이다.
  • pool.terminate()라는 변종도 있다. 이는 작업이 끝났든 말든 풀 객체를 닫아버린다.
• pool.join()은 프로세스 때와 마찬가지로 풀 객체가 닫히면 다음 줄로 넘어갈 수 있다는 말이다.

간단한 예제

마찬가지로 1부터 100 출력을 예제로 들어보겠다.

from multiprocessing import Pool

tosso = [i for i in range(1,101)]

def chul(su):
    print(su)
    return su

if __name__ == "__main__":
  pool = Pool(processes=4)
  pool.map(chul, tosso)
  pool.close()
  pool.join()

• tosso는 1부터 100까지의 수가 들어있는 리스트이다.
• chul 함수는 그냥 인자를 출력해주는 함수이다.
• main 함수에서 풀 객체를 생성한 뒤, chul 함수와 tosso 인자를 할당해 실행시킨다.
• 이제 pool 객체에서는 프로세스들에게 tosso 내 인자들을 하나씩 주어, chul 함수를 실행시키라고 명령한다.
• 프로세서들은 이를 실행하고, 끝나면 pool 객체에서 실행 안된 인자를 또 받아 chul을 실행한다. 각 프로세스에게 어떤 인자를 주는지는 pool 객체가 알아서 정한다.
  • 즉 여기서 chul 함수는 100번 실행되는 것이다.
• 여기서도 또한 출력 순서가 뒤바뀔 수 있다. 인자 할당 순서는 pool 객체 맘대로이기 때문이다.

실전 예제

돌아온 실전 예제다. 이번에는 본인이 다른 프로젝트에서 사용한 업로드 작업을 예시로 들어보겠다.
(마찬가지로 보안상 코드 간략화 및 검열이 있었음을 알린다.)

for (path, dirs, files) in os.walk(file_location):
  for myfile in files:
      param_list.append(
          (path, myfile))

pool = Pool(os.cpu_count()*2)

for _ in tqdm(pool.istarmap(upload_data, param_list), total=len(param_list)):
  pass
pool.close()
pool.join()

• 파일 디렉토리 변수인 file_location 내 파일명 및 해당 파일의 경로들을 인자로 담아 풀에서 upload_data 함수를 실행하는 작업이다.
• 우선 os.walk()를 사용해 디렉토리 내 파일명과 파일 경로를 뽑아낸다.
  • os.walk()는 os 라이브러리에서 제공하는 탐색 기능으로, 괄호 안에 디렉토리 위치를 넣으면 해당 디렉토리 내 모든 하위 디렉토리를 찾아 (하위 디렉토리의 절대경로,절대경로 내 폴더 리스트,절대경로 내 파일 리스트) 이런 투플로 반환해준다.
  • 따라서 for (path,dirs,files) in os.walk() 를 하면 모든 하위 디렉토리의 경로/폴더들/파일들 을 탐색한다.
  • files는 리스트로 반환되므로 for myfile in files 로 파일 하나하나씩 경로(path)와 함께 투플로 만들어 인자 리스트(param_list)에 넣어준다.
• pool 객체 생성…
• tqdm은 pip 라이브러리 중 하나로 반복문과 함께 쓰여 진행상황을 콘솔창에 시각화하여 알려준다. total=() 을 하면 진행상황의 전체 길이를 선언할 수 있다.
  • 즉 인자 리스트의 길이가 10000개인 경우 total=10000 이렇게
• pool.istarmap() 에 upload_data 함수 및 아까 만든 인자 리스트 param_list를 할당해준다.
  • 위에서 말했듯이 인자 리스트의 요소들이 투플((경로,파일) 형태)이기에 pool.map() 대신 pool.starmap() 을 사용해야 한다.
  • map 에는 앞에 i를 붙여 imap으로 사용할 수 있다.
    • pool은 같은 함수를 여러번 실행하기에 함수에 반환값이 있으면 모아놨다가 pool 작업이 끝나면 한번에 반환해 준다.
    • map은 이 반환값들을 리스트 형태로, imap은 이터레이터 형태로 반환해 준다.
    • 본인은 메모리를 줄이기 위해 이터레이터 형태로 반환값을 받길 원했다.
      • 이터레이터는 모든 요소들을 전부 메모리에 미리 불러오는 것이 아닌 호출할 때마다 그 요소만 메모리에 넣기 때문이다.
    • pool에 할당하는 함수는 인자를 2개 받기에 starmap을 사용해야 한다. 하지만 multiprocessing pool 라이브러리는 istarmap이라는 기능을 제공해주지 않는다. 따라서 본인은 https://stackoverflow.com/a/57364423 을 참고하여 istarmap을 직접 만들었다.
• pool 객체 실행을 tqdm 내에 할당했기에 pool 함수가 하나 끝날때마다 tqdm 상황 바가 업데이트된다.

마치며

이상이다.

• 작성 예정 글
  • 쓰레딩-thread
  • 비동기 처리-asyncio
  • 새로운 멀티프로세싱 방법-concurrent.futures

1. 본 글에서는 지난번에 설명한 하나의 작업 단위로서의 프로세스, 파이썬 라이브러리 기능으로서의 프로세스가 혼동되어 사용되고 있다. ↩︎