From 8ab9982851a5cade01cd1e3e41aa3a584a05a918 Mon Sep 17 00:00:00 2001 From: James <445169590@qq.com> Date: Wed, 9 Sep 2026 09:44:19 +0800 Subject: [PATCH] [NPU]support mf device urma and host rdma trans type (#38174) --- .../disaggregation/ascend/transfer_engine.py | 47 +++++++++---------- 1 file changed, 22 insertions(+), 25 deletions(-) diff --git a/python/sglang/srt/disaggregation/ascend/transfer_engine.py b/python/sglang/srt/disaggregation/ascend/transfer_engine.py index 93d1a2956..cac0b9c6e 100644 --- a/python/sglang/srt/disaggregation/ascend/transfer_engine.py +++ b/python/sglang/srt/disaggregation/ascend/transfer_engine.py @@ -8,8 +8,6 @@ from sglang.srt.disaggregation.utils import DisaggregationMode from sglang.srt.distributed.device_communicators.mooncake_transfer_engine import ( MooncakeTransferEngine, ) -from sglang.srt.environ import envs -from sglang.srt.utils.common import run_with_deadline from sglang.srt.utils.network import NetworkAddress try: @@ -22,6 +20,8 @@ except ImportError as e: logger = logging.getLogger(__name__) +_DEFAULT_PROTOCOL = "sdma" + class AscendTransferEngine(MooncakeTransferEngine): def __init__( @@ -63,30 +63,21 @@ class AscendTransferEngine(MooncakeTransferEngine): ) transfer_protocol = self._get_transfer_protocol() - if transfer_protocol is None or transfer_protocol == "sdma": - trans_op_type = TransferEngine.TransDataOpType.SDMA - elif transfer_protocol == "device_urma": - trans_op_type = TransferEngine.TransDataOpType.DEVICE_URMA - elif transfer_protocol == "device_uboe": - trans_op_type = TransferEngine.TransDataOpType.DEVICE_UBOE - else: - trans_op_type = TransferEngine.TransDataOpType.DEVICE_RDMA - """with device RDMA for PD transfer""" + if transfer_protocol == "device_rdma": + # with device RDMA for PD transfer: initialize hccl in advance + # through all_gather to avoid conflicts with rdma initialization. tmp_tensor = torch.zeros(1, device="npu") output_tensor_list = [ torch.empty_like(tmp_tensor) for _ in range(get_world_size()) ] - # Initialize hccl in advance through all_gather to avoid conflicts with rdma initialization. torch.distributed.all_gather( output_tensor_list, tmp_tensor, group=get_world_group().device_group ) + + trans_op_type = self._resolve_trans_op_type(transfer_protocol) """Initialize the ascend transfer instance.""" - ret_value = run_with_deadline( - lambda: self.engine.initialize( - self.store_url, self.session_id, self.role, self.npu_id, trans_op_type - ), - timeout_s=envs.SGLANG_DISAGGREGATION_ENGINE_INIT_TIMEOUT.get(), - what=f"Ascend TransferEngine.initialize({self.store_url!r}, {self.session_id!r})", + ret_value = self.engine.initialize( + self.store_url, self.session_id, self.role, self.npu_id, trans_op_type ) if ret_value != 0: logger.error("Ascend Transfer Engine initialization failed.") @@ -102,13 +93,19 @@ class AscendTransferEngine(MooncakeTransferEngine): logger.debug(f"Ascend memory registration for ptr {ptrs} failed.") @staticmethod - def _get_transfer_protocol(): + def _get_transfer_protocol() -> str: protocol = os.getenv("ASCEND_MF_TRANSFER_PROTOCOL") - allowed_protocols = {"device_rdma", "sdma", "device_urma", "device_uboe"} - if protocol and protocol.lower() in allowed_protocols: - return protocol.lower() - else: + return protocol.strip().lower() if protocol else _DEFAULT_PROTOCOL + + @staticmethod + def _resolve_trans_op_type(protocol: str): + op_type = getattr(TransferEngine.TransDataOpType, protocol.upper(), None) + if op_type is None: logger.warning( - "Invalid or no transfer protocol specified, using default protocol." + "Transfer protocol %r is not supported by the installed " + "memfabric_hybrid, falling back to %r.", + protocol, + _DEFAULT_PROTOCOL, ) - return None + op_type = TransferEngine.TransDataOpType.SDMA + return op_type