PEP 3148 – futures - 계산을 비동기적으로 실행하기

Author:: Brian Quinlan <brian at sweetapp.com>
Status:: Final
Type:: Standards Track
Created:: 16-Oct-2009
Python-Version:: 3.2
Post-History:

Table of Contents

초록
동기
명세
근거
참조 구현
참고 문헌
Copyright

번역·라이선스 안내

이 비공식 한국어 번역은 원문 Copyright 절의 Public Domain 조건에 따라 제공합니다. 원저자와 공식 원문은 그대로 표시합니다. 수정되지 않은 기준 원문 · 공식 최신판

초록

이 PEP는 스레드와 프로세스를 사용해 호출 가능 객체를 쉽게 평가할 수 있게 해 주는 패키지의 설계를 제안합니다.

Python에는 현재 멀티스레드 및 멀티프로세스 애플리케이션을 구성하기 위한 강력한 기본 요소가 있지만, 간단한 작업을 병렬화하려면 많은 작업이 필요합니다. 즉, 프로세스/스레드를 명시적으로 시작하고, 작업/결과 큐를 구성하고, 완료나 다른 종료 조건(예: 실패, 타임아웃)을 기다려야 합니다. 또한 각 구성 요소가 저마다의 병렬 실행 전략을 만들어 내는 경우, 전역 프로세스/스레드 제한이 있는 애플리케이션을 설계하기가 어렵습니다.

명세

명명

제안하는 패키지의 이름은 “futures”이며, 새로운 “concurrent” 최상위 패키지 안에 위치합니다. futures 라이브러리를 “concurrent” 네임스페이스에 넣는 데에는 여러 가지 근거가 있습니다. 첫 번째이자 가장 단순한 이유는 Python에서 오랫동안 사용되어 온 기존의 “from __future__ import x” 관용구와의 모든 혼동을 방지하기 위함입니다. 또한 이름 앞에 “concurrent”를 붙이면 이 라이브러리가 무엇과 관련되어 있는지, 즉 동시성과 관련되어 있음을 온전히 나타낼 수 있다고 봅니다. 커뮤니티의 모든 사람이 Java Futures나, 미국 주식 시장과 관련된 의미 외의 Futures라는 용어에 익숙한 것은 아니라는 지적이 있었으므로, 이렇게 하면 추가적인 모호함도 해소될 것입니다.

마지막으로, 표준 라이브러리에 새로운 네임스페이스를 마련합니다. 이름은 당연히 “concurrent”입니다. 앞으로 동시성 관련 라이브러리를 이곳에 추가하거나 기존 라이브러리를 이곳으로 옮기기를 바랍니다. 대표적인 예로 multiprocessing.Pool 작업과 그 모듈에 포함된 다른 “애드온”이 있으며, 이들은 스레드와 프로세스의 경계를 넘나들며 동작합니다.

인터페이스

제안하는 패키지는 두 가지 핵심 클래스를 제공합니다. 바로 Executor와 Future입니다. Executor는 비동기 작업 요청(호출 가능 객체와 그 인자의 형태)을 받아, 해당 작업 요청의 실행을 나타내는 Future를 반환합니다.

Executor

Executor는 호출을 비동기적으로 실행하는 메서드를 제공하는 추상 클래스입니다.

submit(fn, *args, **kwargs)

호출 가능 객체가 fn(*args, **kwargs)로 실행되도록 예약하고, 호출 가능 객체의 실행을 나타내는 Future 인스턴스를 반환합니다.
이것은 추상 메서드이며 Executor 서브클래스에서 반드시 구현해야 합니다.

map(func, *iterables, timeout=None)

map(func, *iterables)와 같지만, func가 비동기적으로 실행되며 func에 대한 여러 호출이 동시에 이루어질 수 있습니다. 반환된 이터레이터는 __next__()가 호출되었을 때 map()을 처음 호출한 시점부터 타임아웃 timeout 초가 지난 후에도 결과를 사용할 수 없으면 TimeoutError를 발생시킵니다. 타임아웃 timeout이 지정되지 않았거나 None이면 대기 시간에 제한이 없습니다. 호출이 예외를 발생시키면, 이터레이터에서 해당 값을 가져올 때 그 예외가 발생합니다.

