파이썬 동시성 프로그래밍 - (8) Concurrent.Futures & ProcessPoolExecutor

연재 순서 

1. threading

2. Condition & Semaphore

3. Queue

4. multiprocessing

5. 비동기 (gevent) 

6. 분산 (celery)

7. GPGPU (PyCUDA)

8. 코루틴,asyncio,async/awit

9. concurrent.future

 CONCURRENT.FUTURES 모듈 퀵 가이드
 [번역]

concurrent.futures 모듈은 비동기 작업을 시작하기 위한 높은 수준의 API를 제공하는 표준 라이브러리의 일부입니다. 이 모듈의 일반적인 사용법에 대한 코드 샘플을 살펴 보겠습니다.

Executors

이 모듈은 추상클래스인 Executor 클래스를 제공하며, 직접 사용할 수는 없으며 대신 ThreadPoolExecutor와 ProcessPoolExecutor라는 매우 유용한 두 개의 서브 클래스를 사용 할 수 있습니다. 이름에서 알 수 있듯이 하나는 멀티 스레딩을 사용하고 다른 스레드는 멀티프로세스를 사용합니다. 두 경우 모두 스레드 또는 프로세스 풀을 가져 와서 이 풀에 작업을 제출하는 방식으로 작업을 합니다. 

ThreadPoolExecutor

코드를 보시죠

Python

1 2 3 4 5 6 7 8 9 10 11 12 13 14

from concurrent.futures import ThreadPoolExecutor from time import sleep   def return_after_5_secs(message):     sleep(5)     return message   pool = ThreadPoolExecutor(3)   future = pool.submit(return_after_5_secs, ("hello")) print(future.done()) sleep(5) print(future.done()) print(future.result())

먼저 스레드 3개를 가진 ThreadPoolExecutor를 만듭니다. 그런 다음 풀에 작업을 제출하여  5초 후에 첫 번째 인수로 전달 된 메시지를 다시 반환 받게 됩니다. 키 포인트는 태스크를 제출하고 나서 future를 되돌려 받는 코드에 있는데요. Doc에서 볼 수 있듯이 Future 객체에는 태스크가 해결되었는지, 즉 해당 future 객체에 대한 값이 설정되었는지 알려주는 done () 메서드가 있습니다. 태스크가 완료되면 (값을 리턴하거나 예외로 인터럽트 된 경우), 스레드 풀 실행 프로그램은 값을 future 오브젝트에 설정합니다.

위의 예에서 작업은 5 초가 경과 할 때까지 완료되지 않으므로 done()을 처음 호출하면 False가 반환됩니다. 그리고 잠시 대기한 후에 result () 메소드를 호출하여 future의 결과를 얻을 수 있습니다.

Future 객체를 잘 이해하고 그것의 메소드를 아는 것은 Python에서 비동기 프로그래밍을 이해하고 수행하는 데 아주 중요합니다. 그래서 문서를 꼼꼼히 읽어보시는게 좋을 겁니다.

ProcessPoolExecutor

이것은 이전에 설명한 ThreadPoolExecutor 와 매우 유사한 API를 가지고 있습니다. 이제 이전 예제를 수정해서 ProcessPool을 사용해 봅시다.

Python

1 2 3 4 5 6 7 8 9 10 11 12 13 14

from concurrent.futures import ProcessPoolExecutor from time import sleep   def return_after_5_secs(message):     sleep(5)     return message   pool = ProcessPoolExecutor(3)   future = pool.submit(return_after_5_secs, ("hello")) print(future.done()) sleep(5) print(future.done()) print("Result: " + future.result())

 완벽하게 작동합니다! 물론 CPU 집약적인 작업을 위해 ProcessPoolExecutor를 사용하는게 좋습니다. ThreadPoolExecutor는 네트워크 작업 또는 I / O에 더 적합합니다. (역주: 파이썬의 GIL 때문에) 

API는 비슷하지만 ProcessPoolExecutor는 다중 처리 모듈을 사용하며 Global Interpreter Lock의 영향을 받지 않습니다. 그러나 picklable이 아닌 객체는 사용할 수 없습니다. 따라서 우리는 process pool executor에 전달 된 callable 내부에서 무엇을 사용하고 리턴하는지 대해서 신중하게 선택할 필요가 있습니다.

Executor.map()

