파이썬 Asyncio 를 이해하기 위한 여정
파이썬 Asyncio 를 이해하기
위한 여정[번역]
[원제: 파이썬 ASYNCIO / nODE.JS 와 함께 비동기 io 이해하기]
소개
이번 여름에 Node.js에서 실행되는 웹 플랫폼에서 작업했었습니다. Node.js 를 가지고 풀 타임으로 일한 것은 이번이 처음이었고, 몇 주 동안 작업 한 결과 꽤 클리어해 진 것은 그 당시 나 자신을 비롯한 많은 개발자들이 Node의 비동기 기능에 대해 정확히 설명하지 못했다는 점이었습니다. 이 글에서는 js 작업 및 하위 수준에서의 구현 방법에 대해 설명합니다. 라이브러리를 효율적으로 사용하는 유일한 방법은 그것이 어떻게 작동하는지 명확하게 이해하는 것이고 그러기 위해서 깊이 파고 들기로 결정했습니다. 이 호기심은 다른 언어, 특히 파이썬에서 유사한 비동기 기능을 구현하는데도 나에게 도움을 주었습니다. 파이썬은 내가 관심을 가지고 있어서 공부하는 언어로 , 이번 경험으로 Python 3.4의 비동기 IO 라이브러리 asyncio 에 대한 이해도 깊어 졌습니다. 이 라이브러리에서는 코루틴을 적극적으로 사용하므로 코루틴에 대한 기존의 관심사도 믹스되었습니다. 이 게시물은 이 주제에 대해 더 많은 것들을 배우면서 제기된 질문 및 답변을 찾기 위한 여정의 결과물입니다. 다른 사람들이 이 글을 읽고서 몇 가지 의문점을 분명히 하는데 도움이되기를 바랍니다.
모든 파이썬 코드는 파이썬 3.4 용입니다. Python 3.4가 asyncio 뿐만 아니라 selectors 모듈을 도입했기 때문입니다. Python의 3.4 이전 버전에서는 Twisted, gevent 및 tornado와 같은 라이브러리를 통해 유사한 기능을 제공 받을 수 있습니다.
아래의 초기 예제에서는 학습을 단순화하기 위해 오류 처리 및 예외 문제를 거의 완전히 무시했지만, 예외를 적절하게 처리하는 것은 앞으로 보게될 코드 유형들의 매우 중요한 부분이 되어야한다는 점에 유의해야합니다. 이 글의 마지막으로 파이썬 3.4의 asyncio 모듈이 예외를 처리하는 방법에 대한 몇 가지 예를 제공 할 것입니다.
시작하기 : Hello World
아주 간단한 문제를 해결하기 위한 프로그램 부터 작성해 보겠다. 앞으로 이 문제를 해결하기 위해 코드를 요리조리 변경해 가며 설명해 나갈 것이다.
"Hello world!"를 출력 하기위한 프로그램을 작성하라. 3 초마다 출력되며 이와 동시에 사용자로부터의 입력을 기다린다. 각 사용자 입력 행에는 하나의 양수 n이 포함되며 , 입력되면 피보나치 수 F (n)을 계산하여 출력하고 다시 입력을 기다린다. 즉 3초마다 출력 & 입력받아서 어떤 계산 2가지 일을 하게 만드는 것이다
"Hello world!" 문자열이 사용자 입력의 중간에 삽입되지만 그것에 대해 신경쓰지말라. 중요한것은 문자열이 찍히는 타이밍이 어떻게 다른지에 대해 촛점을 맞추라.
Node.js와 JavaScript에 익숙한 사용자라면 다음과 같은 해결책을 생각할 수 있지 않을까?
log_execution_time = require('./utils').log_execution_time;
var fib = function fib(n) {
if (n < 2) return n;
return fib(n - 1) + fib(n - 2);
};
var timed_fib = log_execution_time(fib);
var sayHello = function sayHello() {
console.log(Math.floor((new Date()).getTime() / 1000) + " - Hello world!");
};
var handleInput = function handleInput(data) {
n = parseInt(data.toString());
console.log('fib(' + n + ') = ' + timed_fib(n));
};
process.stdin.on('data', handleInput);
setInterval(sayHello, 3000);
보다시피, Node.js에서는 매우 쉽게 할 수 있다. 우리가 해야 할 일은 "Hello world!"를 출력 할 인터벌 타이머를 설정하는 것 뿐이며 process.stdin의 데이터 이벤트에 이벤트 핸들러를 첨부하면 완료된다. 추상적인 수준에서 이해하기 쉽고 사용하기 쉬우며 잘 작동한다! 그러나 어떻게? 이 질문에 답하기 위해 파이썬 코드로 똑같은 일을 해보려 한다.
먼저 log_execution_time 데코레이터를 사용하여 피보나치 수를 계산하는 데 걸리는 시간을 출력하는 방식을 파이썬에서는 다음 처럼 만든다. (역주: 우리의 목적에는 별로 의미 없는 것들이다. 굳이 파보나치를 이해 할 필요도 없고 functools 같은 것을 알 필요도 없다. 그냥 어떤 것을 계산하는 do_somthing() 함수라고만 이해하고 넘어가도 된다)
from functools import wraps
from time import time
def log_execution_time(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time()
return_value = func(*args, **kwargs)
message = "Executing {} took {:.03} seconds.".format(func.__name__,
time() - start)
print(message)
return return_value
return wrapper
(역주:partial이 그 함수가 할 수 있는 모든 것중 특정 부분을 하는 함수로 변환 시키는 느낌이라면 wraps 는 기능을 추가/꾸미는느낌이다. wraps 는 함수를 래핑할 때 원래 함수의 정보들을 유지 시킬 수 있다. )
자바스크립트에서도 유사하게 코딩 할 수 있다.
// We do not care about handling the "this" parameter correctly in our examples.
// Do not use this decorator where that's needed!
module.exports.log_execution_time = function log_execution_time(func) {
var wrapper = function() {
start = (new Date()).getTime();
return_value = func.apply(this, arguments);
message = "Calculation took " + ((new Date()).getTime() - start) / 1000 + " seconds";
console.log(message);
return return_value;
};
return wrapper;
};여기서 사용 된 피보나치 수를 계산하는 알고리즘은 의도적으로 모든 것 중 가장 느린 것 (지수 실행 시간)으로 선택되었다. 이것은 이 게시물이 피보나치 수에 관한 것이 아니기 때문이다.
from log_execution_time import log_execution_time
def fib(n):
return fib(n - 1) + fib(n - 2) if n > 1 else n
timed_fib = log_execution_time(fib)
이제 당면 과제로 돌아가보자. 파이썬은 자바스크립트와 같은 setInterval 또는 setTimeout을 제공하지 않기 때문에 가능한 첫 번째 해결 방법은 OS 수준의 동시성을 사용하는 것이다. 즉 두 개의 쓰레드를 사용해서 해결해보자.
from threading import Thread
from time import sleep
from time import time
from fib import timed_fib
def print_hello():
while True:
print("{} - Hello world!".format(int(time())))
sleep(3)
def read_and_process_input():
while True:
n = int(input())
print('fib({}) = {}'.format(n, timed_fib(n)))
def main():
# Second thread will print the hello message. Starting as a daemon means
# the thread will not prevent the process from exiting.
t = Thread(target=print_hello)
t.daemon = True
t.start()
# Main thread will read and process input
read_and_process_input()
if __name__ == '__main__':
main()
아주 간단하다. 근데 스레드 기반 Python 솔루션과 Node.js 솔루션은 동등한가? 실험을 해보자. 위에 언급했듯이 피보나치 수 계산 코드는 매우 느리기 때문에 파이썬에 대해서는 37 개, Node.js에 대해서는 45 개 (자바 스크립트는 숫자 계산시 일반 파이썬보다 상당히 빠르다) 라는 다소 큰 숫자를 사용해 보려한다.
$ python3.4 hello_threads.py 1412360472 - Hello world! 37 1412360475 - Hello world! 1412360478 - Hello world! 1412360481 - Hello world! Executing fib took 8.96 seconds. fib(37) = 24157817 1412360484 - Hello world!
파이썬은 계산이 끝나기까지 대략 9 초가 걸리며 "Hello world!" 출력은 3초마다 꾸준히 찍히고 있다.
즉 피보나치 계산을 하는 동안에도 출력은 멈추지 않았다.이제 Node.js로 해보자.
$ node hello.js 1412360534 - Hello world! 1412360537 - Hello world! 45 Calculation took 12.793 seconds fib(45) = 1134903170 1412360551 - Hello world! 1412360554 - Hello world! 1412360557 - Hello world!
반면에 Node.js를 사용하면 피보나치 수를 계산하는 동안 "Hello world!" 메시지가 일시 중지되었다.
왜 그런것일까? 이것이 어떤 의미가 있는지 살펴 보자.
이벤트 루프와 쓰레드
(역주: 앞부분의 회색 문장은 일반 쓰레드 얘기니깐 잘 알고 있다면 건너뛰어도 된다.)
이전 섹션 두 솔루션(node, python)의 동작 차이를 이해하려면 스레드 및 이벤트 루프를 간단히 이해해야한다. 쓰레드 부터 시작해 보자. 쓰레드를 명령 시퀀스와 그것들을 실행할 때의 CPU의 현재 상태 (CPU 상태는 예를 들어 레지스터 값, 특히 다음 인스트럭션 레지스터를 참조 함) 라고 생각하자.
간단한 동기 프로그램의 경우 보통 단일 스레드에서 실행되기 때문에 작업이 IO 작업이나 타이머와 같은 무언가를 기다려야하는 경우 그 작업이 끝날 때까지 프로그램 실행이 일시 중지된다. 가장 간단하게 실행을 차단 할 수 있는 작업 중 하나는 sleep 이다. 사실, 모든 sleep 은 주어진 시간 동안 실행되는 쓰레드를 차단한다. 여러 스레드가 하나의 프로세스에서 실행될 수 있으며 동일한 프로세스의 쓰레드는 메모리와 그 주소 공간, 파일 기술자 등과 같은 프로세스 레벨 자원을 공유한다.
운영 체제는 쓰레드 처리를 담당하고 있으며 운영 체제의 스케줄러는 하나의 쓰레드를 일시 중지하고 다른 스레드로 CPU를 제어하여 실행되도록 선택한다. 이를 컨텍스트 스위치라고하며, 현재 쓰레드 (예 : CPU 레지스터 값)의 컨텍스트를 저장 한 다음 살아 날 대상 스레드의 상태를 로드하는 작업을 한다. 컨텍스트 스위칭은 다소 CPU 를 소비하는 비싼 계산이다.
하나의 쓰레드에서 다른 스레드로 전환되는 많은 이유가 있다. 예를 들어, 우선 순위가 높은 또 다른 프로세스 나 스레드가 즉각적인 주의 (예 : 하드웨어 인터럽트를 처리하는 코드)가 필요하거나 스레드 자체가 일시 중지 (예 : sleep)를 요청하거나 스레드가 전용 시간을 사용했기 때문일 수 있습니다 할당되었지만 (이것은 또한 thread quantum 이라고도 함) 실행을 계속하기 위해 대기열로 돌아 가야합니다.
이제 일반적인 쓰레드 이야기는 여기까지 하고 위의 코드로 돌아가 보자.
위의 파이썬 코드는 멀티 쓰레드였으며 그것은 두 작업이 동시에 실행될 수 있는 이유와 CPU 집중적인 피보나치 계산이 다른 스레드의 실행을 막지 않는 이유를 설명한다. 즉 CPU 집중적인 계산이 이루어지면서 중간 중간 쓰레드 제어권이 "Hello World" 를 찍는 쓰레드로도 변경된 다는 것이다.
하지만 Node.js는 어떤가? 피보나치 계산이 다른 작업("Hello world" 출력) 을 차단 한다는 사실에 미루어 짐작컨데 우리 코드가 단일 스레드에서 실행되고 있는 것을 알 수 있다. 즉 피보나치 계산이 이루어지는 동안에는 다른 어떠한 작업도 실행 될 수 없다는 것이다. 이것은 실제로 Node.js가 구현되는 방법이다.
지금까지 운영 체제가 응용 프로그램을 단일 스레드로 실행하고 있다는 것을 염두에 두면 (플랫폼에 따라 libuv가 일부 IO 이벤트에 스레드 풀을 사용할 수 있지만 즉 로우레벨에서 쓰레드가 여러개가 작동하지만 우리가 코딩하고 실행을 하는 위치에서의 자바 스크립트 코드는 여전히 단일 스레드에서 실행되고 있다는 사실이다)
특정 상황에서 스레드를 피하려고 하는 몇 가지 이유가 있는데 하나는 쓰레드가 계산적으로나 자원적으로 비싸고, 다른 하나는 여러 스레드가 동작할 때 공유 메모리와 함께 교착 상태 및 경쟁 조건과 같은 동시성 문제로 인해 더 복잡하고 위험한 코드가 발생할 소지가 높아진다는 점이다.
그럼 "파이썬 코드"를 멀티 쓰레딩을 사용하지 않고 "단일 쓰레드" 로 문제를 해결할 수 있는지 생각해 보자. 그렇게 하기 위해 우리는 Node.js가 내부에서 사용하는 방식을 모방 할것이다. 장면을 떠올려 보자.
첫번째 이벤트 루프.
먼저, 표준 입력을 폴링(poll) 하는 방법, 즉 파일 디스크립터 (이 경우 stdin)에 입력 가능 여부를 묻는 시스템 호출이 필요하며 운영 체제에 따라 poll, select, kqueue 등과 같은 다양한 시스템 호출이 있어서 사용 할 수 있다. Python 3.4에서 selectors 모듈은 이러한 시스템 호출에 대한 추상화를 제공하므로 다양한 머신 위에서 (다소) 안전하게 사용할 수 있을 것이다.
폴링 기능이 갖추고 나서 (코드에서는 selectors 를 이용함) , 단일쓰레드 하에서 반복되는 이벤트 루프를 매우 간단하게 만들어서 루프를 각각 반복하는 동안 읽을 수 있는 입력이 있는지 확인한다. 그 다음, "Hello world!"의 마지막 출력 이후 3 초 이상 지 났는지 확인하고 "예" 인 경우 출력한다. 코드를 보면 명확해 질 것이다.
(역주: 간단히 말해 이벤트루프는 그냥 while 이다. while 문 돌면서 ,입력이 있으면 입력 받은 값으로 피보나치 계산해주고, 입력받은게 없고 이전에 "Hello world" 찍은지 3초 지났으면 또 다시 찍는 것이다. 입력이 있는지 확인받는 방법을 selector 를 활용한 것. 참고로 selector 는 대략적으로 이렇다. OS 에게 말하길 "혹시 입력을 받았으면 나한테 알려줘" 라는 부탁을 한다. 코드에서는 EVENT_READ 이다. 그리고 나서 selector.select() 로 그 부탁이 실행 됬는지 확인하기 위해 잠시 확인하는 방식이다. python 의 selector 는 SelectSelector / PollSelector / EpoolSelector / KqueueSelector 등 OS 맞춤으로 제공한다. 아쉽게도 윈도우즈는 소켓만 가능하고 파이프 (stdio등) 은 지원되지 않고 있는 듯하다. 따라서 이 예제들은 윈도우에서는 안된다.)
import selectors
import sys
from time import time
from fib import timed_fib
def process_input(stream):
text = stream.readline()
n = int(text.strip())
print('fib({}) = {}'.format(n, timed_fib(n)))
def print_hello():
print("{} - Hello world!".format(int(time())))
def main():
selector = selectors.DefaultSelector()
# Register the selector to poll for "read" readiness on stdin
selector.register(sys.stdin, selectors.EVENT_READ)
last_hello = 0 # Setting to 0 means the timer will start right away
while True:
# Wait at most 100 milliseconds for input to be available
for event, mask in selector.select(0.1):
process_input(event.fileobj)
if time() - last_hello > 3:
last_hello = time()
print_hello()
if __name__ == '__main__':
main()
결과는 아래와 같다.
$ python3.4 hello_eventloop.py 1412376429 - Hello world! 1412376432 - Hello world! 1412376435 - Hello world! 37 Executing fib took 9.7 seconds. fib(37) = 24157817 1412376447 - Hello world! 1412376450 - Hello world!
예상대로, 단일 스레드를 사용하기 때문에 이 코드는 Node.js와 같은 방식으로 작동한다. 즉 피보나치 계산을 하는 동안 "Hello world!" 실행을 차단. 깔끔하다. 그러나 우리의 코드는 특정 문제에 대해 다소 하드 코딩되어 있는데 다음 섹션에서는 이벤트 루프 코드를 좀 더 강력하고 프로그램하기 쉽도록 일반화하는 방법을 살펴볼 것이다. 먼저 콜백을 사용하고 그 다음엔 coroutines를 사용해서~ 겁먹지 마시라~
(역주: 하지만 앞으로의 얘기는 python 에 대한 약간의 지식이 필요 할 것이다.)
이벤트 루프와 Callbacks
이전 섹션의 이벤트 루프를 자연스럽게 일반화하면 일반 이벤트 핸들러를 사용해서 구현 할 수 있게 된다. 즉 콜백을 사용하여 상대적으로 쉽게 달성 할 수 있는데 각각의 이벤트 유형 (지금 경우에는 표준 입력과 타이머 입력 두 개) 에 대해 사용자가 임의의 함수를 이벤트 핸들러로 추가 할 수 있을 것이다. 코드는 꽤 간단하여 바로 코드를 봐도 되지만 하나가 조금 까다로운데 타이머 이벤트를 처리하기 위해 bisect.insort를 사용하는 부분이다. 이 알고리즘은 타이머 이벤트 목록을 정렬 된 상태로 유지하고 타이머를 가장 먼저 실행한다. 이러한 방법으로 이벤트 루프를 반복 할 때마다 타이머가 있는지 확인하고 타이머가 있는 경우 처음부터 시작하여 만료된 타이머를 모두 실행하는 식이다. bisect.insort를 이용하여 항목을 목록의 올바른 색인에 삽입함으로써 이 작업을보다 쉽게 수행할 수 있다. 이것에 대한 여러 가지 다른 접근 방식이 있지만 이것은 그냥 내가 선택한 방식이다.
(역주: 다시 언급하지만 이벤트 루프는 그냥 while 문 도는거라고 생각하면 된다. 동일하다. while 문 내부에서 어떤 조건을 체크해서 a 에 해당하는 이벤트가 발생하면 a_func 함수, 즉 콜백을 호출해준다는 간단한 이야기이다.
동시성 패턴에 자주 사용되는 Actor 라는 것도 사실 비슷하다. 자신의 쓰레드 안에서 while 문 도는 것일 뿐)
from bisect import insort
from collections import namedtuple
from fib import timed_fib
from time import time
import selectors
import sys
Timer = namedtuple('Timer', ['timestamp', 'handler'])
class EventLoop(object):
"""
Implements a callback based single-threaded event loop as a simple
demonstration.
"""
def __init__(self, *tasks):
self._running = False
self._stdin_handlers = []
self._timers = []
self._selector = selectors.DefaultSelector()
self._selector.register(sys.stdin, selectors.EVENT_READ)
def run_forever(self):
self._running = True
while self._running:
# First check for available IO input
for key, mask in self._selector.select(0):
line = key.fileobj.readline().strip()
for callback in self._stdin_handlers:
callback(line)
# Handle timer events
while self._timers and self._timers[0].timestamp < time():
handler = self._timers[0].handler
del self._timers[0]
handler()
def add_stdin_handler(self, callback):
self._stdin_handlers.append(callback)
def add_timer(self, wait_time, callback):
timer = Timer(timestamp=time() + wait_time, handler=callback)
insort(self._timers, timer)
def stop(self):
self._running = False
def main():
loop = EventLoop()
def on_stdin_input(line):
if line == 'exit':
loop.stop()
return
n = int(line)
print("fib({}) = {}".format(n, timed_fib(n)))
def print_hello():
print("{} - Hello world!".format(int(time())))
loop.add_timer(3, print_hello)
def f(x):
def g():
print(x)
return g
loop.add_stdin_handler(on_stdin_input)
loop.add_timer(0, print_hello)
loop.run_forever()
if __name__ == '__main__':
main()(역주: bisect 는 이름이 말하듯이 이진트리로 정렬을 해 놓는 컬렉션이다. 삽입하면 바로 정렬해 준다는 것이다. insort(정렬될 리스트, 값) 이런식으로 활용한다.)
매우 단순한 방식의 이 코드는 실제로는 Node.js 에서 가장 일반적으로 사용되는 방법이다. 그러나 좀 더 복잡한 응용 프로그램에서는 비동기 코드를 작성하는 이러한 스타일이 특히 오류 처리가 추가되면 매우 빠르게 콜백 지옥으로 알려진 모양으로 악화 될 수 있다. Guido van Rossum의 콜백 인용 :
콜백으로 이루어진 복잡한 코드를 읽는 것은 미친짓이다. 슈퍼 울트라 교육을 받은 인간만이 가능할거다. 내 말을 못믿겠거든 널려 있는 자바 스크립트 코드를 보라 - 귀도 반 로섬
promises 와 코루틴 (각 대안에 대해 헤아릴 수 없을 만큼의 NPM 라이브러리) 과 같은 여러 가지 대체 접근법이 있다. 내가 가장 선호하는 것은 coroutines를 사용하는 것이며 다음 섹션에서는 coroutines를 사용해서 유사한 이벤트 루프를 구현하는 방법에 대해 설명 할 예정이다.
이벤트 루프와 코루틴
코루틴 - 한글블로그에도 휼륭한 글이 많으니 참고 합시다 -> 참고
코루틴은 상태 (로컬 변수의 값 및 다음 명령의 의미)를 기억하면서 "return" 할 수있는 함수이다. 이렇게 하면 코 루틴을 다시 반복 호출 할 수있게 되어 결과가 중단된 부분부터 계속됩니다. 이러한 형태의 "return"은 특별히 yielding 이라고 한다. 즉 자신이 잠깐 다른 일을 해야 하니깐 제어를 양보한다는 의미이다. 값을 받는데 사용하는 yield 키워드는 함수 내부의 "=" 오른쪽에 있는 표현식으로 사용 할 수도 있다. send() 메서드를 사용하여 값을 코루틴 함수로 다시 전달할 수 있다. 이때가 양보가 끝마치는 시점이다. 즉 블럭이 해소되는 시점.
"코루틴은 주거니 받거니 하는 함수이다. 먼가 쓰레드 처럼 살아숨쉬는 듯한 함수"
좀 헥깔릴 수 있는데. 함수와 제네레이터 , 코루틴 모두 def 를 통해 만든다. 이것에 대해 모두 다르게 표시해야 한다고 주장하는 사람들이 있으며, 파이썬의 창조자 귀도는 이에 반대하여 def 로 통일되어 있다고 한다. 아무튼 중요한것은 코루틴은 반복이 목적이 아니라. 코루틴은 외부와의 상호 작용이다.
def repeater():
while True:
received = yield
print('Echo:', received)
rp = repeater()
next(rp) # 코루틴 시작
rp.send('Hello')
rp.send('World')
#Echo: First
#Echo: Second
(비동기에서 코루틴은 요긴하게 사용되는데 잠시 생상을 해보세요. 어떻게 사용 될까요? 네 어떤 비동기 작업 (주로 i/o 많을듯) 에 대한 결과를 받아서 전달하는 매개함수 역할을 합니다. 매우 중요합니다. )
Python에서는 yield 키워드를 사용하여 코루틴을 만들 수 있다. yield value 와 같은 간단한 명령문으로 사용되면 주어진 값이 내보내지고 호출자에게 제어가 되돌려진다. yield 문 다음 명령에서 시작하여 coroutine을 계속하려면 호출자는 내장 된 next 함수를 사용하면 된다. y = yield x와 같은 표현식으로 사용될 때는 x 값이 yielded 되고 , coroutine을 계속하려면 coroutine의 send 메소드를 사용할 수 있다.이 경우 send에 지정된 값이 coroutine으로 다시 전송된다.(이 예에서 y에 할당 됨)
이것은 우리가 코루틴으로 비동기 코드를 쓸 수 있고 (역주: 이벤트 핸들링을 위해) 비동기 작업을 기다릴 필요가 있을 때 간단히 yield 할 수 있음을 의미한다. 이렇게 하기 위해서, 단순히 우리가 계속해야 할 가치가 있는 작업이나 다른 코루틴을 yield 하게 된다. 그러면 코드는 매우 순차적이며 동기 코드와 유사하게 보이게 될 것이다. 우리 코드의 피보나치 부분이 코루틴이 된다면 어떻게 생겼는지에 대한 간단한 예가 있다 :
(역주: 글로는 더 헥깔릴 수 있으니 코드를 반복적으로 돌려 보면서 익히자)
def read_input():
while True:
line = yield sys.stdin
n = int(line)
print("fib({}) = {}".format(n, timed_fib(n)))물론 이것이 작동하려면 코루틴을 처리 할 수있는 이벤트 루프가 필요하다. 이를 달성하기 위해 우리는 이벤트 루프에 의해 실행될 태스크 큐를 유지할 것인데 입력 이벤트가 발생했을 때와 타이머가 꺼지면 (또는 더 일반적으로 우리가 신경 쓰이는 다른 이벤트), 우리는 계속해야 할 코루틴 목록을 가지고있을 것이다. 각 작업에는 체인에서 실행할 코루틴 스택을 추적하는 바운드 스택 변수가 있는데 각 변수는 다음 종료마다 다를 것이다. 이것은 PEP 342에서 제공하는 "Trampoline"예제를 기반으로 하는데 JavaScript에서 Function.prototype.bind와 동일한 Python 방식인 functools.partial을 사용한다. 즉, 매개 변수 값을 바인딩하여 함수를 Curry 한다.
다음은 전체 코드이다.
from bisect import insort
from collections import deque
from collections import namedtuple
from fib import timed_fib
from functools import partial
from time import time
import selectors
import sys
import types
Timer = namedtuple('Timer', ['timestamp', 'handler'])
class sleep_for_seconds(object):
"""
Yield an object of this type from a coroutine to have it "sleep" for the
given number of seconds.
"""
def __init__(self, wait_time):
self._wait_time = wait_time
class EventLoop(object):
"""
Implements a simplified coroutine-based event loop as a demonstration.
Very similar to the "Trampoline" example in PEP 342, with exception
handling taken out for simplicity, and selectors added to handle file IO
"""
def __init__(self, *tasks):
self._running = False
self._selector = selectors.DefaultSelector()
# Queue of functions scheduled to run
self._tasks = deque(tasks)
# (coroutine, stack) pair of tasks waiting for input from stdin
self._tasks_waiting_on_stdin = []
# List of (time_to_run, task) pairs, in sorted order
self._timers = []
# Register for polling stdin for input to read
self._selector.register(sys.stdin, selectors.EVENT_READ)
def resume_task(self, coroutine, value=None, stack=()):
result = coroutine.send(value)
if isinstance(result, types.GeneratorType):
self.schedule(result, None, (coroutine, stack))
elif isinstance(result, sleep_for_seconds):
self.schedule(coroutine, None, stack, time() + result._wait_time)
elif result is sys.stdin:
self._tasks_waiting_on_stdin.append((coroutine, stack))
elif stack:
self.schedule(stack[0], result, stack[1])
def schedule(self, coroutine, value=None, stack=(), when=None):
"""
Schedule a coroutine task to be run, with value to be sent to it, and
stack containing the coroutines that are waiting for the value yielded
by this coroutine.
"""
# Bind the parameters to a function to be scheduled as a function with
# no parameters.
task = partial(self.resume_task, coroutine, value, stack)
if when:
insort(self._timers, Timer(timestamp=when, handler=task))
else:
self._tasks.append(task)
def stop(self):
self._running = False
def do_on_next_tick(self, func, *args, **kwargs):
self._tasks.appendleft(partial(func, *args, **kwargs))
def run_forever(self):
self._running = True
while self._running:
# First check for available IO input
for key, mask in self._selector.select(0):
line = key.fileobj.readline().strip()
for task, stack in self._tasks_waiting_on_stdin:
self.schedule(task, line, stack)
self._tasks_waiting_on_stdin.clear()
# Next, run the next task
if self._tasks:
task = self._tasks.popleft()
task()
# Finally run time scheduled tasks
while self._timers and self._timers[0].timestamp < time():
task = self._timers[0].handler
del self._timers[0]
task()
self._running = False
def print_every(message, interval):
"""
Coroutine task to repeatedly print the message at the given interval
(in seconds)
"""
while True:
print("{} - {}".format(int(time()), message))
yield sleep_for_seconds(interval)
def read_input(loop):
"""
Coroutine task to repeatedly read new lines of input from stdin, treat
the input as a number n, and calculate and display fib(n).
"""
while True:
line = yield sys.stdin
if line == 'exit':
loop.do_on_next_tick(loop.stop)
continue
n = int(line)
print("fib({}) = {}".format(n, timed_fib(n)))
def main():
loop = EventLoop()
hello_task = print_every('Hello world!', 3)
fib_task = read_input(loop)
loop.schedule(hello_task)
loop.schedule(fib_task)
loop.run_forever()
if __name__ == '__main__':
main()
(역주: parital 이란 어떤 함수가 있을 때 그 함수가 가진 매개변수들을 미리 채워 넣어서 만들어진 부분함수라고 보면된다. 그리고 이 코드를 번역된 혹은 원본을 읽고 이해하려고 하지말자. 글로 이해하기는 매우 복잡하다. 코드를 반복해서 읽고 이해하자. 그게 더 쉽다. 전체적으로 간략히 설명하면 스케쥴 함수에 내가 해야할 일을 넣어주고 있으며, run_forever 함수가 while 을 돌면서 스케쥴 함수를 통해 들어온 일을 처리하는 모양새이다. 주의 깊게 볼 부분은 resume_task 로 , 코루틴에 값을 전달해서 실행시키기도 하고, 값을 리턴(yield) 받아서 다시 스케쥴함수에 넣어주기도 한다. 즉 계속 task 를 재생산하고 있다는 점)
이 구현은 Node.js에서 process.nextTick 이 하는 일을 어느 정도 수행하는 간단한 do_on_next_tick 함수를 추가하게 한다. 기능을 종료하기 위해 간단한 타입 exit 를 구현하는 데 사용한다. loop.stop ()을 직접 호출 할 수도 있었으나 do_on_next_tick을 사용하였다.또 다른 흥미로운 점은 재귀 호출 대신 coroutines를 사용하여 재귀 피보나치 알고리즘을 다시 구현할 수 있다는 점인데 그렇게 하면 hello를 포함하여 다른 coroutine과 "parallel" 하게 실행할 수 있게 된다.
위의 코드를 라이브러리 처럼 활용한 다음 코드를 살펴보자.
from event_loop_coroutine import EventLoop
from event_loop_coroutine import print_every
import sys
def fib(n):
if n <= 1:
yield n
else:
a = yield fib(n - 1)
b = yield fib(n - 2)
yield a + b
def read_input(loop):
while True:
line = yield sys.stdin
n = int(line)
fib_n = yield fib(n)
print("fib({}) = {}".format(n, fib_n))
def main():
loop = EventLoop()
hello_task = print_every('Hello world!', 3)
fib_task = read_input(loop)
loop.schedule(hello_task)
loop.schedule(fib_task)
loop.run_forever()
if __name__ == '__main__':
main()
결과: (중간에 블럭되지 않았다.)
$ python3.4 fib_coroutine.py
1412727829 - Hello world!
1412727832 - Hello world!
28
1412727835 - Hello world!
1412727838 - Hello world!
fib(28) = 317811
1412727841 - Hello world!
1412727844 - Hello world!
바퀴를 재발명 할 필요없다.
앞의 두 섹션에서 우리는 콜백이나 코루틴을 사용하여 비동기 코드를 작성할 수 있도록 이벤트 루프를 구현하는 일반적인 아이디어를 검토했다. 이것은 아이디어에 대해 실험하고 배우기 위한 목적 이었지만 실제로는 이미 이벤트 루프를 제공하는 Python용 라이브러리가 이미 성숙해 있는데 Python 3.4에는 IO,네트워킹 작업을 하기 위한 코루틴과 이벤트 루프을 가진 asyncio 모듈이 제공된다.
우선 asyncio를 사용하여 위의 문제를 해결하는 모습을 살펴 보자.
import asyncio
import sys
from time import time
from fib import timed_fib
def process_input():
text = sys.stdin.readline()
n = int(text.strip())
print('fib({}) = {}'.format(n, timed_fib(n)))
@asyncio.coroutine
def print_hello():
while True:
print("{} - Hello world!".format(int(time())))
yield from asyncio.sleep(3)
def main():
loop = asyncio.get_event_loop()
loop.add_reader(sys.stdin, process_input)
loop.run_until_complete(print_hello())
if __name__ == '__main__':
main()@ asyncio.coroutine이 coroutine을 장식하는 데 사용되는 방식에 주목하고 yield 와 반대되는 즉 다른 coroutine으로 부터의 값을 사용하는 yield from 을 눈여겨보자.
yield -> 주는
yield from -> 받는 (받기 위해 기다림)
예외 핸들링
파이썬 코루틴은 코루틴의 스택 프레임에 예외가 던져 지도록 허용하고, 코루틴이 일시 중지 한 지점에서 예외를 포착하도록 한다. 간단한 예를 살펴 보자.
def coroutine():
print("Starting")
try:
yield "Let's pause until continued."
print("Continuing")
except Exception as e:
yield "Got an exception: " + str(e)
def main():
c = coroutine()
next(c) # 처음 yield 까지 실행
# Now throw an exception at the point where the coroutine has paused
value = c.throw(Exception("Have an exceptional day!"))
print(value)
if __name__ == '__main__':
main()
결과:
Starting
Got an exception: Have an exceptional day!이벤트 루프가 예외를 적절하게 포착하고 전파하는 경우, 동기 및 비동기 코드 모두에서 예외를 사용하여 오류를 처리하는 통일 된 방법을 얻는 것은 꽤 간단하다.
체인화 된 coroutine과 이벤트 루프를 사용한 예를 살펴 보자.
import asyncio
@asyncio.coroutine
def A():
raise Exception("Something went wrong in A!")
@asyncio.coroutine
def B():
a = yield from A()
yield a + 1
@asyncio.coroutine
def C():
try:
b = yield from B()
print(b)
except Exception as e:
print("C got exception:", e)
def main():
loop = asyncio.get_event_loop()
loop.run_until_complete(C())
if __name__ == '__main__':
main()
결과:
C got exception: Something went wrong in A!이 예에서, 코루틴 C는 B의 결과에 의존하며, 쭈욱 거슬러 올라가 보면 예외를 던지기로 결정하는 A의 결과에 최종 의존한다. 보시다시피, 예외는 C 에게 전달되어 메시지를 캐치하고 메시지를 출력 하게 된다. 반갑게도 동기식 코드와 거의 똑같이 작동한다. 즉 수동으로 콜백을 통해 오류를 포착하고 전달할 필요가 없게 되는 것이다!!
이 예제가 너무 간단해서 만족스럽지 않은 분들을 위한 실제 예를 살펴 보았다. ipify를 사용하여 컴퓨터의 외부 IP 주소를 비동기적으로 잡아내는 코드를 작성해 보려한다. asyncio는 HTTP 클라이언트와 함께 제공되지 않기 때문에 (아직 어쨌든!) TCP 수준으로 가서 HTTP 요청을 작성하고 응답을 직접 구문 분석하려 한다. 실무적으로는 aiohttp과 같은 라이브러리를 사용하는 것은 당연히 더 좋은 생각일 것이다.
이제 코드를 살펴보자.
import asyncio
import json
host = 'api.ipify.org'
request_headers = {'User-Agent': 'python/3.4',
'Host': host,
'Accept': 'application/json',
'Accept-Charset': 'UTF-8'}
@asyncio.coroutine
def write_headers(writer):
for key, value in request_headers.items():
writer.write((key + ': ' + value + '\r\n').encode())
writer.write(b'\r\n')
yield from writer.drain()
@asyncio.coroutine
def read_headers(reader):
response_headers = {}
while True:
line_bytes = yield from reader.readline()
line = line_bytes.decode().strip()
if not line:
break
key, value = line.split(':', 1)
response_headers[key.strip()] = value.strip()
return response_headers
@asyncio.coroutine
def get_my_ip_address(verbose):
reader, writer = yield from asyncio.open_connection(host, 80)
writer.write(b'GET /?format=json HTTP/1.1\r\n')
yield from write_headers(writer)
status_line = yield from reader.readline()
status_line = status_line.decode().strip()
http_version, status_code, status = status_line.split(' ')
if verbose:
print('Got status {} {}'.format(status_code, status))
response_headers = yield from read_headers(reader)
if verbose:
print('Response headers:')
for key, value in response_headers.items():
print(key + ': ' + value)
# Assume the content length is sent by the server, which is the case
# with ipify
content_length = int(response_headers['Content-Length'])
response_body_bytes = yield from reader.read(content_length)
response_body = response_body_bytes.decode()
response_object = json.loads(response_body)
writer.close()
return response_object['ip']
@asyncio.coroutine
def print_my_ip_address(verbose):
try:
ip_address = yield from get_my_ip_address(verbose)
print("My IP address is:")
print(ip_address)
except Exception as e:
print("Error: ", e)
def main():
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(print_my_ip_address(verbose=True))
finally:
loop.close()
if __name__ == '__main__':
main()
다시 말하지만 동기 코드와의 유사점 (콜백 없음, 복잡한 오류 처리 없음, 간단하고 읽기 쉬운 코드)을 확인하라. 오류없이 어떻게 잘 작동지에 대해 살펴보자.
$ python3.4 ipify.py
Got status 200 OK
Response headers:
Content-Length: 21
Server: Cowboy
Connection: keep-alive
Via: 1.1 vegur
Content-Type: application/json
Date: Fri, 10 Oct 2014 03:46:31 GMT
My IP address is:
<my IP address here, hidden for privacy!>물론 인터넷에 연결되어 있지 않은 경우 다음과 같이 출력 될 것이다.
$ python3.4 ipify.py
Error: [Errno 8] nodename nor servname provided, or not known필자의 견해로는 비동기 코드에 coroutines를 사용하면 얻을 수있는 주요 강력한 장점 중 하나으로 생각한다. 오류 처리가 동기 코드와 완벽하게 일치하는 것 아름다운 모습을 보라. 예를 들어 위의 경우 체인화 된 coroutines 중 하나가 실패하거나 동기 호출 중 하나가 실패하면 예외는 잡히고 똑같은 방식으로 처리될 것이다.
여러개의 독립 코루틴들의 결과 다루기
위의 예제에서는 본질적으로 순차적인 비동기 코드를 작성했다. 즉, 코루틴의 각 명령문은 계속 진행하기 전에 완료 한 이전 명령문에 의존한다. 때로는 일련의 독립적인 작업을 실행하고 실행 순서를 신경 쓰지 않고 완료된 상태로 사용하기를 원한다. 예를 들어 웹 크롤러처럼 웹 페이지의 모든 링크에 비동기 요청을 보내고 결과를 받아서 처리 할 대기열에 응답을 추가하자.
코루틴은 매우 순차적으로 흐르는 비동기 코드를 작성할 수 있지만, 독립적인 작업을 실행하고 결과를 한꺼번에 처리하거나 콜백을 수행 할 때 처음에는 콜백이 더 나은 것처럼 보일 수 있다. 그러나 Python 3.4의 asyncio에는 정확하게 asyncio.as_completed 및 asyncio.gather라는 두 가지 시나리오에 대한 내장 함수가 있는데 3 개의 URL을로드해야하는 간단한 예제를 통해 살펴 보겠다. 먼저 asyncio.as_completed를 사용하여 결과를 처리하고 두 번째 방법에서는 asyncio.gather를 사용하여 로드를 완료 한 후 두 가지 방법으로 결과를 처리하려 한다. 실제로 URL을 로드하는 대신에 임의의 초 동안 일시 중지되는 간단한 동시 루틴을 선택했다. 코드는 다음과 같다.
import asyncio
import random
@asyncio.coroutine
def get_url(url):
wait_time = random.randint(1, 4)
yield from asyncio.sleep(wait_time)
print('Done: URL {} took {}s to get!'.format(url, wait_time))
return url, wait_time
@asyncio.coroutine
def process_as_results_come_in():
coroutines = [get_url(url) for url in ['URL1', 'URL2', 'URL3']]
for coroutine in asyncio.as_completed(coroutines):
url, wait_time = yield from coroutine
print('Coroutine for {} is done'.format(url))
@asyncio.coroutine
def process_once_everything_ready():
coroutines = [get_url(url) for url in ['URL1', 'URL2', 'URL3']]
results = yield from asyncio.gather(*coroutines)
print(results)
def main():
loop = asyncio.get_event_loop()
print("First, process results as they come in:")
loop.run_until_complete(process_as_results_come_in())
print("\nNow, process results once they are all ready:")
loop.run_until_complete(process_once_everything_ready())
if __name__ == '__main__':
main()
결과
$ python3.4 gather.py
First, process results as they come in:
Done: URL URL2 took 2s to get!
Coroutine for URL2 is done
Done: URL URL3 took 3s to get!
Coroutine for URL3 is done
Done: URL URL1 took 4s to get!
Coroutine for URL1 is done
Now, process results once they are all ready:
Done: URL URL1 took 1s to get!
Done: URL URL2 took 3s to get!
Done: URL URL3 took 4s to get!
[('URL1', 1), ('URL2', 3), ('URL3', 4)]
더 알아보기
아직 다루지 않은 것들이 많이있다. Futures and libuv 등과 파이썬 3.4에서 비동기 IO에 대한 Guido의 강연( Guido's talk on asynchronous IO in Python 3.4) 도 있다. 심각한 오역이나 오류에 대해서 알려 주시면 반영하겠다.
(역주: 여기까지 읽고 이해가 안가신다면 두어번 더 읽어보시길 바랍니다. 그래도 이해가 안간다면 WINDOWS SELECT, Java nio 의 selector, node.js 를 먼저 공부해 보셨으면 하구요.내용도 많을 것이고 좀 더 근본에 대해서 공부한다면 상위층에 대한 이해도 빨라질 것이기 때문입니다.)
번역 : http://sahandsaba.com/understanding-asyncio-node-js-python-3-4.html