aws-efa: cap auto-QP at 2 on EFA to avoid 128-slot GIN ring overflow

deep_ep/buffers/elastic.py

@@ -21,6 +21,30 @@
 from ..utils.comm import get_nccl_comm_handle
 
 
+# AWS EFA + aws-ofi-nccl GIN plugin uses a 128-slot per-peer request ring
+# (NCCL_OFI_GIN_RING_SIZE, see aws-ofi-nccl/src/rdma/gin/nccl_ofi_gin.cpp).
+# DeepEP V2's auto-QP formula (num_sms*16+1) easily produces 129+ QPs, which
+# overruns the ring and triggers a hard assert ("Next sequence number is in
+# use") surfaced to CUDA as CUDA_ERROR_LAUNCH_FAILED. Cap auto-sized QPs on
+# EFA to stay safely below the ring limit. `EP_EFA_MAX_QPS` overrides the
+# default (empirically validated at 2 on p5en.48xlarge).
+_EFA_MAX_QPS = int(os.environ.get('EP_EFA_MAX_QPS', 2))
+
+
+def _is_efa_fabric() -> bool:
+    """Detect AWS EFA fabric via env var or sysfs rdmap*s0 device presence."""
+    if os.environ.get('FI_PROVIDER', '').lower() == 'efa':
+        return True
+    try:
+        import glob
+        # AWS EFA exposes HCAs as /sys/class/infiniband/rdmap<N>s0
+        if glob.glob('/sys/class/infiniband/rdmap*s0'):
+            return True
+    except Exception:
+        pass
+    return False
+
+
 class EPHandle:
     """
     Communication handle returned by `ElasticBuffer.dispatch`.
@@ -198,6 +222,14 @@ def __init__(self,
                 num_allocated_qps = 65 if check_fast_rdma_atomic_support() else 129
             else:
                 num_allocated_qps = 17
+            # AWS EFA + aws-ofi-nccl GIN plugin caps per-peer outstanding
+            # requests at 128; clamp auto-sized QP count to avoid ring overflow.
+            if _is_efa_fabric() and num_allocated_qps > _EFA_MAX_QPS:
+                import sys
+                print(f'[DeepEP] EFA detected: capping num_allocated_qps '
+                      f'{num_allocated_qps} -> {_EFA_MAX_QPS} to avoid GIN '
+                      f'128-slot ring overflow', file=sys.stderr)
+                num_allocated_qps = _EFA_MAX_QPS
         self.num_allocated_qps = num_allocated_qps
 
         # Create CPP handle
@@ -658,6 +690,10 @@ def get_theoretical_num_qps(self, num_sms: int) -> int:
         if self.allow_hybrid_mode:
             num_qps = num_sms * 16 + 1
 
+        # AWS EFA: cap below the 128-slot aws-ofi-nccl GIN ring (see helper docstring).
+        if _is_efa_fabric():
+            num_qps = min(num_qps, _EFA_MAX_QPS)
+
         return min(num_qps, self.num_allocated_qps)
 
     def dispatch(self,

deep_ep/include/deep_ep/impls/hybrid_dispatch.cuh

@@ -23,7 +23,7 @@ template <bool kDoCPUSync,
           int kNumChannelsPerSM = kNumScaleoutWarps,
           int kNumChannels = kNumScaleoutWarps * kNumSMs,
           int kNumMaxTokensPerChannel = math::constexpr_ceil_div(kNumMaxTokensPerRank, kNumChannels),
-          int kScaleoutUpdateInterval = 3,
+          int kScaleoutUpdateInterval = 16,
           int kNumSlotsPerForwardChunk = kScaleoutUpdateInterval,
           int kNumRanks = kNumScaleoutRanks * kNumScaleupRanks,
           int kNumNotifyThreads = kNumNotifyWarps * 32,

deep_ep/utils/envs.py

@@ -253,6 +253,17 @@ def get_rdma_gbs(nic_name: str = _DEFAULT_NIC_NAME) -> float:
     Returns:
         gbs: the RDMA bandwidth in GB/s (0 if detection fails).
     """
+
+    # AWS EFA fast path: theoretical 16-rail × 200 Gb/s = 400 GB/s, but
+    # aws-ofi-nccl Gin proxy + libfabric SRD delivers ~25 GB/s effective
+    # aggregate per node. Override EP_EFA_RDMA_GBS to adjust.
+    import glob as _glob
+    try:
+        if _glob.glob('/sys/class/infiniband/rdmap*s0'):
+            return float(os.environ.get('EP_EFA_RDMA_GBS', 25.0))
+    except Exception:
+        pass
+
     # noinspection PyBroadException
     try:
         result = subprocess.run(['ibstat'], capture_output=True, text=True, check=True)
@@ -264,5 +275,24 @@ def get_rdma_gbs(nic_name: str = _DEFAULT_NIC_NAME) -> float:
         rate = int(match.group(1))
         return rate / 8
     except Exception as e:
+        # Fall back to sysfs (required on AWS EFA where `ibstat` does not enumerate the rdmap*s0 HCAs)
+        try:
+            import glob
+            rates = []
+            for rate_path in glob.glob('/sys/class/infiniband/*/ports/1/rate'):
+                with open(rate_path) as f:
+                    line = f.read().strip()
+                # Format: "200 Gb/sec (4X HDR)" — take the first int token
+                for tok in line.split():
+                    try:
+                        rates.append(int(tok))
+                        break
+                    except ValueError:
+                        continue
+            if rates:
+                # Aggregate across all rails, convert Gb/s to GB/s
+                return sum(rates) / 8.0
+        except Exception:
+            pass
         print(f'Failed to get RDMA connection speed: {e}')
         return 0