두 executors 는 일반적인 방법인 map ()을 사용합니다. 내장 함수와 마찬가지로 map 메서드는 제공된 함수에 대한 여러 호출을 허용하여 iterable의 각 항목을 해당 함수에 전달합니다. 이 경우를 제외하고는 함수가 동시에 호출됩니다. 다중 처리의 경우,이 반복 가능은 청크로 분리되고 이러한 청크는 각각 별도의 프로세스에서 함수로 전달됩니다. chunk_size 매개 변수를 전달하여 청크 크기를 제어 할 수 있습니다. 기본적으로 청크 크기는 1입니다.

다음은 공식 문서의 ThreadPoolExample입니다.

Python

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

import concurrent.futures import urllib.request   URLS = ['http://www.foxnews.com/',         'http://www.cnn.com/',         'http://europe.wsj.com/',         'http://www.bbc.co.uk/',         'http://some-made-up-domain.com/']   def load_url(url, timeout):     with urllib.request.urlopen(url, timeout=timeout) as conn:         return conn.read()   with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:          future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}     for future in concurrent.futures.as_completed(future_to_url):         url = future_to_url[future]         try:             data = future.result()         except Exception as exc:             print('%r generated an exception: %s' % (url, exc))         else:             print('%r page is %d bytes' % (url, len(data)))

그리고 ProcessPoolExecutor 예제 :

Python

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

import concurrent.futures import math   PRIMES = [     112272535095293,     112582705942171,     112272535095293,     115280095190773,     115797848077099,     1099726899285419]   def is_prime(n):     if n % 2 == 0:         return False       sqrt_n = int(math.floor(math.sqrt(n)))     for i in range(3, sqrt_n + 1, 2):         if n % i == 0:             return False     return True   def main():     with concurrent.futures.ProcessPoolExecutor() as executor:         for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):             print('%d is prime: %s' % (number, prime))   if __name__ == '__main__':     main()

as_completed() & wait()

concurrent.futures 모듈은 executor에 의해 반환 된 future를 다루는 두 개의 함수를 가지고 있습니다. 하나는 as_completed ()이고 다른 하나는 wait ()입니다.

as_completed() 함수는 future 객체를 반복하여 가져오고 future가 resolving을 시작되면 즉시 값을 yielding 하기 시작합니다. 앞서 언급 한 map 메소드와 as_completed의 주요 차이점은 map이 iterable을 전달한 순서대로 결과를 리턴한다는 점이다. 첫 번째 항목에 대한 결과는 map메서드의 첫 번째 결과입니다. 반면, as_completed 함수의 첫 번째 결과는 가장 먼저 완료된 future의 결과입니다.

예제를 보시죠.

Python

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

from concurrent.futures import ThreadPoolExecutor, wait, as_completed from time import sleep from random import randint   def return_after_5_secs(num):     sleep(randint(1, 5))     return "Return of {}".format(num)   pool = ThreadPoolExecutor(5) futures = [] for x in range(5):     futures.append(pool.submit(return_after_5_secs, x))   for x in as_completed(futures):     print(x.result())

wait () 함수는 두 세트를 포함하는 명명된 튜플을 반환합니다. 하나의 세트에는 완료된 futures (결과 또는 예외가 있음)과 완료되지 않은 것을 포함하는 다른 세트가 포함됩니다.

여기에 예제가 있습니다.

Python

1 2 3 4 5 6 7 8 9 10 11 12 13 14

from concurrent.futures import ThreadPoolExecutor, wait, as_completed from time import sleep from random import randint   def return_after_5_secs(num):     sleep(randint(1, 5))     return "Return of {}".format(num)   pool = ThreadPoolExecutor(5) futures = [] for x in range(5):     futures.append(pool.submit(return_after_5_secs, x))   print(wait(futures))

우리는 그것이 반환되어야 할 때를 정의함으로써 wait 함수의 행동을 제어 할 수 있습니다. FIRST_COMPLETED, FIRST_EXCEPTION 및 ALL_COMPLETED 함수의 return_when 매개 변수에 이러한 값 중 하나를 전달할 수 있습니다. 기본적으로 ALL_COMPLETED로 설정되므로 모든 future가 완료 될 때 만 wait 함수가 반환됩니다. 그러나 이 매개 변수를 사용하여 첫 번째 future가 완료되거나 첫 번째 예외가 발생할 때 반환하도록 선택할 수 있습니다.

---

번역:

http://masnun.com/2016/03/29/python-a-quick-introduction-to-the-concurrent-futures-module.html