[DSV4] Avoid host syncs in EAGLE prefill (#33662)
Co-authored-by: weireweire <20922698+weireweire@users.noreply.github.com>
This commit is contained in:
@@ -1824,6 +1824,14 @@ class DeepseekV4AttnBackend(
|
|||||||
if cache is None:
|
if cache is None:
|
||||||
seq_lens_cpu = forward_batch.seq_lens_cpu
|
seq_lens_cpu = forward_batch.seq_lens_cpu
|
||||||
assert seq_lens_cpu is not None
|
assert seq_lens_cpu is not None
|
||||||
|
extend_seq_lens_cpu = forward_batch.extend_seq_lens_cpu
|
||||||
|
assert extend_seq_lens_cpu is not None
|
||||||
|
total_swa = sum(
|
||||||
|
min(int(seq_len), int(extend_len) + SWA_WINDOW - 1)
|
||||||
|
for seq_len, extend_len in zip(
|
||||||
|
seq_lens_cpu.tolist(), extend_seq_lens_cpu, strict=True
|
||||||
|
)
|
||||||
|
)
|
||||||
# ``swa_window_size`` on the pool is its storage page size, not
|
# ``swa_window_size`` on the pool is its storage page size, not
|
||||||
# the model's SWA window — pass both explicitly.
|
# the model's SWA window — pass both explicitly.
|
||||||
cache = SparsePrefillChunkCache.build(
|
cache = SparsePrefillChunkCache.build(
|
||||||
@@ -1836,6 +1844,7 @@ class DeepseekV4AttnBackend(
|
|||||||
swa_page_size=token_to_kv_pool.swa_window_size,
|
swa_page_size=token_to_kv_pool.swa_window_size,
|
||||||
num_qo_tokens=q_flat.shape[0],
|
num_qo_tokens=q_flat.shape[0],
|
||||||
max_seq_len=int(seq_lens_cpu.max().item()),
|
max_seq_len=int(seq_lens_cpu.max().item()),
|
||||||
|
total_swa=total_swa,
|
||||||
)
|
)
|
||||||
self.forward_metadata.sparse_prefill_cache = cache
|
self.forward_metadata.sparse_prefill_cache = cache
|
||||||
|
|
||||||
|
|||||||
@@ -195,6 +195,7 @@ def build_swa_token_ids(
|
|||||||
req_to_token: torch.Tensor,
|
req_to_token: torch.Tensor,
|
||||||
full_to_swa: torch.Tensor,
|
full_to_swa: torch.Tensor,
|
||||||
swa_window: int,
|
swa_window: int,
|
||||||
|
total_swa: int,
|
||||||
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
|
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
|
||||||
"""Build a flat list of physical SWA-cache token IDs covering each
|
"""Build a flat list of physical SWA-cache token IDs covering each
|
||||||
request's positional union of every query's SWA window.
|
request's positional union of every query's SWA window.
|
||||||
@@ -216,6 +217,7 @@ def build_swa_token_ids(
|
|||||||
full_to_swa: (full_pool_size + extra,) int64. Maps full kv id to
|
full_to_swa: (full_pool_size + extra,) int64. Maps full kv id to
|
||||||
SWA-cache id.
|
SWA-cache id.
|
||||||
swa_window: int. SWA window size.
|
swa_window: int. SWA window size.
|
||||||
|
total_swa: Number of token IDs to allocate, computed from CPU lengths.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
swa_token_ids: (total_swa,) int32, flat physical SWA-cache token IDs.
|
swa_token_ids: (total_swa,) int32, flat physical SWA-cache token IDs.
|
||||||
@@ -238,8 +240,6 @@ def build_swa_token_ids(
|
|||||||
swa_first_pos = (seq_lens - swa_gather_lens).to(torch.int32)
|
swa_first_pos = (seq_lens - swa_gather_lens).to(torch.int32)
|
||||||
swa_offsets = torch.zeros(num_reqs + 1, dtype=torch.int32, device=device)
|
swa_offsets = torch.zeros(num_reqs + 1, dtype=torch.int32, device=device)
|
||||||
swa_offsets[1:] = torch.cumsum(swa_gather_lens, dim=0).to(torch.int32)
|
swa_offsets[1:] = torch.cumsum(swa_gather_lens, dim=0).to(torch.int32)
|
||||||
total_swa = int(swa_offsets[-1].item()) # one CPU sync per chunk
|
|
||||||
|
|
||||||
swa_token_ids = torch.empty(total_swa, dtype=torch.int32, device=device)
|
swa_token_ids = torch.empty(total_swa, dtype=torch.int32, device=device)
|
||||||
if total_swa == 0:
|
if total_swa == 0:
|
||||||
return swa_token_ids, swa_first_pos, swa_gather_lens, swa_offsets
|
return swa_token_ids, swa_first_pos, swa_gather_lens, swa_offsets
|
||||||
@@ -322,6 +322,7 @@ class SparsePrefillChunkCache:
|
|||||||
swa_page_size: int,
|
swa_page_size: int,
|
||||||
num_qo_tokens: int,
|
num_qo_tokens: int,
|
||||||
max_seq_len: int,
|
max_seq_len: int,
|
||||||
|
total_swa: int,
|
||||||
) -> "SparsePrefillChunkCache":
|
) -> "SparsePrefillChunkCache":
|
||||||
device = seq_lens.device
|
device = seq_lens.device
|
||||||
num_reqs = seq_lens.shape[0]
|
num_reqs = seq_lens.shape[0]
|
||||||
@@ -337,6 +338,7 @@ class SparsePrefillChunkCache:
|
|||||||
req_to_token=req_to_token,
|
req_to_token=req_to_token,
|
||||||
full_to_swa=full_to_swa,
|
full_to_swa=full_to_swa,
|
||||||
swa_window=swa_window_size,
|
swa_window=swa_window_size,
|
||||||
|
total_swa=total_swa,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -92,7 +92,10 @@ def _eagle_prefill_tail_tokens(
|
|||||||
for i, r in enumerate(batch.reqs):
|
for i, r in enumerate(batch.reqs):
|
||||||
if r is batch.chunked_req:
|
if r is batch.chunked_req:
|
||||||
tail_tokens = tail_tokens.clone()
|
tail_tokens = tail_tokens.clone()
|
||||||
tail_tokens[i] = next_prompt_token
|
# Keep the scalar as a kernel argument. Assigning a Python scalar
|
||||||
|
# through scalar indexing issues a pageable H2D copy and
|
||||||
|
# synchronizes the current CUDA stream before draft extend.
|
||||||
|
tail_tokens[i : i + 1].fill_(next_prompt_token)
|
||||||
break
|
break
|
||||||
return tail_tokens
|
return tail_tokens
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user