From 2c14a6ccd4b5ec70a8219bdd830c46f7e7be6cf4 Mon Sep 17 00:00:00 2001 From: ZLkanyo009 <407120045@qq.com> Date: Wed, 31 Dec 2025 06:12:06 +0000 Subject: [PATCH] [Feat] add ttft measure for qwen3vl --- python/sglang/bench_serving.py | 73 ++-- .../srt/managers/detokenizer_manager.py | 27 ++ python/sglang/srt/managers/io_struct.py | 26 ++ python/sglang/srt/managers/mm_utils.py | 296 ++++++++++++++++- python/sglang/srt/managers/schedule_batch.py | 35 +- python/sglang/srt/managers/scheduler.py | 90 ++++- .../scheduler_output_processor_mixin.py | 96 +++++- .../sglang/srt/managers/tokenizer_manager.py | 285 +++++++++++++++- python/sglang/srt/managers/tp_worker.py | 4 + python/sglang/srt/managers/utils.py | 6 + .../srt/model_executor/forward_batch_info.py | 49 ++- .../sglang/srt/model_executor/model_runner.py | 36 +- .../multimodal/processors/base_processor.py | 313 ++++++++++++++++-- .../srt/multimodal/processors/qwen_vl.py | 64 +++- 14 files changed, 1285 insertions(+), 115 deletions(-) diff --git a/python/sglang/bench_serving.py b/python/sglang/bench_serving.py index cebc645a86ed..e36ad3095164 100644 --- a/python/sglang/bench_serving.py +++ b/python/sglang/bench_serving.py @@ -315,23 +315,27 @@ async def async_request_openai_chat_completions( # image if request_func_input.image_data: - content_items.extend([ - { - "type": "image_url", - "image_url": {"url": img_url}, - } - for img_url in request_func_input.image_data - ]) + content_items.extend( + [ + { + "type": "image_url", + "image_url": {"url": img_url}, + } + for img_url in request_func_input.image_data + ] + ) # audio if request_func_input.audio_data: - content_items.extend([ - { - "type": "audio_url", - "audio_url": {"url": audio_url}, - } - for audio_url in request_func_input.audio_data - ]) + content_items.extend( + [ + { + "type": "audio_url", + "audio_url": {"url": audio_url}, + } + for audio_url in request_func_input.audio_data + ] + ) # text if request_func_input.prompt: @@ -741,8 +745,8 @@ def get_processor( pretrained_model_name_or_path = get_model(pretrained_model_name_or_path) # Disable truncation when loading the processor for a quantized model. - # Truncation affects the `create_mm_data_row` function’s ability to correctly - # count input tokens (text/vision) for bench serving, which may lead to + # Truncation affects the `create_mm_data_row` function’s ability to correctly + # count input tokens (text/vision) for bench serving, which may lead to # inaccurate token statistics. return AutoProcessor.from_pretrained( pretrained_model_name_or_path, @@ -1327,6 +1331,7 @@ def sample_random_requests( print(f"#Output tokens: {np.sum(output_lens)}") return input_requests + # Function from https://github.com/sgl-project/sglang/pull def encode_wav_data_url(y: np.ndarray, sr: int) -> str: import io @@ -1358,6 +1363,7 @@ def encode_wav_data_url(y: np.ndarray, sr: int) -> str: b64 = pybase64.b64encode(buf.getvalue()).decode("utf-8") return f"data:audio/wav;base64,{b64}" + def generate_dummy_audio(duration_s=10, sr=1600): """ Generate a float32 audio array of `duration_s` seconds at sampling rate `sr`. @@ -1369,10 +1375,11 @@ def generate_dummy_audio(duration_s=10, sr=1600): y = np.random.uniform(low=-1.0, high=1.0, size=num_samples).astype(np.float32) return y, sr + def build_no_speical_token_table(tokenizer, special_tokens=None): """ build no special token to replace special token randomly generated - for Qwen3-Omni + for Qwen3-Omni """ vocab = tokenizer.get_vocab().values() special_set = set(special_tokens or []) @@ -1388,6 +1395,7 @@ def build_no_speical_token_table(tokenizer, special_tokens=None): return np.array(safe_tokens, dtype=np.int32) + def sample_random_omni_requests( input_len: int, output_len: int, @@ -1397,15 +1405,15 @@ def sample_random_omni_requests( processor: AutoProcessor, tokenizer: PreTrainedTokenizerBase, return_text: bool = True, - skip_special_tokens: bool = True, + skip_special_tokens: bool = True, ) -> List[DatasetRow]: """ Omni request temporary limitation: ---------------------------------- Currently this benchmark only tests the (audio + text) modality. Audio is generated as a dummy waveform and paired - with the provided random text prompt. - `skip_speical_tokens` default true to avoid common mistakes made + with the provided random text prompt. + `skip_speical_tokens` default true to avoid common mistakes made in Qwen3-Omni's preprocessing """ input_lens = np.random.randint( @@ -1441,8 +1449,8 @@ def sample_random_omni_requests( for j in range(input_lens[i]): tid = (offsets[i] + i + j) % tokenizer.vocab_size if ( - no_special_tokens is not None - and special_tokens is not None + no_special_tokens is not None + and special_tokens is not None and tid in special_tokens ): idx = (tid + j) % len(no_special_tokens) @@ -1455,7 +1463,7 @@ def sample_random_omni_requests( if audio_length and audio_length > 0: y, sr = generate_dummy_audio(audio_length, 1600) data_url = encode_wav_data_url(y, sr) - audio_list = [data_url] + audio_list = [data_url] else: audio_list = None input_requests.append( @@ -1468,10 +1476,13 @@ def sample_random_omni_requests( ) print(f"#Input text tokens: {np.sum(input_lens)}") - print(f"#Total input audio lengths: {audio_length*num_prompts if audio_length and audio_length > 0 else 0} seconds") + print( + f"#Total input audio lengths: {audio_length*num_prompts if audio_length and audio_length > 0 else 0} seconds" + ) print(f"#Output tokens: {np.sum(output_lens)}") return input_requests + def parse_image_resolution(image_resolution: str) -> Tuple[int, int]: """Parse image resolution into (width, height). @@ -1641,11 +1652,11 @@ def _gen_random_image_data_uri( and processor.tokenizer.all_special_ids is not None ): special_tokens = set(processor.tokenizer.all_special_ids) - + # Extract special token strings and convert to token_id special_token_strings = [] special_token_ids = set() - + if hasattr(processor.tokenizer, "special_tokens"): # Extract all special tokens from tokenizer.special_tokens dictionary for key, value in processor.tokenizer.special_tokens.items(): @@ -1653,15 +1664,17 @@ def _gen_random_image_data_uri( special_token_strings.append(value) elif isinstance(value, list): special_token_strings.extend(value) - + # Also extract from additional_special_tokens if hasattr(processor.tokenizer, "additional_special_tokens"): special_token_strings.extend(processor.tokenizer.additional_special_tokens) - + # Convert to token_id if special_token_strings and hasattr(processor.tokenizer, "convert_tokens_to_ids"): - special_token_ids = set(processor.tokenizer.convert_tokens_to_ids(special_token_strings)) - + special_token_ids = set( + processor.tokenizer.convert_tokens_to_ids(special_token_strings) + ) + # Merge all_special_ids and special_token_ids converted from strings if special_tokens is not None and special_token_ids: merged_special_tokens = special_tokens | special_token_ids diff --git a/python/sglang/srt/managers/detokenizer_manager.py b/python/sglang/srt/managers/detokenizer_manager.py index 8db48f82d1fc..977960002de4 100644 --- a/python/sglang/srt/managers/detokenizer_manager.py +++ b/python/sglang/srt/managers/detokenizer_manager.py @@ -150,6 +150,10 @@ def handle_batch_embedding_out(self, recv_obj: BatchEmbeddingOutput): return recv_obj def handle_batch_token_id_out(self, recv_obj: BatchTokenIDOutput): + import time + + detokenizer_receive_time = time.time() + bs = len(recv_obj.rids) # Initialize decode status @@ -246,6 +250,19 @@ def handle_batch_token_id_out(self, recv_obj: BatchTokenIDOutput): s.sent_offset = len(output_str) output_strs.append(incremental_output) + # Record detokenizer send time and create timing lists + detokenizer_send_time = time.time() + detokenizer_receive_times = ( + [detokenizer_receive_time] * bs + if hasattr(recv_obj, "output_send_time") and recv_obj.output_send_time + else None + ) + detokenizer_send_times = ( + [detokenizer_send_time] * bs + if hasattr(recv_obj, "output_send_time") and recv_obj.output_send_time + else None + ) + return BatchStrOutput( rids=recv_obj.rids, http_worker_ipcs=recv_obj.http_worker_ipcs, @@ -274,6 +291,16 @@ def handle_batch_token_id_out(self, recv_obj: BatchTokenIDOutput): placeholder_tokens_idx=None, placeholder_tokens_val=None, token_steps=recv_obj.token_steps, + prefill_preparation_time=recv_obj.prefill_preparation_time, + vit_encoding_time=recv_obj.vit_encoding_time, + llm_prefill_time=recv_obj.llm_prefill_time, + total_prefill_time=recv_obj.total_prefill_time, + scheduler_receive_time=recv_obj.scheduler_receive_time, + model_call_time=recv_obj.model_call_time, + first_token_generated_time=recv_obj.first_token_generated_time, + output_send_time=recv_obj.output_send_time, + detokenizer_receive_time=detokenizer_receive_times, + detokenizer_send_time=detokenizer_send_times, ) def handle_multimodal_decode_req(self, recv_obj: BatchMultimodalDecodeReq): diff --git a/python/sglang/srt/managers/io_struct.py b/python/sglang/srt/managers/io_struct.py index d7671e3e8448..f825c86c9608 100644 --- a/python/sglang/srt/managers/io_struct.py +++ b/python/sglang/srt/managers/io_struct.py @@ -857,6 +857,18 @@ class BatchTokenIDOutput(BaseBatchReq): # The trainer step id. Used to know which step's weights are used for sampling. token_steps: List[List[int]] = None + # TTFT breakdown timing (in seconds) + prefill_preparation_time: List[float] = None + vit_encoding_time: List[float] = None + llm_prefill_time: List[float] = None + total_prefill_time: List[float] = None + + # Scheduler overhead timing (in seconds) + scheduler_receive_time: List[float] = None + model_call_time: List[float] = None + first_token_generated_time: List[float] = None + output_send_time: List[float] = None + @dataclass class BatchMultimodalDecodeReq(BaseBatchReq): @@ -933,6 +945,20 @@ class BatchStrOutput(BaseBatchReq): # The trainer step id. Used to know which step's weights are used for sampling. token_steps: List[List[int]] = None + # TTFT breakdown timing (in seconds) + prefill_preparation_time: List[float] = None + vit_encoding_time: List[float] = None + llm_prefill_time: List[float] = None + total_prefill_time: List[float] = None + + # Scheduler overhead timing (in seconds) + scheduler_receive_time: List[float] = None + model_call_time: List[float] = None + first_token_generated_time: List[float] = None + output_send_time: List[float] = None + detokenizer_receive_time: List[float] = None + detokenizer_send_time: List[float] = None + @dataclass class BatchMultimodalOutput(BaseBatchReq): diff --git a/python/sglang/srt/managers/mm_utils.py b/python/sglang/srt/managers/mm_utils.py index be2d5c145837..a0f53beaf313 100644 --- a/python/sglang/srt/managers/mm_utils.py +++ b/python/sglang/srt/managers/mm_utils.py @@ -2,7 +2,6 @@ Multi-modality utils """ -import os import pickle from abc import abstractmethod from typing import Any, Callable, Dict, List, Literal, Optional, Tuple @@ -25,11 +24,9 @@ from sglang.srt.utils import ( flatten_nested_list, is_cuda_alike, - is_hip, is_npu, print_warning_once, ) -from sglang.utils import logger _is_npu = is_npu() @@ -133,7 +130,7 @@ def __setstate__(self, state: Dict[str, Any]): try: target_device = torch.device(f"cuda:{source_device_index}") - + with torch.cuda.device(target_device): storage = torch.UntypedStorage._new_shared_cuda(*handle) reconstructed_tensor = torch.empty( @@ -385,9 +382,23 @@ def _get_chunked_prefill_embedding( extend_length: List[int], items_offset_list: List[List[Tuple[int, int]]], ) -> Tuple[Optional[torch.Tensor], int]: + import logging + import time + + logger = logging.getLogger(__name__) + + torch.cuda.synchronize() + overall_start = time.time() + # Calculate embedding for each request, try to get it from cache to avoid repeated calculation embedding_list = [] expected_token_count = 0 + cache_hit_count = 0 + cache_miss_count = 0 + total_cache_lookup_time = 0 + total_vit_inference_time = 0 + total_chunk_processing_time = 0 + # FIXME(Xinyuan): temporary workaround for eagle3, which may have len(items_size) > len(prefix_length) if not get_global_server_args().mm_enable_dp_encoder: max_iterations = min(len(items_size) - 1, len(prefix_length)) @@ -397,13 +408,32 @@ def _get_chunked_prefill_embedding( embedding_items_per_req = embedding_items[items_size[i] : items_size[i + 1]] items_offset = items_offset_list[i] assert items_offset is not None, items_offset + + # Cache lookup + torch.cuda.synchronize() + cache_lookup_start = time.time() embedding_items_hash = get_embedding_hash(embedding_items_per_req) # if all items has been prefixed, we do not need to calculate embedding if all([offset_end < prefix_length[i] for _, offset_end in items_offset]): continue embedding_per_req = embedding_cache.get(embedding_items_hash) + torch.cuda.synchronize() + cache_lookup_time = time.time() - cache_lookup_start + total_cache_lookup_time += cache_lookup_time + if embedding_per_req is None: + cache_miss_count += 1 + # ViT inference + vit_inference_start = time.time() + torch.cuda.synchronize() embedding_per_req = data_embedding_func(embedding_items_per_req) + torch.cuda.synchronize() + vit_inference_time = time.time() - vit_inference_start + total_vit_inference_time += vit_inference_time + logger.info( + f"[VIT_DETAIL] Step 2.1.1 - ViT inference (cache miss {cache_miss_count}): {vit_inference_time * 1000:.2f} ms" + ) + if not embedding_cache.put(embedding_items_hash, embedding_per_req): print_warning_once( "Multimodal embedding cache is full. This typically occurs when a single " @@ -411,39 +441,85 @@ def _get_chunked_prefill_embedding( "`SGLANG_VLM_CACHE_SIZE_MB` environment variable or reducing the input " "embedding size." ) + else: + cache_hit_count += 1 + logger.info( + f"[VIT_DETAIL] Step 2.1.1 - ViT inference (cache hit {cache_hit_count}): 0.00 ms (cached)" + ) + # Chunk processing + torch.cuda.synchronize() + chunk_start = time.time() embedding_per_req_chunk, start_idx, end_idx = get_embedding_chunk( embedding=embedding_per_req, extend_prefix_len=prefix_length[i], extend_seq_len=extend_length[i] if i < len(extend_length) else 0, items_offset=items_offset, ) + torch.cuda.synchronize() + chunk_time = time.time() - chunk_start + total_chunk_processing_time += chunk_time + embedding_list.append(embedding_per_req_chunk) expected_token_count += end_idx - start_idx else: + # Hash computation + torch.cuda.synchronize() + hash_start = time.time() embedding_items_hash_list = get_embedding_hash_list(embedding_items) + torch.cuda.synchronize() + hash_time = (time.time() - hash_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 2.1.0 - Hash computation: {hash_time:.2f} ms") embedding_items_uncached = [] embedding_items_hash_list_uncached = [] embedding_items_feature_num = [] + + # Cache lookup + torch.cuda.synchronize() + cache_lookup_start = time.time() for i in range(len(embedding_items)): embedding_per_req = embedding_cache.get(embedding_items_hash_list[i]) embedding_list.append(embedding_per_req) if embedding_per_req is None: + cache_miss_count += 1 embedding_items_uncached.append(embedding_items[i]) if embedding_items[i].modality == Modality.AUDIO: - embedding_items_feature_num.append(1) # audio always has 1 object + embedding_items_feature_num.append(1) # audio always has 1 object elif embedding_items[i].modality == Modality.IMAGE: embedding_items_feature_num.append( embedding_items[i].image_grid_thw.shape[0] ) else: # TODO: Handle the case where embedding_items[i].modality is VIDEO. - print_warning_once(f"Cannot handel type { embedding_items[i].modality}") + print_warning_once( + f"Cannot handle type { embedding_items[i].modality}" + ) embedding_items_hash_list_uncached.append(embedding_items_hash_list[i]) + else: + cache_hit_count += 1 + torch.cuda.synchronize() + cache_lookup_time = time.time() - cache_lookup_start + total_cache_lookup_time = cache_lookup_time + logger.info( + f"[VIT_DETAIL] Step 2.1.1 - Cache lookup: {cache_lookup_time * 1000:.2f} ms (hits: {cache_hit_count}, misses: {cache_miss_count})" + ) if None in embedding_list: + # ViT inference for uncached items + vit_inference_start = time.time() + torch.cuda.synchronize() embeddings = data_embedding_func(embedding_items_uncached) + torch.cuda.synchronize() + vit_inference_time = time.time() - vit_inference_start + total_vit_inference_time = vit_inference_time + logger.info( + f"[VIT_DETAIL] Step 2.1.2 - ViT inference (batch {len(embedding_items_uncached)} items): {vit_inference_time * 1000:.2f} ms" + ) + + # Merge embeddings + torch.cuda.synchronize() + merge_start = time.time() embeddings_merged = [] embeddings_idx = 0 for feature_num in embedding_items_feature_num: @@ -452,14 +528,25 @@ def _get_chunked_prefill_embedding( elif embedding_items[i].modality == Modality.IMAGE: embeddings_merged.append( torch.cat( - embeddings[embeddings_idx : embeddings_idx + feature_num], dim=0 + embeddings[embeddings_idx : embeddings_idx + feature_num], + dim=0, ) ) else: # TODO: Handle the case where embedding_items[i].modality is VIDEO. - print_warning_once(f"Cannot handel type { embedding_items[i].modality}") + print_warning_once( + f"Cannot handle type { embedding_items[i].modality}" + ) embeddings_idx += feature_num + torch.cuda.synchronize() + merge_time = (time.time() - merge_start) * 1000 + logger.info( + f"[VIT_DETAIL] Step 2.1.3 - Merge embeddings: {merge_time:.2f} ms" + ) + # Cache put + torch.cuda.synchronize() + cache_put_start = time.time() for embedding_items_hash, embedding_per_req in zip( embedding_items_hash_list_uncached, embeddings_merged ): @@ -470,12 +557,29 @@ def _get_chunked_prefill_embedding( "`SGLANG_VLM_CACHE_SIZE_MB` environment variable or reducing the input " "embedding size." ) + torch.cuda.synchronize() + cache_put_time = (time.time() - cache_put_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 2.1.4 - Cache put: {cache_put_time:.2f} ms") + else: + logger.info( + f"[VIT_DETAIL] Step 2.1.2 - ViT inference: 0.00 ms (all cached)" + ) + + # Fill in cached embeddings + torch.cuda.synchronize() + fill_start = time.time() embeddings_merged_idx = 0 for i in range(len(embedding_list)): if embedding_list[i] is None: embedding_list[i] = embeddings_merged[embeddings_merged_idx] embeddings_merged_idx += 1 + torch.cuda.synchronize() + fill_time = (time.time() - fill_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 2.1.5 - Fill embeddings: {fill_time:.2f} ms") + # Chunk processing + torch.cuda.synchronize() + chunk_start = time.time() for i in range(len(embedding_items)): embedding_per_req_chunk, start_idx, end_idx = get_embedding_chunk( embedding=embedding_list[i], @@ -485,9 +589,33 @@ def _get_chunked_prefill_embedding( ) embedding_list[i] = embedding_per_req_chunk expected_token_count += end_idx - start_idx + torch.cuda.synchronize() + chunk_time = time.time() - chunk_start + total_chunk_processing_time = chunk_time + logger.info( + f"[VIT_DETAIL] Step 2.1.6 - Chunk processing: {chunk_time * 1000:.2f} ms" + ) + if len(embedding_list) == 0: return None - return torch.concat(embedding_list, dim=0), expected_token_count + + torch.cuda.synchronize() + concat_start = time.time() + result = torch.concat(embedding_list, dim=0) + torch.cuda.synchronize() + concat_time = (time.time() - concat_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 2.1.7 - Concat embeddings: {concat_time:.2f} ms") + + torch.cuda.synchronize() + overall_duration = (time.time() - overall_start) * 1000 + logger.info( + f"[VIT_DETAIL] === _get_chunked_prefill_embedding TOTAL: {overall_duration:.2f} ms ===" + ) + logger.info( + f"[VIT_DETAIL] Summary - Cache hits: {cache_hit_count}, Cache misses: {cache_miss_count}, Total ViT inference time: {total_vit_inference_time * 1000:.2f} ms" + ) + + return result, expected_token_count def _get_multimodal_mask( @@ -557,7 +685,17 @@ def get_embedding_and_mask( - The generated embeddings tensor - A boolean mask tensor indicating where these embeddings should be placed """ + import logging + import time + + logger = logging.getLogger(__name__) + + torch.cuda.synchronize() + overall_start = time.time() + # 1. Get embedding + torch.cuda.synchronize() + step1_start = time.time() embedding = _get_precomputed_embedding(embedding_items) if embedding is None: embedding, expected_token_count = _get_chunked_prefill_embedding( @@ -570,14 +708,41 @@ def get_embedding_and_mask( ) if embedding is None: return None, None + torch.cuda.synchronize() + step1_duration = (time.time() - step1_start) * 1000 + logger.info( + f"[VIT_DETAIL] Step 2.1 - Get embeddings (cache or compute): {step1_duration:.2f} ms" + ) + # 2. Get mask + torch.cuda.synchronize() + step2_start = time.time() if _is_npu: torch.npu.current_stream().synchronize() special_multimodal_mask = _get_multimodal_mask(input_ids, placeholder_tensor) + torch.cuda.synchronize() + step2_duration = (time.time() - step2_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 2.2 - Get multimodal mask: {step2_duration:.2f} ms") + # 3. Adjust embedding length if needed + torch.cuda.synchronize() + step3_start = time.time() embedding = _adjust_embedding_length( embedding, special_multimodal_mask, logger, expected_token_count ) + torch.cuda.synchronize() + step3_duration = (time.time() - step3_start) * 1000 + logger.info( + f"[VIT_DETAIL] Step 2.3 - Adjust embedding length: {step3_duration:.2f} ms" + ) + + torch.cuda.synchronize() + total_steps = step1_duration + step2_duration + step3_duration + overall_duration = (time.time() - overall_start) * 1000 + logger.info( + f"[VIT_DETAIL] === get_embedding_and_mask TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===" + ) + return embedding, special_multimodal_mask @@ -608,21 +773,36 @@ def embed_mm_inputs( Returns: Combined embedding tensor with multimodal content integrated """ + import logging + import time + + logger = logging.getLogger(__name__) + + torch.cuda.synchronize() + overall_start = time.time() + other_info = {} if mm_inputs_list is None: return None # 1. Calculate the multimodal data which exists in input_ids, with the help of pad_values # we assume that multimodal data are represented with its pad_values in input_ids + torch.cuda.synchronize() + step1_start = time.time() item_flatten_list = [] for mm_inputs in mm_inputs_list: item_flatten_list += [item for item in mm_inputs.mm_items if item is not None] + torch.cuda.synchronize() + step1_duration = (time.time() - step1_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 1 - Flatten MM items: {step1_duration:.2f} ms") # deepstack_embeddings: per-modality modalities, embeddings, masks, deepstack_embeddings = [], [], [], [] # 2. Get multimodal embedding separately # Try get mm embedding if any + torch.cuda.synchronize() + step2_start = time.time() for modality in Modality.all(): items = [ item for item in item_flatten_list if item.is_modality(modality=modality) @@ -676,8 +856,13 @@ def embed_mm_inputs( modalities += [modality] embeddings += [embedding] masks += [mask] + torch.cuda.synchronize() + step2_duration = (time.time() - step2_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 2 - Get MM embeddings: {step2_duration:.2f} ms") # 3. Get input embeddings + torch.cuda.synchronize() + step3_start = time.time() vocab_size = input_embedding.num_embeddings # Important: clamp after getting original multimodal regions # Clamp input ids. This is because the input_ids for the multimodal tokens are @@ -701,9 +886,14 @@ def embed_mm_inputs( ) other_info["input_deepstack_embeds"] = input_deepstack_embeds + torch.cuda.synchronize() + step3_duration = (time.time() - step3_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 3 - Get text embeddings: {step3_duration:.2f} ms") # 4. scatter embeddings into input embedding # Use masked_scatter_ to completely avoid D2D synchronization + torch.cuda.synchronize() + step4_start = time.time() for i, modality, embedding, mask in zip( range(len(embeddings)), modalities, embeddings, masks ): @@ -733,6 +923,16 @@ def embed_mm_inputs( inputs_embeds.device, inputs_embeds.dtype ) input_deepstack_embeds.masked_scatter_(mask_1d.unsqueeze(-1), deepstack_emb) + torch.cuda.synchronize() + step4_duration = (time.time() - step4_start) * 1000 + logger.info(f"[VIT_DETAIL] Step 4 - Scatter embeddings: {step4_duration:.2f} ms") + + torch.cuda.synchronize() + total_steps = step1_duration + step2_duration + step3_duration + step4_duration + overall_duration = (time.time() - overall_start) * 1000 + logger.info( + f"[VIT_DETAIL] === embed_mm_inputs TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===" + ) return inputs_embeds, other_info @@ -764,6 +964,11 @@ def general_mm_embed_routine( Returns: Hidden states from language model forward pass """ + import logging + import time + + logger = logging.getLogger(__name__) + assert hasattr(language_model, "get_input_embeddings") embed_tokens = language_model.get_input_embeddings() if ( @@ -784,6 +989,44 @@ def general_mm_embed_routine( for i, seq_len in enumerate(forward_batch.extend_seq_lens_cpu) if forward_batch.mm_inputs[i] is not None ] + + # Synchronize before starting timing to ensure all previous GPU operations are complete + torch.cuda.synchronize() + vit_start_time = time.time() + + # Check how many encoders are cached vs need computation + num_cached = 0 + num_to_encode = 0 + has_encoder_cached_attr = hasattr(forward_batch, "encoder_cached") + print( + f"[VIT_DEBUG] forward_batch has encoder_cached: {has_encoder_cached_attr}", + flush=True, + ) + + if has_encoder_cached_attr and forward_batch.encoder_cached: + print( + f"[VIT_DEBUG] encoder_cached values: {forward_batch.encoder_cached}", + flush=True, + ) + for i, mm_input in enumerate(forward_batch.mm_inputs): + if mm_input is not None: + if ( + i < len(forward_batch.encoder_cached) + and forward_batch.encoder_cached[i] + ): + num_cached += 1 + else: + num_to_encode += 1 + else: + # For models like Qwen3-VL that don't use encoder_cached + num_mm_inputs = sum( + 1 for mm_input in forward_batch.mm_inputs if mm_input is not None + ) + print( + f"[VIT_DEBUG] No encoder_cached, counting mm_inputs: {num_mm_inputs}", + flush=True, + ) + inputs_embeds, other_info = embed_mm_inputs( mm_inputs_list=mm_inputs_list, extend_prefix_lens=extend_prefix_lens, @@ -795,6 +1038,25 @@ def general_mm_embed_routine( placeholder_tokens=placeholder_tokens, use_deepstack=use_deepstack, ) + + # Synchronize after to ensure ViT encoding is complete before recording end time + torch.cuda.synchronize() + vit_end_time = time.time() + vit_encoding_time = vit_end_time - vit_start_time + + # Log with cache information + if num_cached > 0 or num_to_encode > 0: + msg = f"[TTFT_BREAKDOWN] ViT encoding + embedding time: {vit_encoding_time * 1000:.2f} ms (cached: {num_cached}, encoded: {num_to_encode})" + logger.info(msg) + print(msg, flush=True) + else: + msg = f"[TTFT_BREAKDOWN] ViT encoding + embedding time: {vit_encoding_time * 1000:.2f} ms" + logger.info(msg) + print(msg, flush=True) + + # Store timing in forward_batch + forward_batch.vit_encoding_time = vit_encoding_time + # add for qwen3_vl deepstack if use_deepstack: kwargs["input_deepstack_embeds"] = other_info["input_deepstack_embeds"] @@ -804,12 +1066,28 @@ def general_mm_embed_routine( else: inputs_embeds = embed_tokens(input_ids) + # Synchronize before LLM forward timing + torch.cuda.synchronize() + llm_prefill_start_time = time.time() + hidden_states = language_model( input_ids=None, forward_batch=forward_batch, input_embeds=inputs_embeds, **kwargs, ) + + # Synchronize after LLM forward to ensure completion + torch.cuda.synchronize() + llm_prefill_end_time = time.time() + llm_prefill_time = llm_prefill_end_time - llm_prefill_start_time + logger.info( + f"[TTFT_BREAKDOWN] LLM prefill forward time: {llm_prefill_time * 1000:.2f} ms" + ) + + # Store timing in forward_batch + forward_batch.llm_prefill_time = llm_prefill_time + return hidden_states diff --git a/python/sglang/srt/managers/schedule_batch.py b/python/sglang/srt/managers/schedule_batch.py index 0be57a6c40fd..34a83dad36ab 100644 --- a/python/sglang/srt/managers/schedule_batch.py +++ b/python/sglang/srt/managers/schedule_batch.py @@ -667,6 +667,22 @@ def __init__( # For metrics self.metrics_collector = metrics_collector self.time_stats: TimeStats = TimeStats(disagg_mode=disagg_mode) + + # TTFT breakdown timing (in seconds) + self.prefill_preparation_time: float = 0.0 + self.vit_encoding_time: float = 0.0 + self.llm_prefill_time: float = 0.0 + self.total_prefill_time: float = 0.0 + + # Additional timing for scheduler overhead + self.scheduler_receive_time: float = 0.0 # When scheduler receives this request + self.batch_prepare_time: float = 0.0 # When batch preparation is complete + self.model_call_time: float = 0.0 # When model_worker.forward is called + self.first_token_generated_time: float = ( + 0.0 # When first token is generated in scheduler + ) + self.output_send_time: float = 0.0 # When scheduler sends output to detokenizer + self.has_log_time_stats: bool = False self.last_tic = time.monotonic() @@ -1127,6 +1143,10 @@ def is_empty(self): return len(self.reqs) == 0 def prepare_encoder_info_extend(self, input_ids: List[int], seq_lens: List[int]): + import logging + + logger = logging.getLogger(__name__) + self.encoder_lens_cpu = [] self.encoder_cached = [] @@ -1138,10 +1158,21 @@ def prepare_encoder_info_extend(self, input_ids: List[int], seq_lens: List[int]) self.encoder_cached.append(True) else: self.encoder_lens_cpu.append(im.num_image_tokens) - self.encoder_cached.append( + is_cached = ( self.forward_mode.is_decode() or len(req.prefix_indices) >= im.num_image_tokens ) + self.encoder_cached.append(is_cached) + + # Log cache hit/miss for TTFT analysis + if is_cached: + msg = f"[ENCODER_CACHE] Req {req.rid}: ✓ Cache HIT - Skipping ViT encoding ({im.num_image_tokens} tokens cached, prefix_len={len(req.prefix_indices)})" + logger.info(msg) + print(msg, flush=True) + else: + msg = f"[ENCODER_CACHE] Req {req.rid}: ✗ Cache MISS - Will run ViT encoding ({im.num_image_tokens} tokens needed, prefix_len={len(req.prefix_indices)})" + logger.info(msg) + print(msg, flush=True) self.encoder_lens = torch.tensor(self.encoder_lens_cpu, dtype=torch.int64).to( self.device, non_blocking=True @@ -1364,7 +1395,7 @@ def prepare_for_extend(self): mm_item.feature = pixel_values.reconstruct_on_target_device( torch.cuda.current_device() ) - + self.multimodal_inputs = multimodal_inputs self.token_type_ids = token_type_ids_tensor self.seq_lens_sum = sum(seq_lens) diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index ff48f4dce7fa..9ba1e6a180cd 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -1070,6 +1070,15 @@ def event_loop_normal(self): if batch: result = self.run_batch(batch) + # Record model completion time for TTFT tracking (right after run_batch returns) + # Only for prefill/extend batches + if batch.forward_mode.is_extend(): + import time + + model_complete_time = time.time() + for req in batch.reqs: + if req.first_token_generated_time == 0.0: + req.first_token_generated_time = model_complete_time self.process_batch_result(batch, result) else: # When the server is idle, do self-check and re-init some states @@ -1091,12 +1100,44 @@ def event_loop_overlap(self): batch_result = None if batch: + import time + + start = time.time() batch_result = self.run_batch(batch) + end = time.time() + logger.info( + f"[TTFT_BREAKDOWN] run_batch time: {(end - start) * 1000:.2f} ms" + ) + # Record model completion time for TTFT tracking (right after run_batch returns) + # Only for prefill/extend batches + if batch.forward_mode.is_extend(): + import time + + model_complete_time = time.time() + for req in batch.reqs: + if req.first_token_generated_time == 0.0: + req.first_token_generated_time = model_complete_time self.result_queue.append((batch.copy(), batch_result)) if self.last_batch: # Process the results of the last batch + import time + + before_popleft = time.time() tmp_batch, tmp_result = self.result_queue.popleft() + + # Log the delay from when this batch was completed to when we start processing it + if tmp_batch.forward_mode.is_extend() and tmp_batch.reqs: + for req in tmp_batch.reqs: + if req.first_token_generated_time > 0: + delay_time = ( + before_popleft - req.first_token_generated_time + ) * 1000 + logger.info( + f"[OUTPUT_PROC_GAP] Req {req.rid}: Delay from token generation to start processing: {delay_time:.2f} ms" + ) + break + self.process_batch_result(tmp_batch, tmp_result) elif batch is None: # When the server is idle, do self-check and re-init some states @@ -1465,6 +1506,8 @@ def _add_request_to_queue(self, req: Req, is_retracted: bool = False): self._prefetch_kvcache(req) self.waiting_queue.append(req) req.time_stats.wait_queue_entry_time = time.perf_counter() + # Record scheduler receive time for TTFT breakdown + req.scheduler_receive_time = time.time() trace_slice_end("process req", req.rid, auto_next_anon=True) elif self.disaggregation_mode == DisaggregationMode.PREFILL: self._prefetch_kvcache(req) @@ -1975,6 +2018,12 @@ def run_batch( """Run a batch.""" self.forward_ct += 1 + # Record model call time for TTFT breakdown + model_call_start = time.time() + for req in batch.reqs: + if req.model_call_time == 0.0: # Only set once for first forward + req.model_call_time = model_call_start + # Whether to run the profiler self._profile_batch_predicate(batch) if self.forward_sleep_time is not None: @@ -2072,6 +2121,18 @@ def run_batch( batch_result.extend_logprob_start_len_per_req = ( extend_logprob_start_len_per_req ) + + # Fill in model forward timing from batch_result (measured in model_runner) + # Only for extend/prefill batches + if batch.forward_mode.is_extend() and hasattr( + batch_result, "total_model_forward_time" + ): + for req in batch.reqs: + req.prefill_preparation_time = batch_result.prefill_preparation_time + req.vit_encoding_time = batch_result.vit_encoding_time + req.llm_prefill_time = batch_result.llm_prefill_time + req.total_prefill_time = batch_result.total_model_forward_time + return batch_result else: # embedding or reward model model_worker_batch = batch.get_model_worker_batch() @@ -2099,20 +2160,47 @@ def process_batch_result( batch: ScheduleBatch, result: Union[GenerationBatchResult, EmbeddingBatchResult], ): + import time + + process_batch_result_start = time.time() + if batch.forward_mode.is_decode(): self.process_batch_result_decode(batch, result) if self.enable_trace: trace_slice_batch("decode loop", batch.reqs) elif batch.forward_mode.is_extend(): + prefill_start = time.time() self.process_batch_result_prefill(batch, result) + prefill_duration = (time.time() - prefill_start) * 1000 if self.enable_trace: trace_slice_batch("prefill", batch.reqs) + # Log the time from first_token_generated to now for TTFT requests + for req in batch.reqs: + if req.first_token_generated_time > 0: + time_since_generation = ( + time.time() - req.first_token_generated_time + ) * 1000 + logger.info( + f"[OUTPUT_PROC_GAP] Req {req.rid}: From token generation to end of process_batch_result: {time_since_generation:.2f} ms" + ) + logger.info( + f"[OUTPUT_PROC_GAP] Req {req.rid}: process_batch_result_prefill took: {prefill_duration:.2f} ms" + ) + break # Only log once per batch + elif batch.forward_mode.is_idle(): if self.enable_overlap: if result.copy_done is not None: result.copy_done.synchronize() + process_batch_result_duration = ( + time.time() - process_batch_result_start + ) * 1000 + if batch.forward_mode.is_extend(): + logger.info( + f"[OUTPUT_PROC_GAP] process_batch_result total: {process_batch_result_duration:.2f} ms" + ) self.maybe_send_health_check_signal() @@ -2822,8 +2910,6 @@ def run_scheduler_process( # Configure the logger (must be called before logging) configure_logger(server_args, prefix=prefix) suppress_other_loggers() - - # Set cpu affinity to this gpu process if get_bool_env_var("SGLANG_SET_CPU_AFFINITY"): diff --git a/python/sglang/srt/managers/scheduler_output_processor_mixin.py b/python/sglang/srt/managers/scheduler_output_processor_mixin.py index e06fac95aea1..5d0b9012fcf5 100644 --- a/python/sglang/srt/managers/scheduler_output_processor_mixin.py +++ b/python/sglang/srt/managers/scheduler_output_processor_mixin.py @@ -41,9 +41,15 @@ def process_batch_result_prefill( batch: ScheduleBatch, result: Union[GenerationBatchResult, EmbeddingBatchResult], ): + import time + + overall_start = time.time() + skip_stream_req = None if self.is_generation: + # Step 1: GPU synchronization and data extraction + step1_start = time.time() if result.copy_done is not None: result.copy_done.synchronize() @@ -58,8 +64,11 @@ def process_batch_result_prefill( result.extend_input_len_per_req, result.extend_logprob_start_len_per_req, ) + step1_duration = (time.time() - step1_start) * 1000 + # logger.info(f"[OUTPUT_PROC_DETAIL] Step 1 - GPU sync & data extract: {step1_duration:.2f} ms") - # Move next_token_ids and logprobs to cpu + # Step 2: GPU to CPU transfer + step2_start = time.time() next_token_ids = next_token_ids.tolist() if batch.return_logprob: if logits_output.next_token_logprobs is not None: @@ -70,10 +79,12 @@ def process_batch_result_prefill( logits_output.input_token_logprobs = tuple( logits_output.input_token_logprobs.tolist() ) + step2_duration = (time.time() - step2_start) * 1000 + # logger.info(f"[OUTPUT_PROC_DETAIL] Step 2 - GPU->CPU transfer: {step2_duration:.2f} ms") + # Step 3: Process each request + step3_start = time.time() hidden_state_offset = 0 - - # Check finish conditions logprob_pt = 0 for i, (req, next_token_id) in enumerate(zip(batch.reqs, next_token_ids)): @@ -188,6 +199,9 @@ def process_batch_result_prefill( ) logprob_pt += num_input_logprobs + step3_duration = (time.time() - step3_start) * 1000 + # logger.info(f"[OUTPUT_PROC_DETAIL] Step 3 - Process requests: {step3_duration:.2f} ms") + else: # embedding or reward model is_sparse = envs.SGLANG_EMBEDDINGS_SPARSE_HEAD.is_set() @@ -224,7 +238,20 @@ def process_batch_result_prefill( # being chunked reqs' prefill is not finished req.is_chunked -= 1 + overall_duration = (time.time() - overall_start) * 1000 + # Calculate sum of all steps + if self.is_generation: + total_steps = step1_duration + step2_duration + step3_duration + # logger.info(f"[OUTPUT_PROC_DETAIL] === process_batch_result_prefill TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===") + # else: + # logger.info(f"[OUTPUT_PROC_DETAIL] === process_batch_result_prefill TOTAL: {overall_duration:.2f} ms ===") + + stream_output_call_start = time.time() self.stream_output(batch.reqs, batch.return_logprob, skip_stream_req) + stream_output_call_duration = (time.time() - stream_output_call_start) * 1000 + logger.info( + f"[OUTPUT_PROC_GAP] stream_output() call took: {stream_output_call_duration:.2f} ms" + ) def _resolve_spec_overlap_token_ids( self: Scheduler, result: GenerationBatchResult, batch: ScheduleBatch @@ -710,6 +737,12 @@ def stream_output_generation( return_logprob: bool, skip_req: Optional[Req] = None, ): + import time + + overall_start = time.time() + + # Step 1: Collect request data + step1_start = time.time() rids = [] http_worker_ipcs = [] finished_reasons: List[BaseFinishReason] = [] @@ -912,9 +945,38 @@ def stream_output_generation( ): req.log_time_stats() + step1_duration = (time.time() - step1_start) * 1000 + # logger.info(f"[OUTPUT_PROC_DETAIL] Step 4 - Collect request data: {step1_duration:.2f} ms") + + # Step 2: Prepare output structure + step2_start = time.time() + output_send_time = time.time() + + prefill_preparation_times = [] + vit_encoding_times = [] + llm_prefill_times = [] + total_prefill_times = [] + scheduler_receive_times = [] + model_call_times = [] + first_token_generated_times = [] + output_send_times = [] + for req in reqs: + if req.stream: + prefill_preparation_times.append(req.prefill_preparation_time) + vit_encoding_times.append(req.vit_encoding_time) + llm_prefill_times.append(req.llm_prefill_time) + total_prefill_times.append(req.total_prefill_time) + scheduler_receive_times.append(req.scheduler_receive_time) + model_call_times.append(req.model_call_time) + first_token_generated_times.append(req.first_token_generated_time) + output_send_times.append(output_send_time) + # Send to detokenizer if rids: if self.model_config.is_multimodal_gen: + overall_duration = (time.time() - overall_start) * 1000 + total_steps = step1_duration + # logger.info(f"[OUTPUT_PROC_DETAIL] === stream_output_generation TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===") return self.send_to_detokenizer.send_output( @@ -950,8 +1012,36 @@ def stream_output_generation( http_worker_ipcs=http_worker_ipcs, placeholder_tokens_idx=None, placeholder_tokens_val=None, + prefill_preparation_time=( + prefill_preparation_times if prefill_preparation_times else None + ), + vit_encoding_time=( + vit_encoding_times if vit_encoding_times else None + ), + llm_prefill_time=llm_prefill_times if llm_prefill_times else None, + total_prefill_time=( + total_prefill_times if total_prefill_times else None + ), + scheduler_receive_time=( + scheduler_receive_times if scheduler_receive_times else None + ), + model_call_time=model_call_times if model_call_times else None, + first_token_generated_time=( + first_token_generated_times + if first_token_generated_times + else None + ), + output_send_time=output_send_times if output_send_times else None, ) ) + step2_duration = (time.time() - step2_start) * 1000 + # logger.info(f"[OUTPUT_PROC_DETAIL] Step 5 - Send to detokenizer: {step2_duration:.2f} ms") + total_steps = step1_duration + step2_duration + else: + total_steps = step1_duration + + overall_duration = (time.time() - overall_start) * 1000 + # logger.info(f"[OUTPUT_PROC_DETAIL] === stream_output_generation TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===") def stream_output_embedding(self: Scheduler, reqs: List[Req]): rids = [] diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index a18240c3a17d..cb833f3cdbb4 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -136,6 +136,22 @@ class ReqState: last_time: float = 0.0 last_completion_tokens: int = 1 + # For TTFT breakdown timing + text_tokenization_time: float = 0.0 + mm_preprocessing_time: float = 0.0 + prefill_preparation_time: float = 0.0 + vit_encoding_time: float = 0.0 + llm_prefill_time: float = 0.0 + total_prefill_time: float = 0.0 + + # For scheduler overhead timing + scheduler_receive_time: float = 0.0 + model_call_time: float = 0.0 + first_token_generated_time: float = 0.0 + output_send_time: float = 0.0 + detokenizer_receive_time: float = 0.0 + detokenizer_send_time: float = 0.0 + # For streaming output last_output_offset: int = 0 @@ -429,6 +445,9 @@ async def generate_request( request: Optional[fastapi.Request] = None, ): created_time = time.time() + logger.info( + f"[TTFT_MEASUREMENT] Request {obj.rid} received at time: {created_time:.6f}" + ) self.auto_create_handle_loop() obj.normalize_batch_and_arguments() @@ -455,6 +474,11 @@ async def generate_request( if obj.is_single: tokenized_obj = await self._tokenize_one_request(obj) state = self._send_one_request(obj, tokenized_obj, created_time) + # Copy timing info from obj to state + state.text_tokenization_time = getattr( + obj, "text_tokenization_time", 0.0 + ) + state.mm_preprocessing_time = getattr(obj, "mm_preprocessing_time", 0.0) async for response in self._wait_one_response(obj, state, request): yield response else: @@ -562,10 +586,17 @@ async def _tokenize_texts( if not texts or self.tokenizer is None: raise ValueError("texts cannot be empty and tokenizer must be initialized") + overall_start = time.time() + # Step 1: Detect input format and prepare for tokenization + step1_start = time.time() input_format = self._detect_input_format(texts, is_cross_encoder) tokenizer_input = self._prepare_tokenizer_input(texts, input_format) original_batch_size = len(texts) if not isinstance(texts, str) else 1 + step1_duration = (time.time() - step1_start) * 1000 + logger.info( + f"[TEXT_TOK_DETAIL] Step 1 - Input format detection: {step1_duration:.2f} ms" + ) # Step 2: Set up tokenizer arguments tokenizer_kwargs = ( @@ -578,6 +609,7 @@ async def _tokenize_texts( and input_format == "single_string" ) + step3_start = time.time() if use_async_tokenizer: logger.debug("Using async dynamic batch tokenizer for single text") result = await self.async_dynamic_batch_tokenizer.encode( @@ -592,20 +624,40 @@ async def _tokenize_texts( ) else: logger.debug(f"Using regular tokenizer for {len(tokenizer_input)} inputs") + logger.info(f"[TEXT_TOK_DETAIL] tokenizer_input: {tokenizer_input}") encoded = self.tokenizer(tokenizer_input, **tokenizer_kwargs) + logger.info(f"[TEXT_TOK_DETAIL] encoded: {encoded}") input_ids = encoded["input_ids"] token_type_ids = encoded.get("token_type_ids") if is_cross_encoder else None + step3_duration = (time.time() - step3_start) * 1000 + logger.info( + f"[TEXT_TOK_DETAIL] Step 3 - Tokenization execution: {step3_duration:.2f} ms" + ) # Step 4: Extract results based on input format - return self._extract_tokenizer_results( + step4_start = time.time() + result = self._extract_tokenizer_results( input_ids, token_type_ids, input_format, original_batch_size ) + step4_duration = (time.time() - step4_start) * 1000 + logger.info( + f"[TEXT_TOK_DETAIL] Step 4 - Result extraction: {step4_duration:.2f} ms" + ) + + overall_duration = (time.time() - overall_start) * 1000 + total_steps = step1_duration + step3_duration + step4_duration + logger.info( + f"[TEXT_TOK_DETAIL] === _tokenize_texts TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===" + ) + + return result async def _tokenize_one_request( self, obj: Union[GenerateReqInput, EmbeddingReqInput], ): """Tokenize one request.""" + tokenize_start_time = time.time() # Tokenize input_embeds = None input_text = obj.text @@ -632,22 +684,42 @@ async def _tokenize_one_request( "the engine with skip_tokenizer_init=False." ) + # Skip first text tokenization for multimodal requests + # The multimodal processor will do the tokenization with proper placeholders skip_first_tokenize = self.mm_processor and obj.contains_mm_input() + if skip_first_tokenize: # For multimodal requests, skip the first tokenization # The input_ids will be generated by mm_data_processor later input_ids = None + text_tokenization_duration = 0.0 + logger.info( + f"[TTFT_BREAKDOWN] Req {obj.rid}: Text tokenization time: 0.00 ms (skipped for multimodal)" + ) + obj.text_tokenization_time = text_tokenization_duration else: # For text-only requests, perform tokenization input_ids, token_type_ids = await self._tokenize_texts( input_text, is_cross_encoder_request ) + tokenize_text_time = time.time() + text_tokenization_duration = tokenize_text_time - tokenize_start_time + logger.info( + f"[TTFT_BREAKDOWN] Req {obj.rid}: Text tokenization time: {text_tokenization_duration * 1000:.2f} ms" + ) + # Store in obj for later aggregation + obj.text_tokenization_time = text_tokenization_duration if self.mm_processor and obj.contains_mm_input(): + mm_process_start_time = time.time() if obj.image_data is not None and not isinstance(obj.image_data, list): obj.image_data = [obj.image_data] if obj.audio_data is not None and not isinstance(obj.audio_data, list): obj.audio_data = [obj.audio_data] + + # logger.info(f"[TEXT_TOK_DETAIL] input_text: {input_text}") + # logger.info(f"[TEXT_TOK_DETAIL] input_ids: {input_ids}") + # logger.info(f"[TEXT_TOK_DETAIL] input_text or input_ids: {input_text or input_ids}") mm_inputs: Dict = await self.mm_data_processor.process( image_data=obj.image_data, audio_data=obj.audio_data, @@ -655,8 +727,21 @@ async def _tokenize_one_request( request_obj=obj, max_req_input_len=self.max_req_input_len, ) + mm_process_end_time = time.time() + mm_preprocessing_duration = mm_process_end_time - mm_process_start_time + logger.info( + f"[TTFT_BREAKDOWN] Req {obj.rid}: Multimodal preprocessing time: {mm_preprocessing_duration * 1000:.2f} ms" + ) + # Store in obj for later aggregation + obj.mm_preprocessing_time = mm_preprocessing_duration if mm_inputs and "input_ids" in mm_inputs: input_ids = mm_inputs["input_ids"] + # elif input_ids is None: + # # Fallback: if multimodal processing didn't return input_ids, do tokenization now + # logger.warning(f"[TTFT_BREAKDOWN] Req {obj.rid}: Multimodal processing didn't return input_ids, performing fallback tokenization") + # input_ids, token_type_ids = await self._tokenize_texts( + # input_text, is_cross_encoder_request + # ) else: mm_inputs = None @@ -1406,6 +1491,61 @@ def _handle_batch_output( ) continue + # Store model forward timing from scheduler if available + if hasattr(recv_obj, "total_prefill_time") and recv_obj.total_prefill_time: + if i < len(recv_obj.total_prefill_time): + state.total_prefill_time = recv_obj.total_prefill_time[i] + if ( + hasattr(recv_obj, "prefill_preparation_time") + and recv_obj.prefill_preparation_time + ): + if i < len(recv_obj.prefill_preparation_time): + state.prefill_preparation_time = recv_obj.prefill_preparation_time[ + i + ] + if hasattr(recv_obj, "vit_encoding_time") and recv_obj.vit_encoding_time: + if i < len(recv_obj.vit_encoding_time): + state.vit_encoding_time = recv_obj.vit_encoding_time[i] + if hasattr(recv_obj, "llm_prefill_time") and recv_obj.llm_prefill_time: + if i < len(recv_obj.llm_prefill_time): + state.llm_prefill_time = recv_obj.llm_prefill_time[i] + + # Store scheduler overhead timing if available + if ( + hasattr(recv_obj, "scheduler_receive_time") + and recv_obj.scheduler_receive_time + ): + if i < len(recv_obj.scheduler_receive_time): + state.scheduler_receive_time = recv_obj.scheduler_receive_time[i] + if hasattr(recv_obj, "model_call_time") and recv_obj.model_call_time: + if i < len(recv_obj.model_call_time): + state.model_call_time = recv_obj.model_call_time[i] + if ( + hasattr(recv_obj, "first_token_generated_time") + and recv_obj.first_token_generated_time + ): + if i < len(recv_obj.first_token_generated_time): + state.first_token_generated_time = ( + recv_obj.first_token_generated_time[i] + ) + if hasattr(recv_obj, "output_send_time") and recv_obj.output_send_time: + if i < len(recv_obj.output_send_time): + state.output_send_time = recv_obj.output_send_time[i] + if ( + hasattr(recv_obj, "detokenizer_receive_time") + and recv_obj.detokenizer_receive_time + ): + if i < len(recv_obj.detokenizer_receive_time): + state.detokenizer_receive_time = recv_obj.detokenizer_receive_time[ + i + ] + if ( + hasattr(recv_obj, "detokenizer_send_time") + and recv_obj.detokenizer_send_time + ): + if i < len(recv_obj.detokenizer_send_time): + state.detokenizer_send_time = recv_obj.detokenizer_send_time[i] + # Build meta_info and return value meta_info = { "id": rid, @@ -1434,6 +1574,140 @@ def _handle_batch_output( } ) + # Track TTFT (Time To First Token) + if state.first_token_time == 0.0 and recv_obj.completion_tokens[i] > 0: + state.first_token_time = time.time() + ttft_seconds = state.first_token_time - state.created_time + logger.info( + f"[TTFT_MEASUREMENT] Req {rid}: First token generated at time: {state.first_token_time:.6f}" + ) + logger.info( + f"[TTFT_MEASUREMENT] Req {rid}: TTFT: {ttft_seconds:.6f} seconds ({ttft_seconds * 1000:.2f} ms)" + ) + + # Calculate scheduler overhead and return path timing + scheduler_queue_time = 0.0 + scheduler_to_model_time = 0.0 + output_processing_time = 0.0 + scheduler_to_detok_ipc = 0.0 + detokenizer_processing_time = 0.0 + detok_to_tokmgr_ipc = 0.0 + + if state.scheduler_receive_time > 0: + # Time from tokenizer_manager to scheduler + scheduler_queue_time = state.scheduler_receive_time - ( + state.created_time + + state.text_tokenization_time + + state.mm_preprocessing_time + ) + if state.model_call_time > 0 and state.scheduler_receive_time > 0: + # Time from scheduler receiving to calling model + scheduler_to_model_time = ( + state.model_call_time - state.scheduler_receive_time + ) + if ( + state.first_token_generated_time > 0 + and state.output_send_time > 0 + ): + # Time from model generation complete to scheduler sending output + output_processing_time = ( + state.output_send_time - state.first_token_generated_time + ) + if ( + state.detokenizer_receive_time > 0 + and state.output_send_time > 0 + ): + # IPC time from scheduler to detokenizer + scheduler_to_detok_ipc = ( + state.detokenizer_receive_time - state.output_send_time + ) + if ( + state.detokenizer_send_time > 0 + and state.detokenizer_receive_time > 0 + ): + # Detokenizer processing time + detokenizer_processing_time = ( + state.detokenizer_send_time - state.detokenizer_receive_time + ) + if state.detokenizer_send_time > 0: + # IPC time from detokenizer to tokenizer_manager + detok_to_tokmgr_ipc = ( + state.first_token_time - state.detokenizer_send_time + ) + + # Print detailed breakdown summary + logger.info( + f"[TTFT_SUMMARY] ====== Req {rid} TTFT Breakdown ======" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 1. Text tokenization: {state.text_tokenization_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 2. Multimodal preprocessing: {state.mm_preprocessing_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 3. Scheduler queue time: {scheduler_queue_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 4. Scheduler preparation: {scheduler_to_model_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 5. Prefill preparation: {state.prefill_preparation_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 6. ViT encoding + embedding: {state.vit_encoding_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 7. LLM prefill forward: {state.llm_prefill_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 8. Output processing: {output_processing_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 9. Scheduler->Detok IPC: {scheduler_to_detok_ipc * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 10. Detokenizer processing: {detokenizer_processing_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: 11. Detok->TokMgr IPC: {detok_to_tokmgr_ipc * 1000:8.2f} ms" + ) + + # Calculate breakdown sum + breakdown_sum = ( + state.text_tokenization_time + + state.mm_preprocessing_time + + scheduler_queue_time + + scheduler_to_model_time + + state.prefill_preparation_time + + state.vit_encoding_time + + state.llm_prefill_time + + output_processing_time + + scheduler_to_detok_ipc + + detokenizer_processing_time + + detok_to_tokmgr_ipc + ) + + # Also show total model forward time for comparison + logger.info( + f"[TTFT_SUMMARY] Req {rid}: -----------------------------------------------" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: Breakdown sum (1-11): {breakdown_sum * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: Total model forward (5+6+7): {state.total_prefill_time * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: Actual TTFT: {ttft_seconds * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: Difference: {(ttft_seconds - breakdown_sum) * 1000:8.2f} ms" + ) + logger.info( + f"[TTFT_SUMMARY] Req {rid}: =======================================================" + ) + if getattr(recv_obj, "output_hidden_states", None): meta_info["hidden_states"] = recv_obj.output_hidden_states[i] @@ -1681,9 +1955,14 @@ def collect_metrics(self, state: ReqState, recv_obj: BatchStrOutput, i: int): ): state.first_token_time = state.last_time = time.time() state.last_completion_tokens = completion_tokens - self.metrics_collector.observe_time_to_first_token( - labels, state.first_token_time - state.created_time + ttft_seconds = state.first_token_time - state.created_time + logger.info( + f"[TTFT_MEASUREMENT] First token generated at time: {state.first_token_time:.6f}" + ) + logger.info( + f"[TTFT_MEASUREMENT] TTFT: {ttft_seconds:.6f} seconds ({ttft_seconds * 1000:.2f} ms)" ) + self.metrics_collector.observe_time_to_first_token(labels, ttft_seconds) else: num_new_tokens = completion_tokens - state.last_completion_tokens if num_new_tokens: diff --git a/python/sglang/srt/managers/tp_worker.py b/python/sglang/srt/managers/tp_worker.py index 1327c56833a8..2d3166978d4b 100644 --- a/python/sglang/srt/managers/tp_worker.py +++ b/python/sglang/srt/managers/tp_worker.py @@ -374,6 +374,10 @@ def forward_batch_generation( batch_result = GenerationBatchResult( logits_output=logits_output, can_run_cuda_graph=can_run_cuda_graph, + prefill_preparation_time=forward_batch.prefill_preparation_time, + vit_encoding_time=forward_batch.vit_encoding_time, + llm_prefill_time=forward_batch.llm_prefill_time, + total_model_forward_time=forward_batch.total_model_forward_time, ) if is_verify: diff --git a/python/sglang/srt/managers/utils.py b/python/sglang/srt/managers/utils.py index 6e753f1659c4..1259d10e075b 100644 --- a/python/sglang/srt/managers/utils.py +++ b/python/sglang/srt/managers/utils.py @@ -44,6 +44,12 @@ class GenerationBatchResult: # relay path: forward stream -> next step forward next_draft_input: Optional[EagleDraftInput] = None + # TTFT breakdown timing (in seconds) - measured in model_runner + prefill_preparation_time: float = 0.0 + vit_encoding_time: float = 0.0 + llm_prefill_time: float = 0.0 + total_model_forward_time: float = 0.0 + def copy_to_cpu(self, return_logprob: bool = False): """Copy tensors to CPU in overlap scheduling. Only the tensors which are needed for processing results are copied, diff --git a/python/sglang/srt/model_executor/forward_batch_info.py b/python/sglang/srt/model_executor/forward_batch_info.py index b411f4acb53b..c7fa8ed519b0 100644 --- a/python/sglang/srt/model_executor/forward_batch_info.py +++ b/python/sglang/srt/model_executor/forward_batch_info.py @@ -306,6 +306,12 @@ class ForwardBatch: spec_algorithm: SpeculativeAlgorithm = None capture_hidden_mode: CaptureHiddenMode = None + # TTFT breakdown timing (in seconds) - measured in model_runner + prefill_preparation_time: float = 0.0 + vit_encoding_time: float = 0.0 + llm_prefill_time: float = 0.0 + total_model_forward_time: float = 0.0 + # For padding padded_static_len: int = -1 # -1 if not padded num_token_non_padded: Optional[torch.Tensor] = None # scalar tensor @@ -564,14 +570,16 @@ def _compute_mrope_positions( batch_size = self.seq_lens.shape[0] device = model_runner.device if self.forward_mode.is_decode(): - has_mm_input = [batch.multimodal_inputs[i] is not None for i in range(batch_size)] + has_mm_input = [ + batch.multimodal_inputs[i] is not None for i in range(batch_size) + ] if not any(has_mm_input): mrope_positions_list = torch.full( (3, batch_size), 0, dtype=torch.int64, - device = device, + device=device, ) mrope_positions_list += (self.seq_lens - 1).unsqueeze(0) self.mrope_positions = mrope_positions_list @@ -579,12 +587,17 @@ def _compute_mrope_positions( deltas_cpu = torch.zeros((batch_size, 1, 1), dtype=torch.int64) for i in range(batch_size): if has_mm_input[i]: - assert batch.multimodal_inputs[i].mrope_position_delta.shape == (1, 1) + assert batch.multimodal_inputs[i].mrope_position_delta.shape == ( + 1, + 1, + ) deltas_cpu[i] = batch.multimodal_inputs[i].mrope_position_delta deltas_cpu_flat = deltas_cpu.view(batch_size) deltas_gpu = deltas_cpu_flat.to(device, non_blocking=True) - self.mrope_positions = torch.empty((3, batch_size), dtype=torch.int64, device=device) + self.mrope_positions = torch.empty( + (3, batch_size), dtype=torch.int64, device=device + ) compute_mrope_positions_kernel[(batch_size,)]( self.seq_lens, deltas_gpu, @@ -604,27 +617,28 @@ def _compute_mrope_positions( mm_input = batch.multimodal_inputs[batch_idx] if mm_input is None: - positions = torch.arange( - extend_prefix_len, - extend_prefix_len + extend_seq_len, - dtype=torch.int64 - ).unsqueeze(0).expand(3, -1) + positions = ( + torch.arange( + extend_prefix_len, + extend_prefix_len + extend_seq_len, + dtype=torch.int64, + ) + .unsqueeze(0) + .expand(3, -1) + ) else: positions = mm_input.mrope_positions[ :, extend_prefix_len : extend_prefix_len + extend_seq_len, ] - + mrope_positions_cpu_list.append(positions) - + mrope_positions_cpu = torch.cat(mrope_positions_cpu_list, dim=1) self.mrope_positions = mrope_positions_cpu.to( - dtype=torch.int64, - device=device, - non_blocking=True + dtype=torch.int64, device=device, non_blocking=True ) - def get_max_chunk_capacity(self): # Maximum number of tokens in each chunk # TODO: Should be changed to a better value, maybe passed through server args @@ -1130,12 +1144,12 @@ def create_chunked_prefix_cache_kv_indices( tl.store( chunk_kv_indices_ptr + chunk_kv_indices_offset + offset, data, mask=mask ) - + @triton.jit def compute_mrope_positions_kernel( seq_lens_ptr, # (batch_size,) - deltas_ptr, # (batch_size,) + deltas_ptr, # (batch_size,) mrope_positions_ptr, # (3, batch_size) batch_size, ): @@ -1146,4 +1160,3 @@ def compute_mrope_positions_kernel( delta = tl.load(deltas_ptr + pid) for j in range(3): tl.store(mrope_positions_ptr + j * batch_size + pid, base + delta) - diff --git a/python/sglang/srt/model_executor/model_runner.py b/python/sglang/srt/model_executor/model_runner.py index 0dbb6ee86c8d..fef22dd9dede 100644 --- a/python/sglang/srt/model_executor/model_runner.py +++ b/python/sglang/srt/model_executor/model_runner.py @@ -1988,6 +1988,13 @@ def forward_extend( skip_attn_backend_init: bool = False, pp_proxy_tensors=None, ) -> LogitsProcessorOutput: + import logging + import time + + logger = logging.getLogger(__name__) + + prefill_start_time = time.time() + if not skip_attn_backend_init: self.attn_backend.init_forward_metadata(forward_batch) @@ -2003,13 +2010,40 @@ def forward_extend( if self.piecewise_cuda_graph_runner.can_run(forward_batch): return self.piecewise_cuda_graph_runner.replay(forward_batch, **kwargs) - return self.model.forward( + # Synchronize before model forward timing + torch.cuda.synchronize() + model_forward_start = time.time() + prefill_prep_time = model_forward_start - prefill_start_time + logger.info( + f"[TTFT_BREAKDOWN] Prefill preparation time: {prefill_prep_time * 1000:.2f} ms" + ) + + result = self.model.forward( forward_batch.input_ids, forward_batch.positions, forward_batch, **kwargs, ) + # Synchronize after model forward to ensure completion + torch.cuda.synchronize() + model_forward_end = time.time() + model_forward_time = model_forward_end - model_forward_start + total_forward_time = model_forward_end - prefill_start_time + + logger.info( + f"[TTFT_BREAKDOWN] Model forward (ViT + LLM Prefill) time: {model_forward_time * 1000:.2f} ms" + ) + logger.info( + f"[TTFT_BREAKDOWN] Total prefill time: {total_forward_time * 1000:.2f} ms" + ) + + # Store timing in forward_batch for later retrieval + forward_batch.prefill_preparation_time = prefill_prep_time + forward_batch.total_model_forward_time = total_forward_time + + return result + def forward_idle( self, forward_batch: ForwardBatch, pp_proxy_tensors=None ) -> LogitsProcessorOutput: diff --git a/python/sglang/srt/multimodal/processors/base_processor.py b/python/sglang/srt/multimodal/processors/base_processor.py index 22ce6cbe8ad7..1514e0a2eeec 100644 --- a/python/sglang/srt/multimodal/processors/base_processor.py +++ b/python/sglang/srt/multimodal/processors/base_processor.py @@ -241,10 +241,19 @@ def process_mm_data( """ process multimodal data with transformers AutoProcessor """ + import time + + logger.info(f"[DEBUG_TRACE] Entering process_mm_data") + + # Step 5.1.1: Prepare kwargs + logger.info(f"[DEBUG_TRACE] Preparing kwargs for processor") + substep1_start = time.time() if images: kwargs["images"] = images + logger.info(f"[DEBUG_TRACE] Added {len(images)} images to kwargs") if videos: kwargs["videos"] = videos + logger.info(f"[DEBUG_TRACE] Added {len(videos)} videos to kwargs") if audios: if self._processor.__class__.__name__ in { "Gemma3nProcessor", @@ -272,12 +281,38 @@ def process_mm_data( }: # Note: for qwen-vl, processor has some reshape issue because of dims restriction on Ascend. kwargs["device"] = "npu" - result = processor.__call__( - text=[input_text], - padding=True, - return_tensors="pt", - **kwargs, + substep1_duration = (time.time() - substep1_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5.1.1 - Prepare kwargs: {substep1_duration:.2f} ms" + ) + + # Step 5.1.2: Call processor.__call__() + logger.info(f"[DEBUG_TRACE] About to call processor.__call__()") + substep2_start = time.time() + try: + result = processor.__call__( + text=[input_text], + padding=True, + return_tensors="pt", + **kwargs, + ) + logger.info( + f"[DEBUG_TRACE] processor.__call__() completed, result keys: {list(result.keys())}" + ) + except Exception as e: + logger.error( + f"[DEBUG_TRACE] Exception in processor.__call__(): {e}", exc_info=True + ) + raise + # end = time.time() + # logger.info(f"processor call time taken: {end - start}") + substep2_duration = (time.time() - substep2_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5.1.2 - processor.__call__() execution: {substep2_duration:.2f} ms" ) + + # Step 5.1.3: Move feature tensors to CPU (if needed) + substep3_start = time.time() if not self.server_args.keep_mm_feature_on_device: # move feature tensors to cpu for feature_name in self.FEATURE_NAMES: @@ -288,6 +323,10 @@ def process_mm_data( result[feature_name], torch.Tensor ): result[feature_name] = result[feature_name].to("cpu") + substep3_duration = (time.time() - substep3_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5.1.3 - Move tensors to CPU: {substep3_duration:.2f} ms" + ) return result @@ -345,6 +384,7 @@ def _load_single_item( try: if modality == Modality.IMAGE: img_tensor, _ = load_image_tensor(data, discard_alpha_channel) + img_tensor = img_tensor.to("cuda") return img_tensor elif modality == Modality.VIDEO: return load_video(data, frame_count_limit) @@ -368,19 +408,33 @@ def submit_data_loading_tasks( """ load multimodal data parallelly using iterators. """ + logger.info( + f"[DEBUG_TRACE] → submit_data_loading_tasks 开始: text_parts={len(text_parts)}, data_iterators={list(data_iterators.keys())}" + ) + logger.info(f"[DEBUG_TRACE] , text_parts: {text_parts}") futures = [] task_info = [] - for text_part in text_parts: + for idx, text_part in enumerate(text_parts): + logger.info(f"[DEBUG_TRACE] 处理 text_part[{idx}]: {text_part}") modality = multimodal_tokens.get_modality_of_token(text_part) if modality is not None: + logger.info( + f"[DEBUG_TRACE] 处理 text_part[{idx}]: modality={modality}" + ) data_iterator = data_iterators.get(modality) if data_iterator is None: + logger.error( + f"[DEBUG_TRACE] ✗ 没有找到 {modality} 的 data_iterator" + ) raise ValueError(f"No data iterator found for token: {text_part}") try: data = next(data_iterator) except StopIteration: + logger.error( + f"[DEBUG_TRACE] ✗ data_iterator 耗尽: modality={modality}, text_part={text_part}" + ) raise ValueError( f"Mismatch: More '{text_part}' tokens found than corresponding data items provided." ) @@ -396,6 +450,9 @@ def submit_data_loading_tasks( # Ensure we don't exceed the absolute max (redundant if scaling_factor handles it) # frame_count_limit = min(frame_count_limit, max_image_frames) except StopIteration: + logger.error( + f"[DEBUG_TRACE] ✗ image_estimated_frames_iter 耗尽" + ) raise ValueError( "Mismatch between image tokens and estimated frame counts." ) @@ -411,7 +468,12 @@ def submit_data_loading_tasks( ) ) task_info.append((modality, data, frame_count_limit)) + if idx < 5: + logger.info( + f"[DEBUG_TRACE] ✓ 已提交 future[{len(futures)-1}] for {modality}" + ) + logger.info(f"[DEBUG_TRACE] → 检查剩余数据项") for modality, iterator in data_iterators.items(): try: next(iterator) @@ -423,6 +485,9 @@ def submit_data_loading_tasks( except Exception: pass + logger.info( + f"[DEBUG_TRACE] ✓ submit_data_loading_tasks 完成: 返回 {len(futures)} futures, {len(task_info)} task_info" + ) return futures, task_info def load_mm_data( @@ -445,6 +510,12 @@ def load_mm_data( discard_alpha_channel: if True, discards the alpha channel in the returned images """ + import time + + overall_start = time.time() + + # Step 1: Parse prompt and prepare iterators + step1_start = time.time() multimodal_tokens_pattern = multimodal_tokens.get_combined_regex() if isinstance(prompt, list) and return_text: @@ -465,23 +536,57 @@ def load_mm_data( data_iterators[Modality.VIDEO] = iter(video_data) if multimodal_tokens.audio_token and audio_data: data_iterators[Modality.AUDIO] = iter(audio_data) + step1_duration = (time.time() - step1_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 1 - Prompt parsing: {step1_duration:.2f} ms" + ) + logger.info( + f"[DEBUG_TRACE] ✓ Step 1 完成: text_parts={len(text_parts)}, data_iterators={list(data_iterators.keys())}" + ) + + # Step 2: Submit data loading tasks (parallel I/O) + logger.info(f"[DEBUG_TRACE] → 即将开始 Step 2 - submit_data_loading_tasks") + step2_start = time.time() + try: + futures, task_info = self.submit_data_loading_tasks( + text_parts=text_parts, + multimodal_tokens=multimodal_tokens, + data_iterators=data_iterators, + discard_alpha_channel=discard_alpha_channel, + audio_sample_rate=audio_sample_rate, + ) + logger.info( + f"[DEBUG_TRACE] ✓ submit_data_loading_tasks 完成: futures={len(futures)}, task_info={len(task_info)}" + ) + except Exception as e: + logger.error( + f"[DEBUG_TRACE] ✗ submit_data_loading_tasks 异常: {type(e).__name__}: {e}", + exc_info=True, + ) + raise + step2_duration = (time.time() - step2_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 2 - Submit I/O tasks ({len(futures)} files): {step2_duration:.2f} ms" + ) - # futures: the futures of loaded data - # task_info: modality, raw_data, and other metadata of each data - futures, task_info = self.submit_data_loading_tasks( - text_parts=text_parts, - multimodal_tokens=multimodal_tokens, - data_iterators=data_iterators, - discard_alpha_channel=discard_alpha_channel, - audio_sample_rate=audio_sample_rate, + logger.info( + f"[DEBUG_TRACE] → 创建迭代器: task_info={len(task_info)}, futures={len(futures)}" ) task_info_iter = iter(task_info) futures_iter = iter(futures) + logger.info(f"[DEBUG_TRACE] ✓ 迭代器创建完成") - # Process results + # Step 3: Collect I/O results + logger.info(f"[DEBUG_TRACE] → 即将开始 Step 3 - 收集 I/O 结果") + step3_start = time.time() images, videos, audios = [], [], [] new_text_parts = [] - for text_part in text_parts: + logger.info(f"[DEBUG_TRACE] → 开始遍历 {len(text_parts)} 个 text_parts") + for idx, text_part in enumerate(text_parts): + if idx % 10 == 0: # Print every 10 iterations to avoid log spam + logger.info( + f"[DEBUG_TRACE] 处理中 text_part [{idx}/{len(text_parts)}]" + ) try: if multimodal_tokens_pattern.match(text_part): modality, raw_data, frame_limit = next(task_info_iter) @@ -527,15 +632,38 @@ def load_mm_data( new_text_parts += [text_part] except Exception as e: + logger.error( + f"[DEBUG_TRACE] ✗ text_part[{idx}] 处理异常: {type(e).__name__}: {e}", + exc_info=True, + ) raise RuntimeError( f"An exception occurred while loading multimodal data: {e}" ) - return BaseMultiModalProcessorOutput( + + logger.info( + f"[DEBUG_TRACE] ✓ 遍历完成: images={len(images)}, videos={len(videos)}, audios={len(audios)}" + ) + step3_duration = (time.time() - step3_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 3 - Collect I/O results: {step3_duration:.2f} ms" + ) + + logger.info(f"[DEBUG_TRACE] → 计算总耗时") + overall_duration = (time.time() - overall_start) * 1000 + total_steps = step1_duration + step2_duration + step3_duration + logger.info( + f"[MM_PREPROC_DETAIL] === load_mm_data TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===" + ) + + logger.info(f"[DEBUG_TRACE] → 创建 BaseMultiModalProcessorOutput 对象") + result = BaseMultiModalProcessorOutput( images=images, audios=audios, videos=videos, input_text="".join(new_text_parts), ) + logger.info(f"[DEBUG_TRACE] load_mm_data returning successfully") + return result @staticmethod def get_mm_items_offset( @@ -605,13 +733,71 @@ def _process_and_collect_mm_items( Returns: Tuple of (created mm_items, input_ids) """ - ret = self.process_mm_data( - input_text=input_text, images=images, audios=audios, videos=videos, **kwargs + import time + + logger.info( + f"[DEBUG_TRACE] Entering _process_and_collect_mm_items with {len(images) if images else 0} images, {len(audios) if audios else 0} audios, {len(videos) if videos else 0} videos" ) - input_ids = ret["input_ids"].flatten() - collected_items = self.collect_mm_items_from_processor_output(ret) + # Step 5.1: Call HF processor + logger.info(f"[DEBUG_TRACE] About to call process_mm_data") + substep1_start = time.time() + try: + ret = self.process_mm_data( + input_text=input_text, + images=images, + audios=audios, + videos=videos, + **kwargs, + ) + logger.info(f"[DEBUG_TRACE] process_mm_data returned successfully") + except Exception as e: + logger.error( + f"[DEBUG_TRACE] Exception in process_mm_data: {e}", exc_info=True + ) + raise + substep1_duration = (time.time() - substep1_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5.1 - HF processor __call__: {substep1_duration:.2f} ms" + ) + + # Step 5.2: Extract input_ids + logger.info(f"[DEBUG_TRACE] About to extract input_ids") + substep2_start = time.time() + try: + input_ids = ret["input_ids"].flatten() + logger.info( + f"[DEBUG_TRACE] Extracted input_ids with shape: {input_ids.shape}" + ) + except Exception as e: + logger.error( + f"[DEBUG_TRACE] Exception extracting input_ids: {e}", exc_info=True + ) + raise + substep2_duration = (time.time() - substep2_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5.2 - Extract input_ids: {substep2_duration:.2f} ms" + ) + # Step 5.3: Collect mm_items from processor output + logger.info(f"[DEBUG_TRACE] About to collect mm_items from processor output") + substep3_start = time.time() + try: + collected_items = self.collect_mm_items_from_processor_output(ret) + logger.info(f"[DEBUG_TRACE] Collected {len(collected_items)} mm_items") + except Exception as e: + logger.error( + f"[DEBUG_TRACE] Exception collecting mm_items: {e}", exc_info=True + ) + raise + substep3_duration = (time.time() - substep3_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5.3 - Collect mm_items: {substep3_duration:.2f} ms" + ) + + logger.info( + f"[DEBUG_TRACE] _process_and_collect_mm_items returning successfully" + ) return collected_items, input_ids, ret def process_and_combine_mm_data( @@ -627,19 +813,38 @@ def process_and_combine_mm_data( Returns: Tuple of (list of mm_items, input_ids) """ - # Collect all items and categorize them - all_items = base_output.organize_results() + import time + + logger.info(f"[DEBUG_TRACE] Entering process_and_combine_mm_data") + overall_start = time.time() + + # Step 1: Categorize items + logger.info(f"[DEBUG_TRACE] Step 4 - About to organize results") + step1_start = time.time() + try: + all_items = base_output.organize_results() + logger.info(f"[DEBUG_TRACE] Organized {len(all_items)} items") + except Exception as e: + logger.error( + f"[DEBUG_TRACE] Exception in organize_results: {e}", exc_info=True + ) + raise # Handle text-only case if not all_items: + logger.info(f"[DEBUG_TRACE] Text-only case, no multimodal items") input_ids = self._processor.tokenizer( base_output.input_text, return_tensors="pt", add_special_tokens=True, ).input_ids.flatten() + logger.info(f"[DEBUG_TRACE] Returning early for text-only") return [], input_ids, {} + logger.info(f"[DEBUG_TRACE] Categorizing {len(all_items)} items by type") dict_items, raw_images, raw_audios, raw_videos = [], [], [], [] - for modality, item in all_items: + for idx, (modality, item) in enumerate(all_items): + if idx % 10 == 0: + logger.info(f"[DEBUG_TRACE] Categorizing item {idx}/{len(all_items)}") if isinstance(item, dict): dict_items.append(item) elif modality == Modality.IMAGE: @@ -650,21 +855,52 @@ def process_and_combine_mm_data( raw_videos.append(item) else: raise ValueError(f"Unknown multimodal item type: {type(item)}") - # Process items and get input_ids + step1_duration = (time.time() - step1_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 4 - Categorize items: {step1_duration:.2f} ms" + ) + logger.info( + f"[DEBUG_TRACE] Categorized: {len(dict_items)} dict, {len(raw_images)} images, {len(raw_audios)} audios, {len(raw_videos)} videos" + ) + + # Step 2: Process raw items (HF processor call) all_collected_items: list[MultimodalDataItem] = [] input_ids = None + step2_duration = ( + 0.0 # Initialize to handle case where no raw items need processing + ) - # Handle raw items (need processing) if raw_images or raw_audios or raw_videos: - collected_items, input_ids, ret = self._process_and_collect_mm_items( - input_text=base_output.input_text, - images=raw_images, - audios=raw_audios, - videos=raw_videos, - **kwargs, + logger.info( + f"[DEBUG_TRACE] Step 5 - About to call _process_and_collect_mm_items" + ) + step2_start = time.time() + try: + collected_items, input_ids, ret = self._process_and_collect_mm_items( + input_text=base_output.input_text, + images=raw_images, + audios=raw_audios, + videos=raw_videos, + **kwargs, + ) + logger.info( + f"[DEBUG_TRACE] _process_and_collect_mm_items returned {len(collected_items)} items" + ) + except Exception as e: + logger.error( + f"[DEBUG_TRACE] Exception in _process_and_collect_mm_items: {e}", + exc_info=True, + ) + raise + step2_duration = (time.time() - step2_start) * 1000 + logger.info( + f"[MM_PREPROC_DETAIL] Step 5 - HF processor call: {step2_duration:.2f} ms" ) all_collected_items = collected_items else: + logger.info( + f"[DEBUG_TRACE] No raw items to process, skipping HF processor call" + ) ret = None # Handle dict items (already processed) @@ -737,4 +973,17 @@ def process_and_combine_mm_data( sync_buffer_meta=sync_flag, ) + logger.info( + f"[DEBUG_TRACE] About to finalize and return from process_and_combine_mm_data" + ) + overall_duration = (time.time() - overall_start) * 1000 + # Calculate sum of steps + total_steps = step1_duration + step2_duration + logger.info( + f"[MM_PREPROC_DETAIL] === process_and_combine_mm_data TOTAL: {overall_duration:.2f} ms (sum of steps: {total_steps:.2f} ms) ===" + ) + + logger.info( + f"[DEBUG_TRACE] process_and_combine_mm_data returning successfully with {len(all_collected_items)} items" + ) return all_collected_items, input_ids, ret diff --git a/python/sglang/srt/multimodal/processors/qwen_vl.py b/python/sglang/srt/multimodal/processors/qwen_vl.py index 647ca1779e57..f3e156d90b0a 100644 --- a/python/sglang/srt/multimodal/processors/qwen_vl.py +++ b/python/sglang/srt/multimodal/processors/qwen_vl.py @@ -273,41 +273,75 @@ async def process_mm_data_async( *args, **kwargs, ): - base_output = self.load_mm_data( - prompt=input_text, - image_data=image_data, - video_data=request_obj.video_data, - audio_data=request_obj.audio_data, - multimodal_tokens=self.mm_tokens, + logger.info( + f"[DEBUG_TRACE] 🚀 QwenVL.process_mm_data_async 开始: images={len(image_data) if image_data else 0}, videos={len(request_obj.video_data) if request_obj.video_data else 0}, audios={len(request_obj.audio_data) if request_obj.audio_data else 0}" ) + logger.info(f"[DEBUG_TRACE] → 调用 load_mm_data") + try: + base_output = self.load_mm_data( + prompt=input_text, + image_data=image_data, + video_data=request_obj.video_data, + audio_data=request_obj.audio_data, + multimodal_tokens=self.mm_tokens, + ) + logger.info( + f"[DEBUG_TRACE] ✓ load_mm_data 完成: images={len(base_output.images)}, videos={len(base_output.videos)}, audios={len(base_output.audios)}" + ) + except Exception as e: + logger.error( + f"[DEBUG_TRACE] ✗ load_mm_data 异常: {type(e).__name__}: {e}", + exc_info=True, + ) + raise # Qwen-specific: resize images if they are raw Image objects # Only apply default resize logic if mm_processor_kwargs was not set by user + logger.info(f"[DEBUG_TRACE] → 检查是否需要 resize images") if not self.server_args.mm_processor_kwargs: if base_output.images and isinstance(base_output.images[0], Image.Image): + logger.info( + f"[DEBUG_TRACE] → 执行异步 resize {len(base_output.images)} 张图像" + ) resize_tasks = [ resize_image_async(image) for image in base_output.images ] base_output.images = await asyncio.gather(*resize_tasks) + logger.info(f"[DEBUG_TRACE] ✓ 图像 resize 完成") video_metadata = None if base_output.videos: + logger.info(f"[DEBUG_TRACE] → 预处理 {len(base_output.videos)} 个视频") video_results = await asyncio.gather( *[preprocess_video(video) for video in base_output.videos] ) base_output.videos, video_metadata = map(list, zip(*video_results)) + logger.info(f"[DEBUG_TRACE] ✓ 视频预处理完成") # NOTE: for qwen3-vl, video_meta need to be passed in, since do_sample_frames is already done in preprocess_video - if self.hf_config.model_type in ("qwen3_vl", "qwen3_vl_moe"): - mm_items, input_ids, ret = self.process_and_combine_mm_data( - base_output, - self.mm_tokens, - video_metadata=video_metadata, - do_sample_frames=False, + logger.info( + f"[DEBUG_TRACE] → 调用 process_and_combine_mm_data, model_type={self.hf_config.model_type}" + ) + try: + if self.hf_config.model_type in ("qwen3_vl", "qwen3_vl_moe"): + mm_items, input_ids, ret = self.process_and_combine_mm_data( + base_output, + self.mm_tokens, + video_metadata=video_metadata, + do_sample_frames=False, + ) + else: + mm_items, input_ids, ret = self.process_and_combine_mm_data( + base_output, self.mm_tokens + ) + logger.info( + f"[DEBUG_TRACE] ✓ process_and_combine_mm_data 完成: mm_items={len(mm_items)}, input_ids.shape={input_ids.shape}" ) - else: - mm_items, input_ids, ret = self.process_and_combine_mm_data( - base_output, self.mm_tokens + except Exception as e: + logger.error( + f"[DEBUG_TRACE] ✗ process_and_combine_mm_data 异常: {type(e).__name__}: {e}", + exc_info=True, ) + raise audio_feature_lengths = None