Microsoft MVP성태의 닷넷 이야기
글쓴 사람
정성태 (techsharer at outlook.com)
홈페이지
첨부 파일
 

파이썬 - 비동기 호출 함수(run_until_complete, run_in_executor, create_task, run_in_threadpool)

지난 글에 이어,

파이썬 - async/await 기본 사용법
; https://www.sysnet.pe.kr/2/0/13423

이번에는 다양한 비동기 호출 함수에 대해 알아보겠습니다. 우선, 비동기를 수행하는 asyncio의 run 함수 먼저 볼까요? ^^

import asyncio


async def main_async():
    time.sleep(3)


asyncio.run(main_async())

run은 async 함수를 전달받아 그것의 실행이 종료될 때까지 대기하는 역할을 합니다. 사실 이 코드는 다음과 같이 바꿔 쓸 수 있습니다. (run의 실제 내부 구현과는 다를 수 있습니다.)

import asyncio


async def main_async():
    time.sleep(3)


loop = asyncio.get_event_loop()
loop.run_until_complete(main_async())  # 의미상, "await main_async()"

만약 await 예약어 호출을 async 함수 내에서만 허용한다는 규칙이 없었다면 아마도 asyncio.run 함수는 그냥 단순히 await 예약어로 대체되었을지도 모릅니다.




파이썬의 경우, 명확하게 await 호출의 대상 함수가 async이기를 요구하는데요, 그렇다면 기존에 제작된 동기 함수들을 비동기로 호출하고 싶다면 어떻게 해야 할까요?

이런 경우에 대해 asyncio는 run_in_executor 함수를 제공합니다.

import time
import asyncio


def do_sync(delay: int=0):
    time.sleep(delay)


async def main_async():
    print('async calling', time.strftime('%X'))

    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, do_sync, 3)

    print('async called', time.strftime('%X'))


asyncio.run(main_async())

""" 출력 결과
async calling 16:02:46
async called 16:02:49
"""

run_in_executor는 2번째 인자로 받은 (동기) 함수를 내부의 이벤트 루프에 태워 비동기로 실행합니다. 따라서, run_in_executor의 호출에는 await을 해야 do_sync가 완료된 후 그다음의 코드를 이어서 호출하게 됩니다.

만약, 위의 예제에서 "await loop.run_in_executor(...)" 코드를 await 없이 호출하게 되면 실행 후 종료 시점에 아래와 같은 오류가 발생합니다.

exception calling callback for 
Traceback (most recent call last):
  File "/usr/lib/python3.8/concurrent/futures/_base.py", line 328, in _invoke_callbacks
    callback(self)
  File "/usr/lib/python3.8/asyncio/futures.py", line 374, in _call_set_state
    dest_loop.call_soon_threadsafe(_set_state, destination, source)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 764, in call_soon_threadsafe
    self._check_closed()
  File "/usr/lib/python3.8/asyncio/base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

왜냐하면, await이 없으므로 호출 대기(설명은 '대기'라고 했지만 스레드 블록킹은 아닙니다.)를 하지 않을 것이고, 이로 인해 그대로 프로그램이 종료하면서 아직 진행 중인 callback 함수(위의 경우, do_sync)가 강제 종료되므로 예외를 통해 그 사실을 알려주는 것입니다.




자... 그럼 run_in_executor의 병렬 처리도 이전에 쓴 글과 동일한 규칙으로 코딩할 수 있습니다. 우선, 다음의 코드는 차례로 6초가 걸리지만,

await loop.run_in_executor(None, do_sync, 3)  # 3초 지나서 이후의 코드로 진행 (다시 강조하지만, 호출 측의 스레드 블록킹이 되는 것이 아닙니다.)
await loop.run_in_executor(None, do_sync, 3)  # 3초 지나서 이후의 코드로 진행
                                              # 총 6초 소요

이렇게 하면 3초 만에 결과가 나옵니다.

task1 = loop.run_in_executor(None, do_sync, 3)  # 비동기로 do_sync 함수를 전달하고 바로 제어 반환
task2 = loop.run_in_executor(None, do_sync, 3)  # 비동기로 do_sync 함수를 전달하고 바로 제어 반환

await task1  # do_sync 작업 완료 대기 (3초) 후 진행
await task2  # 위의 코드에서 3초 대기를 이미 했으므로 곧바로 제어 반환




그다음, asyncio의 create_task 함수를 볼까요?

# 직렬로 비동기 함수 호출

async def main_async():
    await asyncio.create_task(do_async(3))
    await asyncio.create_task(do_async(3))

# 병렬로 비동기 함수 호출

async def main_async():
    task1 = asyncio.create_task(do_async(3))
    task2 = asyncio.create_task(do_async(3))

    await task1
    await task2

사실 저 코드는 asyncio.create_task 호출 자체를 없애고 await으로 대체해도 동일한 동작을 합니다.

async def main_async():
    await do_async(3)
    await do_async(3)

async def main_async():
    task1 = do_async(3)
    task2 = do_async(3)

    await task1
    await task2

단지, create_task의 경우 부가적으로 name 인자와 함께 문맥 정보를 전달할 수 있는 정도의 서비스가 추가된 정도입니다. (문맥은 설명이 복잡하니 다른 글에서 한번 다루겠습니다.)




이 정도면, 비동기 호출과 관련한 예를 웬만큼 다룬 것 같은데요, 마지막으로 하나 더 남은 것이 있다면 FastAPI에서 제공하는 run_in_threadpool입니다.

from fastapi.concurrency import run_in_threadpool


def myfunc(arg_values):
    delay_number = arg_values['delay']
    time.sleep(delay_number)


@app.get("/asynctest", response_class=HTMLResponse)
async def asynctest(delay: int = 0):
    delay_arg = {'delay': 1}
    await run_in_threadpool(myfunc, delay_arg)  # 1초 후 아래 코드로 진행
    await run_in_threadpool(myfunc, delay_arg)  # 1초 후 아래 코드로 진행
    await run_in_threadpool(myfunc, delay_arg)  # 1초 점유 (총 3초 소요)

    return "TEST"

run_in_threadpool의 이름처럼, run_in_executor와 같이 동기 함수를 비동기에 태워서 실행할 수 있는 데, 하지만 동작 자체는 매우 다릅니다. 왜냐하면, run_in_threadpool에 전달한 시점에 myfunc가 호출되지 않고, await을 한 시점에 실질적인 코드 실행 단계로 넘어가기 때문입니다.

따라서, run_in_threadpool의 경우에는 await만으로는 "병렬 처리"가 되지 않습니다.

task1 = run_in_threadpool(myfunc, delay_arg)  # 작업만 전달 (실행은 안 됨)
task2 = run_in_threadpool(myfunc, delay_arg)  # 작업만 전달 (실행은 안 됨)
task3 = run_in_threadpool(myfunc, delay_arg)  # 작업만 전달 (실행은 안 됨)

await task1  # 이 단계에서 myfunc 작업을 비동기로 실행
await task2  # task1이 완료된 후 task2가 이어서 비동기로 실행
await task3  # task2가 완료된 후 task3이 이어서 비동기로 실행

대신 run_in_threadpool의 작업을 병렬 처리하고 싶다면 asyncio.gather 등의 함수를 이용할 수 있습니다.

await asyncio.gather(task1, task2, task3)  # task1, task2, task3 작업이 모두 한꺼번에 비동기로 실행

한 가지 유의해야 할 점은, run_in_threadpool의 경우 비동기 함수를 인자로 받아도 실행 시 오류가 발생하지 않는다는 점입니다.

@app.get("/asynctest", response_class=HTMLResponse)
async def asynctest(delay: int = 0):
    task1 = run_in_threadpool(my_aio_func)  # 비동기 함수를 인자로 전달
    task2 = run_in_threadpool(my_aio_func)
    task3 = run_in_threadpool(my_aio_func)

    await asyncio.gather(task1, task2, task3)


async def my_aio_func(args=None):
    print('my_aio_func: start:', args)
    await asyncio.sleep(1)
    return "my value(async)"

실제로 위의 코드를 실행해 보면 아무런 오류가 발생하지 않는데, 여기서 문제는 my_aio_func이 아예 실행이 안 된다는 점입니다. (실행해 보시면, 화면 출력에 "my-aio_func: start"라는 문자열이 안 나옵니다.)

하지만, FastAPI 앱을 종료해 보면 그제야 다음과 같은 식으로 오류 메시지를 출력합니다.

/usr/lib/python3.8/threading.py:932: RuntimeWarning: coroutine 'my_aio_func' was never awaited
  self.run()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

이 정도면... 대충 파악이 되셨을 거라 생각합니다. ^^




[이 글에 대해서 여러분들과 의견을 공유하고 싶습니다. 틀리거나 미흡한 부분 또는 의문 사항이 있으시면 언제든 댓글 남겨주십시오.]







[최초 등록일: ]
[최종 수정일: 10/13/2023]

Creative Commons License
이 저작물은 크리에이티브 커먼즈 코리아 저작자표시-비영리-변경금지 2.0 대한민국 라이센스에 따라 이용하실 수 있습니다.
by SeongTae Jeong, mailto:techsharer at outlook.com

비밀번호

댓글 작성자
 




... [16]  17  18  19  20  21  22  23  24  25  26  27  28  29  30  ...
NoWriterDateCnt.TitleFile(s)
13460정성태11/26/20236773닷넷: 2172. .NET 6+ 기반의 COM Server 내에 Type Library를 내장하는 방법파일 다운로드1
13459정성태11/26/20236583닷넷: 2171. .NET Core 3/5+ 기반의 COM Server를 기존의 regasm처럼 등록하는 방법파일 다운로드1
13458정성태11/26/20236710닷넷: 2170. .NET Core/5+ 기반의 COM Server를 tlb 파일을 생성하는 방법(tlbexp)
13457정성태11/25/20236481VS.NET IDE: 187. Visual Studio - 16.9 버전부터 추가된 "Display inline type hints" 옵션
13456정성태11/25/20236927닷넷: 2169. C# - OpenAI를 사용해 PDF 데이터를 대상으로 OpenAI 챗봇 작성 [1]파일 다운로드1
13455정성태11/25/20236812닷넷: 2168. C# - Azure.AI.OpenAI 패키지로 OpenAI 사용파일 다운로드1
13454정성태11/23/20237070닷넷: 2167. C# - Qdrant Vector DB를 이용한 Embedding 벡터 값 보관/조회 (Azure OpenAI) [1]파일 다운로드1
13453정성태11/23/20236373오류 유형: 879. docker desktop 설치 시 "Invalid JSON string. (Exception from HRESULT: 0x83750007)"
13452정성태11/22/20236599닷넷: 2166. C# - Azure OpenAI API를 이용해 사용자가 제공하는 정보를 대상으로 검색하는 방법파일 다운로드1
13451정성태11/21/20236826닷넷: 2165. C# - Azure OpenAI API를 이용해 ChatGPT처럼 동작하는 콘솔 응용 프로그램 제작파일 다운로드1
13450정성태11/21/20236476닷넷: 2164. C# - Octokit을 이용한 GitHub Issue 검색파일 다운로드1
13449정성태11/21/20236655개발 환경 구성: 688. Azure OpenAI 서비스 신청 방법
13448정성태11/20/20236981닷넷: 2163. .NET 8 - Dynamic PGO를 결합한 성능 향상파일 다운로드1
13447정성태11/16/20237004닷넷: 2162. ASP.NET Core 웹 사이트의 SSL 설정을 코드로 하는 방법
13446정성태11/16/20236718닷넷: 2161. .NET Conf 2023 - Day 1 Blazor 개요 정리
13445정성태11/15/20237465Linux: 62. 리눅스/WSL에서 CA 인증서를 저장하는 방법
13444정성태11/15/20237107닷넷: 2160. C# 12 - Experimental 특성 지원
13443정성태11/14/20237003개발 환경 구성: 687. OpenSSL로 생성한 사용자 인증서를 ASP.NET Core 웹 사이트에 적용하는 방법
13442정성태11/13/20236924개발 환경 구성: 686. 비주얼 스튜디오로 실행한 ASP.NET Core 사이트를 WSL 2 인스턴스에서 https로 접속하는 방법
13441정성태11/12/20237437닷넷: 2159. C# - ASP.NET Core 프로젝트에서 서버 Socket을 직접 생성하는 방법파일 다운로드1
13440정성태11/11/20236398Windows: 253. 소켓 Listen 시 방화벽의 Public/Private 제어 기능이 비활성화된 경우
13439정성태11/10/20238207닷넷: 2158. C# - 소켓 포트를 미리 시스템에 등록/예약해 사용하는 방법(Port Exclusion Ranges)파일 다운로드1
13438정성태11/9/20236968닷넷: 2157. C# - WinRT 기능을 이용해 윈도우에서 실행 중인 Media App 제어
13437정성태11/8/20237116닷넷: 2156. .NET 7 이상의 콘솔 프로그램을 (dockerfile 없이) 로컬 docker에 배포하는 방법
13436정성태11/7/20237370닷넷: 2155. C# - .NET 8 런타임부터 (Reflection 없이) 특성을 이용해 public이 아닌 멤버 호출 가능
13435정성태11/6/20237166닷넷: 2154. C# - 네이티브 자원을 포함한 관리 개체(예: 스레드)의 GC 정리
... [16]  17  18  19  20  21  22  23  24  25  26  27  28  29  30  ...