From 0f2df9370a1de1b4fb11b071d39ab3ce2287a350 Mon Sep 17 00:00:00 2001 From: Yingchun Lai Date: Sun, 1 Feb 2026 09:50:42 +0800 Subject: [PATCH] feat: validate ib devices in server args (#17598) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Kangyan Zhou --- python/sglang/srt/server_args.py | 61 +++++++++ .../server_fixtures/disaggregation_fixture.py | 117 +++++++++++++++++- 2 files changed, 173 insertions(+), 5 deletions(-) diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 8173d5594..1c7074715 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -2149,6 +2149,11 @@ class ServerArgs: self.eplb_algorithm == "elasticity_aware" ), "Elastic EP requires eplb_algorithm to be set to 'auto' or 'elasticity_aware'." + if self.elastic_ep_backend == "mooncake": + self.mooncake_ib_device = self._validate_ib_devices( + self.mooncake_ib_device + ) + def _handle_expert_distribution_metrics(self): if self.enable_expert_distribution_metrics and ( self.expert_distribution_recorder_mode is None @@ -2478,6 +2483,62 @@ class ServerArgs: "requires at least one encoder urls to be set via --encoder-urls" ) + # Validate IB devices when mooncake backend is used + if ( + self.disaggregation_transfer_backend == "mooncake" + and self.disaggregation_mode in ("prefill", "decode") + ) or self.encoder_transfer_backend == "mooncake": + self.disaggregation_ib_device = self._validate_ib_devices( + self.disaggregation_ib_device + ) + + def _validate_ib_devices(self, device_str: str) -> Optional[str]: + """ + Validate IB devices before passing to mooncake. + + Args: + device_str: Comma-separated IB device names (e.g., "mlx5_0,mlx5_1") + + Returns: + Normalized comma-separated string of validated device names, or None if input is None. + """ + if device_str is None: + logger.warning( + "No IB devices specified for Mooncake backend, falling back to auto discovery." + ) + return None + + # Strip whitespace from device names + devices = [d.strip() for d in device_str.split(",") if d.strip()] + if len(devices) == 0: + raise ValueError("No valid IB devices specified") + + # Check for duplicates + if len(devices) != len(set(devices)): + raise ValueError(f"Duplicate IB devices specified: {device_str}") + + # Get available IB devices from sysfs + ib_sysfs_path = "/sys/class/infiniband" + if not os.path.isdir(ib_sysfs_path): + raise RuntimeError( + f"InfiniBand sysfs path not found: {ib_sysfs_path}. " + "Please ensure InfiniBand drivers are installed." + ) + + available_devices = set(os.listdir(ib_sysfs_path)) + if len(available_devices) == 0: + raise RuntimeError(f"No IB devices found in {ib_sysfs_path}") + + # Check for invalid devices + invalid_devices = [d for d in devices if d not in available_devices] + if len(invalid_devices) != 0: + raise ValueError( + f"Invalid IB devices specified: {invalid_devices}. " + f"Available devices: {sorted(available_devices)}" + ) + + return ",".join(devices) + def _handle_tokenizer_batching(self): if self.enable_tokenizer_batch_encode and self.enable_dynamic_batch_tokenizer: raise ValueError( diff --git a/python/sglang/test/server_fixtures/disaggregation_fixture.py b/python/sglang/test/server_fixtures/disaggregation_fixture.py index 6b70a0e26..4e2017633 100644 --- a/python/sglang/test/server_fixtures/disaggregation_fixture.py +++ b/python/sglang/test/server_fixtures/disaggregation_fixture.py @@ -103,6 +103,82 @@ class PDDisaggregationServerBase(CustomTestCase): time.sleep(5) +def _get_available_ib_devices(): + """Auto-detect available high-speed RDMA devices from sysfs. + + Filters for devices that are: + 1. Not Ethernet NICs (excludes devices with 'eth' in the name like mlx5_eth0) + 2. Active (port state) + 3. High-speed (rate >= 100 Gbps to exclude regular Ethernet NICs) + """ + ib_sysfs_path = "/sys/class/infiniband" + if not os.path.isdir(ib_sysfs_path): + logger.warning("IB sysfs path %s does not exist", ib_sysfs_path) + return None + + all_devices = sorted(os.listdir(ib_sysfs_path)) + logger.warning("All IB devices in sysfs: %s", all_devices) + + devices = [] + for dev in all_devices: + # Check port 1 state and rate (most devices have single port) + port_path = os.path.join(ib_sysfs_path, dev, "ports", "1") + if not os.path.isdir(port_path): + logger.warning("Device %s: SKIPPED (no port 1)", dev) + continue + + # Read state and rate for logging + state = "unknown" + rate = -1 + state_file = os.path.join(port_path, "state") + rate_file = os.path.join(port_path, "rate") + + try: + with open(state_file) as f: + state = f.read().strip() + except (OSError, IOError): + pass + + try: + with open(rate_file) as f: + rate_str = f.read().strip() + rate = int(rate_str.split()[0]) + except (OSError, IOError, ValueError, IndexError): + pass + + # Log device properties for debugging + logger.warning( + "Device %s: state=%s, rate=%d Gbps, has_eth_in_name=%s", + dev, + state, + rate, + "eth" in dev.lower(), + ) + + # Skip devices with "eth" in the name - these are typically Ethernet NICs + # that don't work properly with RDMA (e.g., mlx5_eth0) + if "eth" in dev.lower(): + logger.warning("Device %s: SKIPPED (contains 'eth' in name)", dev) + continue + + # Check if port is active + # State format is like "4: ACTIVE" or just "ACTIVE" + if "ACTIVE" not in state.upper(): + logger.warning("Device %s: SKIPPED (state=%s)", dev, state) + continue + + # Check rate (filter out low-speed NICs like 10/25 Gbps Ethernet) + if rate >= 0 and rate < 100: # Skip devices slower than 100 Gbps + logger.warning("Device %s: SKIPPED (rate=%d Gbps)", dev, rate) + continue + + devices.append(dev) + logger.warning("Device %s: INCLUDED", dev) + + logger.warning("Filtered IB devices: %s (count=%d)", devices, len(devices)) + return devices if devices else None + + def get_rdma_devices_args(): def _parse_list_env(var_name: str): val = os.getenv(var_name) @@ -114,10 +190,13 @@ def get_rdma_devices_args(): def _pick_default_pair(rdma_all_devices): return [rdma_all_devices[0], rdma_all_devices[len(rdma_all_devices) // 2]] - rdma_all_devices = _parse_list_env("SGLANG_CI_RDMA_ALL_DEVICES") or [ - f"mlx5_roce{i}" for i in range(8) - ] - logger.info("Resolved rdma_all_devices=%s", rdma_all_devices) + # Priority: env var > auto-detect > hardcoded fallback + rdma_all_devices = ( + _parse_list_env("SGLANG_CI_RDMA_ALL_DEVICES") + or _get_available_ib_devices() + or [f"mlx5_roce{i}" for i in range(8)] + ) + logger.warning("Resolved rdma_all_devices=%s", rdma_all_devices) n_rdma = len(rdma_all_devices) @@ -148,9 +227,37 @@ def get_rdma_devices_args(): ) # 3. Generate RDMA device names + # Detect total GPUs on the node (not just visible ones) + try: + import torch + + total_gpus = torch.cuda.device_count() + except Exception: + total_gpus = 8 # Fallback to common 8-GPU setup + + # Handle edge cases + if total_gpus == 0: + total_gpus = 8 + if n_rdma > total_gpus: + logger.warning( + "More RDMA devices (%d) than GPUs (%d), using first and middle device", + n_rdma, + total_gpus, + ) + return ",".join(_pick_default_pair(rdma_all_devices)) + + # Calculate how many GPUs share each RDMA device + gpus_per_rdma = max(1, total_gpus // n_rdma) + logger.warning( + "GPU-to-RDMA mapping: total_gpus=%d, n_rdma=%d, gpus_per_rdma=%d", + total_gpus, + n_rdma, + gpus_per_rdma, + ) + rdma_devices = [] for gpu_idx in gpu_indices: - nic_index = gpu_idx // (8 // n_rdma) + nic_index = min(gpu_idx // gpus_per_rdma, n_rdma - 1) rdma_devices.append(rdma_all_devices[nic_index]) if not rdma_devices: