[Python] asyncio를 사용하여 코루틴 구현해 보기

코루틴(coroutine)이란?

코루틴은 co + routine의 합성어로, 직역하면 함께/협력하여 수행되는 루틴이라고 볼 수 있다. 여기서 루틴이란 실행하는 프로시저나 함수를 의미한다. 다시 말해, 코루틴은 협력하여 실행되는 함수로 설명된다. 이게 어떤 의미일까? 루틴의 종류부터 살펴보자.

 

메인 루틴 - 서브 루틴

함수 내에서 또 다른 함수를 호출하면, 해당 함수가 결과 값을 반환할 때까지 대기했다가 그 이후 코드들을 순차적으로 수행한다. 이때 호출한 함수가 메인 루틴, 호출된 함수가 서브 루틴에 해당한다.

 

코루틴

반면에 코루틴은 비동기적으로 함수 내 여러 다른 지점에서 진입하고 탈출하고 재개할 수 있는 특수한 서브루틴에 해당한다.

코루틴이 완료되지 않아도 또 다른 코루틴을 실행할 수 있다. 코루틴을 활용하면 I/O 작업 단위에 기다리지 않고 다른 작업을 하는 게 가능하다. 

아래의 이미지를 보면 1번처럼 하나의 A스레드에서 Task1을 처리하다가 2번에서 Task 1을 일시정지(suspend) 시켜 놓고 Task2를 진행한다. Task2 역시 suspend 된 후 Task3을 처리할 수도 있고, Task2의 작업이 끝나 다시 Task1을 재개할 수도 있다. 

출처 : https://velog.io/@haero_kim/Thread-vs-Coroutine-비교해보기

정리하면 코루틴은 동시성을 기반으로 여러 루틴들이 함께 실행할 수 있도록 하는, 멀티태스킹이 가능하도록 하는 개념이다. 다음으로 python으로 코루틴 개념을 이해해보자.

Python과 코루틴

python에서는 3.5부터 asyncio라는 라이브러리를 내장하며 동시성 기능을 제공한다. 3.5 이전에는 제너레이터를 사용하여 구현이 가능했는데 두 구현 방식을 분류하기 위해 제너레이터로 구현한 코루틴을 제너레이터 코루틴, asyncio로 구현한 코루틴을 네이티브 코루틴이라고 부른다. python으로 코루틴을 구현한다고 하면 일반적으로 네이티브 코루틴을 떠올리면 된다. 해당 포스팅에서는 네이티브 코루틴을 살펴보겠다.

 

들어가기 앞서 필요한 선수 지식들이 있다. 하나씩 살펴보자.

 

async/await

코루틴 객체는 비동기 함수로 구현한다. async/await 키워드를 사용하여 구현할 수 있다.

• async : 함수 정의 시 async 키워드를 붙이면 함수를 비동기 함수임을 나타낸다. async로 정의된 함수는 호출 시 즉시 실행되는 것이 아니라, 코루틴 함수 내에서 await와 함께 사용될 때 비동기적으로 실행된다.

async def coroutine():
    print('hello world')

# coroutine 함수 자체는 function 객체이다.
type(coroutine)
>>> function

# 그러나 함수 호출 시에는 coroutine 객체를 반환받는다.
type(coroutine())
>>> coroutine

print(coroutine())
>>> <coroutine object coroutine at 0x106d0a800>

• await : async 함수 내에서 사용하며, await을 마주치면 자신의 실행을 중단하고 관련된 상태를 저장한 뒤 해당 함수가 결과를 반환할 때까지 기다린다. await 오른쪽에는 Awaitable 객체만 올 수 있다. coroutine, Task, Future 객체가 해당되며, __await__()을 정의한 객체라면 호출 가능하다.

 

이벤트 루프(event loop)

이벤트 루프는 콜백 또는 코루틴을 실행하고 관리하는 역할을 한다. 하나의 스레드에는 한 개의 이벤트 루프가 생성되고 코루틴을 Task 객체로 등록하여 task의 실행을 예약한다. 이벤트 루프는 루프를 돌며 예약된 task들을 하나씩 처리한다. 최종적으로 모든 task들의 실행이 완료되면 close 된다.

 

Task

Task는 코루틴 함수를 바인딩하여 이벤트 루프에서 실행할 수 있도록 하는 객체이다. Future 클래스를 상속하여 작업의 실행 상태 및 결과를 저장한다. 퓨처 객체와 다른 점은, Task 객체의 _coro 필드에 코루틴 함수를 바인딩한다. 태스크 객체는 생성되는 즉시 이벤트 루프에 자신의 실행을 예약한다. 이벤트 루프는 현재 실행 중인 태스크 객체가 없다면 예약된 태스크를 가져와_coro필드의 코루틴 함수를 실행한다. 해당 코루틴 함수를 실행하다 보면, await을 만나 다른 코루틴 함수를 호출하는 경우가 있을 텐데 결과적으로 Task는 여러 개의 코루틴이 쌓여있는 코루틴 체인을 갖게 된다. Task 객체를 생성하는 방법은 2가지이다.

1. aysnc.run() 함수에 코루틴 함수를 인자로 넘긴다. 넘긴 코루틴 함수는 task 객체로 예약되어 실행된다.
2. asyncio.create_task()로 코루틴 함수로 등록한다. 아래서 예시를 살펴보겠다.

Future

