[Feat]: first impliment for async GPU connector

afd_plugin/compat/npu/feature_validation.py

@@ -7,7 +7,7 @@
 from typing import TYPE_CHECKING
 
 from afd_plugin.config import (
-    AFD_ASYNC_CONNECTOR,
+    AFD_ASYNC_NPU_CONNECTOR,
     AFDConfig,
     is_afd_async_dp,
     parse_afd_config,
@@ -34,7 +34,7 @@ def fail_if_unsupported_npu_afd_features(
         vllm_config,
     )
 
-    if afd_config.connector == AFD_ASYNC_CONNECTOR:
+    if afd_config.connector == AFD_ASYNC_NPU_CONNECTOR:
         _fail_if_unsupported_npu_afd_async_features(
             vllm_config,
             afd_config,

afd_plugin/config.py

@@ -16,14 +16,22 @@
     from vllm.config import VllmConfig
 
 AFD_ADDITIONAL_CONFIG_KEY: Final[str] = "afd"
-AFD_ASYNC_CONNECTOR: Final[str] = "CAMAsyncAFDConnector"
+AFD_ASYNC_NPU_CONNECTOR: Final[str] = "CAMAsyncAFDConnector"
+AFD_ASYNC_GPU_CONNECTOR: Final[str] = "GpuAsyncAFDConnector"
+# Connectors that drive FFN work from the connector receive loop instead of a DP
+# metadata control plane, and therefore need the async-DP engine patches. The
+# patches are platform-neutral; only the connector below them differs.
+AFD_ASYNC_CONNECTORS: Final[frozenset[str]] = frozenset(
+    {AFD_ASYNC_NPU_CONNECTOR, AFD_ASYNC_GPU_CONNECTOR},
+)
 AFDRole = Literal["attention", "ffn"]
 
 SUPPORTED_AFD_ROLES: Final[tuple[str, ...]] = ("attention", "ffn")
 SUPPORTED_AFD_CONNECTORS: Final[tuple[str, ...]] = (
     "P2pNcclAFDConnector",
     "CAMP2pAFDConnector",
-    AFD_ASYNC_CONNECTOR,
+    AFD_ASYNC_NPU_CONNECTOR,
+    AFD_ASYNC_GPU_CONNECTOR,
 )
 
 _ALIASES: Final[dict[str, str]] = {
@@ -283,7 +291,7 @@ def is_afd_async_dp(vllm_config: VllmConfig) -> bool:
     return (
         config is not None
         and config.async_dp
-        and config.connector == AFD_ASYNC_CONNECTOR
+        and config.connector in AFD_ASYNC_CONNECTORS
     )
 
 
@@ -307,9 +315,10 @@ def validate_afd_config(
             "AFD connector must be one of "
             f"{SUPPORTED_AFD_CONNECTORS!r}, got {config.connector!r}",
         )
-    if config.async_dp and config.connector != AFD_ASYNC_CONNECTOR:
+    if config.async_dp and config.connector not in AFD_ASYNC_CONNECTORS:
         raise ValueError(
-            "AFD async mode requires connector='CAMAsyncAFDConnector'",
+            "AFD async mode requires one of "
+            f"{sorted(AFD_ASYNC_CONNECTORS)!r}, got {config.connector!r}",
         )
     if config.connector == "P2pNcclAFDConnector":
         from afd_plugin.distributed import validate_p2p_topology
@@ -331,7 +340,9 @@ def validate_afd_config(
 
 __all__ = [
     "AFDConfig",
-    "AFD_ASYNC_CONNECTOR",
+    "AFD_ASYNC_NPU_CONNECTOR",
+    "AFD_ASYNC_CONNECTORS",
+    "AFD_ASYNC_GPU_CONNECTOR",
     "afd_config_from_mapping",
     "AFD_ADDITIONAL_CONFIG_KEY",
     "AFDRole",

afd_plugin/connectors/async_topology.py

@@ -0,0 +1,94 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the AFD plugin project
+"""Rank layout shared by the asynchronous AFD connectors.
+
+Both async connectors -- Ascend CAM and CUDA NVSHMEM -- lay their world out
+Attention-first and derive expert placement the same way. Keeping that here lets
+the CUDA connector reuse it without importing a backend module.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from afd_plugin.config import AFDConfig
+
+ASYNC_MOE_REQUEST_SPLIT = "request"
+ATTN_RANKS_PER_DP_CONFIG_KEY = "attn_ranks_per_dp"
+
+
+@dataclass(frozen=True, slots=True)
+class AFDAsyncTopology:
+    """Role-local and world rank information for one async participant."""
+
+    role: str
+    role_rank: int
+    world_rank: int
+    attn_size: int
+    ffn_size: int
+    expert_per_rank: int
+
+    @property
+    def world_size(self) -> int:
+        """Return the total number of Attention and FFN ranks."""
+        return self.attn_size + self.ffn_size
+
+
+def build_async_topology(
+    afd_config: AFDConfig,
+    role_rank: int,
+    *,
+    num_routed_experts: int | None = None,
+) -> AFDAsyncTopology:
+    """Validate role-local rank settings and derive the async world rank.
+
+    The world is Attention-first: Attention role rank ``i`` maps to world rank
+    ``i`` and FFN role rank ``j`` maps to ``num_attention_ranks + j``. Routed
+    experts are distributed across FFN ranks using a ceiling division;
+    production model layouts should keep the routed-expert count divisible by
+    the FFN rank count.
+    """
+    attn_size = afd_config.num_attention_ranks
+    ffn_size = afd_config.num_ffn_ranks
+    if attn_size <= 0 or ffn_size <= 0:
+        raise ValueError("AFD async topology sizes must be positive")
+    if role_rank < 0:
+        raise ValueError(f"AFD async role rank must be non-negative, got {role_rank}")
+
+    if afd_config.role == "attention":
+        if role_rank >= attn_size:
+            raise ValueError(
+                "Attention role rank must be within attention size "
+                f"(rank={role_rank}, size={attn_size})",
+            )
+        world_rank = role_rank
+    elif afd_config.role == "ffn":
+        if role_rank >= ffn_size:
+            raise ValueError(
+                "FFN role rank must be within FFN size "
+                f"(rank={role_rank}, size={ffn_size})",
+            )
+        world_rank = attn_size + role_rank
+    else:
+        raise ValueError(f"unknown AFD role {afd_config.role!r}")
+
+    expert_count = num_routed_experts or 1
+    expert_per_rank = (expert_count + ffn_size - 1) // ffn_size
+    return AFDAsyncTopology(
+        role=afd_config.role,
+        role_rank=role_rank,
+        world_rank=world_rank,
+        attn_size=attn_size,
+        ffn_size=ffn_size,
+        expert_per_rank=expert_per_rank,
+    )
+
+
+__all__ = [
+    "ASYNC_MOE_REQUEST_SPLIT",
+    "ATTN_RANKS_PER_DP_CONFIG_KEY",
+    "AFDAsyncTopology",
+    "build_async_topology",
+]

afd_plugin/connectors/factory.py

@@ -100,6 +100,11 @@ def parse_connector_extra_info(
     "afd_plugin.connectors.npu.async_cam",
     "CAMAsyncAFDConnector",
 )
+AFDConnectorFactory.register_connector(
+    "GpuAsyncAFDConnector",
+    "afd_plugin.connectors.gpu.async_gpu",
+    "GpuAsyncAFDConnector",
+)
 
 
 __all__ = ["AFDConnectorFactory"]

afd_plugin/connectors/gpu/__init__.py

@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the AFD plugin project
 """GPU-specific AFD connector implementations."""
 
+from afd_plugin.connectors.gpu.async_gpu import GpuAsyncAFDConnector
 from afd_plugin.connectors.gpu.p2p import P2pNcclAFDConnector
 
-__all__ = ["P2pNcclAFDConnector"]
+__all__ = ["GpuAsyncAFDConnector", "P2pNcclAFDConnector"]