python async IO를 통한 동시성 처리

python에서 동시성을 처리하는 방법을 정리해본다.

 

다루게 될 키워드들.

 

multiprocessing, multithreading, async, await, async IO, coroutine, task, explicitly asynchronous, implicitly asynchronous, futures

 

기존에 동시성을 구현하기 위해 사용하던 방법들

1. Multiprocessing

복수의 CPU를 동시에 사용. multiprocessing 패키지을 이용해서 구현할 수 있다. 하드웨어 사양에 관련이 있고, 동시에 많은 계산이 필요한 작업에 적합하다.

 

2. Threading

다수의 thread가 번갈아가면서 실행되는 방법으로 동시성을 구현한다. 하나의 process가 다수의 thread를 포함할 수 있다. python에서는 threading 패키지을 이용해서 구현할 수 있으며, CPU의 사용보다 기다리는 일이 많은 동시 IO 작업에 더 적합하다.

 

Async IO

C# 과 같은 다른 언어에서 제공하는 async, await 방식을 asyncio 패키지를 통해 python에서도 지원하게 되었다.

async IO 는 multiprocessing 이나 threading 을 사용하지 않으며, single-process, single-thread 환경에서 "cooperative multitasking"이라는 방식을 통해 동작한다.

 

Single-thread 가 중요한 이유는 대부분의 POSIX 기반 OS에서 Socket Connection은 수천, 수만개를 부담 없이 열 수 있지만, thread는 몇 백개만 넘어가도 시스템에 부담을 주기 때문이다. 대량의 Socket Connection을 동시에 처리해야 할 때 multi-thread로 처리하는 것 보다 Single-Thread로 처리하는 것이 훨씬 경제적이다.

 

Asyncronous (비동기)

비동기 함수는 결과적으로 코드 실행의 동시성을 구현하기 위해 사용된다. 지나치게 단순화해서 말하면, async 방식은 어느 부분에서 sleep(or pause) 하고 있는 동안 다른 부분이 실행되도록 하는 것이다.

Async IO는 single-process, single-thread 에서 위에서 말한 방식을 구현할 수 있게 한다.

 

asyncio package

Python 3.4부터 asyncio 패키지가 표준 라이브러리에 포함되었다. asyncio 의 async/ await를 이용해서 비동기 코드를 구현한다.

IO 가 병목인 네트워크 코드를 만드는데 적합하다.

 

coroutine

coroutine은 return되기 전에 실행을 중단하고 다른 coroutine에 제어권을 넘길 수 있는 함수다.

기본적으로 coroutine은 async def 키워드를 이용해서 만든다. 

 

import asyncio
import time

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")

async def main():
    await count()
    await count()
    await count()

if __name__ == "__main__":
    s = time.perf_counter()

    # asyncio.run() 으로 시작해야 한다. main()으로 호출하면 coroutine(main())이 실행되지 않는다.
    asyncio.run(main())
    
    elapsed = time.perf_counter() - s
    print(f"executed in {elapsed:0.2f} seconds.")

 

One
Two
One
Two
One
Two
executed in 3.01 seconds.

one (1초 쉬고) two 가 3회 반복. 총 3(3.01)초가 소요된다. 응?

여기까지만 보면 sync 로 구현된 아래 코드와 동일하다.

 

import time

def count():
    print("One")
    time.sleep(1)
    print("Two")

def main():
    count()
    count()
    count()


if __name__ == "__main__":
    s = time.perf_counter()

    main()
    
    elapsed = time.perf_counter() - s
    print(f"executed in {elapsed:0.2f} seconds.")

One
Two
One
Two
One
Two
executed in 3.01 seconds.

 

자, 첫번째 코드에서 (asyncio 사용) count() 호출하는 부분을 이렇게 바꿔보자.

async def main():
    # count() 3번을 asynchronous 하게 실행한다.
    await asyncio.gather(count(), count(), count())

One
One
One
Two
Two
Two
executed in 1.01 seconds.

 

