diff --git a/python/sglang/srt/managers/detokenizer_manager.py b/python/sglang/srt/managers/detokenizer_manager.py index 9399bbbea..d09f17d2e 100644 --- a/python/sglang/srt/managers/detokenizer_manager.py +++ b/python/sglang/srt/managers/detokenizer_manager.py @@ -279,8 +279,8 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin): token_steps=recv_obj.token_steps, queue_time=recv_obj.queue_time, forward_entry_time=recv_obj.forward_entry_time, - prefill_delay=recv_obj.prefill_delay, - prefill_latency=recv_obj.prefill_latency, + prefill_launch_delay=recv_obj.prefill_launch_delay, + prefill_launch_latency=recv_obj.prefill_launch_latency, ) def handle_multimodal_decode_req(self, recv_obj: BatchMultimodalDecodeReq): @@ -297,8 +297,8 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin): placeholder_tokens_val=None, queue_time=recv_obj.queue_time, forward_entry_time=recv_obj.forward_entry_time, - prefill_delay=recv_obj.prefill_delay, - prefill_latency=recv_obj.prefill_latency, + prefill_launch_delay=recv_obj.prefill_launch_delay, + prefill_launch_latency=recv_obj.prefill_launch_latency, ) def handle_freeze_gc_req(self, recv_req: FreezeGCReq): diff --git a/python/sglang/srt/managers/io_struct.py b/python/sglang/srt/managers/io_struct.py index c6e06dc85..cdecebca3 100644 --- a/python/sglang/srt/managers/io_struct.py +++ b/python/sglang/srt/managers/io_struct.py @@ -80,18 +80,18 @@ class RequestTimingMetricsMixin: # - Prefill instance (P): timestamp when prefill 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 - # 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]]] - # 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 # This represents the delay between when the request enters the forward stage # 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. - # Calculated as: prefill_end_time - prefill_start_time - prefill_latency: Optional[List[Optional[float]]] + # Prefill launch latency: time spent during prefill kernel launch. + # Calculated as: prefill_end_time_host - prefill_start_time_host + prefill_launch_latency: Optional[List[Optional[float]]] @dataclass diff --git a/python/sglang/srt/managers/multi_tokenizer_mixin.py b/python/sglang/srt/managers/multi_tokenizer_mixin.py index 66410571b..84a12957f 100644 --- a/python/sglang/srt/managers/multi_tokenizer_mixin.py +++ b/python/sglang/srt/managers/multi_tokenizer_mixin.py @@ -86,310 +86,199 @@ class SocketMapping: 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): """NOTE: A maintainable method is better here.""" if isinstance(output, BatchTokenIDOutput): new_output = BatchTokenIDOutput( rids=[output.rids[i]], - spec_verify_ct=( - [output.spec_verify_ct[i]] if len(output.spec_verify_ct) > i else None + spec_verify_ct=_extract_field_by_index(output, "spec_verify_ct", i), + spec_accepted_tokens=_extract_field_by_index( + output, "spec_accepted_tokens", i ), - spec_accepted_tokens=( - [output.spec_accepted_tokens[i]] - if len(output.spec_accepted_tokens) > i - else None + queue_time=_extract_field_by_index(output, "queue_time", i), + forward_entry_time=_extract_field_by_index(output, "forward_entry_time", i), + prefill_launch_delay=_extract_field_by_index( + output, "prefill_launch_delay", i ), - queue_time=[output.queue_time[i]] if len(output.queue_time) > i else None, - forward_entry_time=( - [output.forward_entry_time[i]] - if len(output.forward_entry_time) > i - else None + prefill_launch_latency=_extract_field_by_index( + output, "prefill_launch_latency", i ), - prefill_delay=( - [output.prefill_delay[i]] if len(output.prefill_delay) > i else None + finished_reasons=_extract_field_by_index(output, "finished_reasons", i), + 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=( - [output.prefill_latency[i]] if len(output.prefill_latency) > i else None + spaces_between_special_tokens=_extract_field_by_index( + output, "spaces_between_special_tokens", i ), - finished_reasons=( - [output.finished_reasons[i]] - if len(output.finished_reasons) > i - else None + no_stop_trim=_extract_field_by_index(output, "no_stop_trim", 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 ), - decoded_texts=( - [output.decoded_texts[i]] if len(output.decoded_texts) > i else None + input_token_logprobs_idx=_extract_field_by_index( + output, "input_token_logprobs_idx", i, check_length=False ), - decode_ids=([output.decode_ids[i]] if len(output.decode_ids) > i else None), - read_offsets=( - [output.read_offsets[i]] if len(output.read_offsets) > i else None + output_token_logprobs_val=_extract_field_by_index( + output, "output_token_logprobs_val", i, check_length=False ), - output_ids=( - [output.output_ids[i]] - if output.output_ids and len(output.output_ids) > i - else None + output_token_logprobs_idx=_extract_field_by_index( + output, "output_token_logprobs_idx", i, check_length=False ), - skip_special_tokens=( - [output.skip_special_tokens[i]] - if len(output.skip_special_tokens) > i - else None + input_top_logprobs_val=_extract_field_by_index( + output, "input_top_logprobs_val", i, check_length=False ), - spaces_between_special_tokens=( - [output.spaces_between_special_tokens[i]] - if len(output.spaces_between_special_tokens) > i - else None + input_top_logprobs_idx=_extract_field_by_index( + output, "input_top_logprobs_idx", i, check_length=False ), - no_stop_trim=( - [output.no_stop_trim[i]] if len(output.no_stop_trim) > i else None + output_top_logprobs_val=_extract_field_by_index( + output, "output_top_logprobs_val", i, check_length=False ), - prompt_tokens=( - [output.prompt_tokens[i]] if len(output.prompt_tokens) > i else None + output_top_logprobs_idx=_extract_field_by_index( + output, "output_top_logprobs_idx", i, check_length=False ), - completion_tokens=( - [output.completion_tokens[i]] - if len(output.completion_tokens) > i - else None + input_token_ids_logprobs_val=_extract_field_by_index( + output, "input_token_ids_logprobs_val", i, check_length=False ), - cached_tokens=( - [output.cached_tokens[i]] if len(output.cached_tokens) > i else None + input_token_ids_logprobs_idx=_extract_field_by_index( + output, "input_token_ids_logprobs_idx", i, check_length=False ), - input_token_logprobs_val=( - [output.input_token_logprobs_val[i]] - if output.input_token_logprobs_val - else None + output_token_ids_logprobs_val=_extract_field_by_index( + output, "output_token_ids_logprobs_val", i, check_length=False ), - input_token_logprobs_idx=( - [output.input_token_logprobs_idx[i]] - if output.input_token_logprobs_idx - else None + output_token_ids_logprobs_idx=_extract_field_by_index( + output, "output_token_ids_logprobs_idx", i, check_length=False ), - output_token_logprobs_val=( - [output.output_token_logprobs_val[i]] - if output.output_token_logprobs_val - else None + output_token_entropy_val=_extract_field_by_index( + output, "output_token_entropy_val", i, check_length=False ), - output_token_logprobs_idx=( - [output.output_token_logprobs_idx[i]] - 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 + output_hidden_states=_extract_field_by_index( + output, "output_hidden_states", i, check_length=False ), placeholder_tokens_idx=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): new_output = BatchEmbeddingOutput( rids=[output.rids[i]], - finished_reasons=( - [output.finished_reasons[i]] - if len(output.finished_reasons) > i - else None - ), - 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 - ), + finished_reasons=_extract_field_by_index(output, "finished_reasons", i), + embeddings=_extract_field_by_index(output, "embeddings", i), + prompt_tokens=_extract_field_by_index(output, "prompt_tokens", i), + cached_tokens=_extract_field_by_index(output, "cached_tokens", i), placeholder_tokens_idx=None, placeholder_tokens_val=None, ) elif isinstance(output, BatchStrOutput): new_output = BatchStrOutput( rids=[output.rids[i]], - spec_verify_ct=( - [output.spec_verify_ct[i]] if len(output.spec_verify_ct) > i else None + spec_verify_ct=_extract_field_by_index(output, "spec_verify_ct", i), + spec_accepted_tokens=_extract_field_by_index( + output, "spec_accepted_tokens", i ), - spec_accepted_tokens=( - [output.spec_accepted_tokens[i]] - if len(output.spec_accepted_tokens) > i - else None + queue_time=_extract_field_by_index(output, "queue_time", i), + forward_entry_time=_extract_field_by_index(output, "forward_entry_time", i), + prefill_launch_delay=_extract_field_by_index( + output, "prefill_launch_delay", i ), - queue_time=[output.queue_time[i]] if len(output.queue_time) > i else None, - forward_entry_time=( - [output.forward_entry_time[i]] - if len(output.forward_entry_time) > i - else None + prefill_launch_latency=_extract_field_by_index( + output, "prefill_launch_latency", i ), - prefill_delay=( - [output.prefill_delay[i]] if len(output.prefill_delay) > i else None + finished_reasons=_extract_field_by_index(output, "finished_reasons", i), + 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=( - [output.prefill_latency[i]] if len(output.prefill_latency) > i else None + input_token_logprobs_idx=_extract_field_by_index( + output, "input_token_logprobs_idx", i, check_length=False ), - finished_reasons=( - [output.finished_reasons[i]] - if len(output.finished_reasons) > i - else None + output_token_logprobs_val=_extract_field_by_index( + output, "output_token_logprobs_val", i, check_length=False ), - output_strs=( - [output.output_strs[i]] if len(output.output_strs) > i else None + output_token_logprobs_idx=_extract_field_by_index( + output, "output_token_logprobs_idx", i, check_length=False ), - output_ids=( - [output.output_ids[i]] - if output.output_ids and len(output.output_ids) > i - else None + input_top_logprobs_val=_extract_field_by_index( + output, "input_top_logprobs_val", i, check_length=False ), - prompt_tokens=( - [output.prompt_tokens[i]] if len(output.prompt_tokens) > i else None + input_top_logprobs_idx=_extract_field_by_index( + output, "input_top_logprobs_idx", i, check_length=False ), - completion_tokens=( - [output.completion_tokens[i]] - if len(output.completion_tokens) > i - else None + output_top_logprobs_val=_extract_field_by_index( + output, "output_top_logprobs_val", i, check_length=False ), - cached_tokens=( - [output.cached_tokens[i]] if len(output.cached_tokens) > i else None + output_top_logprobs_idx=_extract_field_by_index( + output, "output_top_logprobs_idx", i, check_length=False ), - input_token_logprobs_val=( - [output.input_token_logprobs_val[i]] - if output.input_token_logprobs_val - else None + input_token_ids_logprobs_val=_extract_field_by_index( + output, "input_token_ids_logprobs_val", i, check_length=False ), - input_token_logprobs_idx=( - [output.input_token_logprobs_idx[i]] - if output.input_token_logprobs_idx - else None + input_token_ids_logprobs_idx=_extract_field_by_index( + output, "input_token_ids_logprobs_idx", i, check_length=False ), - output_token_logprobs_val=( - [output.output_token_logprobs_val[i]] - if output.output_token_logprobs_val - else None + output_token_ids_logprobs_val=_extract_field_by_index( + output, "output_token_ids_logprobs_val", i, check_length=False ), - output_token_logprobs_idx=( - [output.output_token_logprobs_idx[i]] - if output.output_token_logprobs_idx - else None + output_token_ids_logprobs_idx=_extract_field_by_index( + output, "output_token_ids_logprobs_idx", i, check_length=False ), - input_top_logprobs_val=( - [output.input_top_logprobs_val[i]] - if output.input_top_logprobs_val - else None + output_token_entropy_val=_extract_field_by_index( + output, "output_token_entropy_val", i, check_length=False ), - 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 + output_hidden_states=_extract_field_by_index( + output, "output_hidden_states", i, check_length=False ), placeholder_tokens_idx=None, placeholder_tokens_val=None, - retraction_counts=( - [output.retraction_counts[i]] - if len(output.retraction_counts) > i - else None + retraction_counts=_extract_field_by_index(output, "retraction_counts", i), + token_steps=_extract_field_by_index( + output, "token_steps", i, check_length=False ), - token_steps=([output.token_steps[i]] if output.token_steps else None), ) elif isinstance(output, BatchMultimodalOutput): new_output = BatchMultimodalOutput( rids=[output.rids[i]], - finished_reasons=( - [output.finished_reasons[i]] - if len(output.finished_reasons) > i - else None - ), - 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 - ), + finished_reasons=_extract_field_by_index(output, "finished_reasons", i), + outputs=_extract_field_by_index(output, "outputs", 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), placeholder_tokens_idx=None, placeholder_tokens_val=None, ) diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index a36ea17f5..4e8612ef5 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -1968,7 +1968,7 @@ class Scheduler( if batch.forward_mode == ForwardMode.EXTEND: current_time = time.perf_counter() 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 if batch.forward_mode.is_prebuilt(): @@ -2077,7 +2077,7 @@ class Scheduler( if batch.forward_mode == ForwardMode.EXTEND: current_time = time.perf_counter() for req in batch.reqs: - req.time_stats.prefill_end_time = current_time + req.time_stats.prefill_end_time_host = current_time return ret diff --git a/python/sglang/srt/managers/scheduler_output_processor_mixin.py b/python/sglang/srt/managers/scheduler_output_processor_mixin.py index 4747167c6..6ad020cab 100644 --- a/python/sglang/srt/managers/scheduler_output_processor_mixin.py +++ b/python/sglang/srt/managers/scheduler_output_processor_mixin.py @@ -724,8 +724,8 @@ class SchedulerOutputProcessorMixin: queue_times = [] forward_entry_times = [] - prefill_delays = [] - prefill_latencies = [] + prefill_launch_delays = [] + prefill_launch_latencies = [] if return_logprob: input_token_logprobs_val = [] @@ -830,24 +830,10 @@ class SchedulerOutputProcessorMixin: queue_times.append(req.time_stats.get_queueing_time()) forward_entry_times.append(req.time_stats.forward_entry_time) - if req.time_stats.prefill_start_time > 0.0: - prefill_delays.append( - req.time_stats.prefill_start_time - - 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) + prefill_launch_delays.append(req.time_stats.get_prefill_launch_delay()) + prefill_launch_latencies.append( + req.time_stats.get_prefill_launch_latency() + ) if not self.spec_algorithm.is_none(): spec_verify_ct.append(req.spec_verify_ct) @@ -946,8 +932,8 @@ class SchedulerOutputProcessorMixin: spec_accepted_tokens=spec_accepted_tokens, queue_time=queue_times, forward_entry_time=forward_entry_times, - prefill_delay=prefill_delays, - prefill_latency=prefill_latencies, + prefill_launch_delay=prefill_launch_delays, + prefill_launch_latency=prefill_launch_latencies, finished_reasons=finished_reasons, decoded_texts=decoded_texts, decode_ids=decode_ids_list, @@ -989,8 +975,8 @@ class SchedulerOutputProcessorMixin: cached_tokens = [] queue_times = [] forward_entry_times = [] - prefill_delays = [] - prefill_latencies = [] + prefill_launch_delays = [] + prefill_launch_latencies = [] retraction_counts = [] for req in reqs: if req.finished(): @@ -1004,24 +990,10 @@ class SchedulerOutputProcessorMixin: queue_times.append(req.time_stats.get_queueing_time()) forward_entry_times.append(req.time_stats.forward_entry_time) - if req.time_stats.prefill_start_time > 0.0: - prefill_delays.append( - req.time_stats.prefill_start_time - - 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) + prefill_launch_delays.append(req.time_stats.get_prefill_launch_delay()) + prefill_launch_latencies.append( + req.time_stats.get_prefill_launch_latency() + ) retraction_counts.append(req.retraction_count) self.send_to_detokenizer.send_output( BatchEmbeddingOutput( @@ -1029,8 +1001,8 @@ class SchedulerOutputProcessorMixin: http_worker_ipcs=http_worker_ipcs, queue_time=queue_times, forward_entry_time=forward_entry_times, - prefill_delay=prefill_delays, - prefill_latency=prefill_latencies, + prefill_launch_delay=prefill_launch_delays, + prefill_launch_latency=prefill_launch_latencies, finished_reasons=finished_reasons, embeddings=embeddings, prompt_tokens=prompt_tokens, diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index c5a2e35ed..4111e16ac 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -140,8 +140,8 @@ class ReqState: finished_time_perf: float = 0.0 first_token_time_perf: float = 0.0 - request_scheduled_ts: float = 0.0 - response_sent_ts: float = 0.0 + request_sent_to_scheduler_ts: float = 0.0 + response_sent_to_client_ts: float = 0.0 # For streaming output last_output_offset: int = 0 @@ -922,7 +922,7 @@ class TokenizerManager(TokenizerCommunicatorMixin): tokenized_obj.trace_context = trace_get_proc_propagate_context(obj.rid) self.send_to_scheduler.send_pyobj(tokenized_obj) 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 trace_slice_end( RequestStage.TOKENIZER_DISPATCH, obj.rid, thread_finish_flag=True @@ -980,11 +980,13 @@ class TokenizerManager(TokenizerCommunicatorMixin): state.out_list = [] 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. - if not state.response_sent_ts: - state.response_sent_ts = time.time() - out["meta_info"]["response_sent_ts"] = state.response_sent_ts + if not state.response_sent_to_client_ts: + state.response_sent_to_client_ts = time.time() + out["meta_info"][ + "response_sent_to_client_ts" + ] = state.response_sent_to_client_ts if self.log_requests: max_length, skip_names, out_skip_names = self.log_request_metadata if self.model_config.is_multimodal_gen: @@ -1036,9 +1038,11 @@ class TokenizerManager(TokenizerCommunicatorMixin): if obj.stream: # Record response sent time right before we send response. - if not state.response_sent_ts: - state.response_sent_ts = time.time() - out["meta_info"]["response_sent_ts"] = state.response_sent_ts + if not state.response_sent_to_client_ts: + state.response_sent_to_client_ts = time.time() + out["meta_info"][ + "response_sent_to_client_ts" + ] = state.response_sent_to_client_ts yield out else: if ( @@ -1420,6 +1424,28 @@ class TokenizerManager(TokenizerCommunicatorMixin): self._result_dispatcher(recv_obj) 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( self, recv_obj: Union[ @@ -1446,26 +1472,14 @@ class TokenizerManager(TokenizerCommunicatorMixin): "total_retractions": recv_obj.retraction_counts[i], } - if ( - hasattr(recv_obj, "queue_time") - and recv_obj.queue_time - and recv_obj.queue_time[i] is not None - ): - meta_info["queue_time"] = recv_obj.queue_time[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 self.enable_metrics: + self._add_metric_if_present(recv_obj, "queue_time", meta_info, i) + self._add_metric_if_present( + recv_obj, "prefill_launch_delay", meta_info, i + ) + self._add_metric_if_present( + recv_obj, "prefill_launch_latency", meta_info, i + ) if getattr(state.obj, "return_logprob", False): self.convert_logprob_style( @@ -1535,8 +1549,8 @@ class TokenizerManager(TokenizerCommunicatorMixin): state.finished_time_perf = time.perf_counter() meta_info["e2e_latency"] = state.finished_time - state.created_time - # Calculate timing metrics - self._calculate_timing_metrics(meta_info, state, recv_obj, i) + if self.enable_metrics: + self._calculate_timing_metrics(meta_info, state, recv_obj, i) trace_req_finish(rid, ts=int(state.finished_time * 1e9)) @@ -1756,16 +1770,18 @@ class TokenizerManager(TokenizerCommunicatorMixin): # Request timing timestamps. if state.created_time > 0: meta_info["request_received_ts"] = state.created_time - if state.request_scheduled_ts > 0: - meta_info["request_scheduled_ts"] = state.request_scheduled_ts + if state.request_sent_to_scheduler_ts > 0: + 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`. if ( not isinstance(recv_obj, BatchEmbeddingOutput) and state.first_token_time > 0 ): meta_info["prefill_finished_ts"] = state.first_token_time - if state.response_sent_ts > 0: - meta_info["response_sent_ts"] = state.response_sent_ts + if state.response_sent_to_client_ts > 0: + meta_info["response_sent_to_client_ts"] = state.response_sent_to_client_ts if state.finished_time > 0: 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 state.finished_time_perf > 0.0 ): - forward_time = state.finished_time_perf - recv_obj.forward_entry_time[i] - meta_info["forward_time"] = forward_time + inference_time = state.finished_time_perf - recv_obj.forward_entry_time[i] + meta_info["inference_time"] = inference_time # Decode throughput, time per token calculation. Only calculated if TTFT is available. if ( @@ -1789,7 +1805,6 @@ class TokenizerManager(TokenizerCommunicatorMixin): decode_time = state.finished_time_perf - state.first_token_time_perf completion_tokens = recv_obj.completion_tokens[i] 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): completion_tokens = ( diff --git a/python/sglang/srt/metrics/collector.py b/python/sglang/srt/metrics/collector.py index 205a3cdbc..69d7dafc5 100644 --- a/python/sglang/srt/metrics/collector.py +++ b/python/sglang/srt/metrics/collector.py @@ -61,12 +61,22 @@ class TimeStats: # TODO: correct set them bootstrap_duration: float = 0.0 alloc_waiting_duration: float = 0.0 - prefill_start_time: float = 0.0 - prefill_end_time: float = 0.0 + prefill_start_time_host: float = 0.0 + prefill_end_time_host: float = 0.0 def get_queueing_time(self) -> float: 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: if self.disagg_mode == DisaggregationMode.NULL: queue_duration = self.forward_entry_time - self.wait_queue_entry_time