백프레셔

파일을 읽는 생산자가 초당 100개를 내보내고 다음 단계의 소비자가 초당 60개를 처리한다고 해보자. 차이가 계속되면 초마다 40개가 쌓인다. 버퍼를 키워 잠깐 더 받아둘 수는 있어도 소비 속도가 그대로라면 그 버퍼도 결국 찬다.

백프레셔(backpressure)를 적용하면 받는 쪽의 여유에 맞춰 보내는 쪽이 생산이나 전송을 조절한다. 데이터는 생산자에서 소비자로 흐르고 더 받을 수 있는지 알리는 신호는 상류로 거슬러 올라간다. TCP의 수신 윈도, 비동기 큐의 대기, 스트림의 요청량은 이 문제를 서로 다른 단위와 실행 방식으로 다룬다.

수신자가 받을 수 있는 양을 알려준다

1981년 TCP 명세인 RFC 793에서 수신 측은 지금 받을 수 있는 바이트 수를 윈도(window)로 알려준다. 송신 측은 보낼 데이터가 더 있어도 이 범위를 고려해 전송한다. 비동기 스트림 라이브러리가 나오기 전부터 수신자의 여유를 송신자에게 알려주는 프로토콜을 사용했던 것이다.¹

수신 측 TCP가 데이터를 받아줬다고 해서 애플리케이션이 그 내용의 주문 처리나 파일 변환을 마친 것은 아니다. 애플리케이션이 소켓에서 계속 읽어 무제한 큐에 넣는다면 TCP 수신 버퍼를 비워주는 동안에도 내부 작업은 쌓인다. 그래서 소켓 바깥의 생산자와 소비자까지 어느 단계에서 더 기다리게 할지 연결해야 한다.

통신 분야에서는 흐름 제어와 혼잡 제어도 구분한다. 수신자는 자신의 수용 능력을 알리고 송신 측의 혼잡 제어는 네트워크 경로가 감당할 전송량을 다룬다. 현재 TCP 기본 명세인 RFC 9293도 송신 윈도와 혼잡 윈도를 구별한다.²

비동기로 나누면 어디에서 기다릴까

한 스레드에서 함수를 차례로 호출하면 앞선 호출이 끝날 때까지 다음 호출을 기다린다. 생산자가 작업을 큐나 별도 Task로 넘기면 소비자의 완료를 기다리지 않고 다음 일을 만들 수 있다. 그 사이의 데이터를 어디에 얼마나 보관할지도 정해야 한다.

생산자가 큐의 각 put 완료를 기다리면 큐가 찼을 때 생산자도 다음 항목을 넣지 못하고 멈춰 선다. 반대로 put을 수행할 Task를 계속 새로 만들면 큐는 작게 유지하더라도 그 바깥에 대기하는 Task가 늘어날 수 있다.

위쪽 화살표는 데이터가 흐르는 방향이고 아래쪽은 대기나 수요 신호가 상류로 전달되는 방향이다. 큐에서 대기하든 요청량을 알리든 그 신호가 생산자의 다음 행동까지 이어진다.

단계별 큐를 제어 지점으로 삼은 서버

Welsh·Culler·Brewer는 2001년 SEDA 논문에서 서버를 여러 처리 단계로 나누고 단계 사이의 큐를 명시적으로 드러냈다. 각 단계의 적체를 보고 스레드 수나 처리 묶음을 조절할 수 있도록 만든 것이다.³

(원문 출처)

이벤트는 입력 큐를 거쳐 처리기로 들어가고 처리기는 다른 단계로 이벤트를 보낸다. 제어기는 이 단계의 자원 사용과 스케줄링을 조절한다.

저자들은 §3.3에서 큐가 새 항목을 거절할 수 있게 하고 애플리케이션이 가득 찬 큐 앞에서 기다리거나 이벤트를 버리는 정책을 고를 수 있다고 설명한다. 큐를 둔 것만으로 상한과 정책이 저절로 정해지지는 않으며 애플리케이션이 그 제어 지점을 이용해 생산을 늦추거나 접수를 줄인다.

이렇게 처리 단계를 나누면 각 단계를 따로 조정하기 쉬워지는 대신 한 단계가 늦어졌을 때 그 제약을 앞단까지 어떻게 전달할지도 정해야 한다. 소켓의 흐름 제어에서 다루던 문제를 서버 내부에서도 만나는 것이다.