shutdown(wait=True)

현재 대기 중인 퓨처의 실행이 끝나면 사용 중인 모든 자원을 해제해야 한다고 실행기에 신호를 보냅니다. 종료 후에 이루어지는 Executor.submit과 Executor.map 호출은 RuntimeError를 발생시킵니다.
wait가 True이면 보류 중인 모든 퓨처의 실행이 끝나고 실행기와 연관된 자원이 해제될 때까지 이 메서드는 반환하지 않습니다. wait가 False이면 이 메서드는 즉시 반환하며, 실행기와 연관된 자원은 보류 중인 모든 퓨처의 실행이 끝났을 때 해제됩니다. wait의 값과 관계없이, 대기 중인 모든 퓨처의 실행이 끝날 때까지 Python 프로그램 전체가 종료되지 않습니다.

__enter__()

__exit__(exc_type, exc_val, exc_tb)

실행기를 컨텍스트 관리자로 사용하면 __exit__가 Executor.shutdown(wait=True)를 호출합니다.

ProcessPoolExecutor

ProcessPoolExecutor 클래스는 프로세스 풀을 사용하여 호출을 비동기적으로 실행하는 Executor 서브클래스입니다. ProcessPoolExecutor.submit에 전달되는 호출 가능 객체와 인자는 multiprocessing 모듈과 같은 제약에 따라 피클 가능해야 합니다.

ProcessPoolExecutor에 제출된 호출 가능 객체 안에서 Executor나 Future 메서드를 호출하면 교착 상태가 발생합니다.

__init__(max_workers)

최대 max_workers개의 프로세스로 이루어진 풀을 사용하여 호출을 비동기적으로 실행합니다. max_workers가 None이거나 주어지지 않으면 머신의 프로세서 수만큼 작업자 프로세스가 생성됩니다.

ThreadPoolExecutor

ThreadPoolExecutor 클래스는 스레드 풀을 사용하여 호출을 비동기적으로 실행하는 Executor 서브클래스입니다.

Future와 연관된 호출 가능 객체가 다른 Future의 결과를 기다리면 교착 상태가 발생할 수 있습니다. 예를 들면:

import time
def wait_on_b():
    time.sleep(5)
    print(b.result())  # b will never complete because it is waiting on a.
    return 5

def wait_on_a():
    time.sleep(5)
    print(a.result())  # a will never complete because it is waiting on b.
    return 6


executor = ThreadPoolExecutor(max_workers=2)
a = executor.submit(wait_on_b)
b = executor.submit(wait_on_a)

그리고:

def wait_on_future():
    f = executor.submit(pow, 5, 2)
    # This will never complete because there is only one worker thread and
    # it is executing this function.
    print(f.result())

executor = ThreadPoolExecutor(max_workers=1)
executor.submit(wait_on_future)

__init__(max_workers)

최대 max_workers개의 스레드로 이루어진 풀을 사용하여 호출을 비동기적으로 실행합니다.

Future 객체

Future 클래스는 호출 가능 객체의 비동기 실행을 캡슐화합니다. Future 인스턴스는 Executor.submit이 반환합니다.

cancel()

호출 취소를 시도합니다. 호출이 현재 실행 중이면 취소할 수 없으며 메서드는 False를 반환하고, 그렇지 않으면 호출이 취소되고 메서드는 True를 반환합니다.

cancelled()

호출이 성공적으로 취소되었으면 True를 반환합니다.

running()

호출이 현재 실행 중이어서 취소할 수 없으면 True를 반환합니다.

done()

호출이 성공적으로 취소되었거나 실행을 마쳤으면 True를 반환합니다.

result(timeout=None)

호출이 반환한 값을 반환합니다. 호출이 아직 완료되지 않았으면 이 메서드는 타임아웃 timeout 초까지 기다립니다. 호출이 타임아웃 timeout 초 안에 완료되지 않으면 TimeoutError가 발생합니다. 타임아웃 timeout이 지정되지 않았거나 None이면 대기 시간에 제한이 없습니다.
퓨처가 완료되기 전에 취소되면 CancelledError가 발생합니다.

호출이 예외를 일으켰으면 이 메서드도 같은 예외를 일으킵니다.

