asyncio

EventLoop in Thread

import asyncio
import threading

# 메인 스레드에서 자식 스레드의 이벤트 루프를 제어
new_loop = asyncio.new_event_loop()

def run_loop(loop: asyncio.AbstractEventLoop) -> None:
    # 현재 스레드에 이벤트 루프 등록
    asyncio.set_event_loop(loop)

    try:
        # 현재 스레드에 등록된 이벤트 루프 가져오기
        # loop = asyncio.get_event_loop()
        loop.run_forever()
    finally:
        try:
            loop.run_until_complete(loop.shutdown_asyncgens())
            loop.run_until_complete(loop.shutdown_default_executor())
        finally:
            # 현재 스레드에 등록된 이벤트 루프 제거
            asyncio.set_event_loop(None)
            loop.close()

t = threading.Thread(target=run_loop, args=(new_loop,))
t.start()

# new_loop에 작업 요청
# ...

# loop.run_forever()를 종료해달라는 요청
new_loop.call_soon_threadsafe(new_loop.stop)
t.join()

EventLoop

• 이벤트 루프 실행

  • loop.run_until_complete(future)
    • future가 완료될 때까지 블로킹 됩니다.
    • 실행전에 등록된 콜백이 있는 경우 해당 콜백들이 실행됩니다.
    • 코루틴을 인자로 전달한 경우 asyncio.Task로 감싸서 실행합니다.
    • future.result()를 반환합니다.
  • loop.run_forever(): 이벤트 루프를 실행시키고 루프가 종료될 때까지 블로킹 됩니다.
  • loop.stop()
    • 루프에 등록된 모든 코루틴을 종료시키고 루프를 종료합니다.
    • 새로 등록한 콜백은 실행되지 않고, 다음 이벤트 루프를 실행시킬 때 실행됩니다.
    • 콜백 내에서 새로운 콜백을 등록한 경우 새로운 콜백이므로 실행되지 않기 때문에 await에 걸려있는 경우 프로그램이 멈추게 됩니다.

• 이벤트 루프에 콜백 등록

  • loop.call_soon(callback, *args, context=None)
    • 콜백은 callback(*args)로 실행됩니다.
    • 컨텍스트를 지정하지 않으면 현재 컨택스트가 사용됩니다.
    • asyncio.Handle를 반환합니다.
  • loop.call_soon_threadsafe(callback, *args, context=None)
    • 스레드 안전한 loop.call_soon()입니다.
    • 다른 스레드에서 콜백을 등록할 때 사용해야합니다.
  • loop.call_later(delay, callback, *args, context=None)
    • delay초 후에 콜백을 등록합니다.
    • asyncio.TimerHandle를 반환합니다.
  • loop.call_at(when, callback, *args, context=None)
    • when == loop.time()에 콜백을 등록합니다.
    • asyncio.TimerHandle를 반환합니다.
  • loop.run_in_executor(executor, callback, *args)
    • executor에서 콜백을 실행합니다.
      • None(기본 실행기), concurrent.futures.ThreadPoolExecutor, concurrent.futures.ProcessPoolExecutor
    • asyncio.Future를 반환합니다.
  • loop.add_signal_handler(sig, callback, *args)
    • signal.SIGINT, signal.SIGTERM, ... 등의 signal이 발생하면 콜백을 실행합니다.
    • 콜백은 callback(*args)로 실행됩니다.
    • 메인 스레드에서만 사용할 수 있습니다.
    • loop.remove_signal_handler(sig)로 콜백을 제거할 수 있습니다.
  • asyncio.run_coroutine_threadsafe(coroutine, loop)
    • 루프에 코루틴을 등록합니다.
    • concurrent.futures.Future를 반환합니다.
  • Keywords 인자를 사용해야하는 경우 functools.partial()을 사용합니다.
    • e.g., loop.call_soon(functools.partial(callback, arg1, key=value))

• 이벤트 루프 상태 확인

  • loop.is_running(): 루프가 실행중이면 True를 반환합니다.
  • loop.is_closed(): 루프가 닫혔으면 True를 반환합니다.

• 이벤트 루프 닫기

  • loop.shutdown_asyncgens()
    • 현재 열려있는 비동기 제너레이터 객체의 aclose()를 호출하여 닫습니다.
    • 이후에 비동기 제너레이터가 열리면 경고가 발생합니다.
  • loop.shutdown_default_executor()
    • 기본 실행기의 종료 콜백을 등록하고, 현재 열려있는 ThreadPoolExecutor의 스레드들을 조인합니다.
    • 이후에 loop.run_in_executor()에 기본 실행기를 사용하려는 경우 RuntimeError를 발생시킵니다.
  • loop.close()
    • 루프를 닫습니다.
    • 등록된 모든 콜백이 삭제 됩니다.