fix(mooncake): honour MOONCAKE_PROTOCOL so EFA hardware can select efa transport (#25083)

Co-authored-by: whn09 <whn09@users.noreply.github.com>

python/sglang/srt/distributed/device_communicators/mooncake_transfer_engine.py

@@ -183,23 +183,20 @@ class MooncakeTransferEngine:
         """Initialize the mooncake instance."""
         if envs.ENABLE_ASCEND_TRANSFER_WITH_MOONCAKE.get():
             npu_phy_id = envs.ASCEND_NPU_PHY_ID.get()
-            if npu_phy_id == -1:
-                hostname += f":{get_free_port()}:npu_{self.gpu_id}"
-            else:
-                hostname += f":{get_free_port()}:npu_{npu_phy_id}"
-            ret_value = self.engine.initialize(
-                hostname,
-                "P2PHANDSHAKE",
-                "ascend",
-                device_name if device_name is not None else "",
-            )
+            suffix = self.gpu_id if npu_phy_id == -1 else npu_phy_id
+            hostname += f":{get_free_port()}:npu_{suffix}"
+            protocol = "ascend"
         else:
-            ret_value = self.engine.initialize(
-                hostname,
-                "P2PHANDSHAKE",
-                "rdma",
-                device_name if device_name is not None else "",
-            )
+            # MOONCAKE_PROTOCOL selects the transport (rdma | efa | tcp | ...).
+            # Default is "rdma"; set MOONCAKE_PROTOCOL=efa on AWS EFA hardware.
+            protocol = envs.MOONCAKE_PROTOCOL.get()
+
+        ret_value = self.engine.initialize(
+            hostname,
+            "P2PHANDSHAKE",
+            protocol,
+            device_name if device_name is not None else "",
+        )
         if ret_value != 0:
             logger.error("Mooncake Transfer Engine initialization failed.")
             raise RuntimeError("Mooncake Transfer Engine initialization failed.")

python/sglang/srt/environ.py

@@ -356,7 +356,7 @@ class Envs:
     MOONCAKE_LOCAL_HOSTNAME = EnvStr("localhost")
     MOONCAKE_TE_META_DATA_SERVER = EnvStr("P2PHANDSHAKE")
     MOONCAKE_GLOBAL_SEGMENT_SIZE = EnvStr("4gb")
-    MOONCAKE_PROTOCOL = EnvStr("tcp")
+    MOONCAKE_PROTOCOL = EnvStr("rdma")
     MOONCAKE_DEVICE = EnvStr("")
     MOONCAKE_MASTER_METRICS_PORT = EnvInt(9003)
     MOONCAKE_CHECK_SERVER = EnvBool(False)

python/sglang/srt/model_executor/model_runner.py

@@ -925,7 +925,10 @@ class ModelRunner(ModelRunnerKVCacheMixin):
         self.remote_instance_transfer_engine = TransferEngine()
         local_ip = get_local_ip_auto()
         self.remote_instance_transfer_engine.initialize(
-            local_ip, "P2PHANDSHAKE", "rdma", envs.MOONCAKE_DEVICE.get()
+            local_ip,
+            "P2PHANDSHAKE",
+            envs.MOONCAKE_PROTOCOL.get(),
+            envs.MOONCAKE_DEVICE.get(),
         )
         self.remote_instance_transfer_engine_session_id = NetworkAddress(
             local_ip, self.remote_instance_transfer_engine.get_rpc_port()