Skip to content
Closed
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 22 additions & 3 deletions python/sglang/srt/layers/dp_attention.py
Original file line number Diff line number Diff line change
Expand Up @@ -355,16 +355,27 @@ def disable_dp_size():


def get_dp_local_info(forward_batch: ForwardBatch) -> Tuple[torch.Tensor, torch.Tensor]:
# `get_dp_local_info` is only called in global DP gather and scatter. We use global DP rank here.
# Compute local token range for current DP rank.
# For LogitsMetadata, use global_num_tokens_for_logprob_gpu (pruned tokens needing logits).
# For ForwardBatch (MLP/Attention), use global_num_tokens_gpu (all tokens in batch).
dp_rank = get_attention_dp_rank()

if forward_batch.dp_local_start_pos is None:
cumtokens = torch.cumsum(forward_batch.global_num_tokens_gpu, dim=0)
# Select metadata source based on context
if (

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How can dp_local_start_pos be None in LogitsMetadata? It is expected to be processed here:

logits_metadata.compute_dp_attention_metadata()
.

LogitsMetadata has a separate processing logic. We should keep it separated. Perhaps we should add assertion like assert isinstance(forward_batch, ForwardBatch) to avoid misuse.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, understand the design! The call logic should indeed stay separated at the outer level. Will have another patch soon after verified

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ch-wan Just pushed another patch, could you take another look? Thanks

type(forward_batch).__name__ == "LogitsMetadata"
and forward_batch.global_num_tokens_for_logprob_gpu is not None
):
global_num_tokens_source = forward_batch.global_num_tokens_for_logprob_gpu
else:
global_num_tokens_source = forward_batch.global_num_tokens_gpu

cumtokens = torch.cumsum(global_num_tokens_source, dim=0)
if dp_rank == 0:
local_start_pos = torch.zeros_like(cumtokens[0])
else:
local_start_pos = cumtokens[dp_rank - 1]
local_num_tokens = forward_batch.global_num_tokens_gpu[dp_rank]
local_num_tokens = global_num_tokens_source[dp_rank]

forward_batch.dp_local_start_pos = local_start_pos
forward_batch.dp_local_num_tokens = local_num_tokens
Expand Down Expand Up @@ -421,6 +432,14 @@ def _dp_gather_via_all_reduce(
):
local_start_pos, local_num_tokens = get_dp_local_info(forward_batch)

# Fix potential metadata mismatch: scheduler's num_tokens_for_logprob may not account
# for pruning done by logits_processor. Use actual tensor size as authoritative source.
actual_local_size = local_tokens.shape[0]
if isinstance(local_num_tokens, torch.Tensor):
local_num_tokens.fill_(actual_local_size) # CUDA graph compatibility
else:
local_num_tokens = actual_local_size

global_tokens.fill_(0)
assert local_tokens.is_contiguous()
assert global_tokens.is_contiguous()
Expand Down
Loading