exception(timeout=None)

호출이 일으킨 예외를 반환합니다. 호출이 아직 완료되지 않았으면 이 메서드는 타임아웃 timeout 초까지 기다립니다. 호출이 타임아웃 timeout 초 안에 완료되지 않으면 TimeoutError가 발생합니다. 타임아웃 timeout이 지정되지 않았거나 None이면 대기 시간에 제한이 없습니다.
퓨처가 완료되기 전에 취소되면 CancelledError가 발생합니다.

호출이 예외 없이 완료되었으면 None을 반환합니다.

add_done_callback(fn)

퓨처가 취소되거나 실행을 마쳤을 때 호출될 호출 가능 객체 fn을 퓨처에 연결합니다. fn은 퓨처를 유일한 인자로 받아 호출됩니다.
추가된 호출 가능 객체는 추가된 순서대로 호출되며, 항상 그것을 추가한 프로세스에 속한 스레드에서 호출됩니다. 호출 가능 객체가 Exception을 일으키면 그 예외는 로그에 기록되고 무시됩니다. 호출 가능 객체가 그 밖의 BaseException을 일으키면 동작은 정의되어 있지 않습니다.

퓨처가 이미 완료되었거나 취소되었으면 fn은 즉시 호출됩니다.

내부 Future 메서드

다음 Future 메서드는 단위 테스트와 Executor 구현에서 사용하기 위한 것입니다.

set_running_or_notify_cancel()

Executor 구현은 Future에 연관된 작업을 실행하기 전에 이 메서드를 호출해야 합니다.
이 메서드가 False를 반환하면 Future가 취소된 것입니다. 즉, Future.cancel이 호출되어 True를 반환한 것입니다. (as_completed() 또는 wait()를 통해) Future의 완료를 기다리던 모든 스레드가 깨어납니다.

이 메서드가 True를 반환하면 Future는 취소되지 않았으며 실행 중 상태가 된 것입니다. 즉, Future.running() 호출은 True를 반환합니다.

이 메서드는 한 번만 호출할 수 있으며, Future.set_result() 또는 Future.set_exception()이 호출된 뒤에는 호출할 수 없습니다.

set_result(result)

Future에 연관된 작업의 결과를 설정합니다.

set_exception(exception)

Future에 연관된 작업의 결과를 주어진 Exception으로 설정합니다.

모듈 함수

wait(fs, timeout=None, return_when=ALL_COMPLETED)

fs로 주어진 Future 인스턴스(서로 다른 Executor 인스턴스가 생성했을 수도 있습니다)가 완료되기를 기다립니다. 집합 두 개로 이루어진 이름 있는 2-튜플을 반환합니다. “done”이라는 이름의 첫 번째 집합에는 대기가 끝나기 전에 완료된(종료되었거나 취소된) 퓨처가 들어 있습니다. “not_done”이라는 이름의 두 번째 집합에는 완료되지 않은 퓨처가 들어 있습니다.
타임아웃 timeout을 사용하여 반환하기 전에 기다릴 최대 초를 제어할 수 있습니다. 타임아웃(timeout)이 지정되지 않았거나 None이면 대기 시간에 제한이 없습니다.

return_when은 메서드가 언제 반환해야 하는지를 나타냅니다. 다음 상수 중 하나여야 합니다:

상수 설명

FIRST_COMPLETED 퓨처 중 하나라도 종료되거나 취소되면 메서드가 반환합니다.

FIRST_EXCEPTION 퓨처 중 하나라도 예외를 발생시키며 종료되면 메서드가 반환합니다. 어떤 퓨처도 예외를 발생시키지 않으면 ALL_COMPLETED와 같습니다.

ALL_COMPLETED 모든 호출이 종료되면 메서드가 반환합니다.

상수	설명
`FIRST_COMPLETED`	퓨처 중 하나라도 종료되거나 취소되면 메서드가 반환합니다.
`FIRST_EXCEPTION`	퓨처 중 하나라도 예외를 발생시키며 종료되면 메서드가 반환합니다. 어떤 퓨처도 예외를 발생시키지 않으면 ALL_COMPLETED와 같습니다.
`ALL_COMPLETED`	모든 호출이 종료되면 메서드가 반환합니다.

