| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | from types import MethodType, SimpleNamespace |
| 5 | from typing import get_args |
| 6 | |
| 7 | import numpy as np |
| 8 | import pytest |
| 9 | import torch |
| 10 | |
| 11 | from vllm.config.model import PROCESSED_LOGPROBS_MODES, LogprobsMode |
| 12 | from vllm.platforms import current_platform |
| 13 | from vllm.v1.worker.gpu.sample.output import SamplingMaskTensors |
| 14 | from vllm.v1.worker.gpu.spec_decode.rejection_sampler import ( |
| 15 | RejectionSampler, |
| 16 | _iter_request_chunks, |
| 17 | ) |
| 18 | |
| 19 | |
| 20 | def _make_input_batch( |
| 21 | cu_num_logits_np: np.ndarray, |
| 22 | idx_mapping_np: np.ndarray, |
| 23 | device: torch.device, |
| 24 | ) -> SimpleNamespace: |
| 25 | """Minimal InputBatch fields read by RejectionSampler._verify_in_chunks.""" |
| 26 | num_reqs = len(idx_mapping_np) |
| 27 | num_logits_per_req = np.diff(cu_num_logits_np) |
| 28 | return SimpleNamespace( |
| 29 | num_reqs=num_reqs, |
| 30 | cu_num_logits_np=cu_num_logits_np, |
| 31 | cu_num_logits=torch.from_numpy(cu_num_logits_np).to(device), |
| 32 | idx_mapping_np=idx_mapping_np, |
| 33 | idx_mapping=torch.from_numpy(idx_mapping_np).to(device), |
| 34 | expanded_idx_mapping=torch.from_numpy( |
| 35 | np.repeat(idx_mapping_np, num_logits_per_req) |
| 36 | ).to(device), |
| 37 | expanded_local_pos=torch.from_numpy( |
| 38 | np.concatenate( |
| 39 | [np.arange(count, dtype=np.int32) for count in num_logits_per_req] |
| 40 | ) |
| 41 | ).to(device), |
| 42 | seq_lens_cpu_upper_bound=torch.full((num_reqs,), 64, dtype=torch.int32), |
| 43 | ) |
| 44 | |
| 45 | |
| 46 | def _make_rejection_sampler( |
| 47 | sampler: SimpleNamespace, num_speculative_steps: int |
| 48 | ) -> RejectionSampler: |
| 49 | """RejectionSampler for standard, fixed-boundary, unwatermarked verification.""" |
| 50 | rejection_sampler = object.__new__(RejectionSampler) |
| 51 | rejection_sampler.sampler = sampler |
| 52 | rejection_sampler.num_speculative_steps = num_speculative_steps |
| 53 | rejection_sampler.enable_adaptive_verification = False |
| 54 | rejection_sampler.synthetic_conditional_rates = None |
| 55 | rejection_sampler.use_block_verification = False |
| 56 | rejection_sampler.watermark_key = None |
| 57 | return rejection_sampler |
| 58 | |
| 59 | |
| 60 | def test_iter_request_chunks_preserves_request_boundaries(): |
| 61 | cu_num_logits = np.array([0, 3, 4, 11, 13], dtype=np.int32) |
| 62 | |
| 63 | assert list(_iter_request_chunks(cu_num_logits, max_chunk_logits=5)) == [ |
| 64 | (0, 2), |
| 65 | (2, 3), |
| 66 | (3, 4), |
| 67 | ] |
| 68 | |
| 69 | |
| 70 | @pytest.mark.skipif(not current_platform.is_cuda(), reason="Requires CUDA") |
| 71 | @pytest.mark.parametrize("logprobs_mode", get_args(LogprobsMode)) |
| 72 | def test_chunked_scores_match_full_batch(logprobs_mode: str): |
| 73 | device = torch.device("cuda") |
| 74 | cu_num_logits_np = np.array([0, 3, 4, 8, 10], dtype=np.int32) |
| 75 | num_logits_per_req = np.diff(cu_num_logits_np) |
| 76 | idx_mapping_np = np.array([7, 2, 9, 1], dtype=np.int32) |
| 77 | input_batch = _make_input_batch(cu_num_logits_np, idx_mapping_np, device) |
| 78 | rejection_sampler = _make_rejection_sampler( |
| 79 | SimpleNamespace(logprobs_mode=logprobs_mode, return_sampling_mask=False), |
| 80 | num_speculative_steps=3, |
| 81 | ) |
| 82 | |
| 83 | def fake_verify( |
| 84 | self, |
| 85 | logits, |
| 86 | _draft_logits, |
| 87 | _draft_sampled, |
| 88 | _pos, |
| 89 | cu_num_logits, |
| 90 | idx_mapping, |
| 91 | *_mappings, |
| 92 | ): |
| 93 | num_sampled = torch.diff(cu_num_logits).to(torch.int32) |
| 94 | sampled = ( |
| 95 | idx_mapping.to(torch.int64).unsqueeze(1) + torch.arange(4, device=device) |
| 96 | ) % logits.shape[1] |
| 97 | return logits.float() + 1, sampled, num_sampled |
| 98 | |
| 99 | rejection_sampler._verify = MethodType(fake_verify, rejection_sampler) |
| 100 | logits = torch.arange(170, dtype=torch.float32, device=device).view(10, 17) |
| 101 | |
| 102 | sampled, num_sampled, chunked_logprobs, sampling_mask_tensors = ( |
| 103 | rejection_sampler._verify_in_chunks( |
| 104 | logits, |
| 105 | input_batch, |
| 106 | draft_logits=None, |
| 107 | draft_sampled=torch.arange(10, device=device), |
| 108 | pos=torch.arange(10, device=device), |
| 109 | max_chunk_logits=5, |
| 110 | max_num_logprobs=2, |
| 111 | ) |
| 112 | ) |
| 113 | score_logits = logits + 1 if logprobs_mode in PROCESSED_LOGPROBS_MODES else logits |
| 114 | full_logprobs = rejection_sampler._get_logprobs_tensors( |
| 115 | sampled, |
| 116 | num_sampled, |
| 117 | score_logits, |
| 118 | input_batch.cu_num_logits, |
| 119 | input_batch.cu_num_logits_np, |
| 120 | max_num_logprobs=2, |
| 121 | ) |
| 122 | |
| 123 | assert sampled[:, 0].tolist() == idx_mapping_np.tolist() |
| 124 | assert num_sampled.tolist() == num_logits_per_req.tolist() |
| 125 | assert chunked_logprobs is not None |
| 126 | assert full_logprobs is not None |
| 127 | assert torch.equal( |
| 128 | chunked_logprobs.logprob_token_ids, |
| 129 | full_logprobs.logprob_token_ids, |
| 130 | ) |
| 131 | assert torch.equal(chunked_logprobs.logprobs, full_logprobs.logprobs) |
| 132 | assert torch.equal( |
| 133 | chunked_logprobs.selected_token_ranks, |
| 134 | full_logprobs.selected_token_ranks, |
| 135 | ) |
| 136 | assert ( |
| 137 | chunked_logprobs.cu_num_generated_tokens |
| 138 | == full_logprobs.cu_num_generated_tokens |
| 139 | ) |
| 140 | assert sampling_mask_tensors is None |
| 141 | |
| 142 | |
| 143 | @pytest.mark.skipif(not current_platform.is_cuda(), reason="Requires CUDA") |
| 144 | def test_replay_on_off_preserves_rejection_sampling_and_rng(monkeypatch): |
| 145 | device = torch.device("cuda") |
| 146 | num_reqs = 3 |
| 147 | num_speculative_steps = 3 |
| 148 | rows_per_request = num_speculative_steps + 1 |
| 149 | vocab_size = 8 |
| 150 | cu_num_logits_np = np.arange(0, 13, rows_per_request, dtype=np.int32) |
| 151 | idx_mapping_np = np.arange(num_reqs, dtype=np.int32) |
| 152 | input_batch = _make_input_batch(cu_num_logits_np, idx_mapping_np, device) |
| 153 | processed_logits = torch.zeros((12, vocab_size), device=device) |
| 154 | processed_logits[0, 0] = -float("inf") |
| 155 | processed_logits[rows_per_request + 1, 1] = -float("inf") |
| 156 | draft_logits = torch.zeros( |
| 157 | (num_reqs, num_speculative_steps, vocab_size), device=device |
| 158 | ) |
| 159 | draft_sampled = torch.zeros( |
| 160 | (num_reqs, rows_per_request), dtype=torch.int64, device=device |
| 161 | ) |
| 162 | draft_sampled[:, 1:] = torch.tensor([0, 1, 2], device=device) |
| 163 | draft_sampled = draft_sampled.flatten() |
| 164 | sampler = SimpleNamespace( |
| 165 | logprobs_mode="processed_logprobs", |
| 166 | return_sampling_mask=False, |
| 167 | sampling_states=SimpleNamespace( |
| 168 | top_k=SimpleNamespace(np=np.full(num_reqs, vocab_size)), |
| 169 | temperature=SimpleNamespace(gpu=torch.ones(num_reqs, device=device)), |
| 170 | seeds=SimpleNamespace( |
| 171 | gpu=torch.arange(num_reqs, dtype=torch.int64, device=device) |
| 172 | ), |
| 173 | ), |
| 174 | use_fp64_gumbel=False, |
| 175 | apply_sampling_params=lambda logits, *_: logits, |
| 176 | ) |
| 177 | rejection_sampler = _make_rejection_sampler(sampler, num_speculative_steps) |
| 178 | pos = torch.arange(12, dtype=torch.int32, device=device) |
| 179 | |
| 180 | pack_calls: list[tuple[list[int], int, int]] = [] |
| 181 | pack = SamplingMaskTensors.from_logits.__func__ |
| 182 | |
| 183 | def track_pack( |
| 184 | cls, |
| 185 | logits, |
| 186 | cu_num_logits, |
| 187 | num_sampled_tokens, |
| 188 | max_num_kept, |
| 189 | rows_per_request=1, |
| 190 | ): |
| 191 | pack_calls.append( |
| 192 | (cu_num_logits.cpu().tolist(), max_num_kept, rows_per_request) |
| 193 | ) |
| 194 | return pack( |
| 195 | cls, |
| 196 | logits, |
| 197 | cu_num_logits, |
| 198 | num_sampled_tokens, |
| 199 | max_num_kept, |
| 200 | rows_per_request, |
| 201 | ) |
| 202 | |
| 203 | monkeypatch.setattr( |
| 204 | SamplingMaskTensors, |
| 205 | "from_logits", |
| 206 | classmethod(track_pack), |
| 207 | ) |
| 208 | |
| 209 | rng_before = torch.cuda.get_rng_state() |
| 210 | off_sampled, off_counts, off_logprobs, off_masks = ( |
| 211 | rejection_sampler._verify_in_chunks( |
| 212 | processed_logits, |
| 213 | input_batch, |
| 214 | draft_logits, |
| 215 | draft_sampled, |
| 216 | pos, |
| 217 | max_chunk_logits=8, |
| 218 | max_num_logprobs=0, |
| 219 | ) |
| 220 | ) |
| 221 | rng_after_off = torch.cuda.get_rng_state() |
| 222 | assert pack_calls == [] |
| 223 | |
| 224 | sampler.return_sampling_mask = True |
| 225 | on_sampled, on_counts, on_logprobs, on_masks = rejection_sampler._verify_in_chunks( |
| 226 | processed_logits, |
| 227 | input_batch, |
| 228 | draft_logits, |
| 229 | draft_sampled, |
| 230 | pos, |
| 231 | max_chunk_logits=8, |
| 232 | max_num_logprobs=0, |
| 233 | ) |
| 234 | rng_after_on = torch.cuda.get_rng_state() |
| 235 | |
| 236 | assert pack_calls == [ |
| 237 | ([0, 4, 8], vocab_size, rows_per_request), |
| 238 | ([0, 4], vocab_size, rows_per_request), |
| 239 | ] |
| 240 | assert off_masks is None and on_masks is not None |
| 241 | assert on_masks.rows_per_request == rows_per_request |
| 242 | assert on_masks.token_ids.shape == (num_reqs * rows_per_request, vocab_size) |
| 243 | assert on_masks.packed_mask.shape[0] == num_reqs * rows_per_request |
| 244 | assert off_counts.tolist() == [1, 2, 4] |
| 245 | assert torch.equal(off_counts, on_counts) |
| 246 | counts = off_counts.cpu().tolist() |
| 247 | for req_idx, count in enumerate(counts): |
| 248 | assert torch.equal(off_sampled[req_idx, :count], on_sampled[req_idx, :count]) |
| 249 | |
| 250 | mapped_rows = [ |
| 251 | int(input_batch.cu_num_logits_np[req_idx]) + slot_idx |
| 252 | for req_idx, count in enumerate(counts) |
| 253 | for slot_idx in range(count) |
| 254 | ] |
| 255 | assert off_logprobs is not None and on_logprobs is not None |
| 256 | assert torch.equal( |
| 257 | off_logprobs.logprob_token_ids[mapped_rows], |
| 258 | on_logprobs.logprob_token_ids[mapped_rows], |
| 259 | ) |
| 260 | assert torch.equal( |
| 261 | off_logprobs.logprobs[mapped_rows], on_logprobs.logprobs[mapped_rows] |
| 262 | ) |
| 263 | assert torch.equal( |
| 264 | off_logprobs.selected_token_ranks[mapped_rows], |
| 265 | on_logprobs.selected_token_ranks[mapped_rows], |
| 266 | ) |
| 267 | assert torch.equal(rng_before, rng_after_off) |
| 268 | assert torch.equal(rng_after_off, rng_after_on) |
| 269 | |
| 270 | masks = ( |
| 271 | on_masks.to_cpu_nonblocking().tolists(on_counts.cpu().numpy()).to_nested_list() |
| 272 | ) |
| 273 | expected_masks = [ |
| 274 | torch.isfinite(processed_logits[row]).nonzero().flatten().tolist() |
| 275 | for row in mapped_rows |
| 276 | ] |
| 277 | emitted = [ |
| 278 | int(on_sampled[req_idx, slot_idx]) |
| 279 | for req_idx, count in enumerate(on_counts.cpu().tolist()) |
| 280 | for slot_idx in range(count) |
| 281 | ] |
| 282 | expected_logprobs = torch.stack( |
| 283 | [ |
| 284 | torch.log_softmax(processed_logits[row], dim=-1)[token_id] |
| 285 | for row, token_id in zip(mapped_rows, emitted) |
| 286 | ] |
| 287 | ) |
| 288 | |
| 289 | assert masks == expected_masks |
| 290 | assert all(token_id in mask for token_id, mask in zip(emitted, masks)) |
| 291 | assert torch.equal(on_logprobs.logprobs[mapped_rows, 0], expected_logprobs) |