Asyncio와 비동기 작업 4) multithreading과 run_in_executor 메서드

1. 동기 코드가 비동기 코드를 block하는 경우

비동기식 코드

먼저 event loop를 활성화해야 호출할 수 있음.

 

동기식 코드

동기 코드는 비동기식 코드블럭에서 쉽게 호출할 수 있음. 

하지만 동기식 코드가 비동기식 코드의 비동기적 실행을 막을 수도 있음.  

 

예) 비동기 함수 counter()가 response = requests.get("www.naver.com")이 실행 완료된 다음에야 재개됨.

=> 비동기 함수는 사용했지만 동기적으로 코드가 실행됨.

import requests

async def counter():
    print("started counter")
    for i in range(0, 5):
        # 3. 현재 태스크 재우고 다른 태스크(main()) 깨움
        await asyncio.sleep(0.001)
        # 6. 실행
        print(f"i is {i}")

async def main():
    # 1. 이벤트 루프에 새 태스크 등록
    task = asyncio.create_task(counter())

    # 2. 현재 태스크 재우고 이벤트 루프에 등록된 태스크(counter()) 깨움
    await asyncio.sleep(0)

    # 4. 실행 - 2초 소요
    print("started http request")
    response = requests.get("https://www.naver.com")
    print("ended http request")

    # 5. 1번에서 등록한 태스크(counter()) 깨우고 실행
    await task

asyncio.get_event_loop().run_until_complete(main())

# started counter
# started http request
# ended http request
# i is 0
# i is 1
# i is 2
# i is 3
# i is 4

 

2. multithreading과 run_in_executor()

asyncio는 기본적으로 single thread 위에서 동작한다. (한 event loop는 한 thread를 가짐)

asyncio를 사용한 I/O bound 작업만 할때는 single thread 방식도 충분히 효율적이다.

 

하지만 asyncio를 사용하지 않은 코드(동기 코드)를 asyncio를 사용한 코드(비동기 코드)와 같이 사용하는 경우

위 경우처럼 동기 코드가 비동기 코드를 blocking하는 경우가 있다.

이 때는 multithread 방식을 사용해야한다.

 

multithread는 여러 thread를 thread pool에 모아놓고 있다가

비동기 코드를 blocking하는 동기 코드에게 thread를 하나씩 할당해서 여러 thread를 동시에 실행하는 방식이다. 

 

이렇게 하면 한 thread는 한 event loop 안에 있는 task를 왔다갔다하며 비동기 코드를 실행하고, 

다른 thread는 시간이 오래 걸리는 동기 코드를 실행하게 만들 수 있다. 

multithread는 asyncio에서 제공하는 run_in_executor 메서드를 사용해서 구현할 수 있다.

async def counter():
    print("started counter")
    for i in range(0, 5):
        # 3. 현재 태스크 재우고 다른 태스크(main()) 깨움
        await asyncio.sleep(0.001)
        # 6. 실행
        print(f"i is {i}")

async def main():
    # 1. 이벤트 루프에 새 태스크 등록
    task = asyncio.get_event_loop().create_task(counter())

    # 2. 현재 태스크 재우고 이벤트 루프에 등록된 태스크(counter()) 깨움
    await asyncio.sleep(0)

    def send_request():
        print("started http request")
        response = requests.get("http://www.naver.com")
        print("ended http request")

    # 4. 다른 thread에 send_request 할당하고 실행
    await asyncio.get_event_loop().run_in_executor(None, send_request)
    
    # 5. 4번 완료 여부와 상관없이 1번에서 등록한 태스크(counter()) 깨우고 실행
    await task


asyncio.get_event_loop().run_until_complete(main())

# started counter
# started http request
# i is 0
# i is 1
# i is 2
# i is 3
# i is 4
# ended http request

run_in_executor 메서드는 인자를 두 개 넣어 호출하고, 호출 시 future를 반환한다.

첫 번째 인자에는 사용할 thread pool을 넣고(None 입력시 기본 thread pool 사용),

두번째 인자에는 새 thread에서 실행할 동기 함수를 넣는다.

 

future는 동기 함수가 실행을 완료하면 done=True 값을 갖는다.

future의 result에는 동기 함수가 값을 반환했으면 반환값이, 에러가 났으면 exception이 저장된다.

 

run_in_executor는 future를 반환한다는 점에서 create_task와 비슷하다.

하지만 create_task는 현재 thread의 event loop에 코루틴을 등록한다는 점,

run_in_executor는 새로운 thread에 동기 함수를 등록한다는 점이 다르다.

 

또 create_task로 생성한 task는 event loop에 이미 실행 예약된 task가 있는 경우 즉시 실행되지 않는다.

run_in_executor는 새로운 thread를 가지므로 대부분의 경우 즉시 실행된다.

 

run_in_executor는 CPU bound 작업에 특히 유용하다.

 

3. Global Interpreter Lock

파이썬은 원칙적으로 single thread 방식을 사용하므로

Global Interpreter Lock(=GIL)을 사용해서 multithreading을 제한한다.

 

GIL은 특정 시점에 하나의 스레드만 공유 자원에 접근할 수 있게 하는 장치이다. (= GIL은 mutex이다.)

GIL 때문에 프로그램에 여러 스레드가 존재하더라도

여러 스레드가 동시에 실행되지 않고 번갈아가며 한번에 하나씩만 실행된다. 

 

하지만 native 코드(OS가 직접 컴파일하는 코드 - C, C++로 작성된 코드)를 실행하는 경우 GIL은 자동으로 해제되는데,

이 때는 여러 스레드를 동시에 실행할 수 있다. 

 

I/O bound 작업에 필요한 라이브러리는 보통 native 코드를 사용한다. 

즉, run_in_executor로 I/O bound 작업을 하는 경우

GIL이 해제되어 있을 가능성이 크므로 multithread를 활용하는 데 큰 문제가 없다.

 

CPU bound 작업에도 run_in_executor를 사용할 수 있다. 

하지만 이 경우 다른 스레드의 작업이 느려지거나 끊길 수 있다.

이 때는 run_in_executor의 첫번째 인자로 concurrent.futures.ProcessPoolExecutor를 넣어주면 된다.

 

 

 

 

참고

python 공식문서 - asyncio

Python Asyncio Part 5 – Mixing Synchronous and Asynchronous Code