Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions afd_plugin/compat/npu/feature_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down
23 changes: 17 additions & 6 deletions afd_plugin/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]] = {
Expand Down Expand Up @@ -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
)


Expand All @@ -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
Expand All @@ -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",
Expand Down
94 changes: 94 additions & 0 deletions afd_plugin/connectors/async_topology.py
Original file line number Diff line number Diff line change
@@ -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",
]
5 changes: 5 additions & 0 deletions afd_plugin/connectors/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
3 changes: 2 additions & 1 deletion afd_plugin/connectors/gpu/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Loading
Loading