support kv cache cpu offload with connector

Signed-off-by: lidenghui <lidenghui1110@gmail.com>
Signed-off-by: AlvisGong <gwly0401@163.com>

Author: AlvisGong

tests/ut/attention/test_mla_v1.py

@@ -554,7 +554,11 @@ def test_mla_preprocess(self, magic_npu_fetch):
         self.impl.num_kv_heads = self.impl.num_heads
 
         decode_res, prefill_res = self.impl._mla_preprocess(
-            hidden_states, kv_cache, attn_metadata, need_gather_q_kv=False)
+            "mock_layer",
+            hidden_states,
+            kv_cache,
+            attn_metadata,
+            need_gather_q_kv=False)
 
         self.assertIsNotNone(decode_res)
         self.assertIsNotNone(prefill_res)

tests/ut/distributed/cpu_offload/test_cpu_kv_cache_manager.py

@@ -22,10 +22,9 @@
 from vllm.distributed.parallel_state import GroupCoordinator
 
 from vllm_ascend.torchair.models.torchair_deepseek_v2 import (
-    TorchairDeepseekV2DecoderLayer, TorchairDeepseekV2ForCausalLM,
-    TorchairDeepseekV2MergedReplicatedLinear, TorchairDeepseekV2MLAAttention,
-    TorchairDeepseekV2MLP, TorchairDeepseekV2MoE,
-    TorchairDeepseekV2RowParallelLinear,
+    TorchairDeepseekV2DecoderLayer, TorchairDeepseekV2MergedReplicatedLinear,
+    TorchairDeepseekV2MLAAttention, TorchairDeepseekV2MLP,
+    TorchairDeepseekV2MoE, TorchairDeepseekV2RowParallelLinear,
     TorchairDeepseekV2RowParallelLinearReplaceAllreduce,
     TorchairDeepseekV2SiluAndMul)
 
@@ -310,22 +309,3 @@ def test_torchair_deepseek_v2_decoder_layer(mock_maybe_chunk_residual,
         model_config=vllm_config.model_config,
         quant_config=None)
     assert isinstance(layer.mlp, TorchairDeepseekV2MLP)
-
-
-def test_torchair_deepseek_v2_for_causal_lm(mock_distributed, vllm_config):
-    model = TorchairDeepseekV2ForCausalLM(vllm_config=vllm_config)
-
-    input_ids = torch.randint(0, 10000, (2, 4))
-    positions = torch.arange(4).repeat(2, 1)
-    with patch.object(model.model,
-                      "forward",
-                      return_value=torch.randn(2, 4, 128)):
-        output = model(input_ids, positions)
-        assert output.shape == (2, 4, 128)
-
-    weights = [("model.embed_tokens.weight", torch.randn(10000, 128))]
-    with patch(
-            "vllm.model_executor.model_loader.weight_utils.default_weight_loader"
-    ):
-        loaded = model.load_weights(weights)
-        assert loaded is not None

tests/ut/distributed/cpu_offload/test_cpu_offload_connector.py

@@ -26,51 +26,19 @@
                                               AttentionLayer, AttentionType)
 from vllm.attention.backends.utils import CommonAttentionState
 from vllm.config import VllmConfig
-from vllm.distributed.kv_transfer import (get_kv_transfer_group,
-                                          has_kv_transfer_group,
-                                          is_v1_kv_transfer_group)
 from vllm.forward_context import ForwardContext, get_forward_context
 from vllm.utils import cdiv, direct_register_custom_op
 from vllm.v1.core.sched.output import SchedulerOutput
 from vllm.v1.kv_cache_interface import AttentionSpec
 
-from vllm_ascend.attention.utils import AscendCommonAttentionMetadata
+from vllm_ascend.attention.utils import (AscendCommonAttentionMetadata,
+                                         maybe_save_kv_layer_to_connector,
+                                         wait_for_kv_layer_from_connector)
 from vllm_ascend.ops.attention import vanilla_chunked_prefill
 from vllm_ascend.utils import (ACL_FORMAT_FRACTAL_NZ, aligned_16, is_310p,
                                nd_to_nz_2d, nd_to_nz_spec)
 
 
-def wait_for_kv_layer_from_connector(layer_name: str):
-    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():
-        return
-
-    connector = get_kv_transfer_group()
-
-    forward_context: ForwardContext = get_forward_context()
-    attn_metadata = forward_context.attn_metadata
-    if attn_metadata is None:
-        return
-    # TODO: assert ascendMetadata
-    connector.wait_for_layer_load(layer_name)
-
-
-def maybe_save_kv_layer_to_connector(
-    layer_name: str,
-    kv_cache_layer: List[torch.Tensor],
-):
-    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():
-        return
-
-    connector = get_kv_transfer_group()
-
-    forward_context: ForwardContext = get_forward_context()
-    attn_metadata = forward_context.attn_metadata
-    if attn_metadata is None:
-        return
-    # TODO: assert ascendMetadata
-    connector.save_kv_layer(layer_name, kv_cache_layer, attn_metadata)
-
-
 class AscendAttentionBackend(AttentionBackend):
     accept_output_buffer: bool = True
 

