개요
이전에 2024.06.26-[Python] asyncio - 비동기 프로그래밍에서 Python에서 비동기 프로그래밍을 하기 위한 라이브러리, 문법, 실행 방법을 알아보았었다.
근데 실제로 사용해 보니 동작 방식에 대한 이래도가 너무 낮다는 느낌이 들었다. 그래서 이번 글에서는 Python에서 비동기 함수를 실행하는 주체인 EventLoop에 대한 개념을 적어두려고 한다.
EventLoop
EventLoop는 비동기 프로그램의 핵심적인 요소로, 비동기 작업과 콜백, 네트워크 I/O 연산, 자식 프로세스 등을 실행한다.
일반적으로 개발자는 asyncio.run과 같은 고수준 함수를 사용하여 비동기 프로그래밍을 개발하게 된다. 즉, EventLoop를 직접 참조하거나 관련 메서드를 호출할 필요가 없다. 하지만 저수준 코드 개발자, 라이브러리 및 프레임워크 개발자는 EventLoop를 세부적으로 제어하는 경우가 존재할 수 있다.
실행 흐름 / 동작 원리
EventLoop의 실행 흐름과 동작 원리를 알면 Python에서의 비동기 프로그래밍이 동작하는 원리를 알 수 있다.
Python 비동기 프로그래밍에서 코루틴을 실행시키는 방법은 크게 await, asyncio.run, asyncio.create_task로 구분할 수 있다. await 키워드는 코루틴 내에서만 사용할 수 있는 방법이고, asyncio.run과 asyncio.create_task을 코루틴 밖에서 코루틴을 실행시킬 수 있는, 즉 코루틴 체인으로 들어가는 일종의 엔트리포인트에 해당한다. 특히 일반적인 asyncio.run은 일반적인 비동기 프로그래밍의 시작점에 해당한다.
asyncio.run은 현재의 스레드에 새 이벤트 루프를 설정하고, 이벤트 루프에서 인자로 넘어오는 코루틴 객체에 해당하는 코루틴을 태스크로 예약하여 실행시킨 뒤, 태스크의 실행이 완료되면 이벤트 루프를 닫는 역할을 수행한다. Python 3.7 미만에서는 아래의 코드로 작성된다.
loop = asyncio.get_event_loop()
loop.run_until_complete(first_coroutine())
loop.close()
단계 별로 동작 원리를 정리한다.
1. loop = asyncio.get_event_loop()
get_event_loop는 현재 스레드에 설정된 이벤트 루프를 가져온다. 만약 현재 스레드에 설정되어 있는 이벤트 루프가 없으면, 새로 생성하여 이를 현재 스레드에 설정한 뒤에 해당 이벤트 루프를 반환한다.
이벤트 루프란 무한 루프를 돌면서 매 루프마다 작업을 하나씩 실행시키는 로직을 의미한다. 달리 말해, 여기서 말하는 현재 스레드에 이벤트 루프를 설정한다는 것은 이벤트 루프를 실행시킬 수 있는 객체를 생성한 것에 해당하며, 작업(Task)이란 하나의 코루틴에서 출발하는 실행 흐름이라고 할 수 있다.
아래는 이벤트 루프가 실행되는 흐름을 간략하게 표현한 것이다.
2. loop.run_until_complete(first_coroutine())
run_until_complete은 이벤트 루프 객체를 이용해 실제로 이벤트 루프를 실행시킨다.
2 - 1. 작업 실행
인자로 넘어오는 코루틴을 이용해 Task 객체를 생성하고, 이벤트 루프에 의해 Task의 실행이 즉시 예약된다.
처음에는 실행이 예약된 Task가 없기 때문에 이벤트 루프는 Task를 바로 실행한다. 여기서 Task를 실행한다는 것은 Task 객체의 __step 메서드를 호출하는 것의 의미한다. __step은 코루틴 객체의 send 메서드를 호출하여 코루틴을 실행시킨다.
이 코루틴을 시작으로 await을 마주칠 때마다 연쇄적으로 코루틴을 호출하여 코루틴 체인을 형성하게 된다.
2 - 2. 코루틴 체인의 종착지 (await)
await을 통해 코루틴을 실행하다 보면 asyncio.sleep 또는 I/O 관련 코루틴을 await 하는 코드를 마주칠 수 있다. 일반적으로 이러한 코루틴을 Future 객체를 await 하도록 구현되어 있다.
I/O 관련 코루틴은 보통 select 함수를 이용해 소켓을 등록한 뒤, 소켓에 바인딩된 Future 객체를 새로 생성하여 await 한다. Future의 __await__ 메서드는 자기 자신을 yield 하도록 구현되어 있기 때문에 Future 객체는 코루틴 체인을 따라 __step 메서드까지 도달한다.
Sleep 코루틴은 이벤트 루프 자체의 타이머를 이용하는데, Future 객체를 하나 생성한 뒤 이벤트 루프에 지정한 시간 뒤에 해당 Future 객체의 결과를 업데이트하도록 요청한다. 그리고 Future 객체를 await 한다. 이후로는 I/O 관련 코루틴과 마찬가지로 Future 객체는 코루틴 체인을 따라 __step 메서드까지 도달한다.
💡 select
Unix의 select를 래핑한 Python 함수
특정 소켓에 대해 데이터를 읽거나 쓸 준비가 될 때까지 대기할 수 있게 하는 Blocking 함수. 원하는 시간만큼 대기한 후 데이터를 읽거나 쓸 준비가 된 소켓을 반환한다.
2 - 3. Task의 Future 처리
yield 된 Future를 받은 Task는 Future를 자신의 _fut_waiter 속성에 저장(바인딩)한다.
그리고 Future의 add_done_callback 메서드를 호출하여 Future가 완료 상태가 될 때 이벤트 루프에게 실행을 예약할 콜백 함수를 등록한다. 등록하는 함수는 자신의 __step 메서드라고 할 수 있다.
이후 Task는 자신의 실행을 중단하고 이벤트 루프에게 제어를 넘긴다. 그러면 이벤트 루프는 실행이 예약된 Task 중 우선순위가 높은 것을 선택하여 실행시킨다.
이러한 과정을 반복하면서 여러 Task를 Concurrent 하게 실행하게 된다.
2 - 4. 이벤트 루프 Polling
이벤트 루프는 실행을 예약한 Task가 존재하지 않으면 select 함수를 사용하여 데이터를 읽거나 쓸 준비가 된 소켓을 찾는다.
소켓을 찾으면 소켓에 바인딩되어 있는 Future의 결과 값을 업데이트하는데, 업데이트되는 순간에 콜백 함수의 실행이 이벤트 루프에 예약된다.
2 - 5. Task 실행 재개
Task의 실행이란 Task의 __step 메서드가 호출되는 것을 말한다. __step 메서드는 Task와 Future의 바인딩을 해제하여 기다리는 Future가 없음을 나타내고, 자신의 코루틴에 대해 send 메서드를 호출하여 해당 코루틴의 실행을 재개하게 한다. 이때 Future 객체의 __await__ 메서드에서 실행이 중단됐던 부분(자기 자신을 yeild 하는 부분)까지 가게 된다.
__await__ 으로 돌아왔을 때, I/O 관련 코루틴을 기다리고 있었던 거라면 해당 소켓에 데이터를 읽거나 쓴 다음 값을 반환한다. 반대로 sleep을 기다리고 있었다면 바로 return 한다.
2 - 6. 최초 코루틴 종료
언젠가는 Task가 실행한 최초의 코루틴이 반환되는 시점에 도달하며, Task의 __step 메서드에서 StopIteration이 발생할 수 있다. 이때 Task는 예외 객체의 value 필드 값을 자신의 결과 값을 업데이트하고 자신의 실행을 종료한다. 이벤트 루프에 의해 실행이 예약되지 않고 버려지는 것이다. 이때가 바로 loop.run_until_complete 함수의 실행이 종료되는 시점이다.
3. loop.close()
loop.run_until_complete의 실행이 끝났다는 것은 이벤트 루프가 실행되지 않는다는 것이므로 이벤트 루프를 닫아주어야 한다. loop.close 함수는 실행이 종료되지 않은 Task 같이 이벤트 루프에 남아있는 데이터를 제거하고 이벤트 루프를 닫아준다.
만약 loop.run_until_complete 실행이 끝나고 loop.close에 의해 이벤트 루프가 닫히는 시점까지 완료되지 않는 Task가 남아있다면 "Task was destroyed but it is pending!"라는 경고가 출력된다.
참고 문서
https://teddylee777.github.io/python/python-async/
https://it-eldorado.com/posts/5ba540c3-98fa-4559-a673-cf232f6978ad