gRPC 스트리밍 네 모양과 기한을 손으로 확인한다

목표

meter.proto 에서 코드를 만들고 서버·클라이언트·양방향 스트리밍을 구현한 뒤, 스트림 도중의 오류와 기한 초과가 클라이언트와 서버에서 각각 어떻게 보이는지 확인합니다.

왜 중요한가

gRPC 스트리밍은 HTTP/2 스트림 하나 위에 메시지를 여러 개 싣는 방식입니다. 그래서 응답 상태는 맨 끝에 오고, 기한은 머리로 서버까지 전해지며, 양쪽이 서로를 기다리면 멈춥니다. 이 세 가지를 모르면 "스트리밍으로 바꿨는데 한꺼번에 온다", "오류가 나면 받은 결과가 다 사라진다", "타임아웃이 났는데 서버 CPU 가 계속 돈다" 같은 일이 생깁니다. 음성 AI 의 부분 인식 결과와 토큰 스트리밍이 바로 이 모양으로 흐릅니다.

단계

계약에서 코드를 만든다 — /opt/fixtures/rt/grpc/meter.proto 에서 파이썬 코드를 만들어 /root/rt/grpc 에 두세요. /opt/rt-lab/bin/python -m grpc_tools.protoc -I/opt/fixtures/rt/grpc --python_out=/root/rt/grpc --grpc_python_out=/root/rt/grpc /opt/fixtures/rt/grpc/meter.proto 한 줄이면 meter_pb2.py(메시지)와 meter_pb2_grpc.py(서비스 뼈대와 스텁)가 생깁니다. 두 파일을 열어 Countdown·Sum·Chat 이 어떤 모양의 호출로 만들어졌는지 확인하세요.
서버 스트리밍 — 만들어지는 대로 보낸다 — /root/rt/grpc/server.py 에 meter_pb2_grpc.MeterServicer 를 상속한 Meter 와 serve(port) 를 만드세요. serve 는 스레드 16개짜리 grpc.server 에 Meter 를 붙여 127.0.0.1:port 에서 듣고 끝날 때까지 기다립니다. Countdown 은 request.n 부터 1 까지 Tick 을 하나씩 보내고, 하나 보낼 때마다 request.interval_ms 밀리초 쉽니다. 채점기는 첫 Tick 이 언제 도착하는지 잽니다.
클라이언트 스트리밍 — 끝까지 받고 한 번 답한다 — Meter 에 Sum 을 추가하세요. 들어오는 Num 스트림을 끝까지 읽어 개수와 합을 Total(count, sum) 으로 한 번 돌려줍니다. 아무것도 오지 않고 끝난 스트림이면 Total(count=0, sum=0) 입니다.
양방향 스트리밍 — 다 받기 전에 답한다 — Meter 에 Chat 을 추가하세요. 들어오는 Line 마다 text 를 대문자로 바꾼 Line 을 곧바로 돌려줍니다. 채점기는 한 줄 보내고 답을 받은 뒤에야 다음 줄을 보냅니다. 그리고 3초 기한을 겁니다.
스트림 도중의 오류는 이미 받은 것을 지우지 않는다 — Countdown 을 고쳐 request.fail_at 이 0 이 아니면 Tick 을 fail_at 개 보낸 뒤 context.abort(grpc.StatusCode.ABORTED, "fail_at") 으로 끝내게 하세요. 그리고 /root/rt/grpc/client.py 에 read_countdown(target, n, interval_ms, fail_at, timeout=10) 을 만드세요. target 에 채널을 열어 Countdown 을 부르고, 받은 value 의 list 와 끝난 상태 코드 이름(정상이면 "OK")을 tuple 로 돌려줍니다. 채점기는 기준 서버를 상대로 부릅니다.
기한은 호출에 건다 — /root/rt/grpc/client.py 에 countdown_deadline(target, n, interval_ms, timeout) 을 추가하세요. Countdown 을 기한 timeout 초로 부르고, 받은 value 의 list, 상태 코드 이름, 호출에 걸린 초를 tuple 로 돌려줍니다. 기한은 스텁 호출의 timeout 인자로 걸어야 합니다.
기한이 지나면 서버의 일도 멈춘다 — Meter 에 Work 와 GetStats 를 추가하세요. Work 는 request.steps 번 request.step_ms 밀리초씩 일하되, 단계마다 context.is_active() 가 거짓이면 멈추고, 마친 단계 수를 WorkDone(steps_done) 으로 돌려줍니다. 서버는 지금까지 모든 Work 가 마친 단계 수의 누계를 세어 두고, GetStats 가 그 누계를 Stats(work_steps) 로 돌려줍니다. 채점기는 100ms 짜리 20단계 일을 0.35초 기한으로 부른 뒤 2.5초 기다려 누계를 봅니다.

