[BugFix][KV Pool] Gate Memcache lazy init by protocol and make register buffer default (vllm-project#12064)

### What this PR does / why we need it?

Memcache initialization currently uses the Ascend device type as a proxy
for the configured transport protocol. This is no longer accurate
because buffer registration decisions are handled by
`store.register_buffer()`, while lazy initialization is only required by
the `device_sdma` protocol.

This PR:

- removes the A2-specific `all_gather` and device-type checks from
`MemcacheBackend`;
- enables requested lazy initialization only when
`MMC_LOCAL_CONFIG_PATH` sets `ock.mmc.local_service.protocol` to
`device_sdma`;
- requires `MMC_LOCAL_CONFIG_PATH`, handles protocol whitespace and
comments, and surfaces missing or unreadable configuration file errors;
- forwards buffer registration decisions to Memcache and defers pending
registrations until a lazy store is initialized.

### Does this PR introduce _any_ user-facing change?

Yes. Memcache lazy initialization is now selected by the configured
`device_sdma` protocol instead of the Ascend device type. Other
protocols initialize the store eagerly.

### How was this patch tested?

- All repository pre-commit hooks passed, including ruff, formatting,
spelling, custom Python checks, and commit sign-off validation.
- Python syntax compilation passed for the modified backend.
- Targeted checks passed for protocol whitespace, comments, missing
environment variables, missing-file error propagation,
`device_sdma`/`device_rdma` selection, and deferred buffer registration
after lazy initialization.

- vLLM version: v0.25.1
- vLLM main:
vllm-project/vllm@fe784ff

Signed-off-by: Pz1116 <zpbzpb123123@gmail.com>

Author: Pz1116

tests/ut/distributed/ascend_store/test_backend.py

@@ -600,9 +600,7 @@ def _make_backend(self):
             # Set internal state to avoid lazy init logic during tests
             backend._lazy_init = False
             backend._store_initialized = True
-            backend._is_a2 = False
-            backend._registered_buffers = None
-            backend._buffers_registered = False
+            backend._pending_buffers = None
             return backend
 
     def test_exists(self):
@@ -612,7 +610,6 @@ def test_exists(self):
 
     def test_register_buffer(self):
         b = self._make_backend()
-        b._is_a2 = True
         b.register_buffer([100], [200])
         b.store.register_buffer.assert_called_once()
 

vllm_ascend/distributed/kv_transfer/kv_pool/ascend_store/backend/memcache_backend.py

@@ -1,4 +1,5 @@
 # Standard
+import os
 import threading
 import time
 from enum import Enum
@@ -10,7 +11,22 @@
 from vllm.logger import logger
 
 from vllm_ascend.distributed.kv_transfer.kv_pool.ascend_store.backend.backend import Backend
-from vllm_ascend.utils import AscendDeviceType, get_ascend_device_type
+
+
+def _is_device_sdma() -> bool:
+    config_path = os.getenv("MMC_LOCAL_CONFIG_PATH")
+    if not config_path:
+        raise ValueError("The environment variable 'MMC_LOCAL_CONFIG_PATH' is not set.")
+    with open(config_path, encoding="utf-8") as config_file:
+        for line in config_file:
+            line = line.strip()
+            if not line or line.startswith(("#", ";")):
+                continue
+            key, separator, value = line.partition("=")
+            if separator and key.strip() == "ock.mmc.local_service.protocol":
+                return value.strip() == "device_sdma"
+    return False
+
 
 MEMCACHE_THREAD_START_WAIT_S = 0.1
 
@@ -32,14 +48,12 @@ def __init__(
     ):
         self.local_rank = local_rank if local_rank is not None else get_world_group().local_rank
         self._init_bm = init_bm
-        self._is_a2 = get_ascend_device_type() in {AscendDeviceType.A2}
-        self._lazy_init = lazy_init and not self._is_a2
+        self._lazy_init = lazy_init and _is_device_sdma()
 
         self.store: Any | None = None
         self._store_initialized = False
         self._store_init_lock = threading.Lock()
-        self._registered_buffers: tuple[list[int], list[int]] | None = None
-        self._buffers_registered = False
+        self._pending_buffers: tuple[list[int], list[int]] | None = None
 
         if not self._lazy_init:
             self.store = self._setup_store()
@@ -68,11 +82,6 @@ def _setup_store(self):
                 "to run vLLM with MemcacheConnector."
             ) from e
 
-        if self._init_bm and self._is_a2:
-            tmp_tensor = torch.zeros(1, device="npu")
-            output_tensor_list = [torch.empty_like(tmp_tensor) for _ in range(torch.distributed.get_world_size())]
-            torch.distributed.all_gather(output_tensor_list, tmp_tensor, group=get_world_group().device_group)
-
         store = DistributedObjectStore()
 
         try:
@@ -108,21 +117,17 @@ def set_device(self):
         torch.npu.set_device(device)
 
     def register_buffer(self, ptrs: list[int], sizes: list[int]):
-        self._registered_buffers = (list(ptrs), list(sizes))
+        self._pending_buffers = (list(ptrs), list(sizes))
         self._register_buffers_if_needed()
 
     def _register_buffers_if_needed(self):
-        if not self._is_a2:
-            return
-        if self._registered_buffers is None or self._buffers_registered:
-            return
-        if not self._store_initialized:
+        if self._pending_buffers is None or not self._store_initialized:
             return
         assert self.store is not None
-        ptrs, sizes = self._registered_buffers
+        ptrs, sizes = self._pending_buffers
         for ptr, size in zip(ptrs, sizes):
             self.store.register_buffer(ptr, size)
-        self._buffers_registered = True
+        self._pending_buffers = None
 
     def exists(self, keys: list[str]) -> list[int]:
         if self._lazy_init and not self._store_initialized: