From 5de66d7fbc17addee15ff52d89157e3984f3cd9d Mon Sep 17 00:00:00 2001 From: Yuang Chen <77919385+cccccya@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:04:03 +0800 Subject: [PATCH] [EPD][BugFix] Fix encoder health check with global cache TP (#29296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: 晟海 Co-authored-by: siyu Co-authored-by: Shangming Cai --- .../srt/disaggregation/encode_server.py | 45 ++++++++++++------- 1 file changed, 30 insertions(+), 15 deletions(-) diff --git a/python/sglang/srt/disaggregation/encode_server.py b/python/sglang/srt/disaggregation/encode_server.py index 6de7918ab..3c20cf791 100644 --- a/python/sglang/srt/disaggregation/encode_server.py +++ b/python/sglang/srt/disaggregation/encode_server.py @@ -3205,22 +3205,37 @@ async def run_encoder( encoder = MMEncoder(server_args, schedule_path, dist_init_method, rank) while True: request = await async_sock_recv(encoder.schedule_socket) - if isinstance(request, ProfileReq): - if request.req_type == ProfileReqType.START_PROFILE: - if encoder.profiler is None: - encoder.profiler = EncoderProfiler(encoder.rank) - encoder.profiler.start(request) - else: - encoder.profiler.stop() - elif isinstance(request, dict) and request.get("type") == "batch_encode": - await encoder.batch_encode( - request["requests"], - Modality.from_str(request["modality"]), - ) + await _handle_encoder_worker_request(encoder, request) + + +async def _handle_encoder_worker_request(encoder: MMEncoder, request): + if isinstance(request, ProfileReq): + if request.req_type == ProfileReqType.START_PROFILE: + if encoder.profiler is None: + encoder.profiler = EncoderProfiler(encoder.rank) + encoder.profiler.start(request) else: - await encoder.encode_request( - request, Modality.from_str(request["modality"]) - ) + encoder.profiler.stop() + elif isinstance(request, dict) and request.get("type") == "batch_encode": + await encoder.batch_encode( + request["requests"], + Modality.from_str(request["modality"]), + ) + elif ( + isinstance(request, dict) + and isinstance(request.get("req_id"), str) + and request["req_id"].startswith(HEALTH_CHECK_RID_PREFIX) + ): + await encoder.encode( + mm_items=request["mm_items"], + modality=Modality.from_str(request["modality"]), + req_id=request["req_id"], + num_parts=request["num_parts"], + part_idx=request["part_idx"], + hashes=request.get("hashes"), + ) + else: + await encoder.encode_request(request, Modality.from_str(request["modality"])) def launch_encoder(server_args, schedule_path, dist_init_method, rank):