라이브러리 사이에서 주고받는 수요

Reactive Streams 참여자들은 비동기 스트림 라이브러리가 수요와 취소를 공통된 규칙으로 주고받도록 인터페이스를 정했다. 1.0.0 태그가 가리키는 커밋은 2015년 4월의 기록으로, 공식 발표에서 비차단 백프레셔를 갖춘 비동기 스트림 처리를 표준화하겠다는 목표를 밝힌다.⁴

(원문 출처)

공식 소개 첫 문장은 참여자들이 내세운 목표를 요약하며 개별 신호의 세부 계약은 명세에서 정의한다.⁵

Publisher가 원소를 보내고 Subscriber가 받는다. Subscriber는 Subscription의 request(n)을 호출해 추가로 받을 수 있는 개수를 알린다. Publisher 규칙 1.1은 누적 전달량이 누적 요청량을 넘지 못하게 하고 Subscription 규칙 3.8은 새 요청량을 남은 수요에 더하도록 정한다.⁶

Subscriber가 request(2)를 호출하고 원소 하나를 받았다면 남은 수요는 하나다. 이어서 request(3)을 호출하면 넷이 된다. 여기서 세는 값은 전달을 허용한 원소 수다. 초당 네 개 같은 시간당 속도는 별도로 정하며 생산자는 데이터가 없거나 스트림을 끝낼 때 요청량보다 적게 보낼 수 있다.

남은 수요가 0인 동안에는 추가 데이터 원소를 보내지 않는다. 새 요청에는 양수를 쓰며 request(0)은 일시 정지 명령이 아니라 잘못된 요청이다. 완료와 오류는 데이터 원소와 다른 신호이므로 수요가 없어도 전달할 수 있다.⁷

Java 9의 java.util.concurrent.Flow도 Publisher·Subscriber·Subscription·Processor를 제공한다. 공식 문서는 이 인터페이스들이 Reactive Streams에 대응한다고 설명한다. 라이브러리 작성자는 데이터를 보내는 메서드뿐 아니라 수요·취소·종료의 계약까지 맞춰야 서로 연결할 수 있다.⁸

기다림을 표현하는 방법

블로킹 구현에서는 스레드가 대기할 수 있고 비동기 구현에서는 코루틴을 보류한 동안 다른 작업을 실행할 수 있다. Reactive Streams는 비동기 구성 요소를 연결하면서 제어 신호 때문에 스레드를 계속 붙잡지 않도록 비차단을 요구한다.⁹

수요를 셀 때는 항목을 받은 시점과 업무를 끝낸 시점도 구별한다. Subscriber가 원소를 받을 때마다 새 DB 작업을 만들고 곧바로 수요를 보충하면 원소 전달 규칙을 지키더라도 실행 중인 DB 작업은 계속 늘어날 수 있다. 자원 사용을 제한하려는 단계는 어떤 일을 마쳤을 때 다음 수요를 요청할지까지 정해야 한다.

큐 두 칸으로 보는 작은 예

다음은 Python 3.11 이상에서 실행한다. 생산자는 maxsize=2인 큐에 값을 하나씩 넣고 각 put의 완료를 기다린다. 큐가 차면 await queue.put()이 생산자 코루틴을 보류하고 이벤트 루프는 소비자 같은 다른 작업을 실행한다.¹⁰

import asyncio

async def produce(queue):
    for item in range(6):
        await queue.put(item)
    await queue.put(None)

async def consume(queue):
    values = []
    while True:
        item = await queue.get()
        if item is None:
            return values
        await asyncio.sleep(0.01)
        values.append(item)

async def main():
    queue = asyncio.Queue(maxsize=2)
    async with asyncio.TaskGroup() as group:
        group.create_task(produce(queue))
        consumer = group.create_task(consume(queue))
    assert consumer.result() == list(range(6))
    print(consumer.result())  # [0, 1, 2, 3, 4, 5]

if __name__ == "__main__":
    asyncio.run(main())

