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

비밀번호

댓글 작성자
 




1  [2]  3  4  5  6  7  8  9  10  11  12  13  14  15  ...
NoWriterDateCnt.TitleFile(s)
13541정성태1/29/20241011VS.NET IDE: 188. launchSettings.json의 useSSL 옵션
13540정성태1/29/20241132Linux: 69. 리눅스 - "Docker Desktop for Windows" Container 환경에서 IPv6 Loopback Address 바인딩 오류
13539정성태1/26/20241402개발 환경 구성: 703. Visual Studio - launchSettings.json을 이용한 HTTP/HTTPS 포트 바인딩
13538정성태1/25/20241531닷넷: 2211. C# - NonGC(FOH) 영역에 .NET 개체를 생성파일 다운로드1
13537정성태1/24/20241550닷넷: 2210. C# - Native 메모리에 .NET 개체를 생성파일 다운로드1
13536정성태1/23/20241646닷넷: 2209. .NET 8 - NonGC Heap / FOH (Frozen Object Heap) [1]
13535정성태1/22/20241490닷넷: 2208. C# - GCHandle 구조체의 메모리 분석
13534정성태1/21/20241313닷넷: 2207. C# - SQL Server DB를 bacpac으로 Export/Import파일 다운로드1
13533정성태1/18/20241379닷넷: 2206. C# - TCP KeepAlive의 서버 측 구현파일 다운로드1
13532정성태1/17/20241354닷넷: 2205. C# - SuperSimpleTcp 사용 시 주의할 점파일 다운로드1
13531정성태1/16/20241398닷넷: 2204. C# - TCP KeepAlive에 새로 추가된 Retry 옵션파일 다운로드1
13530정성태1/15/20241381닷넷: 2203. C# - Python과의 AES 암호화 연동파일 다운로드1
13529정성태1/15/20241283닷넷: 2202. C# - PublishAot의 glibc에 대한 정적 링킹하는 방법
13528정성태1/14/20241401Linux: 68. busybox 컨테이너에서 실행 가능한 C++, Go 프로그램 빌드
13527정성태1/14/20241296오류 유형: 892. Visual Studio - Failed to launch debug adapter. Additional information may be available in the output window.
13526정성태1/14/20241353닷넷: 2201. C# - Facebook 연동 / 사용자 탈퇴 처리 방법
13525정성태1/13/20241301오류 유형: 891. Visual Studio - Web Application을 실행하지 못하는 IISExpress
13524정성태1/12/20241334오류 유형: 890. 한국투자증권 KIS Developers OpenAPI - GW라우팅 중 오류가 발생했습니다.
13523정성태1/12/20241234오류 유형: 889. Visual Studio - error : A project with that name is already opened in the solution.
13522정성태1/11/20241328닷넷: 2200. C# - HttpClient.PostAsJsonAsync 호출 시 "Transfer-Encoding: chunked" 대신 "Content-Length" 헤더 처리
13521정성태1/11/20241456닷넷: 2199. C# - 한국투자증권 KIS Developers OpenAPI의 WebSocket Ping, Pong 처리
13520정성태1/10/20241276오류 유형: 888. C# - Unable to resolve service for type 'Microsoft.Extensions.ObjectPool.ObjectPool`....'
13519정성태1/10/20241347닷넷: 2198. C# - Reflection을 이용한 ClientWebSocket의 Ping 호출파일 다운로드1
13518정성태1/9/20241467닷넷: 2197. C# - ClientWebSocket의 Ping, Pong 처리
13517정성태1/8/20241358스크립트: 63. Python - 공개 패키지를 이용한 위성 이미지 생성 (pystac_client, odc.stac)
13516정성태1/7/20241489닷넷: 2196. IIS - AppPool의 "Disable Overlapped Recycle" 옵션의 부작용
1  [2]  3  4  5  6  7  8  9  10  11  12  13  14  15  ...