Pass per-forward overrides to ForwardBatch.init_new as explicit arguments (#30670)
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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())
|
||||
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user