[ python ] event loop (2)

[ Task 동시 실행 - asyncio.create_task() ]

위에서 이벤트 루프가 태스크들을 동시적으로(Concurrent, not Parallel) 실행한다고 설명하였다. 그런데 사실 asyncio.run() 함수는 기본적으로 하나의 태스크만을 생성하여 실행한다. 따라서 코루틴 체인 과정에서 추가적인 태스크를 생성하여 실행하지 않았다면 현재의 태스크가 중단되었을 때 이벤트 루프는 실행시킬 다른 태스크가 없게 된다. 태스크가 한 개라면 동시적인(Concurrent) 실행을 하는 것이 애초에 말이 되지 않는 것이다.

※ 여기서 말하는 동시 실행이란 Parallel이 아닌 Concurrent를 말한다. 즉, 엄밀한 의미의 동시가 아니라 여러 태스크들을 왔다 갔다 하며 한 쓰레드에서 실행하는 개념인 것이다. 따라서 총 실행 시간은 같거나 오히려 더 늘어난다(문맥 전환 비용 때문).

따라서 동시적인(Concurrent) 실행을 위해서는 asyncio.create_task() 함수를 호출함으로써 태스크를 추가로 생성하여 실행해야 한다. 이 함수를 호출할 때 코루틴 객체를 인자로 넘기면, 해당 코루틴 객체를 이용하여 태스크 객체를 생성하고 이를 반환한다. 그리고 앞서 말했듯 태스크 객체가 생성되면 해당 태스크 객체가 나타내는 태스크의 실행이 이벤트 루프에 의해 즉시 예약된다(즉시 실행이 아니다). 예약 후에는 Taks 객체를 반환한다. 단, 이 함수는 3.7 버전 이상의 Python에서만 사용할 수 있기 때문에, 그 이전 버전에서는 asyncio.ensure_future() 함수를 대신 사용해야 한다.

다음으로, 모든 퓨처 객체(태스크 객체 포함)들이 완료 상태가 될 때까지 기다리는 함수가 asyncio.gather()이다. 이 함수는 인자로 여러 개의 Awaitable 객체들을 받을 수 있는데, 만약 코루틴 객체를 받으면 이는 자동으로 태스크 객체로 래핑이 된다. 따라서 사실상 퓨처 객체(태스크 객체 포함)만 넘어간다고 생각해도 된다. 그리고 모든 퓨처 객체들이 완료 상태가 되면 그것들의 결과 값들을 리스트 형태로 반환한다. 그 순서는 인자로 넘긴 순서와 동일하다. 이 함수는 await 키워드의 뒤에서 호출될 수 있는 코루틴의 일종이다.

 

import asyncio
import time

async def sleep(sec):
    await asyncio.sleep(sec)
    return sec

async def main():
    sec_list = [1, 2]
    tasks = [asyncio.create_task(sleep(sec)) for sec in sec_list]  # [Task 1 객체, Task 2 객체]
    tasks_results = await asyncio.gather(*tasks)  # [Task 1 객체의 결과 값, Task 2 객체의 결과 값]
    return tasks_results

start = time.time()

loop = asyncio.get_event_loop()
result = loop.run_until_complete(main())
loop.close()

end = time.time()

print('result : {}'.format(result))
print('total time : {0:.2f} sec'.format(end - start))

# 출력 결과
# result : [1, 2]

위 예시 코드의 실행 흐름은 다음 링크에서 자세히 설명해주셧습니다.

https://it-eldorado.tistory.com/159

[Python] 비동기 프로그래밍 동작 원리 (asyncio)

 

[ synchronous 함수를 coroutine처럼 사용하기 : loop.run_in_executor() method ]

 

우리가 지금까지 알아본 원리에 따르면, 결국 비동기 프로그래밍의 효과를 보기 위해서는 현재의 쓰레드 실행과 무관하게 다른 곳에서 어떠한 작업을 할 수 있어야 한다. 그 대표적인 예시가 Sleep 혹은 I/O 관련 코루틴이었다. Sleep의 경우에는 이벤트 루프가 자체적으로 타이머를 가지고 있기 때문에, 그리고 I/O 관련 코루틴은 CPU가 열심히 일하는 동안 I/O 장치가 일해주면 되기 때문에 현재의 실행 흐름을 Block 하지 않고 다른 작업을 먼저 할 수 있었던 것이다.

그런데 사실 Python이 가지고 있는 대부분의 API는 동기 방식으로 동작한다. 애초에 동기 방식으로 동작하도록 설계된 언어이기 때문이다. 예를 들어, asyncio.sleep() 함수가 제공되기 전에는 time.sleep() 함수를 사용했는데, 이는 현재의 실행 흐름을 Block 하는 함수였다. 그리고 requests 라이브러리가 제공하는 requests.get(), requests.post() 등의 함수도 현재의 실행 흐름을 Block 하는 함수이다. 이러한 함수들을 이용해서는 비동기 프로그래밍이 불가능할 듯하다. 비동기 프로그래밍이 가능하려면 그러한 작업을 다른 어딘가에 맡겨 놓고 퓨처 객체를 await 하면서 현재 실행 중인 태스크의 제어를 이벤트 루프에게 넘겨야 하기 때문이다.

이때 사용하는 것이 바로 loop.run_in_executor() 메소드이다. loop는 이벤트 루프 객체이다. 어렵게 설명하면 한도 끝도 없겠지만, 간단하게 얘기해서 이 메소드는 동기 함수를 별도의 쓰레드에서 실행시킴으로써 마치 Sleep 혹은 I/O 관련 코루틴처럼 사용할 수 있게 해주는 것이다. 비동기 프로그래밍을 하려면 어떤 작업을 '다른 어딘가(= 별도의 쓰레드)'에 맡겨야 하기 때문이다.

이 함수의 사용 방법을 바로 한 번 알아보자. 이 함수의 반환 값은 퓨처 객체이기 때문에, await 키워드의 뒤에 올 수 있다.

import asyncio
import time

async def sleep(sec):
    await loop.run_in_executor(None, time.sleep, sec)  # time.sleep(sec)
    return sec

async def main():
    sec_list = [1, 2]
    tasks = [asyncio.create_task(sleep(sec)) for sec in sec_list]  # [Task 1 객체, Task 2 객체]
    tasks_results = await asyncio.gather(*tasks)  # [Task 1 객체의 결과 값, Task 2 객체의 결과 값]
    return tasks_results

start = time.time()

loop = asyncio.get_event_loop()
result = loop.run_until_complete(main())
loop.close()

end = time.time()

print('result : {}'.format(result))
print('total time : {0:.2f} sec'.format(end - start))

# 출력 결과
# result : [1, 2]
# total time : 2.03 sec

 

원래는 Blocking 함수인 time.sleep() 함수가 마치 asyncio.sleep() 함수처럼 동작할 수 있도록 하였다. loop.run_in_executor() 메소드의 첫 번째 인자로 넘어가는 None은 실행기를 명시적으로 지정하지 않고 기본 실행기를 사용하겠다는 것인데, 직접 실행기를 지정하면 워커 쓰레드를 원하는 개수만큼 생성하는 것이 가능하다. 두 번째 인자에는 함수 이름을 넘기고, 세 번째 인자부터는 그 함수를 호출할 때 넘길 인자들을 하나씩 넘기면 된다.