minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[EPD][BugFix] Fix encoder health check with global cache TP (#29296)

Browse Source 
Co-authored-by: 晟海 <huangtingwei.htw@antgroup.com>
Co-authored-by: siyu <liusy58@linux.alibaba.com>
Co-authored-by: Shangming Cai <csmthu@gmail.com>
This commit is contained in:
Yuang Chen
2026-07-01 11:04:03 +08:00
committed by GitHub
co-authored by 晟海 siyu Shangming Cai
parent 56f22cd520
commit 5de66d7fbc
1 changed files with 30 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/disaggregation/encode_server.py
+30 -15
View File
@@ -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):