참고

작업 폴더는 /root/rt/grpc 입니다. mkdir -p /root/rt/grpc 로 먼저 만드세요.
서버는 cd /root/rt/grpc && /opt/rt-lab/bin/python -c "import server; server.serve(50051)" 로 띄워 볼 수 있습니다. 채점기는 빈 포트를 골라 따로 띄웁니다.
클라이언트 단계의 상대인 기준 서버는 /opt/rt-lab/bin/python /opt/fixtures/rt/grpc/refserver.py --port 50052 로 직접 띄울 수 있습니다.
흔한 실수 두 가지입니다. 양방향 스트리밍에서 요청을 list 로 먼저 다 모아 서로 멈추는 것, 그리고 기한을 호출이 아니라 클라이언트 쪽 시계로만 재서 서버가 기한을 모르게 하는 것입니다.
파이썬은 반드시 /opt/rt-lab/bin/python 으로 실행합니다. 이 실습의 라이브러리는 그 가상환경에만 들어 있고, 그냥 python3 로 돌리면 ModuleNotFoundError 가 납니다. alias rpy=/opt/rt-lab/bin/python 처럼 줄여 두면 편합니다.
실습 파드는 바깥으로 나가는 연결이 막혀 있습니다. 모든 통신은 같은 파드 안의 127.0.0.1 에서 일어나며, 설치나 다운로드는 필요 없습니다.
채점기는 코드를 별도 프로세스로 불러 실제 연결을 맺어 봅니다. 예시 파일은 함수 틀일 뿐이라 그대로 두면 통과하지 않습니다. 앞 단계에서 완성한 함수는 지우지 마세요.
실습 세션이 끝나면 /root 의 파일은 남지 않습니다. 필요한 코드는 끝내기 전에 따로 보관하세요.

계약에서 코드를 만든다

/opt/fixtures/rt/grpc/meter.proto 에서 파이썬 코드를 만들어 /root/rt/grpc 에 두세요. /opt/rt-lab/bin/python -m grpc_tools.protoc -I/opt/fixtures/rt/grpc --python_out=/root/rt/grpc --grpc_python_out=/root/rt/grpc /opt/fixtures/rt/grpc/meter.proto 한 줄이면 meter_pb2.py(메시지)와 meter_pb2_grpc.py(서비스 뼈대와 스텁)가 생깁니다. 두 파일을 열어 Countdown·Sum·Chat 이 어떤 모양의 호출로 만들어졌는지 확인하세요.

rpc 선언의 stream 이 요청 쪽에 붙었는지 응답 쪽에 붙었는지에 따라 네 모양이 갈립니다. 생성 코드의 unary_stream·stream_unary·stream_stream 이 그 모양입니다. proto 파일은 고치지 않습니다.

서버 스트리밍 — 만들어지는 대로 보낸다

/root/rt/grpc/server.py 에 meter_pb2_grpc.MeterServicer 를 상속한 Meter 와 serve(port) 를 만드세요. serve 는 스레드 16개짜리 grpc.server 에 Meter 를 붙여 127.0.0.1:port 에서 듣고 끝날 때까지 기다립니다. Countdown 은 request.n 부터 1 까지 Tick 을 하나씩 보내고, 하나 보낼 때마다 request.interval_ms 밀리초 쉽니다. 채점기는 첫 Tick 이 언제 도착하는지 잽니다.