count()에서 one 찍고 1초 쉬는 동안 다음 count() 가 2회 더 실행되었고, 총 1.01 초만에 count() 3회가 asynchronous하게 실행되었다. time.sleep() 와 달리 await asyncio.sleep()은 전체 thread를 쉬게 하지 않고, 해당 coroutine만 쉬고 제어권을 넘긴다.

 

정말 single-thread 에서 돌아가는지 한번 확인해보자.

import asyncio
import time, threading

async def count():
    print(f'[{threading.get_ident()}] One')
    await asyncio.sleep(1)
    print(f'[{threading.get_ident()}] Two')

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == '__main__':
    print(f'[{threading.get_ident()}] asyncio.run(main())')
    s = time.perf_counter()

    asyncio.run(main())
    
    elapsed = time.perf_counter() - s
    print(f'[{threading.get_ident()}] executed in {elapsed:0.2f} seconds.')

 

[8270077440] asyncio.run(main())
[8270077440] One
[8270077440] One
[8270077440] One
[8270077440] Two
[8270077440] Two
[8270077440] Two
[8270077440] executed in 1.00 seconds.

 

await

coroutine 내부의 await는 await 뒤의 부분이 완료될 때까지 해당 coroutine의 실행을 잠시 중단하고 제어권을 넘긴다. 준비가 되면 다시 돌아와서 그 부분에서부터 실행이 계속된다.

await 를 사용할 수 있는 조건은 다음과 같다.

• coroutine 함수를 실행하고, 결과를 얻기 위해서는 await를 사용해서 호출해야 한다.
• await는 async 함수 내에서만 사용해야 한다.
• await는 "awaitable" object에만 쓸 수 있다. awaitable object는 coroutine, task, future 3가지이다.

coroutine 함수를 실행하고, 결과를 얻기 위해서는 await를 사용해서 호출해야 하고, (첫번째 예시 코드에서 await count() 대신 count() 를 호출하면 실행되지 않는다.) 또한 await 는 async def 안에서만 사용해야 한다.

 

 

asyncio.run(main())

asyncio.run() 은 호출한 task 들이 모두 종료될 때까지 기다리면서 각 task들이 만드는 이벤트들을 처리하는 역할을 한다.

asyncio.run(main()) 는 아래 코드와 같은 일을 한다.

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

 

asyncio.create_task()

create_task() 는 coroutine의 실행을 준비한다.

asyncio.create_task(count()) 를 호출하면 event loop 에서 count()가 실행되도록 예약된다. 

하지만 create_task() 만 호출하도록 하고 asyncio.run(main())으로 시작하면 loop.run_until_complete()는 count() coroutine이 종료될 때까지 기다리지 않기 때문에 "two" 까지 출력되지 않는다. 

coroutine이 정상적으로 완료되기 위해서는 create_task() 에서 반환되는 Task 객체를 await 해야 한다.

 

import asyncio
import time
import threading


async def count():
    print('One')
    await asyncio.sleep(1)
    print('Two')


async def main():
    f = asyncio.create_task(count())	# count()가 시작은 되지만 sleep() 을 기다려주지 않는다.
    # await f

if __name__ == '__main__':
    s = time.perf_counter()

    asyncio.run(main())

    elapsed = time.perf_counter() - s
    print(f'executed in {elapsed:0.2f} seconds.')

One
[8399526720] executed in 0.00 seconds.

 

async with

with 구문과 유사하게 async with를 사용하면 코드블럭에 진입할 때와 벗어날 때 async 로 선언된 __aenter__, __aexit__ 를 호출한다.

__aenter__, __aexit__ 는 async def 로 선언된 coroutine 이기 때문에, asyncio.sleep() 을 만나면 다른 coroutine으로 제어권을 넘겨주게 된다.

 

 

 

 

 

참고 자료:

https://realpython.com/async-io-python/

Async IO in Python: A Complete Walkthrough – Real Python

https://peps.python.org/pep-0492/

PEP 492 – Coroutines with async and await syntax | peps.python.org

https://python-notes.curiousefficiency.org/en/latest/pep_ideas/async_programming.html

Some Thoughts on Asynchronous Programming — Nick Coghlan's Python Notes 1.0 documentation

https://docs.python.org/3/library/asyncio-task.html