diff --git a/.claude/rules/forward-batch-init-new-purity.md b/.claude/rules/forward-batch-init-new-purity.md new file mode 100644 index 000000000..adee51484 --- /dev/null +++ b/.claude/rules/forward-batch-init-new-purity.md @@ -0,0 +1,17 @@ +--- +paths: + - "**/*.py" +--- + +# `ForwardBatch.init_new` must not mutate the ScheduleBatch + +`init_new` (and any ForwardBatch factory) treats the input `ScheduleBatch` as +read-only. Per-forward overrides go through the kw-only params of `init_new` / +`TpModelWorker.forward_batch_generation`, never a ScheduleBatch field. Batch-prep +writes (`out_cache_loc`, `seq_lens_*`, ...) before `init_new` are out of scope. + +Tolerated exceptions (don't add new ones): + +- `seq_lens_sum` backfill — the `seq_lens` family is slated for removal (→ kv-committed lengths). +- `sampling_info` sub-object writes (grammars, canary ids) — shared object, until the sampling forward-copy op. +- `_expand_mrope_from_input` memoizing `mrope_position_delta_repeated_cache` — pre-existing. diff --git a/python/sglang/benchmark/one_batch.py b/python/sglang/benchmark/one_batch.py index 80d2fbe12..8556becec 100644 --- a/python/sglang/benchmark/one_batch.py +++ b/python/sglang/benchmark/one_batch.py @@ -507,7 +507,11 @@ def extend(reqs, model_runner): ) batch.prefill_input_ids_cpu = None - forward_batch = ForwardBatch.init_new(batch, model_runner) + forward_batch = ForwardBatch.init_new( + batch, + model_runner, + return_hidden_states_before_norm=False, + ) logits_output = model_runner.forward(forward_batch).logits_output next_token_ids = model_runner.sample(logits_output, forward_batch) return next_token_ids, logits_output.next_token_logits, batch @@ -518,7 +522,11 @@ def decode(input_token_ids, batch, model_runner): batch.input_ids = input_token_ids.to(torch.int64) batch.prepare_for_decode() _maybe_prepare_mlp_sync_batch(batch, model_runner) - forward_batch = ForwardBatch.init_new(batch, model_runner) + forward_batch = ForwardBatch.init_new( + batch, + model_runner, + return_hidden_states_before_norm=False, + ) logits_output = model_runner.forward(forward_batch).logits_output next_token_ids = model_runner.sample(logits_output, forward_batch) return next_token_ids, logits_output.next_token_logits diff --git a/python/sglang/srt/hardware_backend/mlx/tp_worker.py b/python/sglang/srt/hardware_backend/mlx/tp_worker.py index ec26937cf..53f9b88c1 100644 --- a/python/sglang/srt/hardware_backend/mlx/tp_worker.py +++ b/python/sglang/srt/hardware_backend/mlx/tp_worker.py @@ -26,7 +26,11 @@ from sglang.srt.hardware_backend.mlx.model_runner import ( from sglang.srt.managers.schedule_batch import ScheduleBatch from sglang.srt.managers.tp_worker import TpModelWorker from sglang.srt.managers.utils import GenerationBatchResult -from sglang.srt.model_executor.forward_batch_info import ForwardBatch, PPProxyTensors +from sglang.srt.model_executor.forward_batch_info import ( + CaptureHiddenMode, + ForwardBatch, + PPProxyTensors, +) logger = logging.getLogger(__name__) @@ -93,6 +97,8 @@ class MlxTpModelWorker(TpModelWorker): pp_proxy_tensors: Optional[PPProxyTensors] = None, is_verify: bool = False, skip_attn_backend_init: Optional[bool] = None, # deprecated + *, + capture_hidden_mode: Optional[CaptureHiddenMode] = None, ) -> GenerationBatchResult: """Override to route through MLX model runner.""" if batch is not None: @@ -106,6 +112,7 @@ class MlxTpModelWorker(TpModelWorker): pp_proxy_tensors, is_verify, skip_attn_backend_init, + capture_hidden_mode=capture_hidden_mode, ) def _cleanup_stale_rids(self, forward_mode, current_rids: set[str]) -> None: diff --git a/python/sglang/srt/managers/schedule_batch.py b/python/sglang/srt/managers/schedule_batch.py index e808d3cf9..72f90779a 100755 --- a/python/sglang/srt/managers/schedule_batch.py +++ b/python/sglang/srt/managers/schedule_batch.py @@ -94,11 +94,7 @@ from sglang.srt.mem_cache.common import ( ) from sglang.srt.mem_cache.memory_pool import ReqToTokenPool from sglang.srt.mem_cache.radix_cache import RadixKey -from sglang.srt.model_executor.forward_batch_info import ( - CaptureHiddenMode, - ForwardBatch, - ForwardMode, -) +from sglang.srt.model_executor.forward_batch_info import ForwardBatch, ForwardMode from sglang.srt.observability.metrics_collector import ( DPCooperationInfo, SchedulerMetricsCollector, @@ -1954,10 +1950,6 @@ class ScheduleBatch(ScheduleBatchDisaggregationDecodeMixin): # spec_info: Optional[SpecInput] = None spec_info: Optional[SpecInput] = None - # === One-shot per-forward overrides; init_new consumes and resets === - capture_hidden_mode: Optional[CaptureHiddenMode] = None - return_hidden_states_before_norm: bool = False - @classmethod def init_new( cls, diff --git a/python/sglang/srt/managers/scheduler_pp_mixin.py b/python/sglang/srt/managers/scheduler_pp_mixin.py index 2cb2c7bf0..30b5bf43e 100644 --- a/python/sglang/srt/managers/scheduler_pp_mixin.py +++ b/python/sglang/srt/managers/scheduler_pp_mixin.py @@ -695,7 +695,11 @@ class SchedulerPPMixin: ) batch.prefill_input_ids_cpu = None - forward_batch = ForwardBatch.init_new(batch, model_runner) + forward_batch = ForwardBatch.init_new( + batch, + model_runner, + return_hidden_states_before_norm=False, + ) set_is_extend_in_batch(batch.forward_mode.is_extend()) _ = model_runner.forward( diff --git a/python/sglang/srt/managers/tp_worker.py b/python/sglang/srt/managers/tp_worker.py index 390025c34..485a897f8 100644 --- a/python/sglang/srt/managers/tp_worker.py +++ b/python/sglang/srt/managers/tp_worker.py @@ -41,7 +41,11 @@ from sglang.srt.managers.schedule_batch import ScheduleBatch from sglang.srt.managers.scheduler import GenerationBatchResult from sglang.srt.mem_cache.allocator import BaseTokenToKVPoolAllocator from sglang.srt.mem_cache.memory_pool import ReqToTokenPool -from sglang.srt.model_executor.forward_batch_info import ForwardBatch, PPProxyTensors +from sglang.srt.model_executor.forward_batch_info import ( + CaptureHiddenMode, + ForwardBatch, + PPProxyTensors, +) from sglang.srt.model_executor.pool_configurator import MemoryPoolConfig from sglang.srt.server_args import ServerArgs from sglang.srt.utils import MultiprocessingSerializer, broadcast_pyobj, set_random_seed @@ -226,7 +230,11 @@ class BaseTpWorker(ABC): return result def forward_batch_embedding(self, batch: ScheduleBatch): - forward_batch = ForwardBatch.init_new(batch, self.model_runner) + forward_batch = ForwardBatch.init_new( + batch, + self.model_runner, + return_hidden_states_before_norm=False, + ) output = self.model_runner.forward(forward_batch).logits_output return output # Returns EmbeddingPoolerOutput @@ -490,16 +498,26 @@ class TpModelWorker(BaseTpWorker): pp_proxy_tensors: Optional[PPProxyTensors] = None, is_verify: bool = False, skip_attn_backend_init: Optional[bool] = None, # deprecated + *, + capture_hidden_mode: Optional[CaptureHiddenMode] = None, ) -> GenerationBatchResult: # Get forward batch from schedule batch if batch is not None: # update the consumer index of hicache to the running batch self.set_hicache_consumer(batch.hicache_consumer_index) - forward_batch = ForwardBatch.init_new(batch, self.model_runner) + forward_batch = ForwardBatch.init_new( + batch, + self.model_runner, + capture_hidden_mode=capture_hidden_mode, + return_hidden_states_before_norm=False, + ) else: # FIXME(lsyin): unify the interface of forward_batch assert forward_batch is not None + assert ( + capture_hidden_mode is None + ), "capture_hidden_mode override requires a ScheduleBatch input" # Deprecated kwarg: pre-planners mark the batch themselves now. forward_batch.apply_deprecated_skip_attn_backend_init(skip_attn_backend_init) @@ -577,7 +595,11 @@ class TpModelWorker(BaseTpWorker): def forward_batch_split_prefill(self, batch: ScheduleBatch): if batch.split_index == 0: - forward_batch = ForwardBatch.init_new(batch, self.model_runner) + forward_batch = ForwardBatch.init_new( + batch, + self.model_runner, + return_hidden_states_before_norm=False, + ) batch.split_forward_batch = forward_batch out = self.model_runner.forward( diff --git a/python/sglang/srt/model_executor/forward_batch_info.py b/python/sglang/srt/model_executor/forward_batch_info.py index 357d80589..07a9e11d6 100644 --- a/python/sglang/srt/model_executor/forward_batch_info.py +++ b/python/sglang/srt/model_executor/forward_batch_info.py @@ -433,7 +433,7 @@ class ForwardBatch(ForwardBatchDeepSeekMHAMixin): # For dumper: request IDs for cross-step sequence tracking rids: Optional[List[str]] = None - # === Resolved from SB one-shot overrides (consumed + reset by init_new) === + # === Per-forward overrides passed explicitly to init_new === capture_hidden_mode: CaptureHiddenMode = None # For hidden states before normal return_hidden_states_before_norm: bool = False @@ -624,17 +624,15 @@ class ForwardBatch(ForwardBatchDeepSeekMHAMixin): cls, batch: ScheduleBatch, model_runner: ModelRunner, + *, + capture_hidden_mode: Optional[CaptureHiddenMode] = None, + return_hidden_states_before_norm: bool, ): - # Consume one-shot per-forward overrides from SB; reset to defaults so - # the next forward on the same SB starts clean. See SB field comment - # for the contract. - capture_hidden_mode = batch.capture_hidden_mode - batch.capture_hidden_mode = None - return_hidden_states_before_norm = batch.return_hidden_states_before_norm - batch.return_hidden_states_before_norm = False + # init_new must not mutate the input ScheduleBatch; per-forward + # overrides go through explicit keyword arguments. - # capture_hidden_mode default: derive from SB.return_hidden_states / - # spec_info.capture_hidden_mode when caller did not override. + # capture_hidden_mode=None means no override: derive from + # SB.return_hidden_states / spec_info.capture_hidden_mode. if capture_hidden_mode is None: if batch.return_hidden_states: capture_hidden_mode = CaptureHiddenMode.FULL @@ -666,6 +664,10 @@ class ForwardBatch(ForwardBatchDeepSeekMHAMixin): # block there). Use it directly. seq_lens_cpu = batch.seq_lens_cpu + # TODO(seq-lens-removal): the whole ScheduleBatch seq_lens family + # (incl. seq_lens_sum) is slated for removal in favor of kv-committed + # lengths, so this init_new-time backfill onto the ScheduleBatch is + # tolerated for now despite the init_new-must-not-mutate-SB rule. if batch.seq_lens_sum is None and seq_lens_cpu is not None: batch.seq_lens_sum = int(seq_lens_cpu.sum()) diff --git a/python/sglang/srt/speculative/base_spec_worker.py b/python/sglang/srt/speculative/base_spec_worker.py index 06c51fa36..fa1d9ca70 100644 --- a/python/sglang/srt/speculative/base_spec_worker.py +++ b/python/sglang/srt/speculative/base_spec_worker.py @@ -114,6 +114,8 @@ class EagleDraftWorkerBase(ABC): num_draft_tokens: int, draft_model_runner: Any, cuda_graph_runner: Any, + *, + return_hidden_states_before_norm: bool, ): from sglang.srt.model_executor.forward_batch_info import ( CaptureHiddenMode, @@ -161,8 +163,12 @@ class EagleDraftWorkerBase(ABC): if batch.forward_mode.is_idle() else ForwardMode.DRAFT_EXTEND_V2 ) - batch.capture_hidden_mode = capture_mode - forward_batch = ForwardBatch.init_new(batch, draft_model_runner) + forward_batch = ForwardBatch.init_new( + batch, + draft_model_runner, + capture_hidden_mode=capture_mode, + return_hidden_states_before_norm=return_hidden_states_before_norm, + ) # Forward sees post-write length (draft extend writes num_draft_tokens # slots); mutation stays on forward_batch to preserve SB.seq_lens. forward_batch.seq_lens = forward_batch.seq_lens + num_draft_tokens @@ -292,8 +298,12 @@ class EagleDraftWorkerBase(ABC): else CaptureHiddenMode.LAST ) draft_input.positions = batch.seq_lens.repeat_interleave(topk, dim=0) - batch.capture_hidden_mode = capture_mode - forward_batch = ForwardBatch.init_new(batch, draft_model_runner) + forward_batch = ForwardBatch.init_new( + batch, + draft_model_runner, + capture_hidden_mode=capture_mode, + return_hidden_states_before_norm=False, + ) can_cuda_graph = cuda_graph_runner and cuda_graph_runner.can_run_graph( forward_batch ) diff --git a/python/sglang/srt/speculative/dflash_info.py b/python/sglang/srt/speculative/dflash_info.py index 6fb382ecb..8ec5420bc 100644 --- a/python/sglang/srt/speculative/dflash_info.py +++ b/python/sglang/srt/speculative/dflash_info.py @@ -69,8 +69,12 @@ class DFlashVerifyInput(SpecInput): if batch.forward_mode.is_idle() else ForwardMode.TARGET_VERIFY ) - batch.capture_hidden_mode = self.capture_hidden_mode - verify_forward_batch = ForwardBatch.init_new(batch, target_worker.model_runner) + verify_forward_batch = ForwardBatch.init_new( + batch, + target_worker.model_runner, + capture_hidden_mode=self.capture_hidden_mode, + return_hidden_states_before_norm=False, + ) can_run_cuda_graph = bool( target_worker.model_runner.decode_cuda_graph_runner diff --git a/python/sglang/srt/speculative/dflash_worker_v2.py b/python/sglang/srt/speculative/dflash_worker_v2.py index 8d983b4ce..f4a5e331e 100644 --- a/python/sglang/srt/speculative/dflash_worker_v2.py +++ b/python/sglang/srt/speculative/dflash_worker_v2.py @@ -1226,8 +1226,9 @@ class DFlashWorkerV2(BaseSpecWorker): if batch.forward_mode.is_extend() or batch.is_extend_in_batch: # Target prefill: capture DFlash aux hidden states for prompt tokens. - batch.capture_hidden_mode = CaptureHiddenMode.FULL - batch_output = self.target_worker.forward_batch_generation(batch) + batch_output = self.target_worker.forward_batch_generation( + batch, capture_hidden_mode=CaptureHiddenMode.FULL + ) logits_output, next_token_ids = ( batch_output.logits_output, diff --git a/python/sglang/srt/speculative/dspark_components/dspark_worker_v2.py b/python/sglang/srt/speculative/dspark_components/dspark_worker_v2.py index 91ce4f42d..dfbff5730 100644 --- a/python/sglang/srt/speculative/dspark_components/dspark_worker_v2.py +++ b/python/sglang/srt/speculative/dspark_components/dspark_worker_v2.py @@ -376,12 +376,14 @@ class DSparkWorkerV2(BaseSpecWorker): ) -> GenerationBatchResult: if batch.forward_mode.is_idle(): if self.server_args.enable_dp_attention: - batch.capture_hidden_mode = CaptureHiddenMode.FULL - self.target_worker.forward_batch_generation(batch) + self.target_worker.forward_batch_generation( + batch, capture_hidden_mode=CaptureHiddenMode.FULL + ) return self._decode_idle_result(on_publish=on_publish) - batch.capture_hidden_mode = CaptureHiddenMode.FULL - batch_output = self.target_worker.forward_batch_generation(batch) + batch_output = self.target_worker.forward_batch_generation( + batch, capture_hidden_mode=CaptureHiddenMode.FULL + ) logits_output = batch_output.logits_output next_token_ids = batch_output.next_token_ids batch_output.new_seq_lens = batch.seq_lens diff --git a/python/sglang/srt/speculative/eagle_utils.py b/python/sglang/srt/speculative/eagle_utils.py index 277b15831..a25b9cbf7 100644 --- a/python/sglang/srt/speculative/eagle_utils.py +++ b/python/sglang/srt/speculative/eagle_utils.py @@ -556,8 +556,12 @@ def eagle_prepare_for_verify( if target_worker.model_runner.spec_algorithm.is_standalone() else CaptureHiddenMode.FULL ) - batch.capture_hidden_mode = capture_mode - verify_forward_batch = ForwardBatch.init_new(batch, target_worker.model_runner) + verify_forward_batch = ForwardBatch.init_new( + batch, + target_worker.model_runner, + capture_hidden_mode=capture_mode, + return_hidden_states_before_norm=False, + ) # Run attention backend plan and cuda graph preparation can_run_cuda_graph = bool( diff --git a/python/sglang/srt/speculative/eagle_worker_v2.py b/python/sglang/srt/speculative/eagle_worker_v2.py index d059cc2c4..4c937c26f 100644 --- a/python/sglang/srt/speculative/eagle_worker_v2.py +++ b/python/sglang/srt/speculative/eagle_worker_v2.py @@ -802,8 +802,12 @@ class EagleDraftWorker(EagleDraftWorkerBase): if self.speculative_algorithm.is_standalone() else CaptureHiddenMode.LAST ) - batch.capture_hidden_mode = capture_hidden_mode - forward_batch = ForwardBatch.init_new(batch, self.draft_runner) + forward_batch = ForwardBatch.init_new( + batch, + self.draft_runner, + capture_hidden_mode=capture_hidden_mode, + return_hidden_states_before_norm=False, + ) forward_batch.return_logprob = False if mm_input_embeds is not None: forward_batch.mm_input_embeds = mm_input_embeds @@ -916,6 +920,7 @@ class EagleDraftWorker(EagleDraftWorkerBase): self.speculative_num_draft_tokens, self.draft_runner, self.cuda_graph_runner_for_draft_extend, + return_hidden_states_before_norm=False, ) if self.plan_stream: @@ -1138,8 +1143,9 @@ class EAGLEWorkerV2(BaseSpecWorker): if self.speculative_algorithm.is_standalone() else CaptureHiddenMode.FULL ) - batch.capture_hidden_mode = target_capture_mode - batch_output = self.target_worker.forward_batch_generation(batch) + batch_output = self.target_worker.forward_batch_generation( + batch, capture_hidden_mode=target_capture_mode + ) # Spec_v2 convention: batch.seq_lens = length BEFORE this iter's tokens. # Extend processed L prompt tokens; next verify iter expects same L. diff --git a/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py b/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py index 2dcd76cb6..e85853aac 100644 --- a/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py +++ b/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py @@ -427,7 +427,11 @@ class FrozenKVMTPDraftWorker(EagleDraftWorkerBase, TpModelWorker): batch.seq_lens_sum = torch.sum(batch.seq_lens).item() batch.return_hidden_states = False - forward_batch = ForwardBatch.init_new(batch, self.draft_model_runner) + forward_batch = ForwardBatch.init_new( + batch, + self.draft_model_runner, + return_hidden_states_before_norm=False, + ) assert forward_batch.capture_hidden_mode == CaptureHiddenMode.LAST self._set_positions(forward_batch) self._expand_for_topk_draft(forward_batch) @@ -710,8 +714,9 @@ class FrozenKVMTPWorkerV2(EAGLEWorkerV2): # size). The draft / seed-based draft-extend hooks are FrozenKVMTPDraftWorker's. if batch.forward_mode.is_extend() or batch.is_extend_in_batch: # Target prefill (frozen is never standalone -> capture FULL hidden). - batch.capture_hidden_mode = CaptureHiddenMode.FULL - batch_output = self.target_worker.forward_batch_generation(batch) + batch_output = self.target_worker.forward_batch_generation( + batch, capture_hidden_mode=CaptureHiddenMode.FULL + ) # Spec_v2 convention: batch.seq_lens = length BEFORE this iter's tokens. batch_output.new_seq_lens = batch.seq_lens diff --git a/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py b/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py index 68699eca6..0d8762593 100644 --- a/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py +++ b/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py @@ -432,9 +432,12 @@ class MultiLayerEagleDraftWorker(EagleDraftWorkerBase): draft_capture_hidden_mode = CaptureHiddenMode.LAST # Run forward - batch.capture_hidden_mode = draft_capture_hidden_mode - batch.return_hidden_states_before_norm = True - forward_batch = ForwardBatch.init_new(batch, self.draft_runner_list[0]) + forward_batch = ForwardBatch.init_new( + batch, + self.draft_runner_list[0], + capture_hidden_mode=draft_capture_hidden_mode, + return_hidden_states_before_norm=True, + ) # Construct input_ids # TODO: same chunked-prefill chain divergence as PR #26329. @@ -530,8 +533,8 @@ class MultiLayerEagleDraftWorker(EagleDraftWorkerBase): self.speculative_num_draft_tokens, self.draft_runner_list[0], self.cuda_graph_runner_for_draft_extend, + return_hidden_states_before_norm=True, ) - forward_batch.return_hidden_states_before_norm = True if self.plan_stream: torch.get_device_module(self.device).current_stream().wait_stream( @@ -721,8 +724,9 @@ class MultiLayerEagleWorkerV2(BaseSpecWorker): if self.speculative_algorithm.is_standalone() else CaptureHiddenMode.FULL ) - batch.capture_hidden_mode = target_capture_mode - batch_output = self.target_worker.forward_batch_generation(batch) + batch_output = self.target_worker.forward_batch_generation( + batch, capture_hidden_mode=target_capture_mode + ) # Spec_v2 convention: batch.seq_lens = length BEFORE this iter's tokens. # Extend processed L prompt tokens; next verify iter expects same L. diff --git a/test/manual/test_forward_split_prefill.py b/test/manual/test_forward_split_prefill.py index 61bd632ff..c54c6456c 100644 --- a/test/manual/test_forward_split_prefill.py +++ b/test/manual/test_forward_split_prefill.py @@ -120,7 +120,11 @@ class TestForwardSplitPrefill(CustomTestCase): batch.forward_mode = ForwardMode.SPLIT_PREFILL # Create forward batch - forward_batch = ForwardBatch.init_new(batch, self.model_runner) + forward_batch = ForwardBatch.init_new( + batch, + self.model_runner, + return_hidden_states_before_norm=False, + ) return forward_batch