제너레이터로 yield 하면 gRPC 가 하나씩 내보냅니다. 전부 list 에 모아 return 하면 첫 메시지가 마지막 메시지와 같은 시각에 도착하고, 그러면 스트리밍으로 만든 의미가 없습니다.

클라이언트 스트리밍 — 끝까지 받고 한 번 답한다

Meter 에 Sum 을 추가하세요. 들어오는 Num 스트림을 끝까지 읽어 개수와 합을 Total(count, sum) 으로 한 번 돌려줍니다. 아무것도 오지 않고 끝난 스트림이면 Total(count=0, sum=0) 입니다.

요청 반복자는 클라이언트가 스트림을 닫아야(half-close) 끝납니다. 끝을 기다리는 것이 이 모양의 계약입니다.

양방향 스트리밍 — 다 받기 전에 답한다

Meter 에 Chat 을 추가하세요. 들어오는 Line 마다 text 를 대문자로 바꾼 Line 을 곧바로 돌려줍니다. 채점기는 한 줄 보내고 답을 받은 뒤에야 다음 줄을 보냅니다. 그리고 3초 기한을 겁니다.

list(request_iterator) 처럼 요청을 먼저 다 모으면 클라이언트는 답을 기다리고 서버는 요청이 끝나기를 기다려 서로 멈춥니다. 반복자를 도는 루프 안에서 곧바로 yield 하세요.

스트림 도중의 오류는 이미 받은 것을 지우지 않는다

Countdown 을 고쳐 request.fail_at 이 0 이 아니면 Tick 을 fail_at 개 보낸 뒤 context.abort(grpc.StatusCode.ABORTED, "fail_at") 으로 끝내게 하세요. 그리고 /root/rt/grpc/client.py 에 read_countdown(target, n, interval_ms, fail_at, timeout=10) 을 만드세요. target 에 채널을 열어 Countdown 을 부르고, 받은 value 의 list 와 끝난 상태 코드 이름(정상이면 "OK")을 tuple 로 돌려줍니다. 채점기는 기준 서버를 상대로 부릅니다.

상태 코드는 메시지 뒤, 트레일러에 실려 옵니다. 그래서 오류는 반복 도중에 grpc.RpcError 로 튀어나오고, 그 전에 받은 메시지는 이미 여러분 손에 있습니다. list(...) 한 번으로 받으면 그것을 잃습니다. e.code().name 이 상태 코드 이름입니다.

기한은 호출에 건다

/root/rt/grpc/client.py 에 countdown_deadline(target, n, interval_ms, timeout) 을 추가하세요. Countdown 을 기한 timeout 초로 부르고, 받은 value 의 list, 상태 코드 이름, 호출에 걸린 초를 tuple 로 돌려줍니다. 기한은 스텁 호출의 timeout 인자로 걸어야 합니다.

timeout 을 호출에 걸면 grpc-timeout 머리로 서버에도 전해집니다. 클라이언트에서 따로 시계를 재서 멈추면 서버는 기한을 모른 채 계속 일합니다. 기한이 지나면 상태는 DEADLINE_EXCEEDED 입니다.

기한이 지나면 서버의 일도 멈춘다

Meter 에 Work 와 GetStats 를 추가하세요. Work 는 request.steps 번 request.step_ms 밀리초씩 일하되, 단계마다 context.is_active() 가 거짓이면 멈추고, 마친 단계 수를 WorkDone(steps_done) 으로 돌려줍니다. 서버는 지금까지 모든 Work 가 마친 단계 수의 누계를 세어 두고, GetStats 가 그 누계를 Stats(work_steps) 로 돌려줍니다. 채점기는 100ms 짜리 20단계 일을 0.35초 기한으로 부른 뒤 2.5초 기다려 누계를 봅니다.

클라이언트가 기한 초과를 받았다고 서버의 스레드가 멈추지는 않습니다. 서버 코드가 스스로 확인해야 합니다. 확인하지 않으면 아무도 받지 않을 결과를 위해 CPU 와 DB 연결을 끝까지 씁니다. 여러 스레드가 누계를 올리므로 락을 쓰세요.