파이썬 코루틴과 태스크 (Coroutines, Tasks)

파이썬에서 async, await 그리고 future라는 코드가 보입니다. 이것들이 무엇을 하는지 알아야, 코드가 제대로 되었는지 확인할 수가 있어서 정리해 보았습니다.

Table Of Contents

• 코루틴(Coroutines)
• TaskGroup 지원

코루틴(Coroutines)

Source code: Lib/asyncio/coroutines.py

 

async/await 구문으로 선언된 코루틴(corutines)은 비동기 애플리케이션을 작성하는 데 선호되는 방식입니다. 예를 들어 다음 코드는 "hello"를 출력하고 1초간 기다린 다음 "world"를 출력합니다.

import asyncio

>>> async def main():
...     print('hello')
...     await asyncio.sleep(1)
...     print('world')

>>> asyncio.run(main())
hello
world

 

코루틴을 호출하는 것만으로는 실행 되지 않는다는 점을 주의해야합니다.

main()
<coroutine object main at 0x1053bb7c8>

asyncio.run(coro, *, debug=None, loop_factory=None)

코루틴 coro를 실행하고 결과를 반환합니다.

이 함수는 전달된 코루틴을 실행하여 asyncio 이벤트 루프 관리, 비동기 제너레이터 마무리, 실행기 종료 등을 처리합니다.

이 함수는 동일한 스레드에서 다른 비동기 이벤트 루프가 실행 중일 때는 호출할 수 없습니다.

debug가 True이면 이벤트 루프가 디버그 모드로 실행됩니다. False는 디버그 모드를 명시적으로 비활성화합니다. None은 전역 디버그 모드 설정에 따라가기 위해 사용됩니다.

loop_factory가 None이 아닌 경우 새 이벤트 루프를 만드는 데 사용되며, 그렇지 않으면 asyncio.new_event_loop()가 사용됩니다. 루프는 마지막에 닫힙니다. 이 함수는 비동기 프로그램의 주요 진입점으로 사용해야 하며, 이상적으로는 한 번만 호출하는 것이 좋습니다. 정책 대신 loop_factory를 사용하여 이벤트 루프를 구성하는 것이 좋습니다.

실행기가 종료될 때까지 5분의 시간 초과 기간이 주어집니다. 실행기가 이 기간 내에 종료되지 않으면 경고가 표시되고 실행기가 닫힙니다.

 

 

실제로 코루틴을 실행하기 위해 asyncio는 다음과 같은 메커니즘을 제공합니다.

• 최상위 진입점 "main()" 함수를 실행하는 asyncio.run() 함수(위 예제 참조).
• 아래 코드는 1초를 기다린 후 "hello"를 출력한 다음 2초를 더 기다린 후 "world"를 출력합니다.

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at {time.strftime('%X')}")

    await say_after(1, 'hello')
    await say_after(2, 'world')

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

asyncio.run(main())

 

<수행결과>

started at 17:13:52
hello
world
finished at 17:13:55

 

코루틴을 비동기 작업으로 동시에 실행하는 asyncio.create_task() 함수.

위의 예제를 수정하여 두 개의 say_after 코루틴을 동시에 실행해 보겠습니다.

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')}")

 

이제 아래 출력에 따르면 코드가 동시에 실행되어 이전보다 1초 더 빠르게 실행됩니다. 즉, 2초 만에 두 task가 모두 실행됩니다.

이것을 볼 때, 실제로 async 작업은 create_task를 호출한 시점에 다른 작업 흐름에 실행을 이미 넘겨서 언제든지 실행될 수 있는 상태에 있다는 것을 말합니다. 그럼 await은 어떤 역할을 할까요?

% python3 test.py
started at 11:00:14
hello
world
finished at 11:00:16

 

아래는 https://kdw9502.tistory.com/6의 티스토리 글에서 발췌한 내용입니다.

태스크를 사용하면 create_task 시점에 실행을 한다고 합니다.

다른 블로그의 정보

 

그럼, 한가지 궁금한 점이 생깁니다. await을 하지 않고 asyncio.run 후에 다른 코드를 수행하면 해당 코드가 동시에 수행될 수 있는가가 궁금해 졌습니다.

아래와 같이 await을 주석으로 모두 막고, run 뒤에 print를 하나 한 후 10초를 기다려 봤습니다.

% cat test.py
import asyncio
import time


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')}")
    print(f"call at {time.strftime('%X')}")
    time.sleep(10)

asyncio.run(main())

 

하지만, 결과는 출력되지 않습니다.

% python3 test.py
started at 13:14:38
call at 13:14:38

 

asyncio.create_task()를 사용하여 생성한 task는 비동기적으로 실행되도록 예약됩니다. 그러나 await 키워드를 사용하지 않으면 해당 task가 실제로 실행되지 않습니다. 이는 비동기 코드에서의 특성으로, await 키워드를 사용하지 않으면 다른 코드가 실행되지 않고 해당 task가 완료될 때까지 기다리지 않습니다.

 

 

TaskGroup 지원

asyncio.TaskGroup 클래스는 create_task()에 보다 더 나은 방법을 제공합니다. 이 API를 사용하면 마지막 예제가 됩니다.

TaskGroup 클래스를 사용한 코드는 파이썬 공식메뉴얼에는 있지만, 제 버전에는 존재하지 않아서 테스트는 진행하지 못했습니다. 참고용으로만 확인해 보시기 바랍니다.

% python3
Python 3.10.12 (main, Nov 20 2023, 15:14:05) [GCC 11.4.0] on linux

 

TaskGroup은 3.11 부터 있습니다. "New in version 3.11: asyncio.TaskGroup."

< asyncio.TaskGroup을 사용한 예제 >

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(
            say_after(1, 'hello'))

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

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

    # The await is implicit when the context manager exits.

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

 

타이밍과 출력은 이전 버전과 동일하다고 합니다.