Home [Python] asyncio 파헤치기 - ③ 동시성이 달성되지 않는 상황
Post
Cancel

[Python] asyncio 파헤치기 - ③ 동시성이 달성되지 않는 상황

Posted Mar 4, 2024
By Seungri
12 min read

본문은 “Python Concurrency with Asyncio”를 읽고 재구성한 글임을 밝힙니다.


이번 포스팅에서는 실제로 asyncio를 사용하는 상황을 연출하면서 주의해야 할 점에 대해 알아보도록 하겠다.


1. 실습 준비

실습을 진행하기 전, 코루틴의 실행 시간을 측정하기 위한 데코레이터를 작성하자.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# util/async_timer.py

import functools
import time
from typing import Callable, Any

def async_timed():
    def wrapper(func: Callable) -> Callable:
        @functools.wraps(func)
        async def wrapped(*args, **kwargs) -> Any:
            print(f"starting {func} with args {args} {kwargs}")
            start = time.time()
            try:
                return await func(*args, **kwargs)
            finally:
                end = time.time()
                total = end - start
                print(f"finished {func} in {total:.4f} second(s)")
        
        return wrapped
    
    return wrapper

1
2
3
4

# util/__init__.py

...
from util.async_timer import async_timed

데코레이터(decorator)는 어떤 함수의 코드를 수정하지 않으면서 기능을 추가할 수 있도록 돕는 패턴이다. 데코레이터를 통해 어떤 함수를 intercept 하여, 실행 전후에 추가 동작을 수행할 수 있다.

더 자세한 내용은 [Python] 데코레이터(Decorator) 포스팅을 참고한다.


이렇게 생성한 async_timed 데코레이터의 동작을 확인하기 위해 다음과 같이 동시성이 달성된 코드에 적용해보겠다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import asyncio
from util import async_timed

@async_timed()
async def delay(delay_seconds: int) -> int:
    print(f"sleeping for {delay_seconds} second(s)")
    await asyncio.sleep(delay_seconds)
    print(f"finished sleeping for {delay_seconds} second(s)")
    
    return delay_seconds

@async_timed()
async def main():
    task_one = asyncio.create_task(delay(2))
    task_two = asyncio.create_task(delay(3))
    
    await task_one
    await task_two
    
asyncio.run(main())

결과를 살펴보면 두 개의 delay 작업이 각각 주어진 시간(각각 2초, 3초)만큼 잘 수행되었으나, 이들이 순차적이 아닌 동시적으로 수행되었기에 main 작업이 5초(= 2+3)가 아닌 3초만에 완료되었음을 확인할 수 있다.

1
2
3
4
5
6
7
8
9
10

starting <function main at 0x103944700> with args () {}
starting <function delay at 0x102a03370> with args (2,) {}
sleeping for 2 second(s)
starting <function delay at 0x102a03370> with args (3,) {}
sleeping for 3 second(s)
finished sleeping for 2 second(s)
finished <function delay at 0x102a03370> in 2.0022 second(s)
finished sleeping for 3 second(s)
finished <function delay at 0x102a03370> in 3.0015 second(s)
finished <function main at 0x103944700> in 3.0033 second(s)

이렇게 데코레이터를 통해 측정된 시간을 기반으로, 예시 코드를 실행해보면서 동시성이 달성되었는지 여부를 판단해보자.


2. asyncio를 사용하더라도 동시성을 달성할 수 없는 경우: Blocking 상황

이러한 성능 상의 이점을 맛보고 나면 어플리케이션의 모든 곳에 asyncio를 사용하고 싶어질 수 있다. 하지만 다음의 두 가지 경우에는 asyncio를 적용하더라도 성능 향상을 달성할 수 없다.

  1. 태스크 혹은 코루틴 내의 CPU-bound 코드를 멀티 프로세싱 없이 사용하는 경우
  2. blocking I/O-bound API를 멀티 쓰레딩 없이 사용하는 경우


그리고 이를 한 마디로 쉽게 정리하면 “blocking 연산이 멀티 프로세싱 및 멀티 쓰레딩 없이 수행될 때는 asyncio를 적용하더라도 동시성을 달성할 수 없다”고 할 수 있다.

일반적으로, (1) 긴 시간이 소요되는 CPU-bound 연산과 (2) 코루틴이 아닌 I/O-bound 연산을 blocking 연산이라 표현한다.

이러한 blocking 연산은 이벤트 루프 쓰레드를 block 하며, 다른 코루틴이 실행될 수 없도록 하기 때문이다.


이 두 가지 경우에 대해 예시와 함께 살펴보자.


2-1. CPU-Bound 코드 (w/o 멀티 프로세싱)

다음과 같이 CPU-bound 코드 두 개를 asyncio를 통해 동시에 실행하려 해보자.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import asyncio
from util import async_timed

@async_timed()
async def cpu_bound_work() -> int:
    counter = 0
    for i in range(100_000_000):
        counter = counter + 1
    return counter

@async_timed()
async def main():
    task_one = asyncio.create_task(cpu_bound_work())
    task_two = asyncio.create_task(cpu_bound_work())
    await task_one
    await task_two

asyncio.run(main())

하지만 태스크를 생성 및 예약했음에도 불구하고, 다음과 같이 동시적이 아닌 순차적으로 수행되는 것을 확인할 수 있다. 따라서 main 작업이 두 번의 cpu_bound_work 작업 시간의 합으로 나타난다.

1
2
3
4
5
6

