[Python] Asyncio_02

Writing Code

async def는 비동기 코루틴 함수에 있어서 일반적인 동기 함수를 정의하는데 사용되는 def와 같은 방식으로 선언된다.

async def example_coroutine_function(a, b):
    # Asynchronous code goes here
    ...

def example_function(a, b):
    # Synchronous code goes here
    ...

비동기 함수에 사용되는 async def는 일반 함수에 사용되는 def와 별다를 것 없어 보이지만 몇 가지 주요한 차이점이 있다.

async def example_coroutine_function(a, b):
	...
    
r = example_coroutine_function(1, 2) 

example_coroutine_function는 2개의 파라미터를 사용하는 호출 가능한 오브젝트이다. 호출하게 되면 def로 지정했을 때 서브루틴 호출로써 즉시 실행되고 반환값이 r에 할당되는 것과 달리 실행되지 않고 코루틴 객체가 생성되어 할당된다.

코드 블록을 실제로 실행하려면 coroutine을 실행하기 위해 asyncio가 제공하는 기능 중 하나를 사용해야 한다.

await

asyncio를 지원하기 위해 언어에 추가된 새 키워드 await. 비동기 코드블록 내에서만 사용할 수 있으며 단일 매개변수를 취하고 값을 리턴한다.

r = await a

이것은 객체 a에 대해 await을 수행하고 반환 값을 r에 할당할 유효한 파이썬 문(statement)이다.

코루틴 객체는 awaitable이다(이것은 await 문에서 사용할 수 있다). 비동기 코드를 실행할 때 이벤트루프에 의해 유지되는 객체인 Task의 context에서 실행되고 각 Task는 고유의 call stack을 가지고 있다. Coroutine 객체가 처음 대기하면 해당 코드 블록이 현재 Task에서 실행되며 일반적인 함수 호출과 마찬가지로 이 Task에 대한 호출 스택의 맨 위에 새 코드 context가 추가됩니다. 코드 블록이 마지막에 도달하면(또는 return), 실행은 그것을 호출한 await 문으로 돌아간다. await 문의 반환 값은 코드 블록에서 반환하는 값이다.

코루틴 객체가 두 번째로 기다리면 예외가 발생한다. 이런 식으로 코루틴 객체를 기다리는 것은 함수를 호출하는 것과 매우 유사하다고 생각할 수 있다. 그러나 코루틴 객체의 코드 블록에는 비동기 코드를 포함할 수 있기 때문에 실행 중에 현재 작업을 일시 중지할 수 있다.

세가지 유형의 awaitable한 객체

1. 코루틴 객체 - 기다릴 때 현재 Task에서 코루틴의 코드 블록을 실행. await문은 코드 블록에서 반환된 값을 반환.
2. syncio.Future의 객체 - 특정 조건이 발생할 때까지 현재 task를 일시 정지.
3. 매직 메소드 __awiat__을 구현하는 객체 - 대기 시 무슨 일이 일어나는지는 그 메소드에 의해 정의.

3번째 객체는 라이브러리의 작성자들이 awaitable 객체의 독자적인 새로운 클래스를 작성해, 대기 시에 무언가 특별한 일을 할 수 있도록 하기 위해서이다. 통상, 커스텀의 awaitable 객체를 Coroutine 오브젝트 또는 Future 오브젝트와 같이 동작 하도록 하고 문서화(docstring)하는 것이 좋다.

중요한 포인트는 현재 실행 중인 Task를 future(또는 future처럼 동작하는 사용자 지정 대기 가능 객체)를 기다리는 것 외에는 어떤 방법으로도 일시 중지할 수 없다. 이것은 비동기 코드 안에서만 일어나는 일이며 await 문은 현재 작업을 일시 중지할 수 있지만 보증할 수 없다. 반대로 await 문(또는 특정 상황에서의 async for, async with)이 아닌 statement는 현재 작업을 일시 중지할 수 없다.

이것은 서로 다른 실행 스레드가 모두 같은 값을 변경하는 데이터 경쟁의 기존 다중 스레드 코드 문제가 비동기 코드에서 심각하게 감소하지만 완전히 제거되지는 않는다는 것을 의미한다. 특히 같은 이벤트 루프의 작업 간에 데이터를 공유하기 위한 모든 동기화 코드를 "atomic"로 간주할 수 있다.

import asyncio

async def get_some_values_from_io():
    # Some IO code which returns a list of values
    ...

vals = []

async def fetcher():
    while True:
        io_vals = await get_some_values_from_io()

        for val in io_vals:
            vals.append(io_vals)

async def monitor():
    while True:
        print (len(vals))

        await asyncio.sleep(1)

async def main():
    t1 = asyncio.create_task(fetcher())
    t2 = asyncio.create_task(monitor())
    await asyncio.gather(t1, t2)

asyncio.run(main())

fetcher와 monitor는 전역 변수 vals에 액세스하지만 동일한 이벤트 루프에서 실행되는 두 가지 작업으로 액세스한다. 따라서 페처가 현재 잠자기 상태로 io를 기다리고 있지 않는 한, monitor의 print문을 실행할 수 없고 이것은 for loop가 중간까지만 실행되고 있는 동안에는 vals의 길이를 출력할 수 없다는 것을 의미한다. 따라서, get_some_values_from_io 가 항상 한 번에 10개의 값을 돌려주는 경우 표시되는 vals 의 길이는 항상 10의 배수여야 한다. vals의 길이가 10의 배수가 아닌 경우 print문을 실행하는 것은 불가능하다.