Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 0 additions & 14 deletions python/sglang/srt/configs/model_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -178,12 +178,6 @@ def __init__(
self.is_multimodal = enable_multimodal and is_multimodal_model(
self.hf_config.architectures
)
self.is_multimodal_gen = enable_multimodal and is_multimodal_gen_model(
self.hf_config.architectures
)
self.is_image_gen = enable_multimodal and is_image_gen_model(
self.hf_config.architectures
)
self.is_audio_model = enable_multimodal and is_audio_model(
self.hf_config.architectures
)
Expand Down Expand Up @@ -1351,14 +1345,6 @@ def is_multimodal_model(model_architectures: List[str]):
return False


def is_multimodal_gen_model(model_architectures: List[str]):
return False


def is_image_gen_model(model_architectures: List[str]):
return False


def is_audio_model(model_architectures: List[str]):
models = [
"WhisperForConditionalGeneration",
Expand Down
2 changes: 2 additions & 0 deletions python/sglang/srt/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,3 +8,5 @@
GPU_MEMORY_TYPE_WEIGHTS,
GPU_MEMORY_TYPE_CUDA_GRAPH,
]

HEALTH_CHECK_RID_PREFIX = "HEALTH_CHECK"
3 changes: 2 additions & 1 deletion python/sglang/srt/entrypoints/http_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import ORJSONResponse, Response, StreamingResponse

from sglang.srt.constants import HEALTH_CHECK_RID_PREFIX
from sglang.srt.disaggregation.utils import FAKE_BOOTSTRAP_HOST, DisaggregationMode
from sglang.srt.entrypoints.anthropic.protocol import (
AnthropicCountTokensRequest,
Expand Down Expand Up @@ -509,7 +510,7 @@ async def health_generate(request: Request) -> Response:
return Response(status_code=200)

sampling_params = {"max_new_tokens": 1, "temperature": 0.0}
rid = f"HEALTH_CHECK_{time.time()}"
rid = f"{HEALTH_CHECK_RID_PREFIX}_{time.time()}"

if _global_state.tokenizer_manager.is_image_gen:
gri = _global_state.tokenizer_manager.get_image_gen_health_check_request(
Expand Down
83 changes: 30 additions & 53 deletions python/sglang/srt/managers/detokenizer_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,10 @@
import setproctitle
import zmq

from sglang.srt.constants import HEALTH_CHECK_RID_PREFIX
from sglang.srt.environ import envs
from sglang.srt.managers.io_struct import (
BatchEmbeddingOutput,
BatchMultimodalDecodeReq,
BatchStrOutput,
BatchTokenIDOutput,
FreezeGCReq,
Expand Down Expand Up @@ -88,16 +88,9 @@ def __init__(
# Init running status
self.init_running_status(server_args)

if server_args.enable_metrics:
start_cpu_monitor_thread("detokenizer")

# Init dispatcher
self.init_request_dispatcher()

@staticmethod
def is_health_check_request(rid: Optional[str]) -> bool:
return isinstance(rid, str) and rid.startswith("HEALTH_CHECK")

def init_ipc_channels(self, port_args: PortArgs):
context = zmq.Context(2)
self.recv_from_scheduler = get_zmq_socket(
Expand All @@ -120,9 +113,8 @@ def init_tokenizer(self, server_args: ServerArgs):

def init_running_status(self, server_args: ServerArgs):
self.decode_status = LimitedCapacityDict(capacity=DETOKENIZER_MAX_STATES)
self.is_dummy = False
self.is_tool_call_parser_gpt_oss = server_args.tool_call_parser == "gpt-oss"
self.disable_tokenizer_batch_decode = server_args.disable_tokenizer_batch_decode
self.is_tool_call_parser_gpt_oss = server_args.tool_call_parser == "gpt-oss"

self.soft_watchdog = Watchdog.create(
debug_name="DetokenizerManager",
Expand All @@ -131,12 +123,14 @@ def init_running_status(self, server_args: ServerArgs):
test_stuck_time=envs.SGLANG_TEST_STUCK_DETOKENIZER.get(),
)

if server_args.enable_metrics:
start_cpu_monitor_thread("detokenizer")

def init_request_dispatcher(self):
self._request_dispatcher = TypeBasedDispatcher(
[
(BatchEmbeddingOutput, self.handle_batch_embedding_out),
(BatchTokenIDOutput, self.handle_batch_token_id_out),
(BatchMultimodalDecodeReq, self.handle_multimodal_decode_req),
(FreezeGCReq, self.handle_freeze_gc_req),
]
)
Expand Down Expand Up @@ -190,8 +184,6 @@ def _grouped_batch_decode(
) -> List[str]:
"""Batch decode with grouping by (skip_special_tokens, spaces_between_special_tokens)."""

assert self.tokenizer is not None

# fast path
first_skip, first_space = skip_list[0], space_list[0]
if all(s == first_skip for s in skip_list) and all(
Expand Down Expand Up @@ -236,9 +228,6 @@ def _decode_batch_token_id_output(self, recv_obj: BatchTokenIDOutput):
surr_offset=0,
read_offset=recv_obj.read_offsets[i],
)
if not self.is_health_check_request(rid):
# for health check requests, we do not store the decode status
self.decode_status[rid] = s
Comment thread
merrymercy marked this conversation as resolved.
else:
s = self.decode_status[rid]
s.decode_ids.extend(recv_obj.decode_ids[i])
Expand All @@ -254,22 +243,16 @@ def _decode_batch_token_id_output(self, recv_obj: BatchTokenIDOutput):

# Decode token ids to strings
if not self.disable_tokenizer_batch_decode:
if not self.is_dummy:
# Run normal batch decode
surr_texts = self._grouped_batch_decode(
surr_ids,
recv_obj.skip_special_tokens,
recv_obj.spaces_between_special_tokens,
)
read_texts = self._grouped_batch_decode(
read_ids,
recv_obj.skip_special_tokens,
recv_obj.spaces_between_special_tokens,
)
else:
# If it is dummy weights, just return dummy strings to prevent potential detokenization edge cases
surr_texts = ["dog" for _ in surr_ids]
read_texts = ["cat" for _ in read_ids]
surr_texts = self._grouped_batch_decode(
surr_ids,
recv_obj.skip_special_tokens,
recv_obj.spaces_between_special_tokens,
)
read_texts = self._grouped_batch_decode(
read_ids,
recv_obj.skip_special_tokens,
recv_obj.spaces_between_special_tokens,
)
else:
# Do not use batch decode to prevent some detokenization edge cases (e.g., gpt-oss).
surr_texts = [
Expand Down Expand Up @@ -297,25 +280,17 @@ def _decode_batch_token_id_output(self, recv_obj: BatchTokenIDOutput):
output_strs = []
for i in range(bs):
rid = recv_obj.rids[i]
if self.is_health_check_request(rid):
s = DecodeStatus(
decoded_text=recv_obj.decoded_texts[i],
decode_ids=recv_obj.decode_ids[i],
surr_offset=0,
read_offset=recv_obj.read_offsets[i],
try:
s = self.decode_status[rid]
except KeyError:
raise RuntimeError(
f"Decode status not found for request {rid}. "
"It may be due to the request being evicted from the decode status due to memory pressure. "
"Please increase the maximum number of requests by setting "
"the SGLANG_DETOKENIZER_MAX_STATES environment variable to a bigger value than the default value. "
f"The current value is {DETOKENIZER_MAX_STATES}. "
"For more details, see: https://github.com/sgl-project/sglang/issues/2812"
)
else:
try:
s = self.decode_status[rid]
except KeyError:
raise RuntimeError(
f"Decode status not found for request {rid}. "
"It may be due to the request being evicted from the decode status due to memory pressure. "
"Please increase the maximum number of requests by setting "
"the SGLANG_DETOKENIZER_MAX_STATES environment variable to a bigger value than the default value. "
f"The current value is {DETOKENIZER_MAX_STATES}. "
"For more details, see: https://github.com/sgl-project/sglang/issues/2812"
)
new_text = read_texts[i][len(surr_texts[i]) :]
if recv_obj.finished_reasons[i] is None:
# Streaming chunk: update the decode status
Expand All @@ -335,6 +310,7 @@ def _decode_batch_token_id_output(self, recv_obj: BatchTokenIDOutput):
recv_obj.finished_reasons[i],
recv_obj.no_stop_trim[i],
)

# Incrementally send text.
incremental_output = output_str[s.sent_offset :]
s.sent_offset = len(output_str)
Expand Down Expand Up @@ -404,14 +380,15 @@ def handle_batch_token_id_out(self, recv_obj: BatchTokenIDOutput):
time_stats=recv_obj.time_stats,
)

def handle_multimodal_decode_req(self, recv_obj: BatchMultimodalDecodeReq):
raise NotImplementedError()

def handle_freeze_gc_req(self, recv_req: FreezeGCReq):
freeze_gc("Detokenizer Manager")
return None


def is_health_check_request(rid: Optional[str]) -> bool:
return isinstance(rid, str) and rid.startswith(HEALTH_CHECK_RID_PREFIX)


class LimitedCapacityDict(OrderedDict):
def __init__(self, capacity: int, *args, **kwargs):
super().__init__(*args, **kwargs)
Expand Down
3 changes: 2 additions & 1 deletion python/sglang/srt/managers/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

from sglang.jit_kernel.ngram_embedding import update_token_table
from sglang.srt.configs.model_config import ModelConfig
from sglang.srt.constants import HEALTH_CHECK_RID_PREFIX
from sglang.srt.constrained.grammar_manager import GrammarManager
from sglang.srt.disaggregation.decode import (
DecodePreallocQueue,
Expand Down Expand Up @@ -3406,7 +3407,7 @@ def maybe_sleep(self):

def is_health_check_generate_req(recv_req):
rid = getattr(recv_req, "rid", None)
return rid is not None and rid.startswith("HEALTH_CHECK")
return rid is not None and rid.startswith(HEALTH_CHECK_RID_PREFIX)


def is_work_request(recv_req):
Expand Down
16 changes: 2 additions & 14 deletions python/sglang/srt/managers/scheduler_output_processor_mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -953,10 +953,6 @@ def stream_output_generation(
if req is skip_req:
continue

# Multimodal partial stream chunks break the detokenizer, so drop aborted requests here.
if self.model_config.is_multimodal_gen and req.to_finish:
continue

if req.finished():
if req.finished_output:
# With the overlap schedule, a request will try to output twice and hit this line twice
Expand All @@ -975,8 +971,7 @@ def stream_output_generation(
# origin stream_interval logic
should_output = (
len(req.output_ids) % stream_interval == 1
if not self.model_config.is_multimodal_gen
and stream_interval > 1
if stream_interval > 1
else len(req.output_ids) % stream_interval == 0
)

Expand All @@ -986,8 +981,6 @@ def stream_output_generation(
else:
should_output = (
len(req.output_ids) % DEFAULT_FORCE_STREAM_INTERVAL == 0
if not self.model_config.is_multimodal_gen
else False
)

if should_output:
Expand All @@ -1003,10 +996,7 @@ def stream_output_generation(
decoded_texts.append(req.decoded_text)
decode_ids, read_offset = req.init_incremental_detokenize()

if self.model_config.is_multimodal_gen:
decode_ids_list.append(decode_ids)
else:
decode_ids_list.append(decode_ids[req.send_decode_id_offset :])
decode_ids_list.append(decode_ids[req.send_decode_id_offset :])

# Exclude the tokens after stop condition
output_ids_ = req.output_ids_through_stop
Expand Down Expand Up @@ -1132,8 +1122,6 @@ def stream_output_generation(

# Send to detokenizer
if reqs or is_idle_batch:
if self.model_config.is_multimodal_gen:
return
self.send_to_detokenizer.send_output(
BatchTokenIDOutput(
rids=rids,
Expand Down
3 changes: 1 addition & 2 deletions python/sglang/srt/managers/tokenizer_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ def init_model_config(self):
self.served_model_name = server_args.served_model_name
self.model_config = model_config_class.from_server_args(server_args)
self.is_generation = self.model_config.is_generation
self.is_image_gen = self.model_config.is_image_gen
self.is_image_gen = getattr(self.model_config, "is_image_gen", False)
self.context_len = self.model_config.context_len
self.image_token_id = self.model_config.image_token_id
self.max_req_input_len = None # Will be set later in engine.py
Expand Down Expand Up @@ -1194,7 +1194,6 @@ async def _wait_one_response(
self.request_logger.log_finished_request(
obj,
out,
is_multimodal_gen=self.model_config.is_multimodal_gen,
request=request,
)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from datetime import datetime
from typing import List, Optional, Union

from sglang.srt.constants import HEALTH_CHECK_RID_PREFIX
from sglang.srt.managers.io_struct import EmbeddingReqInput, GenerateReqInput
from sglang.srt.server_args import ServerArgs

Expand Down Expand Up @@ -128,7 +129,7 @@ async def write_record(
self, obj: Union[GenerateReqInput, EmbeddingReqInput], out_dict: dict
):
# Do not log health check requests, since they don't represent real user requests.
if isinstance(obj.rid, str) and "HEALTH_CHECK" in obj.rid:
if isinstance(obj.rid, str) and HEALTH_CHECK_RID_PREFIX in obj.rid:
return

try:
Expand Down
5 changes: 1 addition & 4 deletions python/sglang/srt/utils/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import builtins
import ctypes
import functools
import gc
import importlib
import inspect
import io
Expand Down Expand Up @@ -2945,8 +2946,6 @@ def gc_callback(phase, info):


def freeze_gc(context: str):
import gc

g0_before, g1_before, g2_before = gc_object_counts()
gc.freeze()
g0_after, g1_after, g2_after = gc_object_counts()
Expand All @@ -2961,8 +2960,6 @@ def freeze_gc(context: str):
def configure_gc_logger():
logger.info("Enable GC Logger")

import gc

gc_start_time = {}

def gc_callback(phase, info):
Expand Down
14 changes: 4 additions & 10 deletions python/sglang/srt/utils/request_logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,6 @@ def log_finished_request(
self,
obj: Union["GenerateReqInput", "EmbeddingReqInput"],
out: Any,
is_multimodal_gen: bool = False,
request: Optional["fastapi.Request"] = None,
) -> None:
if not self.log_requests:
Expand All @@ -181,20 +180,15 @@ def log_finished_request(
}
if headers:
log_data["headers"] = headers
if not is_multimodal_gen:
log_data["out"] = _transform_data_for_logging(
out, max_length, out_skip_names
)
log_data["out"] = _transform_data_for_logging(
out, max_length, out_skip_names
)
log_json(self.targets, "request.finished", log_data)
else:
obj_str = _dataclass_to_string_truncated(
obj, max_length, skip_names=skip_names
)
out_str = (
""
if is_multimodal_gen
else f", out={_dataclass_to_string_truncated(out, max_length, skip_names=out_skip_names)}"
)
out_str = f", out={_dataclass_to_string_truncated(out, max_length, skip_names=out_skip_names)}"
headers_str = f", headers={headers}" if headers else ""
self._log(f"Finish: obj={obj_str}{headers_str}{out_str}")

Expand Down
3 changes: 2 additions & 1 deletion test/registered/bench_fn/test_bench_serving_functionality.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from sglang.bench_serving import run_benchmark
from sglang.benchmark.utils import parse_custom_headers
from sglang.srt.constants import HEALTH_CHECK_RID_PREFIX
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.test_utils import (
Expand Down Expand Up @@ -80,7 +81,7 @@ def _verify_multi_turn_logs(self, content: str):
continue
text = obj.get("obj", {}).get("text")
rid = obj.get("rid", "")
if text and not rid.startswith("HEALTH_CHECK"):
if text and not rid.startswith(HEALTH_CHECK_RID_PREFIX):
reqs.append(text)

self.assertGreaterEqual(len(reqs), NUM_CONVERSATIONS * NUM_TURNS)
Expand Down
Loading
Loading