diff --git a/.agents/skills/deploy-afd-k8s/SKILL.md b/.agents/skills/deploy-afd-k8s/SKILL.md
index f51ae4190..a7ef9b68e 100644
--- a/.agents/skills/deploy-afd-k8s/SKILL.md
+++ b/.agents/skills/deploy-afd-k8s/SKILL.md
@@ -1,6 +1,6 @@
---
name: deploy-afd-k8s
-description: Use when the user asks to deploy, serve, or stand up an AFD GPU recipe on Kubernetes/OpenShift from a recipe .sh script - creating the model PVC, recipe ConfigMap, serve pod(s), and Service(s), and waiting until the endpoint answers. Placement across pods is never asked for or assumed -- it is read directly off the recipe's own top-of-file comment block (however that block describes it, strict or prose): a recipe whose comment describes a per-pod placement deploys exactly that many pods with exactly that per-pod setup; a recipe without one deploys as a single pod. Do not use for local (non-k8s) serving, NPU recipes, prefill-decode disaggregation recipes, or E2E correctness testing (see run-e2e).
+description: Use when the user asks to deploy, serve, or stand up an AFD GPU recipe on Kubernetes/OpenShift from a recipe .sh script - creating the model PVC, recipe ConfigMap, serve pod(s), and Service(s), and waiting until the endpoint answers. Placement across pods is never asked for or assumed -- it is the recipe's LAYOUT (the caller's setting, else the script's default): one pod per LAYOUT entry, a one-entry LAYOUT being a single pod. Do not use for local (non-k8s) serving, NPU recipes, prefill-decode disaggregation recipes, or E2E correctness testing (see run-e2e).
---
# Deploy an AFD GPU recipe on Kubernetes
@@ -15,36 +15,38 @@ On success this skill guarantees, in the target namespace:
| | |
|---|---|
-| Endpoint | `http://vllm-service:${CLIENT_PORT}` (OpenAI-compatible, in-namespace; `CLIENT_PORT` defaults to `18305`) |
+| Endpoint | `http://vllm-service:${CLIENT_PORT}` (OpenAI-compatible, in-namespace; `CLIENT_PORT` is the launcher's Attention port, `18000`) |
| Serving | the model named by `MODEL_ID` |
-| Pod(s) / Service | `vllm-pod`, or one pod per pod the recipe's header comment describes (named `vllm-pod-
-| Documentation | Recipe | Slides | DeepWiki | User Forum | Developer Slack | WeChat | +| Documentation | Recipe | Slides | DeepWiki | User Forum | Developer Slack | WeChat |
## Overview @@ -18,342 +18,46 @@ tests for GPU and Ascend NPU deployments. > This project is still experimental and needs more large-scale testing across > different hardware backends. -The GPU and Ascend target is **vLLM `v0.32.0`** at commit `5df63a72`. -The Ascend integration also targets vLLM-Ascend `bd42eea0`. The plugin does not modify the vLLM -source tree. AFD behavior is installed through the `vllm.general_plugins` entry -point, `--additional-config`, automatically selected role workers, plugin-owned -model wrappers, and narrow version-scoped compatibility shims. +The plugin extends vLLM through its general-plugin entry point without +modifying the vLLM source tree. ## Architecture  -## Current Status +## Getting started -Core runtime support: +1. [Install AFD](docs/installation.md) for CUDA or Ascend NPU. +2. Choose a [connector and supported model](docs/support.md), then follow its + [deployment recipe](recipe/README.md). +3. Pass role/topology settings through + [`--additional-config`](docs/configuration.md). Send requests to Attention; + FFN serves connector work. -- vLLM plugin registration, AFD configuration, and runtime validation. -- Attention/FFN workers, model runners, model wrappers, and connector-driven - execution for CUDA and Ascend NPU. -- Eager and `FULL_DECODE_ONLY` graph execution, plus backend-specific profiling - support. -- Native DBO with exactly two ubatches on CUDA V2. Ascend constraints are - recorded in the existing [NPU runtime matrix](docs/design/module/execution_platforms.md#tested-runtime-matrix). -- DeepSeek MoE handoff at the remote-experts boundary on CUDA, with the gate - placed on either Attention or FFN. -- Expert-routed M:N exchange on CUDA (`DeepEPM2NAFDConnector`) for Qwen3 MoE: - Attention routes each token and dispatches it only to the FFN ranks that own - its experts. +## Support matrix -Model support: +See the [support matrix](docs/support.md) for models, connectors, tested +configurations, and known accuracy/shutdown limits. -| Model family | Registered architectures | Plugin model wrappers | Notes | -| --- | --- | --- | --- | -| DeepSeekV2 / DeepSeekV3 / DeepSeekV3.2 | `DeepseekForCausalLM`, `DeepseekV2ForCausalLM`, `DeepseekV3ForCausalLM`, `DeepseekV32ForCausalLM` | `AFDDeepseekForCausalLM`, `AFDDeepseekV2ForCausalLM`, `AFDDeepseekV3ForCausalLM` | DeepSeekV3.2 uses `AFDDeepseekV3ForCausalLM`. Each AFD role constructs and loads only its role-required model components, while shared embedding, normalization, and output components remain available where required by the model lifecycle. | -| DeepSeek V4 Flash | `DeepseekV4ForCausalLM` | `AFDDeepseekV4ForCausalLM` | CUDA synchronous and Ascend CAM paths; see the [runtime matrix](docs/design/module/execution_platforms.md#tested-runtime-matrix) for backend-specific validation limits. | -| Qwen3 MoE | `Qwen3MoeForCausalLM` | `AFDQwen3MoeForCausalLM` | CUDA; DeepEP M2N routes on Attention, P2P routes on FFN. | -| Qwen3.5 / Qwen3.6 MoE | `Qwen3_5MoeForConditionalGeneration` | `AFDQwen3_5MoeForConditionalGeneration` | CUDA text-only Qwen3.6-35B-A3B with `--language-model-only`, native DP4/TP1/EP4, and synchronous AFD 2A1F. Eager and graph execution remain supported; GDN DBO is temporarily withdrawn; native hybrid/multimodal validation applies. Earlier DBO experiments are historical evidence only; see the [E2E limits](docs/design/module/e2e_testing.md). | +## Documentation -Connector support: - -See the [recipe index](recipe/README.md) for deployment and benchmark examples. - -| Connector | Platform | Recommend Stage | Sync or Async | Graph Support | Notes | -| --- | --- | --- | --- | --- | --- | -| `P2pNcclAFDConnector` | CUDA | Decode | Sync | `FULL_DECODE_ONLY` CUDA graph | FFN ranks are ordered before Attention ranks. `num_attention_ranks` must be greater than or equal to `num_ffn_ranks`; the Attention ranks are spread over the FFN ranks in blocks that differ in size by at most one. See the [DeepSeek V2 Lite recipe](recipe/gpu/P2pNcclAFDConnector/deepseek_v2_lite/README.md). | -| `DeepEPM2NAFDConnector` | CUDA (SM90+) | Decode | Sync | `FULL_DECODE_ONLY` CUDA graph | Expert-routed M2N transport: Attention computes the gate and top-k and dispatches each token only to the FFN ranks that own its experts; the FFN runs only routed experts. Requires `compute_gate_on_attention=true`, FFN TP 1, and the same DBO setting on both roles; supports two-microbatch DBO, Attention TP, BF16 or block-128 FP8 payloads, and one or several nodes. Qwen3 MoE only. Dispatch and combine use [afdEP-comm](https://github.com/afdEP/afdEP-comm) (the `afdep` extra); the default expert backend needs [DeepGEMM](https://github.com/deepseek-ai/DeepGEMM) installed separately. See the [DeepEP M2N guide](docs/gpu/DEEPEP_M2N_CONNECTOR_USER_GUIDE.md) and the [Qwen3 recipes](recipe/gpu/DeepEPM2NAFDConnector/README.md). | -| `CAMP2pAFDConnector` | Ascend NPU | Decode | Sync | `FULL` or `FULL_DECODE_ONLY`; plain MLA graph DBO requires `FULL_DECODE_ONLY` | Uses HCCL/CAMP2P custom ops and native MRV2. DBO requires two ubatches and Attention DP > 1. See the [CAMP2P guide](docs/npu/CAM_P2P_CONNECTOR_USER_GUIDE.md). | -| `CAMAsyncAFDConnector` | Ascend NPU | Prefill / decode | Async | FFN-only experimental `FULL` | Eager Attention supports DP+TP/SP and two native DBO microbatches, including DP1. FFN FULL requires the [W4A8 runtime contract](docs/npu/ASYNC_FFN_RUNTIME.md) and accepts eager Attention DBO; FFN does not join native DBO. PCP/DCP are unsupported. See the [Async CAM guide](docs/npu/CAM_ASYNC_CONNECTOR_USER_GUIDE.md) for configuration and validation limits. | - -Connector implementations are grouped by backend package: -`afd_plugin.connectors.gpu` for GPU-only connectors, -`afd_plugin.connectors.npu` for NPU-only connectors. - -Known gaps: - -- The GPU target is exactly vLLM `0.32.0`; other vLLM versions are not - claimed as supported on GPU. -- NPU hardware evidence covers the specific single-Pod DeepSeek-V2-Lite and - DSV4 configurations in the [0.32 validation record](https://github.com/vllm-project/afd-plugin/pull/507). - Other models and topologies remain unverified. AFD post-request shutdown - failures remain open; successful inference does not imply graceful shutdown. -- Synchronous DSV4 E2E profiles are smoke cases; A5 concurrent answers remain - incorrect. See the [profile limits](tests/e2e/README.md#dsv4-flash-sync-camp2p-concurrent-requests-local-4-or-16-npus). -- GPU Attention requires native ModelRunnerV2; there is no GPU V1 fallback. - FFN remains connector-driven. Synchronous `P2pNcclAFDConnector` supports - two-microbatch DBO and `FULL_DECODE_ONLY`, with model-specific gate placement. - See the [runtime matrix](docs/design/module/execution_platforms.md#tested-runtime-matrix) - for tested configurations and known DeepSeek-V2-Lite DBO shutdown failures. -- `DeepEPM2NAFDConnector` requires SM90 or newer GPUs, Attention-side - routing, and FFN TP 1, and supports only Qwen3 MoE models whose layers are - all MoE and have no shared expert. Its GPU device tests - ([`tests/gpu`](tests/gpu/README.md)) and E2E cases need SM90+ hardware and - run outside CI. -- GPU and NPU E2E tests are opt-in and require real hardware plus model weights. -- GPU CUDA graph support is limited to `FULL_DECODE_ONLY`. -- Native DBO is limited to exactly two ubatches. NPU CAMP2P supports eager - and graph DBO; Async CAM enables eager DBO only on Attention, including - when FFN uses FULL. FFN stays outside Attention's native DBO thread schedule. - Model evidence remains specific to its workload and topology. -- Qwen3 MoE requires Attention-side gate placement for DeepEP M2N and - FFN-side gate placement for P2P. It rejects sequence-parallel MoE, EPLB, - pipeline parallelism, speculative decoding, LoRA, and NPU. -- Qwen3.5/Qwen3.6 MoE adapter-family support is limited to the CUDA text-only - lane; repository hardware E2E evidence covers Qwen3.6-35B-A3B. - GDN DBO is unsupported; upstream hybrid/multimodal DBO validation applies. - NPU and multimodal execution are unsupported/unverified; - `compute_gate_on_attention=true`, pipeline parallelism, asynchronous and - multi-node execution are unsupported; quantization is unverified; no - performance claim is made. - -## Install - -Requires Python **3.11-3.13** and [uv](https://docs.astral.sh/uv/). - -```bash -git clone https://github.com/vllm-project/afd-plugin.git -cd afd-plugin -uv sync --group dev -``` - -`vllm` is an optional runtime extra so CPU-only or macOS development -environments can still run import/config tests without a CUDA wheel. - -### GPU installation - -For Linux / CUDA-capable environments only; Ascend users should skip this -command: - -```bash -uv sync --group dev --extra vllm -``` - -The optional GPU extra pins `vllm==0.32.0`. - -`DeepEPM2NAFDConnector` also needs the `afdep` extra (afdEP-comm) and -DeepGEMM, which is built from source; see -[Installation](docs/gpu/DEEPEP_M2N_CONNECTOR_USER_GUIDE.md#installation) in -the connector guide. - -```bash -uv sync --group dev --extra vllm --extra afdep -``` - -### Ascend NPU installation - -The Ascend integration uses the source pair below. See the -[0.32 NPU validation record](https://github.com/vllm-project/afd-plugin/pull/507) for the tested -scope and retained shutdown limitations. Earlier -[0.30 results](https://github.com/vllm-project/afd-plugin/pull/425#issuecomment-6063910923) -retain their historical environment and sample counts. -Install a compatible driver and firmware, and confirm the devices with -`npu-smi info`. - -| Component | Version | -| --- | --- | -| Python | Package range `3.11–3.13`; tested `3.12.13` | -| vLLM | `0.32.0`, commit `5df63a7203cdb9328d824a27b682e77c7b9928be` | -| vLLM-Ascend | commit [`bd42eea0`](https://github.com/vllm-project/vllm-ascend/commit/bd42eea01d8312c4df469fabaed45f77a12df4d3) | -| Tested CANN / torch / torch-npu | `9.2` / `2.10.0+cpu` / `2.10.0.post7` | -| Tested Triton Ascend / CAM | `3.6` / `1.3` | - -> Hardware evidence is scoped to the recorded models, configurations, and -> sample counts; it does not cover every NPU deployment or guarantee clean -> device shutdown. - -#### Environment - -The recorded A3 environment uses the pinned pair and stack above. The -repository does not claim a released container tag for the Ascend target. Use the -[installation guide at that source snapshot](https://github.com/vllm-project/vllm-ascend/blob/bd42eea01d8312c4df469fabaed45f77a12df4d3/docs/source/getting_started/installation.md) -to prepare a matching A3/openEuler environment, then install AFD from the -repository root. - -#### Install AFD - -From the AFD repository root: - -```bash -AFD_BUILD_ASCEND_OPS=1 \ -SOC_VERSION=ascend910_9391 \ -python -m pip install -v --no-build-isolation --no-deps -e . -``` - -`--no-deps` preserves the matched NPU runtime, and `--no-build-isolation` uses -its CANN/torch-npu toolchain. The command above forces the Ascend op build for -reproducibility. Otherwise, leave `AFD_BUILD_ASCEND_OPS` unset to use the -auto-detection described in [Development](#development), set it to `1` if -detection misses an Ascend environment, or set it to `0` to skip the ops. - -#### Verify - -```bash -python - <<'PY' -import torch -import torch_npu -import vllm_ascend - -from afd_plugin.compat.npu import ensure_afd_ascend_ops_loaded - -assert torch.npu.is_available(), "torch-npu cannot see an Ascend device" -ensure_afd_ascend_ops_loaded() -print("AFD_OPS_OK") -PY -``` - -After `AFD_OPS_OK`, the environment is ready to run the NPU examples and E2E -tests. See the -[NPU testing guide](docs/npu/TESTING.md). -For implementation details, see the -[Attention runtime design](docs/design/module/attention_runtime.md) and -[FFN runtime design](docs/design/module/ffn_runtime.md). - -## Using the Plugin - -Install or sync the distribution as `vllm-afd-plugin`. Python imports use the -`afd_plugin` package name. - -AFD is configured through vLLM `--additional-config`. There is no separate -`--afd-config` flag. When AFD is configured and `--worker-cls` is omitted, the -plugin automatically selects the Attention or FFN worker for the active CUDA -or standard Ascend NPU platform. Explicit AFD worker paths remain accepted for -compatibility with existing commands, but are not required or stable launch -interfaces. - -GPU and NPU launches require `VLLM_USE_V2_MODEL_RUNNER=1` on both roles. -Attention uses native ModelRunnerV2; FFN keeps its connector-driven runner. -Neither backend has a V1 fallback. GPU DBO requires two microbatches and accepts Attention -DP1 or DP > 1. -See the [dedicated E2E scenarios](tests/e2e/README.md#gpu-mrv2-dbo-development-comparison). - -```bash -export VLLM_USE_V2_MODEL_RUNNER=1 -``` - -GPU Attention-side shape: - -```bash -vllm serve /path/to/DeepSeek-V2-Lite \ - --served-model-name deepseek-v2-lite-afd-attention \ - --data-parallel-size 1 \ - --tensor-parallel-size 1 \ - --enable-expert-parallel \ - --enforce-eager \ - --host 127.0.0.1 \ - --port 18000 \ - --additional-config '{"afd":{"role":"attention","connector":"P2pNcclAFDConnector","host":"127.0.0.1","port":6239,"num_attention_ranks":1,"num_ffn_ranks":1}}' -``` - -GPU FFN-side shape: - -```bash -vllm serve /path/to/DeepSeek-V2-Lite \ - --served-model-name deepseek-v2-lite-afd-ffn \ - --data-parallel-size 1 \ - --tensor-parallel-size 1 \ - --enable-expert-parallel \ - --enforce-eager \ - --host 127.0.0.1 \ - --port 18001 \ - --additional-config '{"afd":{"role":"ffn","connector":"P2pNcclAFDConnector","host":"127.0.0.1","port":6239,"num_attention_ranks":1,"num_ffn_ranks":1}}' -``` - -NPU uses the same config channel with `CAMP2pAFDConnector`; the plugin selects -the NPU worker automatically. Attention uses native Ascend ModelRunnerV2: - -```bash -VLLM_USE_V2_MODEL_RUNNER=1 vllm serve /path/to/DeepSeek-V2-Lite \ - --served-model-name deepseek-v2-lite-afd-attention \ - --data-parallel-size 1 \ - --tensor-parallel-size 1 \ - --enable-expert-parallel \ - --enforce-eager \ - --host 127.0.0.1 \ - --port 18000 \ - --additional-config '{"afd":{"role":"attention","connector":"CAMP2pAFDConnector","host":"127.0.0.1","port":6239,"num_attention_ranks":1,"num_ffn_ranks":1}}' -``` - -Attention and FFN may be started in either order. Send requests only to the -Attention API server. FFN workers are connector-driven; scheduler-driven FFN -`execute_model()` calls fail fast. - -For repeatable local smoke testing, prefer the bundled runner: - -```bash -export HF_HOME=/path/to/huggingface -python -c 'from datasets import load_dataset; load_dataset("openai/gsm8k", "main")' - -uv run python tests/e2e/runner.py \ - --model /path/to/DeepSeek-V2-Lite \ - --device-backend gpu \ - --scenario afd-eager-2a1f \ - --attention-devices 0,1 \ - --ffn-devices 2 \ - --gsm8k-output-path /tmp/afd-e2e-gsm8k \ - --api-port-base 18000 \ - --afd-port 6239 \ - --common-vllm-arg=--trust-remote-code -``` - -For NPU, use `--device-backend npu`; the runner maps the same device arguments -to `ASCEND_RT_VISIBLE_DEVICES` and selects `CAMP2pAFDConnector`. - -## AFD Config - -The canonical config shape is: - -```json -{ - "afd": { - "role": "attention", - "connector": "P2pNcclAFDConnector", - "host": "127.0.0.1", - "port": 1239, - "num_attention_ranks": 2, - "num_ffn_ranks": 1, - "compute_gate_on_attention": false, - "connector_extra_config": {} - } -} -``` - -`role` must be `attention` or `ffn`. `connector` must be `P2pNcclAFDConnector`, -`DeepEPM2NAFDConnector`, `CAMP2pAFDConnector`, or `CAMAsyncAFDConnector`. -AFD is active when `additional_config["afd"]` is present and passes common AFD config validation; -omit `additional_config["afd"]` to disable AFD. Connector-owned -`connector_extra_config` is strictly validated by the selected connector parser -when the connector/runtime is constructed. The plugin also accepts selected -compatibility aliases such as `afd_role`, `afd_connector`, `afd_host`, and -`afd_port`. +Browse the [documentation index](docs/README.md) for user guides, module design, +validation and troubleshooting. ## Development -Before adding, changing, moving, or removing tests, or changing the behavior they -cover, read and follow the [test maintenance guide](docs/UT_MAINTENANCE.md). - -Run the default CPU-safe checks: +For CPU-only or macOS development, follow [source setup](docs/installation.md#source-setup), +then run: ```bash uv run pytest uv run ruff check . ``` -The NPU-dependent unit suite must run in an Ascend development environment; -see [Running NPU unit tests](docs/npu/TESTING.md). - -Native C/C++ sources are grouped by backend under `csrc/`: Ascend/CANN sources -live in `csrc/npu`, including the `a2e` and `e2a` ACLNN operators, and -`csrc/gpu` is reserved for GPU native sources. - -Ascend custom ops are built automatically only when the build environment looks -like an Ascend NPU platform, for example when `torch_npu`, CANN environment -variables, or the default Ascend toolkit path are present. GPU builds skip -Ascend ops by default. Set `AFD_BUILD_ASCEND_OPS=1` or -`AFD_BUILD_ASCEND_OPS=0` to override the auto-detection. - -## E2E Test - -To run E2E tests, use the [`run-e2e` skill](.agents/skills/run-e2e/SKILL.md). +Read the [contributor rules](AGENTS.md) and [test maintenance guide](docs/UT_MAINTENANCE.md) +before changing code or tests. See [module design](docs/design/module/index.md) +for ownership, [native builds](csrc/README.md) for backend operators, and the +[E2E guide](tests/e2e/README.md) for hardware validation. ## License diff --git a/afd_plugin/__init__.py b/afd_plugin/__init__.py index eeff9c467..a586088c5 100644 --- a/afd_plugin/__init__.py +++ b/afd_plugin/__init__.py @@ -125,6 +125,9 @@ def get_spawn_context(method: str | None = None): "Qwen3MoeForCausalLM": ( "afd_plugin.model_executor.models.qwen3_moe:AFDQwen3MoeForCausalLM" ), + "Qwen3MoeRoutedExperts": ( + "afd_plugin.model_executor.models.qwen3_moe:AFDQwen3MoeRoutedExperts" + ), } _QWEN3_5_MODEL_REGISTRATIONS = MappingProxyType( diff --git a/afd_plugin/compat/npu/feature_validation.py b/afd_plugin/compat/npu/feature_validation.py index cf151fac1..80f8de270 100644 --- a/afd_plugin/compat/npu/feature_validation.py +++ b/afd_plugin/compat/npu/feature_validation.py @@ -14,7 +14,11 @@ is_afd_async_dp, parse_afd_config, ) -from afd_plugin.validation import cam_receive_capacity, requested_ffn_mode +from afd_plugin.validation import ( + cam_receive_capacity, + requested_ffn_mode, + validate_async_cam_attention_graph, +) if TYPE_CHECKING: from vllm.config import VllmConfig @@ -56,8 +60,15 @@ def fail_if_unsupported_npu_afd_features( return if afd_config.compute_gate_on_attention: - raise RuntimeError( - "AFD NPU runtime does not support compute_gate_on_attention=true yet", + # The sync Attention-side gate prototype is deliberately narrow: one + # connector, the target model and BF16 transport. Widening it without the + # corresponding transport work would produce wrong numerics instead of an + # error, so anything outside that shape stays rejected. + _fail_if_unsupported_sync_attention_gate( + vllm_config, + afd_config, + extra_info, + is_dsv4=is_dsv4, ) if afd_config.connector == CAMP2P_CONNECTOR: from afd_plugin.connectors.npu.camp2p import CAMP2PExtraInfo @@ -151,6 +162,53 @@ def _fail_if_unsupported_dsv4_connector(afd_config: AFDConfig) -> None: ) +def _fail_if_unsupported_sync_attention_gate( + vllm_config: VllmConfig, + afd_config: AFDConfig, + extra_info: object, + *, + is_dsv4: bool, +) -> None: + """Restrict the sync Attention-side gate prototype to what it supports today. + + DBO and ubatching are rejected rather than merely unused: the sync shell does not + call the DBO yield, so enabling DBO here would interleave microbatches against a + transport that never yields. Not calling the yield is not the same as forbidding + the feature. + """ + + parallel = vllm_config.parallel_config + # The routing travels on each rank's own activation, and only the TP=1 shape has + # been verified end to end; a wider TP would split the activation without a + # matching routing split on the Attention side. + if int(parallel.tensor_parallel_size) != 1: + raise RuntimeError( + "Attention-side gate is only supported at tensor-parallel size 1; got " + f"TP={int(parallel.tensor_parallel_size)}", + ) + if bool(parallel.enable_dbo) or bool(parallel.use_ubatching): + raise RuntimeError( + "Attention-side gate with the synchronous connector does not support " + "DBO or ubatching yet", + ) + if afd_config.connector != CAMP2P_CONNECTOR: + raise RuntimeError( + "Attention-side gate is only supported with " + f"{CAMP2P_CONNECTOR}, got {afd_config.connector!r}", + ) + if not is_dsv4: + raise RuntimeError( + "Attention-side gate is only supported for the DeepSeek V4 target", + ) + from afd_plugin.connectors.npu.camp2p import CAMP2PExtraInfo + + if not isinstance(extra_info, CAMP2PExtraInfo): + raise TypeError( + "CAMP2pAFDConnector requires CAMP2PExtraInfo, got " + f"{type(extra_info).__name__}", + ) + + def _fail_if_unsupported_npu_afd_async_features( vllm_config: VllmConfig, afd_config: AFDConfig, @@ -170,22 +228,6 @@ def _fail_if_unsupported_npu_afd_async_features( ) parallel_config = vllm_config.parallel_config - if extra_info.ffn_two_stream: - if not afd_config.is_ffn_server: - raise RuntimeError("ffn_two_stream may only be enabled on the FFN role") - if ( - not vllm_config.use_v2_model_runner - or not afd_config.compute_gate_on_attention - or extra_info.dynamic_quant != 1 - or parallel_config.tensor_parallel_size != 1 - or parallel_config.use_sequence_parallel_moe - or os.environ.get("VLLM_ASCEND_ENABLE_FLASHCOMM1", "0").strip().lower() - in ("1", "true") - ): - raise RuntimeError( - "FFN two-stream requires MRV2 layered GMM, dynamicQuant=1, " - "Attention-side gate, F TP=1 and SP disabled" - ) if not is_afd_async_dp(vllm_config): raise RuntimeError( "CAMAsyncAFDConnector requires additional_config['afd'] " @@ -193,14 +235,13 @@ def _fail_if_unsupported_npu_afd_async_features( ) if not bool(vllm_config.model_config.enforce_eager): if afd_config.is_attention_server: - raise RuntimeError( - "CAMAsyncAFDConnector supports only eager Attention execution" - ) - requested_ffn_mode(vllm_config) - if not vllm_config.use_v2_model_runner: - raise RuntimeError("Async FFN FULL requires MRV2") - if parallel_config.tensor_parallel_size != 1: - raise RuntimeError("Async FFN FULL requires F TP=1") + validate_async_cam_attention_graph(vllm_config) + else: + requested_ffn_mode(vllm_config) + if not vllm_config.use_v2_model_runner: + raise RuntimeError("Async FFN FULL requires MRV2") + if parallel_config.tensor_parallel_size != 1: + raise RuntimeError("Async FFN FULL requires F TP=1") if bool(parallel_config.enable_dbo) or bool(parallel_config.use_ubatching): if not afd_config.is_attention_server: raise RuntimeError("Async CAM DBO/ubatching requires the Attention role") @@ -222,30 +263,34 @@ def _fail_if_unsupported_npu_afd_async_features( if speculative_config.method == "dspark": if vllm_config.model_config.hf_text_config.model_type != "deepseek_v4": raise RuntimeError("AFD DSpark supports only DeepSeek-V4 on NPU") - if extra_info.ffn_two_stream: - raise RuntimeError("AFD DSpark does not support FFN two-stream") # The shared MRV2 validator owns PP/CP and static EP constraints. # Keep only the additional target/draft restrictions here. - # F may replay its expert transaction in FULL; Attention and native - # draft proposal execution stay eager. + # Both graphs use the native decode path. Draft capture is owned + # by native target capture, so draft-only graph mode is unsupported. + # F keeps its independent TP1 expert execution. if ( - ( - afd_config.is_attention_server - and not vllm_config.model_config.enforce_eager + speculative_config.draft_parallel_config.tensor_parallel_size + != parallel_config.tensor_parallel_size + or ( + afd_config.is_ffn_server + and parallel_config.tensor_parallel_size != 1 ) - or not speculative_config.enforce_eager - or parallel_config.tensor_parallel_size != 1 - or speculative_config.draft_parallel_config.tensor_parallel_size != 1 ): raise RuntimeError( - "AFD DSpark requires eager Attention/draft and target/draft TP=1" + "AFD DSpark requires matching target/draft TP and F TP=1" ) + if ( + afd_config.is_attention_server + and not speculative_config.enforce_eager + and vllm_config.model_config.enforce_eager + ): + raise RuntimeError("DSpark draft Graph requires target Attention graph") elif afd_config.is_attention_server and ( speculative_config.use_eagle() or speculative_config.uses_draft_model() ): raise RuntimeError( "CAMAsyncAFDConnector does not support native draft models or MTP; " - "only DSV4 DSpark eager is supported", + "only DSV4 DSpark is supported", ) factor_value = os.environ.get("BATCH_SIZE_FACTOR", "1") @@ -273,9 +318,10 @@ def _fail_if_unsupported_npu_afd_async_features( "AFD DSpark draft query capacity exceeds max_num_batched_tokens: " f"{draft_rows} > {max_num_batched_tokens}" ) - # whole-expert chunks need a full TP group's batch; smaller - # buffers require native sub-expert chunking. - if receive_capacity < max_num_batched_tokens: + if receive_capacity < 1: + raise RuntimeError("BATCH_SIZE_FACTOR must allocate at least one receive row") + # Push chunks whole experts. Pull can split one expert across receive rounds. + if not extra_info.ffn_mega_moe and receive_capacity < max_num_batched_tokens: raise RuntimeError( "CAMAsyncAFDConnector requires BATCH_SIZE_FACTOR to allocate at least " f"max_num_batched_tokens={max_num_batched_tokens} " diff --git a/afd_plugin/connectors/README.md b/afd_plugin/connectors/README.md index e77e20ede..2c5ed3bb8 100644 --- a/afd_plugin/connectors/README.md +++ b/afd_plugin/connectors/README.md @@ -23,7 +23,7 @@ AFD connector implementations are grouped by backend: Both backends target vLLM `0.32.0`; Ascend additionally targets vLLM-Ascend `bd42eea0`. See the -[runtime matrix](../../docs/design/module/execution_platforms.md#tested-runtime-matrix) +[runtime matrix](../../docs/support.md#tested-runtime-matrix) for tested configurations and backend-specific validation limits. Shared connector contracts, metadata containers, factory registration, and diff --git a/afd_plugin/connectors/__init__.py b/afd_plugin/connectors/__init__.py index 2b270cc6e..00384908f 100644 --- a/afd_plugin/connectors/__init__.py +++ b/afd_plugin/connectors/__init__.py @@ -2,27 +2,49 @@ # SPDX-FileCopyrightText: Copyright contributors to the AFD plugin project """AFD connector namespace.""" -from afd_plugin.connectors.base import ( - AFDConnectorBase, - AFDControlPlane, - AFDRoutedExpertsConnector, - ConnectorExtraInfo, -) -from afd_plugin.connectors.factory import AFDConnectorFactory -from afd_plugin.connectors.metadata import ( - AFDA2FTransferPayload, - AFDControlPayload, - AFDDPMetadata, - AFDExpertRoutingSpec, - AFDF2ATransferPayload, - AFDForwardContextMetadata, - AFDSingleDPMetadata, - AFDTransferContext, - AFDTransferMetadata, - AFDTransferState, - FFNLocalExperts, - LocalExpertWeights, -) +from importlib import import_module +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from afd_plugin.connectors.base import ( + AFDConnectorBase, + AFDControlPlane, + AFDRoutedExpertsConnector, + ConnectorExtraInfo, + ) + from afd_plugin.connectors.factory import AFDConnectorFactory + from afd_plugin.connectors.metadata import ( + AFDA2FTransferPayload, + AFDControlPayload, + AFDDPMetadata, + AFDExpertRoutingSpec, + AFDF2ATransferPayload, + AFDForwardContextMetadata, + AFDSingleDPMetadata, + AFDTransferContext, + AFDTransferMetadata, + AFDTransferState, + FFNLocalExperts, + LocalExpertWeights, + ) + + +def __getattr__(name: str): + if name in { + "AFDConnectorBase", + "AFDControlPlane", + "AFDRoutedExpertsConnector", + "ConnectorExtraInfo", + }: + module = "base" + elif name == "AFDConnectorFactory": + module = "factory" + elif name in __all__: + module = "metadata" + else: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + return getattr(import_module(f"{__name__}.{module}"), name) + __all__ = [ "AFDRoutedExpertsConnector", diff --git a/afd_plugin/connectors/base.py b/afd_plugin/connectors/base.py index e0fd3528d..4bf86a7e2 100644 --- a/afd_plugin/connectors/base.py +++ b/afd_plugin/connectors/base.py @@ -5,24 +5,25 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence +from contextlib import AbstractContextManager, nullcontext from dataclasses import dataclass from typing import TYPE_CHECKING, Any -import torch - from afd_plugin.config import AFDConfig, connector_extra_config_from_source -from afd_plugin.connectors.metadata import ( - AFDA2FTransferPayload, - AFDControlPayload, - AFDTransferContext, - FFNLocalExperts, - LocalExpertWeights, -) if TYPE_CHECKING: + import torch from vllm.config import VllmConfig + from afd_plugin.connectors.metadata import ( + AFDA2FTransferPayload, + AFDControlPayload, + AFDTransferContext, + FFNLocalExperts, + LocalExpertWeights, + ) + @dataclass(frozen=True) class ConnectorExtraInfo: @@ -36,6 +37,37 @@ def to_mapping(self) -> dict[str, Any]: return {} +class FFNExecutionSession: + """Backend lifecycle used by FFN runners and workers. + + By default, serving stops on local shutdown and the remaining hooks do + nothing. Backends with remote producers override the hooks and set + ``requires_drain``; pending work and the receiver handshake must finish + before communication or model resources are released. + """ + + requires_drain = False + + def keep_serving(self, shutdown_requested: bool) -> bool: + """Continue until local shutdown; remote producers may extend serving.""" + return not shutdown_requested + + def before_first_step(self) -> None: + """Prepare backend state once before the first FFN step.""" + + def after_step(self) -> None: + """Check backend status after a completed FFN step.""" + + def drain(self, run_one: Callable[[], None]) -> None: + """Drain remote work when overridden; by default do not invoke run_one.""" + + def finish_receiver(self) -> None: + """Finish the receiver handshake after pending work has drained.""" + + def validate_shutdown(self) -> None: + """Reject release when backend resources are still in use.""" + + class AFDConnectorBase(ABC): """Base class for plugin-owned AFD Attention/FFN connectors. @@ -112,6 +144,17 @@ def __init__( # Lifecycle methods # ============================== + def attention_step(self) -> AbstractContextManager[None]: + """Return a context for ``with connector.attention_step():`` around forward. + + Backends may defer transport status checks until the Attention model + step completes. The default context performs no additional work. + """ + return nullcontext() + + def ffn_session(self) -> FFNExecutionSession: + return FFNExecutionSession() + @abstractmethod def close(self) -> None: """Release connector-owned communication resources. @@ -378,4 +421,5 @@ def recv_dp_metadata_list(self) -> AFDControlPayload: "AFDControlPlaneClosedError", "AFDRoutedExpertsConnector", "ConnectorExtraInfo", + "FFNExecutionSession", ] diff --git a/afd_plugin/connectors/factory.py b/afd_plugin/connectors/factory.py index 4cc19400d..20b0a08c7 100644 --- a/afd_plugin/connectors/factory.py +++ b/afd_plugin/connectors/factory.py @@ -37,7 +37,10 @@ def register_connector( def loader() -> type[AFDConnectorBase]: module = importlib.import_module(module_path) - connector_cls = vars(module)[class_name] + # ``getattr`` rather than ``vars(module)``: packages in this tree resolve + # their names through a module-level ``__getattr__``, so the class is not + # necessarily present in the module dictionary. + connector_cls = getattr(module, class_name) if not issubclass(connector_cls, AFDConnectorBase): raise TypeError( f"{module_path}.{class_name} is not an AFDConnectorBase", diff --git a/afd_plugin/connectors/gpu/deepep_m2n/connector.py b/afd_plugin/connectors/gpu/deepep_m2n/connector.py index c0d2fa917..4eb11dca3 100644 --- a/afd_plugin/connectors/gpu/deepep_m2n/connector.py +++ b/afd_plugin/connectors/gpu/deepep_m2n/connector.py @@ -9,12 +9,22 @@ weighted results back, where combine sums each token's top-k contributions. Topology: - One AFD world ordered ``[F0, F1, ..., A0, A1, ...]`` (as the P2P - connector). DeepEP sees it as a symmetric EP group in which FFN rank ``g`` + One AFD role world ordered ``[F0, F1, ..., A0, A1, ...]`` (as the P2P + connector) for the handshake, the model checks and the control plane. + DeepEP sees the ranks as a symmetric EP group in which FFN rank ``g`` owns real experts ``[g * E/F, (g + 1) * E/F)`` and Attention ranks own dummy expert slots (see ``layout.py``). Any Attention:FFN ratio works; ``num_experts`` must be divisible by the FFN rank count. +Placement: + Every NVLink domain (on HGX, every pod) holds the same number of ranks, + and either the FFN ranks fill the first domains or every domain holds the + same number of FFN ranks, with FFN DP ranks rising in pod order (2 x 4A1F + is two pods of one FFN and four Attention ranks). The handshake derives + the transport's physical rank order from the ranks' NVLink domains + (``plan_transport_layout``) and picks afdEP-comm's FFN-prefix or + per-domain mode; nothing is configured. + Data path (per MoE layer and stage, identical order on every rank): Attention: ``send_attn_output(h, ctx, topk_ids=, topk_weights=)`` dispatches; ``recv_ffn_output(ref, ubatch_idx)`` combines and returns @@ -88,7 +98,12 @@ DeepEPM2NExperts, ExpertBackend, ) -from afd_plugin.connectors.gpu.deepep_m2n.layout import DeepEPM2NLayout +from afd_plugin.connectors.gpu.deepep_m2n.layout import ( + DeepEPM2NLayout, + RankPlacement, + TransportLayout, + plan_transport_layout, +) from afd_plugin.connectors.gpu.deepep_m2n.network import ( NetworkConfig, NetworkPlan, @@ -143,6 +158,9 @@ # CPU (gloo) group for the handshake and DeepEP's communicator bootstrap, # created before any NCCL communicator so the network plan can still set NCCL_*. BOOTSTRAP_GROUP_NAME = "afd_deepep_m2n_bootstrap" +# CPU (gloo) group in transport order for the lanes, created only when that +# order differs from the role world (per-domain FFN placement). +TRANSPORT_GROUP_NAME = "afd_deepep_m2n_transport" # FastAFD passes 120 s for both the CPU wait and the GPU spin timeouts. DEFAULT_TIMEOUT_S = 120 # Only Attention rank 0 sends the per-step control payload. @@ -433,6 +451,7 @@ class HandshakeResult: attn_tp_size: int capacity_tokens_per_rank: int bucket_ladder: tuple[int, ...] + transport: TransportLayout def expert_weight_format(quant_config: QuantizationConfig | None) -> str: @@ -626,9 +645,26 @@ def validate_handshake( raise RuntimeError( f"Attention ranks disagree on bucket_ladder: {sorted(map(str, ladders))}" ) - # Every rank evaluates the same plan on the same list. + # Every rank evaluates the same plans on the same list. The network plan + # sees the ranks in the transport order afdEP-comm will see. + topologies = [RankTopology.from_mapping(info["topology"]) for info in infos] + transport = plan_transport_layout( + [ + RankPlacement(info["role"], info["role_rank"], topology.nvlink_domain) + for info, topology in zip(infos, topologies, strict=True) + ], + layout.ffn_size, + ) plan = plan_network( - [RankTopology.from_mapping(info["topology"]) for info in infos], network + [ + RankTopology( + world_rank=transport.transport_rank(info["role"], info["role_rank"]), + host=topology.host, + nvlink_domain=topology.nvlink_domain, + ) + for info, topology in zip(infos, topologies, strict=True) + ], + network, ) return HandshakeResult( expert_alignment=int(alignments.pop()), @@ -636,6 +672,7 @@ def validate_handshake( attn_tp_size=attn_tp_size, capacity_tokens_per_rank=int(capacities.pop()), bucket_ladder=ladders.pop() or (), + transport=transport, ) @@ -962,6 +999,9 @@ def __init__( self._timeout = timedelta(seconds=afd_config.afd_process_group_timeout_s) self.group: dist.ProcessGroup | None = None self.bootstrap_group: dist.ProcessGroup | None = None + # The lanes' group: the bootstrap group, or one in transport order. + self.transport_group: dist.ProcessGroup | None = None + self.transport: TransportLayout | None = None self.network_plan: NetworkPlan | None = None self.lanes: list[ElasticBuffer] = [] self.lane_streams: list[torch.cuda.Stream] = [] @@ -1041,8 +1081,25 @@ def _handshake(self) -> None: self.attn_tp_size = result.attn_tp_size self.capacity_tokens_per_rank = result.capacity_tokens_per_rank self.bucket_ladder = result.bucket_ladder + self.transport = result.transport + if self.transport.is_identity: + self.transport_group = self.bootstrap_group + else: + # Transport rank 0 is FFN rank 0, the process that hosts the + # rendezvous store, so this group joins the same store. + self.transport_group = init_afd_process_group( + backend="gloo", + init_method=self._rendezvous, + world_size=self.layout.union_size, + rank=self.transport.transport_rank( + self.afd_config.role, self.role_rank + ), + group_name=TRANSPORT_GROUP_NAME, + timeout=self._timeout, + ) if self.world_rank == 0: logger.info("DeepEP M2N network plan: %s", self.network_plan.summary()) + logger.info("DeepEP M2N transport: %s", self.transport.summary()) # DeepEP reads both when each buffer is created: the GIN switch, and # with several lanes a communication stream per buffer. apply_gin_environment(self.network_plan) @@ -1117,10 +1174,11 @@ def init_afd_connector( def _new_lane(self) -> ElasticBuffer: """One lane's afdEP-comm transport.""" - assert self.bootstrap_group is not None and self.network_plan is not None + assert self.transport_group is not None and self.transport is not None + assert self.network_plan is not None assert self.capacity_tokens_per_rank is not None return ElasticBuffer( - group=self.bootstrap_group, + group=self.transport_group, capacity_tokens_per_rank=self.capacity_tokens_per_rank, hidden_size=self.hidden_size, num_experts=self.layout.deepep_num_experts, @@ -1132,8 +1190,10 @@ def _new_lane(self) -> ElasticBuffer: rdma_traffic_class=self.network_plan.rdma_traffic_class, cpu_timeout_s=self.config.cpu_timeout_s, gpu_timeout_s=self.config.gpu_timeout_s, - # The DeepEP M2N layout puts FFN ranks first, as afdEP-comm expects. + # FFN ranks are transport ranks [0, F) (FFN prefix), or the first + # ``ffn_ranks_per_domain`` of every NVLink domain (per-domain mode). ffn_ranks=self.ffn_size, + ffn_ranks_per_domain=self.transport.ffn_ranks_per_domain, ) def close(self) -> None: @@ -1143,6 +1203,12 @@ def close(self) -> None: _CONNECTORS.pop(self._op_id, None) for lane in self.lanes: lane.destroy() + if ( + self.transport_group is not None + and self.transport_group is not self.bootstrap_group + ): + dist.destroy_process_group(self.transport_group) + self.transport_group = None self.lanes = [] self.lane_streams = [] self._pending_combines.clear() @@ -1157,8 +1223,9 @@ def is_initialized(self) -> bool: @property def domain(self) -> tuple[int, int]: - """DeepEP ``(scale-out ranks, scale-up ranks)``: nodes and GPUs per node - in hybrid mode, ``(1, world)`` when everything is in one NVLink domain.""" + """DeepEP ``(scale-out ranks, scale-up ranks)``: NVLink domains and + ranks per domain in hybrid mode (2 x 4A1F: ``(2, 5)``), ``(1, world)`` + when everything is in one NVLink domain.""" self._check_initialized() return self.lanes[0].num_scaleout_ranks, self.lanes[0].num_scaleup_ranks diff --git a/afd_plugin/connectors/gpu/deepep_m2n/layout.py b/afd_plugin/connectors/gpu/deepep_m2n/layout.py index 24d1bde89..d8bb225d2 100644 --- a/afd_plugin/connectors/gpu/deepep_m2n/layout.py +++ b/afd_plugin/connectors/gpu/deepep_m2n/layout.py @@ -7,16 +7,28 @@ DeepEP treats the union of Attention and FFN ranks as one symmetric EP group in which every rank owns ``experts_per_rank`` experts. FastAFD orders the union ``[Attention.., FFN..]`` and remaps real expert ids by an offset. The -plugin orders AFD worlds FFN first (``[F0.., A0..]``, as the P2P connector -does), so FFN rank ``g`` owns DeepEP experts ``[g * E_local, (g + 1) * E_local)``, -which are exactly the real expert ids: no remap. Attention ranks own dummy -expert slots after the real ones that routing never selects. +plugin orders the AFD role world FFN first (``[F0.., A0..]``, as the P2P +connector does), so FFN rank ``g`` owns real experts +``[g * E_local, (g + 1) * E_local)``. Attention ranks own dummy expert slots +that routing never selects. + +The transport world (``plan_transport_layout``) holds the same ranks in +physical order: NVLink domain by domain, as DeepEP numbers ranks, with each +domain's FFN ranks first. When the FFN ranks fill the first domains, it equals +the role world and DeepEP expert ids are the real ones (afdEP-comm's FFN +prefix). When every domain holds ``f`` FFN ranks, afdEP-comm's per-domain mode +places FFN rank ``j`` at transport rank ``(j // f) * U + j % f`` (``U`` ranks +per domain) and translates the real expert ids itself. """ from __future__ import annotations +from collections.abc import Sequence from dataclasses import dataclass +FFN_ROLE = "ffn" +ATTENTION_ROLE = "attention" + @dataclass(frozen=True) class DeepEPM2NLayout: @@ -50,10 +62,11 @@ def deepep_num_experts(self) -> int: return self.union_size * self.experts_per_rank def world_rank(self, role: str, role_rank: int) -> int: - if role == "ffn": + """Rank in the role world ``[F0.., A0..]``.""" + if role == FFN_ROLE: _check_rank("ffn role rank", role_rank, self.ffn_size) return role_rank - if role == "attention": + if role == ATTENTION_ROLE: _check_rank("attention role rank", role_rank, self.attn_size) return self.ffn_size + role_rank raise ValueError(f"unknown AFD role {role!r}") @@ -68,3 +81,124 @@ def expert_range(self, ffn_rank: int) -> range: def _check_rank(name: str, rank: int, size: int) -> None: if not 0 <= rank < size: raise ValueError(f"{name} must be in [0, {size}), got {rank}") + + +@dataclass(frozen=True) +class RankPlacement: + """Where one rank runs, as it reports in the handshake.""" + + role: str + role_rank: int + nvlink_domain: str + + +@dataclass(frozen=True) +class TransportLayout: + """The physical rank order afdEP-comm's transport sees. + + Attributes: + order: ``(role, role_rank)`` of each transport rank. + ffn_ranks_per_domain: afdEP-comm's ``ffn_ranks_per_domain``: 0 when the + FFN ranks are transport ranks ``[0, F)`` (FFN prefix), else the FFN + rank count of every NVLink domain (per-domain mode). + num_domains: NVLink domains. + domain_size: ranks per NVLink domain. + shape: ``A