tests/ut/torchair/models/test_torchair_deepseek_v2.py

@@ -16,7 +16,9 @@
 from vllm_ascend.ascend_config import get_ascend_config
 from vllm_ascend.attention.attention_v1 import AscendAttentionState
 from vllm_ascend.attention.utils import (AscendCommonAttentionMetadata,
-                                         split_decodes_and_prefills)
+                                         maybe_save_kv_layer_to_connector,
+                                         split_decodes_and_prefills,
+                                         wait_for_kv_layer_from_connector)
 from vllm_ascend.multistream.base import MSAttentionMetadataSplitConfig
 from vllm_ascend.multistream.context import get_multistream_comm_context
 from vllm_ascend.multistream.ms_split import model_input_split_v1_mla_attn
@@ -857,8 +859,8 @@ def _forward_decode(
                 current_ms_metadata.before_comm_event.wait()
                 return self._v_up_proj(attn_output)
 
-    def _mla_preprocess(self, hidden_states, kv_cache, attn_metadata,
-                        need_gather_q_kv):
+    def _mla_preprocess(self, layer_name, hidden_states, kv_cache,
+                        attn_metadata, need_gather_q_kv):
         # MLA Preprocess:
         # 1. Perform q_a_proj and q_a_layernorm to obtain q_c
         # 2. Perform kv_a_proj_with_mqa to obtain kv_no_split
@@ -887,6 +889,8 @@ def _mla_preprocess(self, hidden_states, kv_cache, attn_metadata,
             kv_no_split = get_tp_group().all_gather(kv_no_split, 0)
         decode_preprocess_res = None
         prefill_preprocess_res = None
+        if has_prefill:
+            wait_for_kv_layer_from_connector(layer_name)
         # Preprocess for decode tokens
         if has_decode:
             decode_q_c = q_c[:num_decode_tokens]
@@ -933,6 +937,7 @@ def _mla_preprocess(self, hidden_states, kv_cache, attn_metadata,
 
     def forward(
         self,
+        layer_name,
         hidden_states: torch.Tensor,  # query in unified attn
         kv_cache: Tuple[torch.Tensor],
         attn_metadata: M,
@@ -959,7 +964,8 @@ def forward(
 
         # MLA Preprocess
         decode_preprocess_res, prefill_preprocess_res = self._mla_preprocess(
-            hidden_states, kv_cache, attn_metadata, need_gather_q_kv)
+            layer_name, hidden_states, kv_cache, attn_metadata,
+            need_gather_q_kv)
 
         if decode_preprocess_res is not None:
             # MLA Preprocess for decoding
@@ -1017,4 +1023,8 @@ def forward(
                     is_force_scatter=self.enable_shared_expert_dp)[0]
                 current_ms_metadata.after_comm_event.record()
         del o_proj_input
+
+        has_prefill = attn_metadata.num_prefills > 0
+        if has_prefill:
+            maybe_save_kv_layer_to_connector(layer_name, list(kv_cache))
         return output_padded

vllm_ascend/attention/attention_v1.py

@@ -1,7 +1,11 @@
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, List
 
 import torch
+from vllm.distributed.kv_transfer import (get_kv_transfer_group,
+                                          has_kv_transfer_group,
+                                          is_v1_kv_transfer_group)
+from vllm.forward_context import ForwardContext, get_forward_context
 
 
 @dataclass
@@ -100,3 +104,34 @@ def split_decodes_and_prefills(
     num_decode_tokens = query_start_loc[first_prefill].item()
     num_prefill_tokens = num_tokens - num_decode_tokens
     return (num_decodes, num_prefills, num_decode_tokens, num_prefill_tokens)
+
+
+def wait_for_kv_layer_from_connector(layer_name: str):
+    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():
+        return
+
+    connector = get_kv_transfer_group()
+
+    forward_context: ForwardContext = get_forward_context()
+    attn_metadata = forward_context.attn_metadata
+    if attn_metadata is None:
+        return
+    # TODO: assert ascendMetadata
+    connector.wait_for_layer_load(layer_name)
+
+
+def maybe_save_kv_layer_to_connector(
+    layer_name: str,
+    kv_cache_layer: List[torch.Tensor],
+):
+    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():
+        return
+
+    connector = get_kv_transfer_group()
+
+    forward_context: ForwardContext = get_forward_context()
+    attn_metadata = forward_context.attn_metadata
+    if attn_metadata is None:
+        return
+    # TODO: assert ascendMetadata
+    connector.save_kv_layer(layer_name, kv_cache_layer, attn_metadata)