diff --git a/services/whisper-gpu/server.py b/services/whisper-gpu/server.py index c00824b3..8b9b8f7f 100644 --- a/services/whisper-gpu/server.py +++ b/services/whisper-gpu/server.py @@ -42,6 +42,24 @@ logger = logging.getLogger("whisper-gpu") _gpu_lock = asyncio.Lock() +# --------------------------------------------------------------------------- +# GPU 큐 즉시 거절 한도 (task-3018) +# +# _gpu_lock 을 쓰는 모든 전사 경로(/v1/transcribe, /v1/transcribe/url, +# /v1/youtube-transcribe)가 공통으로 통과하는 run_transcription() 진입점에서 +# 검사한다. 한도 초과 시 **대기시키지 않고 즉시 429** 를 반환한다 — +# 클라이언트가 타임아웃까지 매달렸다가 포기하고, 서버는 아무도 읽지 않을 +# 결과를 GPU 100%로 계속 만들어내던 사고(원인 C)를 끊는 것이 목적이다. +# +# 카운터는 "락 대기자 + 락 보유자" 합계다. 기본 2 = 실행 1건 + 대기 1건 허용. +# --------------------------------------------------------------------------- + +WHISPER_MAX_QUEUE_DEFAULT = 2 +WHISPER_RETRY_AFTER_SEC_DEFAULT = 300 + +# 현재 _gpu_lock 을 보유 중이거나 대기 중인 요청 수 (이벤트 루프 단일 스레드에서만 변경) +_gpu_queue_depth = 0 + # API 키 인증 (환경변수가 없으면 인증 비활성화) WHISPER_API_KEY: Optional[str] = os.environ.get("WHISPER_API_KEY") or None @@ -169,6 +187,32 @@ def select_model_for_duration(duration_seconds: float) -> str: return os.environ.get("WHISPER_YT_MODEL", DEFAULT_TRANSCRIBE_MODEL) +def _env_int(name: str, default: int, minimum: int = 1) -> int: + """환경변수를 int로 읽습니다. 미설정/파싱 실패/범위 미달 시 default.""" + raw = os.environ.get(name) + if raw is None or raw.strip() == "": + return default + try: + value = int(raw) + except ValueError: + logger.warning("%s 값이 정수가 아님(%r) — 기본값 %d 사용", name, raw, default) + return default + if value < minimum: + logger.warning("%s 값이 최소치(%d) 미만(%d) — 기본값 %d 사용", name, minimum, value, default) + return default + return value + + +def get_max_queue() -> int: + """GPU 큐 즉시 거절 한도. 환경변수 WHISPER_MAX_QUEUE 로 override.""" + return _env_int("WHISPER_MAX_QUEUE", WHISPER_MAX_QUEUE_DEFAULT) + + +def get_retry_after_seconds() -> int: + """429 응답의 Retry-After(초). 환경변수 WHISPER_RETRY_AFTER_SEC 로 override.""" + return _env_int("WHISPER_RETRY_AFTER_SEC", WHISPER_RETRY_AFTER_SEC_DEFAULT) + + def get_model_instance(model_name: str) -> Any: """모델 이름에 해당하는 인스턴스를 반환합니다. lazy-load 적용.""" global _last_used @@ -347,34 +426,74 @@ async def run_transcription( language: str, model_name: str, ) -> dict[str, Any]: - """GPU Lock을 획득한 후 전사를 실행합니다.""" + """GPU Lock을 획득한 후 전사를 실행합니다. + + task-3018 (원인 A): + faster-whisper 의 `model.transcribe()` 는 **제너레이터를 즉시 반환**한다 + (lazy). 실제 GPU 연산은 그 제너레이터를 소비할 때 일어난다. 예전 코드는 + `transcribe()` 호출만 executor 로 보내고 세그먼트 리스트화는 이벤트 루프 + 스레드에서 했기 때문에, 전사가 도는 내내(실측 43분) ASGI 서버 전체가 + 정지해 신규 요청 accept·/v1/health 응답조차 불가했다. + → 호출과 소비를 하나의 동기 함수로 묶어 **executor 안에서 전부** 수행한다. + 이벤트 루프에서 제너레이터를 소비하는 코드는 0이어야 한다. + + task-3018 (원인 C): + 대기 큐 한도(WHISPER_MAX_QUEUE) 초과 시 대기시키지 않고 즉시 429. + """ + global _gpu_queue_depth + + max_queue = get_max_queue() + if _gpu_queue_depth >= max_queue: + logger.warning( + "GPU 큐 한도 초과 — 즉시 거절(429): 대기/실행 %d건, 한도 %d건, file=%s", + _gpu_queue_depth, + max_queue, + file_path, + ) + raise HTTPException( + status_code=429, + detail=( + f"Whisper GPU 큐 포화 (대기/실행 {_gpu_queue_depth}건 >= 한도 {max_queue}건). " + "잠시 후 다시 시도하세요." + ), + headers={"Retry-After": str(get_retry_after_seconds())}, + ) + loop = asyncio.get_event_loop() # 한국어(또는 언어 미지정)일 때만 보험 도메인 initial_prompt 주입 initial_prompt = INSURANCE_INITIAL_PROMPT if (not language or language == "ko") else None - async with _gpu_lock: - logger.info("전사 시작: model=%s language=%s file=%s", model_name, language, file_path) - model = get_model_instance(model_name) - segments_raw, info = await loop.run_in_executor( - None, - lambda: model.transcribe( - file_path, - language=language if language else None, - beam_size=5, - initial_prompt=initial_prompt, - ), - ) - segments = [ - { - "start": round(seg.start, 3), - "end": round(seg.end, 3), - "text": seg.text.strip(), - } - for seg in segments_raw - ] - full_text = postprocess_transcript(" ".join(seg["text"] for seg in segments)) - detected_language = getattr(info, "language", language) or language - duration = float(getattr(info, "duration", 0.0)) + _gpu_queue_depth += 1 + try: + async with _gpu_lock: + logger.info("전사 시작: model=%s language=%s file=%s", model_name, language, file_path) + model = get_model_instance(model_name) + + def _transcribe_and_collect() -> tuple[list[dict[str, Any]], Any]: + """executor 스레드에서 실행: 제너레이터 소비까지 여기서 끝낸다.""" + raw, transcribe_info = model.transcribe( + file_path, + language=language if language else None, + beam_size=5, + initial_prompt=initial_prompt, + ) + collected = [ + { + "start": round(seg.start, 3), + "end": round(seg.end, 3), + "text": seg.text.strip(), + } + for seg in raw + ] + return collected, transcribe_info + + segments, info = await loop.run_in_executor(None, _transcribe_and_collect) + full_text = postprocess_transcript(" ".join(seg["text"] for seg in segments)) + detected_language = getattr(info, "language", language) or language + duration = float(getattr(info, "duration", 0.0)) + finally: + # 예외·취소 상황에서도 반드시 복원 (누수 시 서비스가 영구 429가 된다) + _gpu_queue_depth -= 1 logger.info( "전사 완료: language=%s duration=%.1fs segments=%d", @@ -704,6 +823,10 @@ async def youtube_transcribe(request: YouTubeTranscribeRequest) -> JSONResponse: try: result = await run_transcription(audio_path, request.language, model_name) + except HTTPException: + # 큐 한도 초과(429) 등 의도된 상태코드를 500으로 뭉개지 않는다 (task-3018). + # 다른 엔드포인트(/v1/transcribe, /v1/transcribe/url)와 동일한 규약. + raise except Exception as e: logger.exception("전사 실패: video_id=%s error=%s", video_id, e) raise HTTPException(