In this post, concepts of Coroutine will be provided.

Subroutine(=Routine)

• 프로그램의 호출 가능한 단위 (≈ 메서드, 함수, 프로시저)
• 서브루틴을 호출하면, 해당 서브루틴이 끝나기까지 호출자(Caller)는 대기한다.
• 서브루틴은 고유한 스택 프레임을 갖는다.

Coroutine

• 코루틴은 중단/재개 가능한 서브루틴 (Non-preemptive Scheduling)
  • 여러개의 프로세스를 실행할 때 운영체제가 알아서 스케쥴링 해줌. 운영체제가 해주는 스케줄링은 어느 코드에서 멈추고 어느 코드에서 시작되는지 우리가 모름. 운영체제가 알아서 프로세스의 권한을 조작하는 스케줄링을 선점형 스케쥴링이라 함.
  • 반면 코루틴은 우리가 멈추고 우리가 재개함. 비선점형 스케쥴링이라고도 부름.
• 경량 스레드라고도 불린다.
• 동작이나 구현에 따라 여러 가지로 구분할 수 있다.
  • 중단 시 재개할 코루틴을 선택할 수 있으면 Symmetric Coroutine
  • 중단 시 Caller가 재개되면 Asymmetric Coroutine
  • 코루틴이 별도의 스택에서 실행되면 Stackful Coroutine
  • 코루틴이 Caller 스택 위에서 실행되면 Stackless Coroutine
• 파이썬은 Asymmetric, Stackless Coroutine

Iterator

이터레이터도 코루틴이다. next()와 next() 사이에 이터레이터의 코드가 멈췄다가 실행되고 멈췄다고 실행됨.

class MyIterator:
    def __init__(self):
        self.values = [1,2]
        self.index = 0

    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index >= len(self.values):
            raise StopIteration()
        value = self.values[self.index]
        self.index += 1
        return value

it = MyIterator()
a = next(it) # a = 1
b = next(it) # b = 2
c = next(it) # StopIteration 발생!

Generator

제너레이터도 코루틴이다.

def gen():
    # some task
    x = yield 1
    # some task
    yield x

it = gen()
a = next(it) # a = 1
it.send(10)
b = next(it) # b = 10
c = next(it) # StopIteration 발생!

• asyncio 코루틴의 원시적인 형태
• 비대칭 코루틴이며, 중단 및 재개 시마다 서로 데이터를 주고 받을 수 있다.

Asyncio

• Asynchronous I/O를 기반으로 작성된 코루틴 라이브러리
• 내부적으로 Event Loop와 Generator를 기반으로 동작함
  • 왼쪽에 여러개의 task를 등록해두고 Event Loop가 각각의 task를 실행하다가 중단이 되면 다음 task를 실행시킴.
  • 중단이 발생하는 대부분의 경우는 I/O가 발생하는 것.
• 순서와 관계 없이, 한 태스크가 I/O를 기다리는 동안 준비된 다른 태스크를 실행

import asyncio

async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

asyncio.run(main())

• 함수 선언 시에 async 키워드를 붙인다.
• 코루틴의 반환값을 얻으려면 await 키워드를 붙인다. return 값에도 await를 붙인다.
• await 키워드가 사용되는 시점이 중단이 발생하는 시점
• await 키워드는 코루틴 내부에서만 사용할 수 있다.
• 일반 함수에서 코루틴 함수를 호출하려면, asyncio.run 등을 사용한다.
  • FastAPI 자체가 이미 asyncio 함수 루프 위에서 돌아가고 있기 때문에 쓸 일은 많이 없음.

FastAPI and Asyncio

• FastAPI는 기본적으로 asyncio의 Event Loop위에서 실행된다.
• 우리가 선언한 엔드포인트들이 async 함수냐 일반 함수냐에 따라 FastAPI 동작이 다른다.
  • async 함수인 경우 -> await 으로 호출
  • 일반 함수인 경우 -> threading 사용
• 만약 async 키워드를 붙였는데 blocking io를 호출하면 서버 전체가 blocking이 됨. 그래서 우리가 어떤 엔드포인트 안에서 블로킹 io가 호출이 될 것 같다면 애초에 일반 함수로 선언하는 것이 좋음.

@app.get("/normal-endpoint")
def normal_endpoint():
    time.sleep(5)

@app.get("/async-endpoint")
async def async_endpoint():
    await asyncio.sleep(5)

@app.get("/dump-endpoint")
async def async_endpoint():
    time.sleep(5)

• async 없이 쓰면 time.sleep 이라는 블로킹 io가 발생해도 다른 스레드에서 실행되고 있기 때문에 서버가 블로킹 되지 않음.
• 만약 async 함수에서 blocking이 발생한다면? -> 해당 워커가 block됨
  • asyncio 내부에서 sleep이라는 명령어가 있음. 이는 비동기 sleep이기 때문에 문제가 없음.
  • 세 번째 처럼 적으면 안됨. 서버 전체가 블로킹 됨.
• 따라서 blocking이 발생할 수 있는 경우, 별도 스레드에서 실행해야 함
  • 엔드 포인트 자체를 일반 함수로 선언하거나 run_in_executor 을 이용
  • asyncio도 호출하고 블로킹 io도 호출하는 경우 엔드포인드 자체는 async로 선언하되, 블로킹 io를 호출하는 부분만 run_in_executor 이용.