main
py 639 lines 19.9 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 from types import SimpleNamespace
5 from unittest.mock import Mock
6
7 import pytest
8 import torch
9
10 from vllm.config.mamba import MambaBackendEnum, MambaConfig, MambaSSUAlgorithm
11 from vllm.model_executor.layers.mamba.mamba_mixer2 import MambaMixer2
12 from vllm.model_executor.layers.mamba.mamba_utils import MambaStateShapeCalculator
13 from vllm.model_executor.layers.mamba.ops.ssu_dispatch import (
14 FlashInferSSUBackend,
15 TritonSSUBackend,
16 commit_replayssm_ring_trackers,
17 get_mamba_ssu_backend,
18 initialize_mamba_ssu_backend,
19 reset_replayssm_ring_trackers,
20 selective_state_update,
21 selective_state_update_replayssm_flashinfer,
22 update_replayssm_ring_trackers,
23 )
24 from vllm.utils.torch_utils import set_random_seed
25 from vllm.v1.attention.backends.mamba2_attn import Mamba2AttentionMetadata
26 from vllm.v1.attention.backends.registry import MambaAttentionBackendEnum
27 from vllm.v1.attention.backends.utils import NULL_BLOCK_ID
28 from vllm.v1.kv_cache_interface import (
29 KVCacheConfig,
30 KVCacheGroupSpec,
31 MambaSpec,
32 )
33
34 try:
35 import flashinfer.mamba # noqa: F401
36
37 HAS_FLASHINFER = True
38 except ImportError:
39 HAS_FLASHINFER = False
40
41 try:
42 from flashinfer.mamba.checkpointing_ssu import CheckpointingSSURunner
43
44 HAS_FLASHINFER_CHECKPOINTING_SSU = CheckpointingSSURunner is not None
45 except ImportError:
46 HAS_FLASHINFER_CHECKPOINTING_SSU = False
47
48
49 @pytest.fixture(autouse=True)
50 def restore_backend_state():
51 import vllm.model_executor.layers.mamba.ops.ssu_dispatch as mod
52
53 old_backend = mod._mamba_ssu_backend
54 old_replayssm_kernel = mod._flashinfer_replayssm_kernel
55 yield
56 mod._mamba_ssu_backend = old_backend
57 mod._flashinfer_replayssm_kernel = old_replayssm_kernel
58
59
60 def test_flashinfer_replayssm_ring_tracker_lifecycle():
61 ring_start = torch.zeros(2, dtype=torch.int32, device="cuda")
62 prev_num_accepted = torch.zeros(2, dtype=torch.int32, device="cuda")
63 prev_query_len = torch.zeros(2, dtype=torch.int32, device="cuda")
64 state_batch_indices = torch.tensor([1], dtype=torch.int32, device="cuda")
65
66 observed = []
67 for _ in range(33):
68 update_replayssm_ring_trackers(
69 ring_start,
70 prev_num_accepted,
71 prev_query_len,
72 state_batch_indices,
73 logical_window=16,
74 ring_buffer_len=17,
75 )
76 observed.append((int(ring_start[1]), int(prev_num_accepted[1])))
77
78 assert observed[4] == (0, 5)
79 assert observed[15] == (0, 16)
80 assert observed[16] == (16, 1)
81 assert observed[31] == (16, 16)
82 assert observed[32] == (15, 1)
83
84 reset_replayssm_ring_trackers(
85 ring_start,
86 prev_num_accepted,
87 prev_query_len,
88 state_batch_indices,
89 )
90 assert (
91 ring_start[1].item(),
92 prev_num_accepted[1].item(),
93 prev_query_len[1].item(),
94 ) == (0, 0, 0)
95
96
97 @pytest.mark.parametrize(
98 ("accepted_sequence", "expected"),
99 [
100 pytest.param(
101 [4] * 22,
102 [
103 (0, 0, 4),
104 (0, 4, 4),
105 (0, 8, 4),
106 (0, 12, 4),
107 (0, 16, 4),
108 (16, 4, 4),
109 (16, 8, 4),
110 (16, 12, 4),
111 (16, 16, 4),
112 (12, 4, 4),
113 (12, 8, 4),
114 (12, 12, 4),
115 (12, 16, 4),
116 (8, 4, 4),
117 (8, 8, 4),
118 (8, 12, 4),
119 (8, 16, 4),
120 (4, 4, 4),
121 (4, 8, 4),
122 (4, 12, 4),
123 (4, 16, 4),
124 (0, 4, 4),
125 ],
126 id="all-accepted",
127 ),
128 pytest.param(
129 [4, 4, 0, 3, 4, 2, 4, 1],
130 [
131 (0, 0, 4),
132 (0, 4, 4),
133 (0, 4, 4),
134 (0, 7, 4),
135 (0, 11, 4),
136 (0, 13, 4),
137 (13, 4, 4),
138 (13, 5, 4),
139 ],
140 id="mixed",
141 ),
142 ],
143 )
144 def test_replayssm_commit_tracker_acceptance_sequence(accepted_sequence, expected):
145 logical_window = 16
146 num_speculative_tokens = 3
147 query_len = 1 + num_speculative_tokens
148 ring_buffer_len = logical_window + 1 + num_speculative_tokens
149 ring_start = torch.zeros(2, dtype=torch.int32, device="cuda")
150 prev_num_accepted = torch.zeros(2, dtype=torch.int32, device="cuda")
151 prev_query_len = torch.zeros(2, dtype=torch.int32, device="cuda")
152 state_batch_indices = torch.tensor([1], dtype=torch.int32, device="cuda")
153 query_start_loc = torch.tensor([0, query_len], dtype=torch.int32, device="cuda")
154
155 observed = []
156 for accepted in accepted_sequence:
157 commit_replayssm_ring_trackers(
158 ring_start,
159 prev_num_accepted,
160 prev_query_len,
161 state_batch_indices,
162 torch.tensor([accepted], dtype=torch.int32, device="cuda"),
163 query_start_loc,
164 logical_window,
165 ring_buffer_len,
166 )
167 snapshot = (
168 ring_start[1].item(),
169 prev_num_accepted[1].item(),
170 prev_query_len[1].item(),
171 )
172 observed.append(snapshot)
173 assert snapshot[1] + snapshot[2] <= ring_buffer_len
174
175 assert observed == expected
176
177
178 def test_replayssm_resume_resets_commit_history():
179 ring_start = torch.tensor([0, 13], dtype=torch.int32, device="cuda")
180 prev_num_accepted = torch.tensor([0, 13], dtype=torch.int32, device="cuda")
181 prev_query_len = torch.tensor([0, 4], dtype=torch.int32, device="cuda")
182 state_batch_indices = torch.tensor([1], dtype=torch.int32, device="cuda")
183
184 reset_replayssm_ring_trackers(
185 ring_start,
186 prev_num_accepted,
187 prev_query_len,
188 state_batch_indices,
189 )
190 assert (
191 ring_start[1].item(),
192 prev_num_accepted[1].item(),
193 prev_query_len[1].item(),
194 ) == (0, 0, 0)
195
196 commit_replayssm_ring_trackers(
197 ring_start,
198 prev_num_accepted,
199 prev_query_len,
200 state_batch_indices,
201 torch.tensor([3], dtype=torch.int32, device="cuda"),
202 torch.tensor([0, 4], dtype=torch.int32, device="cuda"),
203 logical_window=16,
204 ring_buffer_len=20,
205 )
206 assert (
207 ring_start[1].item(),
208 prev_num_accepted[1].item(),
209 prev_query_len[1].item(),
210 ) == (0, 0, 4)
211
212
213 def test_replayssm_commit_tracker_ragged_query_lengths():
214 ring_start = torch.zeros(3, dtype=torch.int32, device="cuda")
215 prev_num_accepted = torch.zeros(3, dtype=torch.int32, device="cuda")
216 prev_query_len = torch.zeros(3, dtype=torch.int32, device="cuda")
217 state_batch_indices = torch.tensor([1, 2], dtype=torch.int32, device="cuda")
218 query_start_loc = torch.tensor([0, 4, 6], dtype=torch.int32, device="cuda")
219
220 observed = []
221 for accepted in ([4, 2], [3, 1]):
222 commit_replayssm_ring_trackers(
223 ring_start,
224 prev_num_accepted,
225 prev_query_len,
226 state_batch_indices,
227 torch.tensor(accepted, dtype=torch.int32, device="cuda"),
228 query_start_loc,
229 logical_window=16,
230 ring_buffer_len=20,
231 )
232 observed.append(
233 [
234 (
235 ring_start[slot].item(),
236 prev_num_accepted[slot].item(),
237 prev_query_len[slot].item(),
238 )
239 for slot in (1, 2)
240 ]
241 )
242
243 assert observed == [[(0, 0, 4), (0, 0, 2)], [(0, 3, 4), (0, 1, 2)]]
244
245
246 @pytest.mark.parametrize("operation", ["commit", "reset"])
247 def test_replayssm_tracker_kernels_mask_invalid_slots(operation):
248 num_states = 3
249 ring_start = torch.tensor([11, 2, 33], dtype=torch.int32, device="cuda")
250 prev_num_accepted = torch.tensor([11, 3, 33], dtype=torch.int32, device="cuda")
251 prev_query_len = torch.tensor([11, 4, 33], dtype=torch.int32, device="cuda")
252 state_batch_indices = torch.tensor(
253 [-1, num_states, NULL_BLOCK_ID, 1], dtype=torch.int32, device="cuda"
254 )
255
256 if operation == "commit":
257 commit_replayssm_ring_trackers(
258 ring_start,
259 prev_num_accepted,
260 prev_query_len,
261 state_batch_indices,
262 torch.tensor([4, 4, 4, 2], dtype=torch.int32, device="cuda"),
263 torch.tensor([0, 4, 8, 12, 16], dtype=torch.int32, device="cuda"),
264 logical_window=16,
265 ring_buffer_len=20,
266 )
267 expected_valid = (2, 5, 4)
268 else:
269 reset_replayssm_ring_trackers(
270 ring_start,
271 prev_num_accepted,
272 prev_query_len,
273 state_batch_indices,
274 )
275 expected_valid = (0, 0, 0)
276
277 assert (
278 ring_start.tolist(),
279 prev_num_accepted.tolist(),
280 prev_query_len.tolist(),
281 ) == (
282 [11, expected_valid[0], 33],
283 [11, expected_valid[1], 33],
284 [11, expected_valid[2], 33],
285 )
286
287
288 def _kv_cache_config_with_ssu(
289 mamba_type: MambaAttentionBackendEnum = MambaAttentionBackendEnum.MAMBA2,
290 ) -> KVCacheConfig:
291 spec = MambaSpec(
292 block_size=16,
293 shapes=((16, 64),),
294 dtypes=(torch.float16,),
295 mamba_type=mamba_type,
296 )
297 return KVCacheConfig(
298 num_blocks=1,
299 kv_cache_tensors=[],
300 kv_cache_groups=[KVCacheGroupSpec(layer_names=["l0"], kv_cache_spec=spec)],
301 )
302
303
304 def test_default_backend_is_triton():
305 initialize_mamba_ssu_backend(MambaConfig(), _kv_cache_config_with_ssu())
306 backend = get_mamba_ssu_backend()
307 assert isinstance(backend, TritonSSUBackend)
308 assert backend.name == "triton"
309
310
311 def test_explicit_triton_backend():
312 initialize_mamba_ssu_backend(
313 MambaConfig(backend=MambaBackendEnum.TRITON), _kv_cache_config_with_ssu()
314 )
315 backend = get_mamba_ssu_backend()
316 assert isinstance(backend, TritonSSUBackend)
317
318
319 @pytest.mark.skipif(not HAS_FLASHINFER, reason="flashinfer not installed")
320 def test_flashinfer_backend_init():
321 initialize_mamba_ssu_backend(
322 MambaConfig(backend=MambaBackendEnum.FLASHINFER), _kv_cache_config_with_ssu()
323 )
324 backend = get_mamba_ssu_backend()
325 assert isinstance(backend, FlashInferSSUBackend)
326 assert backend.name == "flashinfer"
327
328
329 @pytest.mark.skipif(not HAS_FLASHINFER, reason="flashinfer not installed")
330 @pytest.mark.parametrize(
331 ("algorithm", "expected"),
332 [
333 (None, "auto"),
334 ("auto", "auto"),
335 ("simple", "simple"),
336 ("vertical", "vertical"),
337 ("horizontal", "horizontal"),
338 ],
339 )
340 def test_flashinfer_forwards_ssu_algorithm(
341 algorithm: MambaSSUAlgorithm | None,
342 expected: MambaSSUAlgorithm,
343 monkeypatch,
344 ):
345 import flashinfer.mamba
346
347 kernel = Mock()
348 monkeypatch.setattr(flashinfer.mamba, "selective_state_update", kernel)
349 backend = FlashInferSSUBackend(
350 MambaConfig(
351 backend=MambaBackendEnum.FLASHINFER,
352 ssu_algorithm=algorithm,
353 )
354 )
355
356 tensor = torch.empty(1)
357 backend(
358 tensor,
359 tensor,
360 tensor,
361 tensor,
362 tensor,
363 tensor,
364 tensor,
365 tensor,
366 )
367
368 assert kernel.call_args.kwargs["algorithm"] == expected
369
370
371 def test_uninitialized_backend_raises():
372 import vllm.model_executor.layers.mamba.ops.ssu_dispatch as mod
373
374 # restore_backend_state (autouse) puts the global back afterwards.
375 mod._mamba_ssu_backend = None
376 with pytest.raises(RuntimeError, match="not been initialized"):
377 get_mamba_ssu_backend()
378
379
380 @pytest.mark.parametrize(
381 "mamba_type",
382 [
383 MambaAttentionBackendEnum.LINEAR,
384 MambaAttentionBackendEnum.GDN_ATTN,
385 MambaAttentionBackendEnum.SHORT_CONV,
386 ],
387 )
388 def test_init_is_noop_for_non_ssu_mamba_type(mamba_type):
389 import vllm.model_executor.layers.mamba.ops.ssu_dispatch as mod
390
391 old = mod._mamba_ssu_backend
392 mod._mamba_ssu_backend = None
393 try:
394 initialize_mamba_ssu_backend(
395 MambaConfig(), _kv_cache_config_with_ssu(mamba_type)
396 )
397 assert mod._mamba_ssu_backend is None
398 with pytest.raises(RuntimeError, match="not been initialized"):
399 get_mamba_ssu_backend()
400 finally:
401 mod._mamba_ssu_backend = old
402
403
404 @pytest.mark.skipif(HAS_FLASHINFER, reason="flashinfer is installed")
405 def test_flashinfer_import_error():
406 with pytest.raises(ImportError, match="FlashInfer is required"):
407 FlashInferSSUBackend(MambaConfig())
408
409
410 def test_triton_basic_call():
411 set_random_seed(0)
412 initialize_mamba_ssu_backend(
413 MambaConfig(backend=MambaBackendEnum.TRITON), _kv_cache_config_with_ssu()
414 )
415 device = "cuda"
416 batch_size = 2
417 dim = 64
418 dstate = 16
419
420 state = torch.randn(batch_size, dim, dstate, device=device)
421 x = torch.randn(batch_size, dim, device=device)
422 out = torch.empty_like(x)
423 dt = torch.randn(batch_size, dim, device=device)
424 dt_bias = torch.rand(dim, device=device) - 4.0
425 A = -torch.rand(dim, dstate, device=device)
426 B = torch.randn(batch_size, dstate, device=device)
427 C = torch.randn(batch_size, dstate, device=device)
428 D = torch.randn(dim, device=device)
429
430 selective_state_update(
431 state,
432 x,
433 dt,
434 A,
435 B,
436 C,
437 D=D,
438 dt_bias=dt_bias,
439 dt_softplus=True,
440 out=out,
441 )
442 assert not torch.isnan(out).any()
443
444
445 @pytest.mark.parametrize("layout", ["packed", "dense"])
446 def test_replayssm_flashinfer_call_forwards_mtp_layout(monkeypatch, layout):
447 import vllm.model_executor.layers.mamba.ops.ssu_dispatch as mod
448
449 kernel = Mock(return_value=torch.empty(0))
450 monkeypatch.setattr(mod, "_flashinfer_replayssm_kernel", kernel)
451
452 batch, max_seqlen, nheads, dim, dstate, ngroups = 2, 4, 2, 4, 8, 1
453 state = torch.empty(2, nheads, dim, dstate)
454 x_shape: tuple[int, ...]
455 B_shape: tuple[int, ...]
456 expected_x_shape: tuple[int, ...]
457 expected_B_shape: tuple[int, ...]
458 if layout == "packed":
459 x_shape = (6, nheads, dim)
460 B_shape = (6, ngroups, dstate)
461 expected_x_shape = (1, 6, nheads, dim)
462 expected_B_shape = (1, 6, ngroups, dstate)
463 cu_seqlens = torch.tensor([0, 4, 6], dtype=torch.int32)
464 kernel_max_seqlen = max_seqlen
465 else:
466 x_shape = (batch, max_seqlen, nheads, dim)
467 B_shape = (batch, max_seqlen, ngroups, dstate)
468 expected_x_shape = x_shape
469 expected_B_shape = B_shape
470 cu_seqlens = None
471 kernel_max_seqlen = None
472 x = torch.empty(x_shape)
473 dt = torch.empty_like(x)
474 A = torch.empty(nheads, dim, dstate)
475 B = torch.empty(B_shape)
476 C = torch.empty_like(B)
477 out = torch.empty_like(x)
478 x_cache = torch.empty(2, nheads, 20, dim)
479 dt_cache = torch.empty(2, nheads, 20)
480 B_cache = torch.empty(2, ngroups, 20, dstate)
481 ring_start = torch.zeros(2, dtype=torch.int32)
482 prev_num_accepted = torch.zeros(2, dtype=torch.int32)
483 prev_query_len = torch.zeros(2, dtype=torch.int32)
484 selective_state_update_replayssm_flashinfer(
485 state,
486 x,
487 dt,
488 A,
489 B,
490 C,
491 out,
492 x_cache,
493 B_cache,
494 dt_cache,
495 ring_start,
496 prev_num_accepted,
497 prev_query_len,
498 logical_window=16,
499 state_batch_indices=torch.tensor([0, 1], dtype=torch.int32),
500 cu_seqlens=cu_seqlens,
501 max_seqlen=kernel_max_seqlen,
502 update_trackers=False,
503 )
504
505 args = kernel.call_args.args
506 assert args[6].shape == expected_x_shape
507 assert args[7].shape == expected_x_shape
508 assert args[9].shape == expected_B_shape
509 assert args[10].shape == expected_B_shape
510 assert args[11].shape == expected_x_shape
511 assert kernel.call_args.kwargs["cu_seqlens"] is cu_seqlens
512 assert kernel.call_args.kwargs["max_seqlen"] == kernel_max_seqlen
513
514
515 @pytest.mark.parametrize(
516 ("query_start_loc", "expected_shape", "expected_max_seqlen"),
517 [
518 pytest.param([0, 4, 8], (2, 4, 2, 4), None, id="dense"),
519 pytest.param([0, 4, 6], (6, 2, 4), 4, id="packed"),
520 ],
521 )
522 def test_replayssm_mixer_selects_mtp_layout(
523 monkeypatch, query_start_loc, expected_shape, expected_max_seqlen
524 ):
525 import vllm.model_executor.layers.mamba.mamba_mixer2 as mod
526
527 mixer = MambaMixer2.__new__(MambaMixer2)
528 torch.nn.Module.__init__(mixer)
529 mixer.prefix = "mixer"
530 mixer.tped_intermediate_size = 0
531 mixer.tped_conv_size = 1
532 mixer.tped_dt_size = 2
533 mixer.num_heads = 2
534 mixer.head_dim = 4
535 mixer.n_groups = mixer.tp_size = 1
536 mixer.ssm_state_size = 8
537 mixer.num_spec = 3
538 mixer.use_replayssm = True
539 mixer.replayssm_buffer_len = 16
540 mixer._commits_replayssm_trackers = True
541 mixer._updates_replayssm_trackers = False
542 mixer.mamba_config = MambaConfig(backend=MambaBackendEnum.FLASHINFER)
543 mixer.cache_config = SimpleNamespace(mamba_block_size=16, mamba_cache_mode="none")
544 mixer.conv_weights = torch.empty(0)
545 mixer.conv1d = SimpleNamespace(bias=None)
546 mixer.activation = "silu"
547 mixer.A = torch.empty(2)
548 mixer.dt_bias = torch.empty(2)
549 mixer.D = torch.empty(2)
550 mixer._replayssm_ring_start = torch.zeros(3, dtype=torch.int32)
551 mixer._replayssm_prev_num_accepted = torch.zeros(3, dtype=torch.int32)
552 mixer._replayssm_prev_query_len = torch.zeros(3, dtype=torch.int32)
553 mixer.kv_cache = (
554 torch.empty(3, 1),
555 torch.empty(3, 2, 4, 8),
556 torch.empty(3, 2, 20, 4),
557 torch.empty(3, 2, 20),
558 torch.empty(3, 1, 20, 8),
559 )
560
561 num_decode_tokens = query_start_loc[-1]
562 query_start_loc_d = torch.tensor(query_start_loc, dtype=torch.int32)
563 metadata = Mamba2AttentionMetadata(
564 num_prefills=0,
565 num_prefill_tokens=0,
566 num_decodes=2,
567 num_decode_tokens=num_decode_tokens,
568 num_reqs=2,
569 has_initial_states_p=None,
570 query_start_loc_p=None,
571 state_indices_tensor_p=None,
572 state_indices_tensor_d=torch.tensor([[1], [2]], dtype=torch.int32),
573 query_start_loc_d=query_start_loc_d,
574 num_accepted_tokens=torch.tensor([4, 2], dtype=torch.int32),
575 seq_lens=torch.tensor([104, 102], dtype=torch.int32),
576 replayssm_scratch=(torch.empty(0), torch.empty(0), torch.empty(0)),
577 replayssm_state_indices_d=torch.tensor([1, 2], dtype=torch.int32),
578 )
579
580 def split_hidden_states_B_C(values):
581 tokens = values.size(0)
582 return (
583 torch.empty(tokens, 8),
584 torch.empty(tokens, 8),
585 torch.empty(tokens, 8),
586 )
587
588 mixer.split_hidden_states_B_C_fn = split_hidden_states_B_C
589 kernel = Mock()
590 monkeypatch.setattr(
591 mod,
592 "get_forward_context",
593 lambda: SimpleNamespace(attn_metadata={mixer.prefix: metadata}),
594 )
595 monkeypatch.setattr(mod, "commit_replayssm_ring_trackers", Mock())
596 monkeypatch.setattr(
597 mod, "causal_conv1d_update", lambda values, *args, **kwargs: values
598 )
599 monkeypatch.setattr(mod, "selective_state_update_replayssm_flashinfer", kernel)
600
601 mixer.conv_ssm_forward(
602 torch.empty(num_decode_tokens, 3), torch.empty(num_decode_tokens, 8)
603 )
604
605 assert kernel.call_args.args[1].shape == expected_shape
606 if expected_max_seqlen is None:
607 assert kernel.call_args.kwargs["cu_seqlens"] is None
608 else:
609 assert kernel.call_args.kwargs["cu_seqlens"] is query_start_loc_d
610 assert kernel.call_args.kwargs["max_seqlen"] == expected_max_seqlen
611
612
613 @pytest.mark.parametrize(
614 ("backend", "num_speculative_tokens", "expected_ring_len"),
615 [
616 (MambaBackendEnum.TRITON, 0, 16),
617 (MambaBackendEnum.FLASHINFER, 0, 17),
618 (MambaBackendEnum.FLASHINFER, 3, 20),
619 ],
620 )
621 def test_replayssm_physical_ring_shape(
622 backend, num_speculative_tokens, expected_ring_len
623 ):
624 base_shapes = ((64, 3), (8, 4, 16))
625
626 shapes = MambaStateShapeCalculator.append_replayssm_ring(
627 base_shapes,
628 n_groups=4,
629 tp_world_size=2,
630 logical_window=16,
631 backend=backend,
632 num_speculative_tokens=num_speculative_tokens,
633 )
634
635 assert shapes[2:] == (
636 (8, expected_ring_len, 4),
637 (8, expected_ring_len),
638 (2, expected_ring_len, 16),
639 )