입력은 0부터 5까지 여섯 값이고 None은 종료 표지다. 단일 소비자는 앞선 값을 순서대로 처리한 뒤 표지를 받고 끝난다. 부모는 TaskGroup을 나갈 때 두 작업의 종료를 기다리므로 이 흐름에서는 완료 항목 수를 따로 세는 join과 task_done을 쓰지 않았다.

소비자가 일반 예외로 실패하면 TaskGroup은 생산자도 취소한다. 두 작업을 함께 중단할 때는 부모 작업을 취소하는 등 수명을 함께 끝내는 경로를 쓴다. 개별 소비자의 CancelledError는 다른 자식을 취소하는 일반 실패와 다르게 처리되기 때문이다.¹¹

이 예제는 한 프로세스에서 큐 대기를 전달한다. Reactive Streams는 별도의 인터페이스와 신호 규칙으로 수요를 전달한다. 같은 문제를 다루되 각각의 API와 종료 계약을 따라 읽으면 된다.

큐가 찼을 때 바로 거절하는 동작도 비교해보자. 다음 Python 3 예제는 put_nowait을 쓰므로 두 번째 항목을 기다려 넣는 대신 QueueFull을 받아 접수하지 않았다고 알린다. 마지막에 꺼내는 값은 먼저 넣은 first다.¹²

import asyncio

queue = asyncio.Queue(maxsize=1)
queue.put_nowait("first")
try:
    queue.put_nowait("second")
except asyncio.QueueFull:
    print("second: not accepted")

print(queue.get_nowait())  # first

같은 한 칸짜리 큐에 await put으로 넣으면 다음처럼 기다린다. Python 3.11 이상에서 실행하며, put_started로 두 번째 항목을 넣기 시작한 시점을 맞춘다. 첫 항목이 큐에 남아 있는 동안에는 두 번째 put이 완료되지 않는다.

import asyncio

async def main():
    queue = asyncio.Queue(maxsize=1)
    await queue.put("first")
    put_started = asyncio.Event()

    async def put_second():
        put_started.set()
        await queue.put("second")

    async with asyncio.TaskGroup() as group:
        pending = group.create_task(put_second())
        await put_started.wait()
        print(pending.done())  # False
        first = await queue.get()
        await pending
        print(first, await queue.get())  # first second

if __name__ == "__main__":
    asyncio.run(main())

첫 get이 항목을 꺼내면 생산자가 두 번째 항목을 넣을 공간이 생긴다. 여기서는 첫 항목을 꺼내 변수에 담기만 했고 별도의 업무 처리를 시작하지도 않았다. 큐의 빈칸을 기다리는 것과 소비자의 처리 완료를 기다리는 것이 어느 지점에서 갈리는지 볼 수 있다.

버퍼가 차면 무엇을 할 것인가

큐의 두 칸은 대기 중인 항목 수를 제한한다. 소비자가 이미 꺼내 처리하는 항목과 생산자가 다음에 넣으려고 들고 있는 항목은 그 바깥에 있다. 위 소비자의 values도 결과를 모으는 별도 목록으로 여기서는 여섯 값만 저장한다. 항목 수의 상한과 프로세스 전체 메모리의 바이트 상한은 이렇게 다른 범위를 가리킨다.

센서 측정이나 마우스 이동처럼 원래의 발생원을 늦출 수 없는 입력도 있다. 이때는 스트림 경계에서 잠시 보관하거나 값을 합치거나 일부를 버리는 정책을 선택한다. Reactive Streams 명세도 원래의 생산 속도를 바꿀 수 없는 소스에서는 수요 상한을 지키기 위해 버퍼링이나 드롭이 필요할 수 있다고 설명한다.¹³

처리 능력을 넘는 입력이 계속 들어오면 유한한 저장소는 언젠가 찬다. 더 오래 보관할 장치를 붙이는 일과 지속적으로 감당할 처리량을 늘리는 일은 구별해야 하며 모든 입력을 보존해야 한다면 그만큼 생산을 늦추거나 처리 능력을 확보해야 한다.

TCP에서는 수신자가 윈도를 알리고 SEDA의 애플리케이션은 단계별 큐를 제어하며 스트림의 Subscriber는 받을 원소 수를 요청한다. 구현을 읽을 때 데이터가 가는 방향과 제약이 돌아오는 방향을 함께 따라가면 어느 생산자가 무엇을 기다리는지 확인할 수 있다.