asyncio – 일반 함수를 비동기로 사용하기

지난 글에서 urlopen()과 같은 표준 라이브러리 함수를 어떻게 비동기 코루틴처럼 asyncio에서 사용할 수 있는지 살펴보았다. aiohttp 등의 비동기 라이브러리를 사용해서 여러 핸들러를 작성해야 할 때, 이와 같은 처리를 많이 해야 한다면 빈번하게 런루프 메소드를 호출하는 것보다, 간단히 데코레이터를 만들어서 활용하는 것이 어떨까?

asyncio의 이벤트 루프에는 run_in_executor(executor, fn, *args) 가 있다. 이 메소드는 concurrent.futures 모듈의 ThreadPoolExecutor나 ProcessPoolExecutor를 사용하여 일반적인 blocking 함수를 다른 스레드 및 프로세스에서 실행하도록 하고 그 자신은 처리를 기다리는 코루틴을 생성한다. 이 기능을 사용하면 일반적인 blocking-I/O 함수를 non-blocking 함수처럼 사용할 수 있다. 단, 파이썬은 GIL에 의해 여러 스레드가 동시에 실행될 수는 없으므로 I/O와 관련된 작업에서만 실질적인 성능향상을 얻을 수 있다.

여기서는 한 반 더 나아가 사용자 정의 함수를 이렇게 사용할 수 있도록하는 데코레이터를 작성해보도록 하겠다. 여기서는 ThreadPoolExecutor를 사용하기 때문에 스레드에서 함수를 실행하지만, ProcessPoolExecutor를 사용하도록 변경하면, 별도의 프로세스에서 함수가 실행될 수 있고 이 때에는 CPU 부하가 큰 작업을 비동기로 실행하여 성능 향상의 효과를 볼 수 있다.

from functools import partial, wraps
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor)
from urllib.request import urlopen
import asyncio


def run_async(loop=None, pool=None):
  _loop = loop if loop is not None else\
          asyncio.get_current_loop()
  def decorator(fn):
    @wraps(fn)
    async def coro(*args, **kwds):
      # `run_in_executor()`는 *args 만 넘겨줄 수 있음
      _fn = partial(**kwds)
   
      # pool=None 이면 기본적으로 ThreadPoolExecutor를 생성해서 실행한다.
      res = await _loop.run_in_executor(pool, _fn, *args)
      return res
    return coro
  return decorator


@run_async(pool=ProcessPoolExecutor(max_worker=4))
def test(url):
  res = urlopen(url)
  return res.read().decode()[:1000]


asyncio.run(test('https://soooprmx.com'))

파이썬 3.9부터는 asyncio.to_thread(fn, *args, **kwds) 함수가 새로 추가되었다. 이 함수의 내부 구현은 사실상 앞서 표현한 것과 동일하다. 현재 런루프를 얻어 run_in_executor()를 호출하게 되어 있다.

async def test():
  res = await asyncio.to_thread(urlopen, 'https://soooprmx.com')
  body = res.read().decode()[:1000]
  print(body)

asyncio.run(test())