코드노예

이번에는 futures에 대해서 알아보겠습니다.

futures는 요청들을 비동기적으로 수행하기 위한 인터페이스를 제공해 주는 모듈입니다. 

Executor

concurrent.futures.Excutor 는 비동기 요청을 수행할 때 필요한 메써드들을 정의해 둔 추상 클래스 입니다. 여기에는 submit, map, shutdown 메써드들이 정의되어 있습니다.

python (버전 3.12.4 기준)에서 기본으로 제공되는 executor에는 ThreadPoolExecutor와 ProcessPoolExecutor 가 있습니다. 이름에서 유추할 수 있듯이, ThreadPoolExecutor는 요청을 수행할 때, ThreadPool을 사용하고, ProcessPoolExecutor는 요청을 수행할 때, ProcessPool을 사용합니다.

Process 와 Thread의 차이를 간략하게 정리하면, Process는 독립적으로 수행되며, 메모리 등의 리소스를 따로 할당해서 수행하게 됩니다. Thread는 Process 내에서 수행이 되며, 하나의 Process 안에서 수행되는 Thread들 간에는 리소스를 공유하게 됩니다. 

따라서 ProcessPoolExecutor 의 경우, 요청들이 Process 단위로 수행되기 때문에, 수행시에 Process에게 리소스를 따로 할당되어야 하기 때문에, overhead가 발생됩니다. 반면 ThreadPoolExecutor의 경우, 리소스가 공유되기 때문에 리소스 할당에 따른 overhead는 없지만, 잘못 사용하게 되면 deadlock이 발생할 수 있습니다.

결론을 이야기하면, 요청 하나 하나의 수행 작업이 오래걸리는 경우 (CPU-bound tasks)에는 리소스 할당에 시간이 소요되더라도 ProcessPoolExecutor가 적합하고, 처리할 데이터가 많지만, 하나 하나의 수행 작업은 빠르게 처리되는 경우 (I/O bound tasks)에는 ThreadPoolExecutor를 사용하는 것이 좋습니다.

Future

executor에게 요청을 submit하면, Future 인스턴스를 반환해 줍니다. 이 Future 인스턴스를 통해서, executor에게 전달한 요청의 상태를 확인할 수 있습니다. 다음은 Future 객체에서 사용할 수 있는 method 들 입니다. (일부만 발췌했으며 전체 리스트를 확인하고 싶으시면 다음의 링크를 참하세요. https://docs.python.org/3/library/concurrent.futures.html)

Module functions

concurrent.futures에 정의되어 있는 futures 에서 사용하는 funciton들 입니다. future 요청은 한 가지 요청만 비동기로 처리하는 경우도 있지만, 보통의 경우 여러 요청을 동시에 처리하기 위해서 많이 사용되기 때문에, 전체 요청들에 대한 결과를 받아서 처리해야 하는 경우, 아래의 function들을 사용하게 됩니다.

Futures 사용 예제

아래 코드는 간단히 작성한 사용 예제 입니다. (docs.python.org 에 있는 예시를 활용하였습니다.)

먼저 map을 이용해서 요청을 처리하는 예시입니다. 

import concurrent.futures
import urllib.request

TEST_URLS = ['https://www.google.com', 'https://www.naver.com', 'https://www.tistory.com']

def load_url(url):
    try:
        with urllib.request.urlopen(url) as conn:
            return url, conn.read(), None
    except Exception as exc:
        return url, None, str(exc)

with concurrent.futures.ThreadPoolExecutor() as executor:
    for url, data, err in executor.map(load_url, TEST_URLS, timeout=60):
        if data:
            print('%r page is %d bytes' % (url, len(data)))
        else:
            print('%r generated an exception: %s' % (url, err))

map은 as_completed 와 비슷한데, 결과 iterator에는 Future 인스턴스가 아닌, 인스턴스의 result 들을 반환해 줍니다. map을 억지로 풀어쓴다면 다음과 같습니다.

with concurrent.futures.ThreadPoolExecutor() as executor:
    for url, data, err in [f_instance.result() for f_instance in 
        concurrent.futures.as_completed([executor.submit(load_url, url) for url in TEST_URLS])]:
        if data:
            print('%r page is %d bytes' % (url, len(data)))
        else:
            print('%r generated an exception: %s' % (url, err))

여기서 이야기하고자 하는 바는, map의 경우에는 Future 인스턴스에서 result를 읽는 과정이 포함되어 있기 때문에, 특정요청에서 exception 이 발생하는 경우, iterator를 수행하는 과정에서 exception이 발생하게 됩니다. 이 exception으로 문제가 생기는 것을 막기 위해서는, 요청 내에서 exception이 발생하더라도 같은 포맷으로 결과를 리턴해 줄 수 있게 디자인해야 합니다.

위의 예시를 as_completed 를 사용해서 변경하면 다음과 같습니다.

import concurrent.futures
import urllib.request

TEST_URLS = ['https://www.google.com', 'https://www.naver.com', 'https://www.tistory.com']

def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

with concurrent.futures.ThreadPoolExecutor() as executor:
    future_to_url = {executor.submit(load_url, url, 60): url for url in TEST_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)))

여기서는 as_completed가 future instance 에서 결과를 받아오는 시점에서 exception을 처리할 수 있기 때문에 요청에서 Exception 이 발생한다고 해도, iterator에서 처리가 가능합니다.

마지막으로 wait을 사용하면, list에 future instance를 호출한 순서대로 저장하여, list를 이용해서 호출한 순서대로 결과를 처리할 수 있습니다. 참고로 wait function은 반환값으로 as_completed와 같이 처리된 순서대로 결과를 처리할 수 있는 generator 형태의 iterator를 돌려줍니다.

import concurrent.futures
import urllib.request

TEST_URLS = ['https://www.google.com', 'https://www.naver.com', 'https://www.tistory.com']

def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

with concurrent.futures.ThreadPoolExecutor() as executor:
    future_list = [executor.submit(load_url, url, 60) for url in TEST_URLS]
    concurrent.futures.wait(future_list)
    for idx, future in enumerate(future_list):
        url = TEST_URLS[idx]
        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)))