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를 반환했습니다. Future의 완료를 기다리는 모든 스레드, 즉 as_completed()또는 wait()를 통한 스레드가 깨어납니다.

메서드가 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라고 함)에서 생성됩니다. 참조 구현은 프로세스 풀이나 스레드 풀을 사용하여 계산을 즉시 평가하는 클래스를 제공합니다.

Python에서 Futures는 인기 있는 Python cookbook 레시피 [2]의 일부로 이미 소개되었으며, Python-3000 메일링 리스트 [3]에서 논의되었습니다.

제안된 설계는 명시적입니다. 즉, 클라이언트가 Futures를 사용하고 있음을 인식해야 합니다. weakref 스타일의 프록시 객체를 반환하여 투명하게 사용할 수 있는 모듈을 설계할 수 있습니다. 제안된 명시적 메커니즘을 기반으로 프록시 구현을 구축할 수 있습니다.

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

Anh Hai Trinh은 더 단순하지만 제한적인 API 개념을 제안했으며 [5], API는 stdlib-sig [6]에서 상당히 자세히 논의되었습니다.

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

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