이 문서는 2주차 part 1 - 모델 서빙 시스템 설계의 /generate 경로를 코드 순서대로 따라간 추적 기록입니다. 스트리밍 경로는 request_id의 여정에서 다룹니다.

/generate_stream이 비켜주며 기다리는 경로라면, /generate는 붙잡고 기다리는 경로입니다. 워커 스레드도, 코루틴도, 큐 브릿지도 쓰지 않습니다. 요청을 받은 그 자리에서 다 될 때까지 돌립니다.

세 갈래: 같은 워커, 다른 길

엔드포인트 세 개가 같은 모델·같은 워커 프로세스를 쓰지만 거기 도달하는 경로가 전부 다릅니다.

가장 큰 갈림은 실행 주체 행입니다. 앞의 둘은 요청을 받은 그 자리에서 끝까지 돌리고, 스트리밍만 일을 데몬 스레드에 넘기고 자기는 잠듭니다.

시퀀스: 프롬프트 4개를 보내고 응답 하나를 받기까지

스트리밍 쪽과 나란히 놓고 보면 차이가 분명합니다. 여기엔 asyncio.Queue도, run_coroutine_threadsafe도, 깨어나는 코루틴도 없습니다. 루프 하나가 전부입니다.

워커와의 왕복은 배치가 나뉜 횟수만큼만 일어납니다. 프롬프트 4개면 대개 한 번, 5개면 두 번입니다. 스트리밍의 21회 왕복과 대비됩니다.

배치는 요청 단위로 끊기지 않는다

llm.py에 이런 주석이 달려 있습니다.

Execute the next batch in one go, it may not be the same prompts as the prompts in the request.

배치를 채우는 건 WorkloadManager의 대기열이지 내 요청이 아닙니다.

내 요청은 5개지만 배치는 4칸이라 두 번에 나뉩니다. 두 번째 배치의 남은 칸은 대기열에 있던 누구의 프롬프트든 채울 수 있습니다.

그래서 결과를 request_id로 되찾아야 합니다. 배치에 뭐가 같이 실렸는지는 호출한 쪽이 알 수 없습니다.

_is_batch_finished(request_ids)   # 내 id 들이 전부 finished 인지만 확인한다

배치 경계와 요청 경계는 별개입니다. 루프가 도는 조건은 "배치가 끝났는가"가 아니라 "내 request_id들이 전부 끝났는가"이고, 그 판정은 sequence_map을 통해 이뤄집니다.

두 경로가 큐 한 쌍을 나눠 쓴다

ModelExecutor는 하나뿐이고, task_queue와 result_queue도 한 쌍뿐입니다. /generate를 처리하는 요청 스레드와 requests_processing_loop 데몬 스레드가 같은 큐에 넣고 같은 큐에서 꺼냅니다.

큐는 누가 넣었는지 기억하지 않습니다. 먼저 get()을 부른 스레드가 맨 앞의 항목을 가져갑니다. 튜플 앞의 태그가 붙어 있는 이유입니다.

두 경로의 확인 강도는 다릅니다.

# execute_forward_batch 는 태그를 검사한다
result_type, results = self._wait_for_result()
if result_type == 'stream':
    return results
else:
    raise UnexpectedResultTypeError(...)

# execute_batch 는 튜플째 돌려주고, 호출부가 인덱싱한다
results = self._wait_for_result()
return results
...
return results[1][0]['generated_text']

두 경로를 동시에 쓰지 않으면 이 구조는 드러나지 않습니다. 스트리밍만 쓰거나 일반 요청만 쓰면 큐에는 한 종류만 오갑니다. 저장소의 repro_lockstep.py가 겨냥한 지점이 여기입니다.

루프 본체

세 줄로 압축하면 이렇습니다.

# llm.py 의 generate()
request_ids = [add_request(p) for p in prompts]        # ① 대기열에 담고 id 를 챙긴다

while not _is_batch_finished(request_ids):             # ② 내 id 들이 다 끝날 때까지
    sequences = get_next_batch()                       #    대기열에서 최대 4개
    if not sequences:
        time.sleep(0.1); continue                      #    없으면 잠깐 쉬고 다시
    results = execute_batch(sequences)                 # ③ 워커 왕복. 여기서 막힌다
    for r in results[1]:
        update_sequence_output(r['request_id'], ...)   #    finished=True 로 표시

return [get_sequence(i).output[0] for i in request_ids]  # ④ 내 id 순서대로 꺼낸다

말로 풀면