as_completed(fs, timeout=None)

fs로 주어진 Future 인스턴스에 대한 이터레이터를 반환하며, 이 이터레이터는 퓨처가 완료(종료 또는 취소)되는 대로 해당 퓨처를 산출합니다. as_completed()가 호출되기 전에 이미 완료된 퓨처가 가장 먼저 산출됩니다. 반환된 이터레이터는 __next__()가 호출되었을 때 as_completed()를 처음 호출한 시점부터 타임아웃 timeout 초가 지난 후에도 결과를 사용할 수 없으면 TimeoutError를 발생시킵니다. 타임아웃 timeout이 지정되지 않았거나 None이면 대기 시간에 제한이 없습니다.
Future 인스턴스는 서로 다른 Executor 인스턴스가 생성한 것일 수 있습니다.

소수 판별 예제

from concurrent import 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 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()

웹 크롤링 예제

from concurrent import 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):
    return urllib.request.urlopen(url, timeout=timeout).read()

def main():
    with futures.ThreadPoolExecutor(max_workers=5) as executor:
        future_to_url = dict(
            (executor.submit(load_url, url, 60), url)
             for url in URLS)

        for future in futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                print('%r page is %d bytes' % (
                          url, len(future.result())))
            except Exception as e:
                print('%r generated an exception: %s' % (
                          url, e))

if __name__ == '__main__':
    main()

근거

이 모듈의 제안된 설계는 Java의 java.util.concurrent 패키지 [1]의 영향을 크게 받았습니다. Java에서와 마찬가지로 이 모듈의 개념적 기반은 Future 클래스이며, 이 클래스는 비동기 계산의 진행 상황과 결과를 나타냅니다. Future 클래스는 사용되는 평가 방식에 거의 제약을 두지 않습니다. 예를 들어 지연 평가나 즉시 평가를 나타내는 데 사용할 수 있고, 스레드, 프로세스 또는 원격 프로시저 호출을 이용한 평가에도 사용할 수 있습니다.

퓨처는 Executor 클래스(Java에서는 ExecutorService라고 부릅니다)의 구체적인 구현이 생성합니다. 참조 구현은 프로세스 풀이나 스레드 풀을 사용하여 계산을 즉시 평가하는 클래스를 제공합니다.

Future는 인기 있는 Python 쿡북 레시피 [2]의 일부로 이미 Python에 등장한 바 있으며, Python-3000 메일링 리스트 [3]에서도 논의되었습니다.

제안된 설계는 명시적입니다. 즉, 클라이언트가 Future를 사용하고 있다는 사실을 인지하고 있어야 합니다. 투명하게 사용할 수 있는 (weakref 방식의) 프록시 객체를 반환하는 모듈을 설계하는 것도 가능할 것입니다. 제안된 명시적 메커니즘 위에 프록시 구현을 구축할 수 있습니다.

제안된 설계는 Python 언어의 문법이나 의미론에 어떠한 변경도 도입하지 않습니다. 함수 및 메서드 호출을 비동기로 표시하기 위한 특수 문법을 도입할 수도 있습니다 [4]. 연산이 비동기적으로 즉시 평가되는 동안 프록시 결과가 반환되며, 연산이 완료되기 전에 프록시 객체가 사용되는 경우에만 실행이 블록됩니다.

Anh Hai Trinh는 더 단순하지만 더 제한된 API 개념을 제안했으며 [5], 이 API는 stdlib-sig(표준 라이브러리 SIG)에서 어느 정도 자세히 논의되었습니다 [6].

제안된 설계는 Python-Dev 메일링 리스트 [7]에서 논의되었습니다. 이러한 논의에 따라 다음과 같은 변경이 이루어졌습니다:

Executor 클래스를 추상 베이스 클래스로 만들었습니다
설득력 있는 사용 사례가 부족하여 Future.remove_done_callback 메서드를 제거했습니다
같은 호출 가능 객체를 여러 번 추가할 수 있도록 Future.add_done_callback 메서드를 수정했습니다
Future 클래스의 변경 메서드들이 해당 Future를 생성한 Executor에서만 사용하는 비공개 메서드임을 나타내도록 문서를 보완했습니다