작업의 실행 상태와 결과를 저장하는 객체이며 마찬가지로 루프에 의해 관리된다. 실행 상태로는 PENDING, CANCELLED, FINISHED 중 하나를 가진다. 퓨처 객체는 태스크 객체처럼 실행할 코루틴 함수를 갖고 있지는 않고, 결괏값을 설정하는 등의 상태 변화로 이벤트루프에 완료 여부를 전달할 수 있다. 예를 들어 sleep()에 대해 퓨처 객체를 생성하게 되는데, 설정한 대기 시간이 지나면 결과값 없이 return 하여 해당 작업이 완료했음을 전달한다. 또한 퓨처 객체에는 add_done_callback() 메서드가 있어, 해당 메서드를 호출하면 Future 객체의 작업이 완료될 때 호출할 함수를 등록할 수 있다.

 

이벤트 루프의 Task 관리

이벤트 루프에서 콜백 또는 Task를 하나씩 실행하는데 Task의 대기와 예약을 반복하며 비동기적으로 여러 코루틴 함수들이 실행된다. 그 방법을 간략히 살펴보자.

1. 이벤트 루프 생성
   • asyncio.run()을 호출하면 이벤트 루프를 생성한다. 만약 동일 스레드에 이벤트 루프가 이미 존재한다면, 에러를 반환한다.
   • run() 메서드의 코루틴 함수를 인자로 넘기면, Task 객체를 생성하며 이벤트루프에 예약을 건다.
2. Task 실행
   • 이벤트 루프는 루프를 돌며 scheduled 된 Task를 실행한다.
   • coroutine을 실행하며 await 구문에서 연쇄적으로 코루틴 객체를 실행한다.
3. 퓨처 객체 생성
   • await 구문에서 I/0 관련 코루틴이나 asyncio.sleep() 구문을 만나면, 퓨처 객체를 하나 생성한 뒤 await 한다. 그리고 Future 객체는 코루틴 체인을 따라 태스크 객체에 전달될 것이다.
4. 퓨처 객체 처리 및 제어권 반납
   • 태스크 객체는 퓨처객체를 받으면, 퓨처 객체의 add_done_callback()을 호출하여 해당 퓨처 객체가 완료 상태가 될 때 이벤트 루프가 실행할 콜백 함수를 예약한다. 이러한 콜백 함수의 실행을 이벤트 루프에 예약한다는 것은 곧 해당 태스크의 실행을 예약하는 것이다.
   • 위 작업을 마친 후 태스크 객체는 자신의 실행을 중단하고 제어권을 이벤트 루프에게 넘긴다. 이벤트 루프는 예약되어 있는 태스크 중 적절히 선택하여 실행한다.
   • 다시 예약된 태스크를 실행하게 되면, I/0 작업의 경우 읽고 쓴 다음의 값을 리턴할 것이고, sleep 관련 코루틴이라면 바로 리턴한다.
5. 이벤트 루프 종료
   • 위 과정이 반복되면 제일 처음 실행한 코루틴이 return 하게 되고, 이벤트 루프도 종료한다.

 

예제를 보며 한 번 더 살펴보자.

 

Task 단일 등록

import asyncio

async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

asyncio.run(main())

1. asyncio.run(main()) : asyncio.run()을 호출하면 이벤트 루프(event loop)를 생성하며 main 코루틴 함수에 대한 Task 객체를 예약한다. 최초로 등록된 Task객체이므로 대기 없이 바로 실행된다.
2. await asyncio.sleep(1) : sleep 코루틴을 만났지만 이벤트루프에 따로 예약된 태스크가 없으므로 제어권을 반납하지 않고 1초 대기한다.

결과는 아래와 같이 간단하다.

Hello...
# 1초 sleep
...World!

 

Task 동시 등록

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)
    
async def main():
    task1 = asyncio.create_task(
        say_after(1, 'hello'))

    task2 = asyncio.create_task(
        say_after(2, 'world'))

    print(f"started at {time.strftime('%X')}")

    # Wait until both tasks are completed (should take
    # around 2 seconds.)
    await task1
    await task2

    print(f"finished at {time.strftime('%X')}")

asyncio.run(main())

1. asyncio.run(main()) : 마찬가지로 이벤트 루프를 생성하며 main 코루틴 함수에 대한 태스크 객체를 예약한다. 
2. asyncio.create_task() : asyncio.create_task()를 사용하여 태스크 객체를 생성과 동시에 예약한다. 이벤트 루프가 실행하는 태스크가 없다면 await을 도달하기 전에 해당 태스크를 바로 실행시켜 버릴 것이다. say_after가 sleep 코루틴이니 task1을 실행한 뒤 제어권을 반납할 테고 이벤트루프는 바로 이어서 task2를 실행시킬 것이다.
3. await task1 : await에서는 해당 태스크가 종료하길 기다린다.

결과는 다음과 같다.

started at 00:44:26
hello
world
finished at 00:44:28

task1은 1초 sleep, task2는 2초 sleep이기 때문에 동기적으로는 3초가 걸리지만, 코루틴으로 제어권을 바로 반납하면서 2초가 소요된 점을 확인할 수 있다.

 

끄-읕

---

참고로 jupyter notebook에서 asyncio.run을 실행하면 다음과 같은 에러가 발생한다. jupyter 7.0 이상에는 이미 이벤트루프가 동작하여 발생하는 문제로 await을 사용하여 코루틴을 호출하면 된다! (참고)

RuntimeError: asyncio.run() cannot be called from a running event loop

 

참조

https://docs.python.org/3/library/asyncio-task.html

https://github.com/python/cpython/tree/main/Lib/asyncio

https://blog.humminglab.io/posts/python-coroutine-programming-1/

https://stackoverflow.com/questions/1934715/difference-between-a-coroutine-and-a-thread

https://it-eldorado.tistory.com/159?category=749661

https://velog.io/@haero_kim/Thread-vs-Coroutine-비교해보기