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

refine stdout logging codes (#13015)

Browse Source
This commit is contained in:
yinghui
2025-11-12 00:16:14 +08:00
committed by GitHub
parent a06c44f905
commit 38a704bccb
7 changed files with 227 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/managers/detokenizer_manager.py
+4 -4
View File
@@ -279,8 +279,8 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin):
token_steps=recv_obj.token_steps, token_steps=recv_obj.token_steps,
queue_time=recv_obj.queue_time, queue_time=recv_obj.queue_time,
forward_entry_time=recv_obj.forward_entry_time, forward_entry_time=recv_obj.forward_entry_time,
prefill_delay=recv_obj.prefill_delay, prefill_launch_delay=recv_obj.prefill_launch_delay,
prefill_latency=recv_obj.prefill_latency, prefill_launch_latency=recv_obj.prefill_launch_latency,
) )
def handle_multimodal_decode_req(self, recv_obj: BatchMultimodalDecodeReq): def handle_multimodal_decode_req(self, recv_obj: BatchMultimodalDecodeReq):
@@ -297,8 +297,8 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin):
placeholder_tokens_val=None, placeholder_tokens_val=None,
queue_time=recv_obj.queue_time, queue_time=recv_obj.queue_time,
forward_entry_time=recv_obj.forward_entry_time, forward_entry_time=recv_obj.forward_entry_time,
prefill_delay=recv_obj.prefill_delay, prefill_launch_delay=recv_obj.prefill_launch_delay,
prefill_latency=recv_obj.prefill_latency, prefill_launch_latency=recv_obj.prefill_launch_latency,
) )
def handle_freeze_gc_req(self, recv_req: FreezeGCReq): def handle_freeze_gc_req(self, recv_req: FreezeGCReq):
python/sglang/srt/managers/io_struct.py
+6 -6
View File
@@ -80,18 +80,18 @@ class RequestTimingMetricsMixin:
# - Prefill instance (P): timestamp when prefill forward pass begins # - Prefill instance (P): timestamp when prefill forward pass begins
# - Decode instance (D): timestamp when decode forward pass begins # - Decode instance (D): timestamp when decode forward pass begins
# Note: This is NOT the same as prefill_start_time. There may be a delay between # Note: This is NOT the same as prefill_start_time. There may be a delay between
# forward_entry_time and prefill_start_time (see prefill_delay). # forward_entry_time and prefill_start_time (see prefill_launch_delay).
forward_entry_time: Optional[List[Optional[float]]] forward_entry_time: Optional[List[Optional[float]]]
# Prefill delay: time spent waiting between forward entry and prefill start. # Prefill launch delay: time spent waiting between forward entry and prefill start.
# Calculated as: prefill_start_time - forward_entry_time # Calculated as: prefill_start_time - forward_entry_time
# This represents the delay between when the request enters the forward stage # This represents the delay between when the request enters the forward stage
# and when prefill computation actually begins. # and when prefill computation actually begins.
prefill_delay: Optional[List[Optional[float]]] prefill_launch_delay: Optional[List[Optional[float]]]
# Prefill latency: time spent during prefill computation. # Prefill launch latency: time spent during prefill kernel launch.
# Calculated as: prefill_end_time - prefill_start_time # Calculated as: prefill_end_time_host - prefill_start_time_host
prefill_latency: Optional[List[Optional[float]]] prefill_launch_latency: Optional[List[Optional[float]]]
@dataclass @dataclass
python/sglang/srt/managers/multi_tokenizer_mixin.py
+133 -244
View File
@@ -86,310 +86,199 @@ class SocketMapping:
self._mapping[ipc_name].send_pyobj(output) self._mapping[ipc_name].send_pyobj(output)
def _extract_field_by_index(
output: Any, field_name: str, index: int, check_length: bool = True
) -> Any:
"""Extract a field value from output by index, handling None and length checks.
Args:
output: The output object containing the field
field_name: The name of the field to extract
index: The index to access in the field list
check_length: If True, check both field existence and length. If False, only check field existence.
Returns:
A list containing the field value at index, or None if not available.
"""
field = getattr(output, field_name, None)
if field is None:
return None
if check_length:
if len(field) <= index:
return None
return [field[index]]
def _handle_output_by_index(output, i): def _handle_output_by_index(output, i):
"""NOTE: A maintainable method is better here.""" """NOTE: A maintainable method is better here."""
if isinstance(output, BatchTokenIDOutput): if isinstance(output, BatchTokenIDOutput):
new_output = BatchTokenIDOutput( new_output = BatchTokenIDOutput(
rids=[output.rids[i]], rids=[output.rids[i]],
spec_verify_ct=( spec_verify_ct=_extract_field_by_index(output, "spec_verify_ct", i),
[output.spec_verify_ct[i]] if len(output.spec_verify_ct) > i else None spec_accepted_tokens=_extract_field_by_index(
output, "spec_accepted_tokens", i
), ),
spec_accepted_tokens=( queue_time=_extract_field_by_index(output, "queue_time", i),
[output.spec_accepted_tokens[i]] forward_entry_time=_extract_field_by_index(output, "forward_entry_time", i),
if len(output.spec_accepted_tokens) > i prefill_launch_delay=_extract_field_by_index(
else None output, "prefill_launch_delay", i
), ),
queue_time=[output.queue_time[i]] if len(output.queue_time) > i else None, prefill_launch_latency=_extract_field_by_index(
forward_entry_time=( output, "prefill_launch_latency", i
[output.forward_entry_time[i]]
if len(output.forward_entry_time) > i
else None
), ),
prefill_delay=( finished_reasons=_extract_field_by_index(output, "finished_reasons", i),
[output.prefill_delay[i]] if len(output.prefill_delay) > i else None decoded_texts=_extract_field_by_index(output, "decoded_texts", i),
decode_ids=_extract_field_by_index(output, "decode_ids", i),
read_offsets=_extract_field_by_index(output, "read_offsets", i),
output_ids=_extract_field_by_index(output, "output_ids", i),
skip_special_tokens=_extract_field_by_index(
output, "skip_special_tokens", i
), ),
prefill_latency=( spaces_between_special_tokens=_extract_field_by_index(
[output.prefill_latency[i]] if len(output.prefill_latency) > i else None output, "spaces_between_special_tokens", i
), ),
finished_reasons=( no_stop_trim=_extract_field_by_index(output, "no_stop_trim", i),
[output.finished_reasons[i]] prompt_tokens=_extract_field_by_index(output, "prompt_tokens", i),
if len(output.finished_reasons) > i completion_tokens=_extract_field_by_index(output, "completion_tokens", i),
else None cached_tokens=_extract_field_by_index(output, "cached_tokens", i),
input_token_logprobs_val=_extract_field_by_index(
output, "input_token_logprobs_val", i, check_length=False
), ),
decoded_texts=( input_token_logprobs_idx=_extract_field_by_index(
[output.decoded_texts[i]] if len(output.decoded_texts) > i else None output, "input_token_logprobs_idx", i, check_length=False
), ),
decode_ids=([output.decode_ids[i]] if len(output.decode_ids) > i else None), output_token_logprobs_val=_extract_field_by_index(
read_offsets=( output, "output_token_logprobs_val", i, check_length=False
[output.read_offsets[i]] if len(output.read_offsets) > i else None
), ),
output_ids=( output_token_logprobs_idx=_extract_field_by_index(
[output.output_ids[i]] output, "output_token_logprobs_idx", i, check_length=False
if output.output_ids and len(output.output_ids) > i
else None
), ),
skip_special_tokens=( input_top_logprobs_val=_extract_field_by_index(
[output.skip_special_tokens[i]] output, "input_top_logprobs_val", i, check_length=False
if len(output.skip_special_tokens) > i
else None
), ),
spaces_between_special_tokens=( input_top_logprobs_idx=_extract_field_by_index(
[output.spaces_between_special_tokens[i]] output, "input_top_logprobs_idx", i, check_length=False
if len(output.spaces_between_special_tokens) > i
else None
), ),
no_stop_trim=( output_top_logprobs_val=_extract_field_by_index(
[output.no_stop_trim[i]] if len(output.no_stop_trim) > i else None output, "output_top_logprobs_val", i, check_length=False
), ),
prompt_tokens=( output_top_logprobs_idx=_extract_field_by_index(
[output.prompt_tokens[i]] if len(output.prompt_tokens) > i else None output, "output_top_logprobs_idx", i, check_length=False
), ),
completion_tokens=( input_token_ids_logprobs_val=_extract_field_by_index(
[output.completion_tokens[i]] output, "input_token_ids_logprobs_val", i, check_length=False
if len(output.completion_tokens) > i
else None
), ),
cached_tokens=( input_token_ids_logprobs_idx=_extract_field_by_index(
[output.cached_tokens[i]] if len(output.cached_tokens) > i else None output, "input_token_ids_logprobs_idx", i, check_length=False
), ),
input_token_logprobs_val=( output_token_ids_logprobs_val=_extract_field_by_index(
[output.input_token_logprobs_val[i]] output, "output_token_ids_logprobs_val", i, check_length=False
if output.input_token_logprobs_val
else None
), ),
input_token_logprobs_idx=( output_token_ids_logprobs_idx=_extract_field_by_index(
[output.input_token_logprobs_idx[i]] output, "output_token_ids_logprobs_idx", i, check_length=False
if output.input_token_logprobs_idx
else None
), ),
output_token_logprobs_val=( output_token_entropy_val=_extract_field_by_index(
[output.output_token_logprobs_val[i]] output, "output_token_entropy_val", i, check_length=False
if output.output_token_logprobs_val
else None
), ),
output_token_logprobs_idx=( output_hidden_states=_extract_field_by_index(
[output.output_token_logprobs_idx[i]] output, "output_hidden_states", i, check_length=False
if output.output_token_logprobs_idx
else None
),
input_top_logprobs_val=(
[output.input_top_logprobs_val[i]]
if output.input_top_logprobs_val
else None
),
input_top_logprobs_idx=(
[output.input_top_logprobs_idx[i]]
if output.input_top_logprobs_idx
else None
),
output_top_logprobs_val=(
[output.output_top_logprobs_val[i]]
if output.output_top_logprobs_val
else None
),
output_top_logprobs_idx=(
[output.output_top_logprobs_idx[i]]
if output.output_top_logprobs_idx
else None
),
input_token_ids_logprobs_val=(
[output.input_token_ids_logprobs_val[i]]
if output.input_token_ids_logprobs_val
else None
),
input_token_ids_logprobs_idx=(
[output.input_token_ids_logprobs_idx[i]]
if output.input_token_ids_logprobs_idx
else None
),
output_token_ids_logprobs_val=(
[output.output_token_ids_logprobs_val[i]]
if output.output_token_ids_logprobs_val
else None
),
output_token_ids_logprobs_idx=(
[output.output_token_ids_logprobs_idx[i]]
if output.output_token_ids_logprobs_idx
else None
),
output_token_entropy_val=(
[output.output_token_entropy_val[i]]
if output.output_token_entropy_val
else None
),
output_hidden_states=(
[output.output_hidden_states[i]]
if output.output_hidden_states
else None
), ),
placeholder_tokens_idx=None, placeholder_tokens_idx=None,
placeholder_tokens_val=None, placeholder_tokens_val=None,
token_steps=([output.token_steps[i]] if output.token_steps else None), token_steps=_extract_field_by_index(
output, "token_steps", i, check_length=False
),
) )
elif isinstance(output, BatchEmbeddingOutput): elif isinstance(output, BatchEmbeddingOutput):
new_output = BatchEmbeddingOutput( new_output = BatchEmbeddingOutput(
rids=[output.rids[i]], rids=[output.rids[i]],
finished_reasons=( finished_reasons=_extract_field_by_index(output, "finished_reasons", i),
[output.finished_reasons[i]] embeddings=_extract_field_by_index(output, "embeddings", i),
if len(output.finished_reasons) > i prompt_tokens=_extract_field_by_index(output, "prompt_tokens", i),
else None cached_tokens=_extract_field_by_index(output, "cached_tokens", i),
),
embeddings=([output.embeddings[i]] if len(output.embeddings) > i else None),
prompt_tokens=(
[output.prompt_tokens[i]] if len(output.prompt_tokens) > i else None
),
cached_tokens=(
[output.cached_tokens[i]] if len(output.cached_tokens) > i else None
),
placeholder_tokens_idx=None, placeholder_tokens_idx=None,
placeholder_tokens_val=None, placeholder_tokens_val=None,
) )
elif isinstance(output, BatchStrOutput): elif isinstance(output, BatchStrOutput):
new_output = BatchStrOutput( new_output = BatchStrOutput(
rids=[output.rids[i]], rids=[output.rids[i]],
spec_verify_ct=( spec_verify_ct=_extract_field_by_index(output, "spec_verify_ct", i),
[output.spec_verify_ct[i]] if len(output.spec_verify_ct) > i else None spec_accepted_tokens=_extract_field_by_index(
output, "spec_accepted_tokens", i
), ),
spec_accepted_tokens=( queue_time=_extract_field_by_index(output, "queue_time", i),
[output.spec_accepted_tokens[i]] forward_entry_time=_extract_field_by_index(output, "forward_entry_time", i),
if len(output.spec_accepted_tokens) > i prefill_launch_delay=_extract_field_by_index(
else None output, "prefill_launch_delay", i
), ),
queue_time=[output.queue_time[i]] if len(output.queue_time) > i else None, prefill_launch_latency=_extract_field_by_index(
forward_entry_time=( output, "prefill_launch_latency", i
[output.forward_entry_time[i]]
if len(output.forward_entry_time) > i
else None
), ),
prefill_delay=( finished_reasons=_extract_field_by_index(output, "finished_reasons", i),
[output.prefill_delay[i]] if len(output.prefill_delay) > i else None output_strs=_extract_field_by_index(output, "output_strs", i),
output_ids=_extract_field_by_index(output, "output_ids", i),
prompt_tokens=_extract_field_by_index(output, "prompt_tokens", i),
completion_tokens=_extract_field_by_index(output, "completion_tokens", i),
cached_tokens=_extract_field_by_index(output, "cached_tokens", i),
input_token_logprobs_val=_extract_field_by_index(
output, "input_token_logprobs_val", i, check_length=False
), ),
prefill_latency=( input_token_logprobs_idx=_extract_field_by_index(
[output.prefill_latency[i]] if len(output.prefill_latency) > i else None output, "input_token_logprobs_idx", i, check_length=False
), ),
finished_reasons=( output_token_logprobs_val=_extract_field_by_index(
[output.finished_reasons[i]] output, "output_token_logprobs_val", i, check_length=False
if len(output.finished_reasons) > i
else None
), ),
output_strs=( output_token_logprobs_idx=_extract_field_by_index(
[output.output_strs[i]] if len(output.output_strs) > i else None output, "output_token_logprobs_idx", i, check_length=False
), ),
output_ids=( input_top_logprobs_val=_extract_field_by_index(
[output.output_ids[i]] output, "input_top_logprobs_val", i, check_length=False
if output.output_ids and len(output.output_ids) > i
else None
), ),
prompt_tokens=( input_top_logprobs_idx=_extract_field_by_index(
[output.prompt_tokens[i]] if len(output.prompt_tokens) > i else None output, "input_top_logprobs_idx", i, check_length=False
), ),
completion_tokens=( output_top_logprobs_val=_extract_field_by_index(
[output.completion_tokens[i]] output, "output_top_logprobs_val", i, check_length=False
if len(output.completion_tokens) > i
else None
), ),
cached_tokens=( output_top_logprobs_idx=_extract_field_by_index(
[output.cached_tokens[i]] if len(output.cached_tokens) > i else None output, "output_top_logprobs_idx", i, check_length=False
), ),
input_token_logprobs_val=( input_token_ids_logprobs_val=_extract_field_by_index(
[output.input_token_logprobs_val[i]] output, "input_token_ids_logprobs_val", i, check_length=False
if output.input_token_logprobs_val
else None
), ),
input_token_logprobs_idx=( input_token_ids_logprobs_idx=_extract_field_by_index(
[output.input_token_logprobs_idx[i]] output, "input_token_ids_logprobs_idx", i, check_length=False
if output.input_token_logprobs_idx
else None
), ),
output_token_logprobs_val=( output_token_ids_logprobs_val=_extract_field_by_index(
[output.output_token_logprobs_val[i]] output, "output_token_ids_logprobs_val", i, check_length=False
if output.output_token_logprobs_val
else None
), ),
output_token_logprobs_idx=( output_token_ids_logprobs_idx=_extract_field_by_index(
[output.output_token_logprobs_idx[i]] output, "output_token_ids_logprobs_idx", i, check_length=False
if output.output_token_logprobs_idx
else None
), ),
input_top_logprobs_val=( output_token_entropy_val=_extract_field_by_index(
[output.input_top_logprobs_val[i]] output, "output_token_entropy_val", i, check_length=False
if output.input_top_logprobs_val
else None
), ),
input_top_logprobs_idx=( output_hidden_states=_extract_field_by_index(
[output.input_top_logprobs_idx[i]] output, "output_hidden_states", i, check_length=False
if output.input_top_logprobs_idx
else None
),
output_top_logprobs_val=(
[output.output_top_logprobs_val[i]]
if output.output_top_logprobs_val
else None
),
output_top_logprobs_idx=(
[output.output_top_logprobs_idx[i]]
if output.output_top_logprobs_idx
else None
),
input_token_ids_logprobs_val=(
[output.input_token_ids_logprobs_val[i]]
if output.input_token_ids_logprobs_val
else None
),
input_token_ids_logprobs_idx=(
[output.input_token_ids_logprobs_idx[i]]
if output.input_token_ids_logprobs_idx
else None
),
output_token_ids_logprobs_val=(
[output.output_token_ids_logprobs_val[i]]
if output.output_token_ids_logprobs_val
else None
),
output_token_ids_logprobs_idx=(
[output.output_token_ids_logprobs_idx[i]]
if output.output_token_ids_logprobs_idx
else None
),
output_token_entropy_val=(
[output.output_token_entropy_val[i]]
if output.output_token_entropy_val
else None
),
output_hidden_states=(
[output.output_hidden_states[i]]
if output.output_hidden_states
else None
), ),
placeholder_tokens_idx=None, placeholder_tokens_idx=None,
placeholder_tokens_val=None, placeholder_tokens_val=None,
retraction_counts=( retraction_counts=_extract_field_by_index(output, "retraction_counts", i),
[output.retraction_counts[i]] token_steps=_extract_field_by_index(
if len(output.retraction_counts) > i output, "token_steps", i, check_length=False
else None
), ),
token_steps=([output.token_steps[i]] if output.token_steps else None),
) )
elif isinstance(output, BatchMultimodalOutput): elif isinstance(output, BatchMultimodalOutput):
new_output = BatchMultimodalOutput( new_output = BatchMultimodalOutput(
rids=[output.rids[i]], rids=[output.rids[i]],
finished_reasons=( finished_reasons=_extract_field_by_index(output, "finished_reasons", i),
[output.finished_reasons[i]] outputs=_extract_field_by_index(output, "outputs", i),
if len(output.finished_reasons) > i prompt_tokens=_extract_field_by_index(output, "prompt_tokens", i),
else None completion_tokens=_extract_field_by_index(output, "completion_tokens", i),
), cached_tokens=_extract_field_by_index(output, "cached_tokens", i),
outputs=([output.outputs[i]] if len(output.outputs) > i else None),
prompt_tokens=(
[output.prompt_tokens[i]] if len(output.prompt_tokens) > i else None
),
completion_tokens=(
[output.completion_tokens[i]]
if len(output.completion_tokens) > i
else None
),
cached_tokens=(
[output.cached_tokens[i]] if len(output.cached_tokens) > i else None
),
placeholder_tokens_idx=None, placeholder_tokens_idx=None,
placeholder_tokens_val=None, placeholder_tokens_val=None,
) )
python/sglang/srt/managers/scheduler.py
+2 -2
View File
@@ -1968,7 +1968,7 @@ class Scheduler(
if batch.forward_mode == ForwardMode.EXTEND: if batch.forward_mode == ForwardMode.EXTEND:
current_time = time.perf_counter() current_time = time.perf_counter()
for req in batch.reqs: for req in batch.reqs:
req.time_stats.prefill_start_time = current_time req.time_stats.prefill_start_time_host = current_time
# Place holder handling for pd-disagg decode event loop # Place holder handling for pd-disagg decode event loop
if batch.forward_mode.is_prebuilt(): if batch.forward_mode.is_prebuilt():
@@ -2077,7 +2077,7 @@ class Scheduler(
if batch.forward_mode == ForwardMode.EXTEND: if batch.forward_mode == ForwardMode.EXTEND:
current_time = time.perf_counter() current_time = time.perf_counter()
for req in batch.reqs: for req in batch.reqs:
req.time_stats.prefill_end_time = current_time req.time_stats.prefill_end_time_host = current_time
return ret return ret
python/sglang/srt/managers/scheduler_output_processor_mixin.py
+14 -42
View File
@@ -724,8 +724,8 @@ class SchedulerOutputProcessorMixin:
queue_times = [] queue_times = []
forward_entry_times = [] forward_entry_times = []
prefill_delays = [] prefill_launch_delays = []
prefill_latencies = [] prefill_launch_latencies = []
if return_logprob: if return_logprob:
input_token_logprobs_val = [] input_token_logprobs_val = []
@@ -830,24 +830,10 @@ class SchedulerOutputProcessorMixin:
queue_times.append(req.time_stats.get_queueing_time()) queue_times.append(req.time_stats.get_queueing_time())
forward_entry_times.append(req.time_stats.forward_entry_time) forward_entry_times.append(req.time_stats.forward_entry_time)
if req.time_stats.prefill_start_time > 0.0: prefill_launch_delays.append(req.time_stats.get_prefill_launch_delay())
prefill_delays.append( prefill_launch_latencies.append(
req.time_stats.prefill_start_time req.time_stats.get_prefill_launch_latency()
- req.time_stats.forward_entry_time
) )
else:
prefill_delays.append(None)
if (
req.time_stats.prefill_start_time > 0.0
and req.time_stats.prefill_end_time > 0.0
):
prefill_latencies.append(
req.time_stats.prefill_end_time
- req.time_stats.prefill_start_time
)
else:
prefill_latencies.append(None)
if not self.spec_algorithm.is_none(): if not self.spec_algorithm.is_none():
spec_verify_ct.append(req.spec_verify_ct) spec_verify_ct.append(req.spec_verify_ct)
@@ -946,8 +932,8 @@ class SchedulerOutputProcessorMixin:
spec_accepted_tokens=spec_accepted_tokens, spec_accepted_tokens=spec_accepted_tokens,
queue_time=queue_times, queue_time=queue_times,
forward_entry_time=forward_entry_times, forward_entry_time=forward_entry_times,
prefill_delay=prefill_delays, prefill_launch_delay=prefill_launch_delays,
prefill_latency=prefill_latencies, prefill_launch_latency=prefill_launch_latencies,
finished_reasons=finished_reasons, finished_reasons=finished_reasons,
decoded_texts=decoded_texts, decoded_texts=decoded_texts,
decode_ids=decode_ids_list, decode_ids=decode_ids_list,
@@ -989,8 +975,8 @@ class SchedulerOutputProcessorMixin:
cached_tokens = [] cached_tokens = []
queue_times = [] queue_times = []
forward_entry_times = [] forward_entry_times = []
prefill_delays = [] prefill_launch_delays = []
prefill_latencies = [] prefill_launch_latencies = []
retraction_counts = [] retraction_counts = []
for req in reqs: for req in reqs:
if req.finished(): if req.finished():
@@ -1004,24 +990,10 @@ class SchedulerOutputProcessorMixin:
queue_times.append(req.time_stats.get_queueing_time()) queue_times.append(req.time_stats.get_queueing_time())
forward_entry_times.append(req.time_stats.forward_entry_time) forward_entry_times.append(req.time_stats.forward_entry_time)
if req.time_stats.prefill_start_time > 0.0: prefill_launch_delays.append(req.time_stats.get_prefill_launch_delay())
prefill_delays.append( prefill_launch_latencies.append(
req.time_stats.prefill_start_time req.time_stats.get_prefill_launch_latency()
- req.time_stats.forward_entry_time
) )
else:
prefill_delays.append(None)
if (
req.time_stats.prefill_start_time > 0.0
and req.time_stats.prefill_end_time > 0.0
):
prefill_latencies.append(
req.time_stats.prefill_end_time
- req.time_stats.prefill_start_time
)
else:
prefill_latencies.append(None)
retraction_counts.append(req.retraction_count) retraction_counts.append(req.retraction_count)
self.send_to_detokenizer.send_output( self.send_to_detokenizer.send_output(
BatchEmbeddingOutput( BatchEmbeddingOutput(
@@ -1029,8 +1001,8 @@ class SchedulerOutputProcessorMixin:
http_worker_ipcs=http_worker_ipcs, http_worker_ipcs=http_worker_ipcs,
queue_time=queue_times, queue_time=queue_times,
forward_entry_time=forward_entry_times, forward_entry_time=forward_entry_times,
prefill_delay=prefill_delays, prefill_launch_delay=prefill_launch_delays,
prefill_latency=prefill_latencies, prefill_launch_latency=prefill_launch_latencies,
finished_reasons=finished_reasons, finished_reasons=finished_reasons,
embeddings=embeddings, embeddings=embeddings,
prompt_tokens=prompt_tokens, prompt_tokens=prompt_tokens,
python/sglang/srt/managers/tokenizer_manager.py
+53 -38
View File
@@ -140,8 +140,8 @@ class ReqState:
finished_time_perf: float = 0.0 finished_time_perf: float = 0.0
first_token_time_perf: float = 0.0 first_token_time_perf: float = 0.0
request_scheduled_ts: float = 0.0 request_sent_to_scheduler_ts: float = 0.0
response_sent_ts: float = 0.0 response_sent_to_client_ts: float = 0.0
# For streaming output # For streaming output
last_output_offset: int = 0 last_output_offset: int = 0
@@ -922,7 +922,7 @@ class TokenizerManager(TokenizerCommunicatorMixin):
tokenized_obj.trace_context = trace_get_proc_propagate_context(obj.rid) tokenized_obj.trace_context = trace_get_proc_propagate_context(obj.rid)
self.send_to_scheduler.send_pyobj(tokenized_obj) self.send_to_scheduler.send_pyobj(tokenized_obj)
state = ReqState([], False, asyncio.Event(), obj, created_time=created_time) state = ReqState([], False, asyncio.Event(), obj, created_time=created_time)
state.request_scheduled_ts = time.time() state.request_sent_to_scheduler_ts = time.time()
self.rid_to_state[obj.rid] = state self.rid_to_state[obj.rid] = state
trace_slice_end( trace_slice_end(
RequestStage.TOKENIZER_DISPATCH, obj.rid, thread_finish_flag=True RequestStage.TOKENIZER_DISPATCH, obj.rid, thread_finish_flag=True
@@ -980,11 +980,13 @@ class TokenizerManager(TokenizerCommunicatorMixin):
state.out_list = [] state.out_list = []
if state.finished: if state.finished:
# For non-streaming cases, response has not been sent yet (`response_sent_ts` has not been set yet). # For non-streaming cases, response has not been sent yet (`response_sent_to_client_ts` has not been set yet).
# Record response sent time right before we log finished results and metrics. # Record response sent time right before we log finished results and metrics.
if not state.response_sent_ts: if not state.response_sent_to_client_ts:
state.response_sent_ts = time.time() state.response_sent_to_client_ts = time.time()
out["meta_info"]["response_sent_ts"] = state.response_sent_ts out["meta_info"][
"response_sent_to_client_ts"
] = state.response_sent_to_client_ts
if self.log_requests: if self.log_requests:
max_length, skip_names, out_skip_names = self.log_request_metadata max_length, skip_names, out_skip_names = self.log_request_metadata
if self.model_config.is_multimodal_gen: if self.model_config.is_multimodal_gen:
@@ -1036,9 +1038,11 @@ class TokenizerManager(TokenizerCommunicatorMixin):
if obj.stream: if obj.stream:
# Record response sent time right before we send response. # Record response sent time right before we send response.
if not state.response_sent_ts: if not state.response_sent_to_client_ts:
state.response_sent_ts = time.time() state.response_sent_to_client_ts = time.time()
out["meta_info"]["response_sent_ts"] = state.response_sent_ts out["meta_info"][
"response_sent_to_client_ts"
] = state.response_sent_to_client_ts
yield out yield out
else: else:
if ( if (
@@ -1420,6 +1424,28 @@ class TokenizerManager(TokenizerCommunicatorMixin):
self._result_dispatcher(recv_obj) self._result_dispatcher(recv_obj)
self.last_receive_tstamp = time.time() self.last_receive_tstamp = time.time()
def _add_metric_if_present(
self,
recv_obj: Any,
attr_name: str,
meta_info: Dict[str, Any],
index: int,
) -> None:
"""Add a metric to meta_info if it exists and is not None.
Args:
recv_obj: The received object that may contain the metric attribute
attr_name: The name of the attribute to check
meta_info: The dictionary to add the metric to
index: The index to access the metric value in the attribute list
"""
if (
hasattr(recv_obj, attr_name)
and getattr(recv_obj, attr_name)
and getattr(recv_obj, attr_name)[index] is not None
):
meta_info[attr_name] = getattr(recv_obj, attr_name)[index]
def _handle_batch_output( def _handle_batch_output(
self, self,
recv_obj: Union[ recv_obj: Union[
@@ -1446,26 +1472,14 @@ class TokenizerManager(TokenizerCommunicatorMixin):
"total_retractions": recv_obj.retraction_counts[i], "total_retractions": recv_obj.retraction_counts[i],
} }
if ( if self.enable_metrics:
hasattr(recv_obj, "queue_time") self._add_metric_if_present(recv_obj, "queue_time", meta_info, i)
and recv_obj.queue_time self._add_metric_if_present(
and recv_obj.queue_time[i] is not None recv_obj, "prefill_launch_delay", meta_info, i
): )
meta_info["queue_time"] = recv_obj.queue_time[i] self._add_metric_if_present(
recv_obj, "prefill_launch_latency", meta_info, i
if ( )
hasattr(recv_obj, "prefill_delay")
and recv_obj.prefill_delay
and recv_obj.prefill_delay[i] is not None
):
meta_info["prefill_delay"] = recv_obj.prefill_delay[i]
if (
hasattr(recv_obj, "prefill_latency")
and recv_obj.prefill_latency
and recv_obj.prefill_latency[i] is not None
):
meta_info["prefill_latency"] = recv_obj.prefill_latency[i]
if getattr(state.obj, "return_logprob", False): if getattr(state.obj, "return_logprob", False):
self.convert_logprob_style( self.convert_logprob_style(
@@ -1535,7 +1549,7 @@ class TokenizerManager(TokenizerCommunicatorMixin):
state.finished_time_perf = time.perf_counter() state.finished_time_perf = time.perf_counter()
meta_info["e2e_latency"] = state.finished_time - state.created_time meta_info["e2e_latency"] = state.finished_time - state.created_time
# Calculate timing metrics if self.enable_metrics:
self._calculate_timing_metrics(meta_info, state, recv_obj, i) self._calculate_timing_metrics(meta_info, state, recv_obj, i)
trace_req_finish(rid, ts=int(state.finished_time * 1e9)) trace_req_finish(rid, ts=int(state.finished_time * 1e9))
@@ -1756,16 +1770,18 @@ class TokenizerManager(TokenizerCommunicatorMixin):
# Request timing timestamps. # Request timing timestamps.
if state.created_time > 0: if state.created_time > 0:
meta_info["request_received_ts"] = state.created_time meta_info["request_received_ts"] = state.created_time
if state.request_scheduled_ts > 0: if state.request_sent_to_scheduler_ts > 0:
meta_info["request_scheduled_ts"] = state.request_scheduled_ts meta_info["request_sent_to_scheduler_ts"] = (
state.request_sent_to_scheduler_ts
)
# For embeddings, there's no separate prefill phase, so omit `prefill_finished_ts`. # For embeddings, there's no separate prefill phase, so omit `prefill_finished_ts`.
if ( if (
not isinstance(recv_obj, BatchEmbeddingOutput) not isinstance(recv_obj, BatchEmbeddingOutput)
and state.first_token_time > 0 and state.first_token_time > 0
): ):
meta_info["prefill_finished_ts"] = state.first_token_time meta_info["prefill_finished_ts"] = state.first_token_time
if state.response_sent_ts > 0: if state.response_sent_to_client_ts > 0:
meta_info["response_sent_ts"] = state.response_sent_ts meta_info["response_sent_to_client_ts"] = state.response_sent_to_client_ts
if state.finished_time > 0: if state.finished_time > 0:
meta_info["decode_finished_ts"] = state.finished_time meta_info["decode_finished_ts"] = state.finished_time
@@ -1776,8 +1792,8 @@ class TokenizerManager(TokenizerCommunicatorMixin):
and recv_obj.forward_entry_time[i] is not None and recv_obj.forward_entry_time[i] is not None
and state.finished_time_perf > 0.0 and state.finished_time_perf > 0.0
): ):
forward_time = state.finished_time_perf - recv_obj.forward_entry_time[i] inference_time = state.finished_time_perf - recv_obj.forward_entry_time[i]
meta_info["forward_time"] = forward_time meta_info["inference_time"] = inference_time
# Decode throughput, time per token calculation. Only calculated if TTFT is available. # Decode throughput, time per token calculation. Only calculated if TTFT is available.
if ( if (
@@ -1789,7 +1805,6 @@ class TokenizerManager(TokenizerCommunicatorMixin):
decode_time = state.finished_time_perf - state.first_token_time_perf decode_time = state.finished_time_perf - state.first_token_time_perf
completion_tokens = recv_obj.completion_tokens[i] completion_tokens = recv_obj.completion_tokens[i]
meta_info["decode_throughput"] = completion_tokens / decode_time meta_info["decode_throughput"] = completion_tokens / decode_time
meta_info["time_per_token"] = decode_time / completion_tokens
def collect_metrics(self, state: ReqState, recv_obj: BatchStrOutput, i: int): def collect_metrics(self, state: ReqState, recv_obj: BatchStrOutput, i: int):
completion_tokens = ( completion_tokens = (
python/sglang/srt/metrics/collector.py
+12 -2
View File
@@ -61,12 +61,22 @@ class TimeStats:
# TODO: correct set them # TODO: correct set them
bootstrap_duration: float = 0.0 bootstrap_duration: float = 0.0
alloc_waiting_duration: float = 0.0 alloc_waiting_duration: float = 0.0
prefill_start_time: float = 0.0 prefill_start_time_host: float = 0.0
prefill_end_time: float = 0.0 prefill_end_time_host: float = 0.0
def get_queueing_time(self) -> float: def get_queueing_time(self) -> float:
return self.forward_entry_time - self.wait_queue_entry_time return self.forward_entry_time - self.wait_queue_entry_time
def get_prefill_launch_delay(self) -> Optional[float]:
if self.prefill_start_time_host > 0.0:
return self.prefill_start_time_host - self.forward_entry_time
return None
def get_prefill_launch_latency(self) -> Optional[float]:
if self.prefill_start_time_host > 0.0 and self.prefill_end_time_host > 0.0:
return self.prefill_end_time_host - self.prefill_start_time_host
return None
def convert_to_duration(self) -> str: def convert_to_duration(self) -> str:
if self.disagg_mode == DisaggregationMode.NULL: if self.disagg_mode == DisaggregationMode.NULL:
queue_duration = self.forward_entry_time - self.wait_queue_entry_time queue_duration = self.forward_entry_time - self.wait_queue_entry_time