starting <function main at 0x1035a0700> with args () {}
starting <function cpu_bound_work at 0x102f8b370> with args () {}
finished <function cpu_bound_work at 0x102f8b370> in 2.7899 second(s)
starting <function cpu_bound_work at 0x102f8b370> with args () {}
finished <function cpu_bound_work at 0x102f8b370> in 2.7878 second(s)
finished <function main at 0x1035a0700> in 5.5781 second(s)

asyncio는 싱글 쓰레드 동시성 모델을 가지고 있기 때문에, 여전히 싱글 쓰레드와 GIL의 제약을 받기 때문이다.


이러한 CPU-bound 코드를 동시에 처리하려면 멀티 프로세싱을 사용해야 하며, async / await 문법을 사용하고 싶다면 asyncio에게 프로세스 풀(process pool)에서 태스크를 실행할 것을 명시해야 한다.

관련된 자세한 내용은 본 포스팅에서 다루지 않겠다.


2-2. Blocking I/O-Bound API (w/o 멀티 쓰레딩)

기존 라이브러리의 I/O-bound 연산을 코루틴으로 수행하는 것 또한 CPU-bound 연산과 동일한 양상을 보인다. 왜냐하면 이러한 I/O-bound API는 메인 쓰레드를 block 하기 때문이다.

코루틴 내에서 blocking API를 호출하면 이벤트 루프 쓰레드 자체가 block 되기 때문에 다른 코루틴이나 태스크도 수행될 수 없다.

이러한 blocking API에는 requests 라이브러리에서 제공하는 연산이나 time.sleep과 같은 것들이 있다.


requests 라이브러리를 사용한 예시 코드는 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import asyncio
import requests
from util import async_timed

@async_timed()
async def get_example_status() -> int:
    return requests.get("http://www.example.com").status_code

@async_timed()
async def main():
    task_1 = asyncio.create_task(get_example_status())
    task_2 = asyncio.create_task(get_example_status())
    task_3 = asyncio.create_task(get_example_status())
    
    await task_1
    await task_2
    await task_3

asyncio.run(main())

결과를 살펴보면 동시성이 달성되지 못하고 순차적으로 수행되었음을 확인할 수 있다.

1
2
3
4
5
6
7
8

starting <function main at 0x10605a9e0> with args () {}
starting <function get_example_status at 0x105e09bd0> with args () {}
finished <function get_example_status at 0x105e09bd0> in 0.3163 second(s)
starting <function get_example_status at 0x105e09bd0> with args () {}
finished <function get_example_status at 0x105e09bd0> in 0.3171 second(s)
starting <function get_example_status at 0x105e09bd0> with args () {}
finished <function get_example_status at 0x105e09bd0> in 0.2895 second(s)
finished <function main at 0x10605a9e0> in 0.9233 second(s)


이처럼 대부분의 API는 blocking 방식으로 동작하기 때문에 asyncio와 사용하더라도 동작하지 않는다. 따라서 (1) 코루틴을 지원하며 (2) non-blocking 소켓을 사용하는 라이브러리를 찾아 사용해야 한다. 그렇지 않으면 여전히 blocking 방식으로 동작하여 동시에(concurrently) 실행되지 않고 순차적(sequentially)으로 실행된다.

이러한 라이브러리의 예시로는 aiohttp 같은 것들이 있다. 이는 non-blocking 소켓을 사용하고 코루틴을 반환하여 동시성을 달성할 수 있도록 지원한다.

sqlalchemy를 asyncio와 함께 사용할 때, aiopg, aiomysql와 같이 비동기를 지원하는 드라이버를 사용해야 했던 이유가 바로 이것이다!


만약 requests 라이브러리를 사용해야 하고 async 문법 또한 사용하고 싶은 경우라면 멀티 쓰레딩을 사용해야 하며, asyncio에게 thread pool executor에서 실행할 것을 명시해야 한다.

관련된 자세한 내용은 본 포스팅에서 다루지 않겠다.


3. 디버그 모드

이전 포스팅에서 다루었듯, 코루틴은 항상 어플리케이션의 어딘가에서 await와 함께 호출되어야 한다. 하지만 개발 과정에서 실수로 await를 빼먹는 경우가 발생할 수 있는데, 이를 방지하기 위해 디버그(debug) 모드를 사용할 수 있다. 디버그 모드에서 지원되는 기능은 다음과 같다.

  1. 100ms 이상 소요되는 코루틴 혹은 태스크가 수행될 때 로그 메시지를 남겨준다.
  2. 코루틴을 await 하지 않으면 exception을 발생시킨다.


이러한 디버그 모드를 사용하는 방법에는 세 가지가 있다.

  1. asyncio.run에 debug 파라미터 True로 넘기기

    1

     asyncio.run(coroutine(), debug=True)

  2. command-line argument -X dev 지정하기

    1

     python3 -X dev program.py

  3. 환경 변수 PYTHONASYNCIODEBUG 사용하기

    1

     PYTHONASYINCIODEBUG=1 python3 program.py


지금까지 asyncio에 대한 아주 기초적인 내용들을 다뤄보았다. 어려우면서도 방대한 내용이다보니 앞으로도 asyncio에 대해 더 알아봐야 할 것 같다.


References

Python, Concurrency in Python
python asyncio nonconcurrent blocking cpu bound io bound
This post is licensed under CC BY 4.0 by the author.
Recently Updated
Contents

[Python] asyncio 파헤치기 - ② 코루틴, 태스크, 퓨처, Awaitable

[Python] list와 dict의 내부 구현과 시간 복잡도 (+ 동적 배열, 해시 충돌)