diff --git a/.agents/skills/run-e2e/SKILL.md b/.agents/skills/run-e2e/SKILL.md index 5cc692df..1d3f860d 100644 --- a/.agents/skills/run-e2e/SKILL.md +++ b/.agents/skills/run-e2e/SKILL.md @@ -1,6 +1,6 @@ --- name: run-e2e -description: Use when the user asks to run, validate, or diagnose the AFD plugin's DeepSeek-V2-Lite end-to-end tests on GPU or Ascend NPU hardware, including PR-gate E2E, GSM8K-7 accuracy, graph, eager, or DBO scenarios. +description: Use when the user asks to run, validate, or diagnose the AFD plugin's DeepSeek-V2-Lite end-to-end tests on GPU or Ascend NPU hardware, including PR-gate E2E, GSM8K-7 accuracy, graph, eager, DBO, or 2A2F scenarios. --- # Run AFD E2E Tests @@ -15,9 +15,8 @@ tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py: - afd-graph - afd-graph-dbo -Each scenario evaluates the first 7 GSM8K samples with 2 Attention ranks and 1 -FFN rank. Do not use removed markers, categories, TP, or 2A2F workflows. This -skill does not cover unit tests or source edits. +Each default scenario evaluates the first 7 GSM8K samples with 2 Attention +ranks and 1 FFN rank. `afd-graph-dbo-2a2f` is a separate opt-in case. ## Workflow @@ -80,11 +79,25 @@ From the repository root, stream output in the foreground: ~~~bash python -m pytest -q -s \ - tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-eager]" \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph]" \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo]" \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[baseline-graph]" ~~~ Do not add backend markers or run scenarios in parallel; they share devices. +For the opt-in GPU/NPU 2A2F case, set four unique device IDs and run: + +~~~bash +export AFD_E2E_DEVICES=0,1,2,3 +python -m pytest -q -s \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo-2a2f]" +~~~ + +The first two devices run Attention DP2/TP1. The last two run FFN +DP2/TP1/EP2. + On cancellation, forward SIGTERM and allow over 90 seconds for cleanup. ### 5. Report @@ -97,7 +110,7 @@ actionable error, and cleanup status. Any skip is a gate failure. | Variable | Backend | Required | |---|---|---| | AFD_E2E_BACKEND | both | yes: gpu or npu | -| AFD_E2E_DEVICES | both | yes: exactly 3 unique IDs | +| AFD_E2E_DEVICES | both | yes: 3 default or 4 for 2A2F | | AFD_GPU_E2E_MODEL | GPU | yes | | AFD_GPU_E2E_VLLM_BIN | GPU | no; defaults to vllm | | AFD_NPU_E2E_MODEL | NPU | yes | diff --git a/docs/design/module/e2e_testing.md b/docs/design/module/e2e_testing.md index 220abe28..5acb0f8b 100644 --- a/docs/design/module/e2e_testing.md +++ b/docs/design/module/e2e_testing.md @@ -24,6 +24,7 @@ validation_paths: - "tests/unit/test_e2e_runner.py" - "tests/unit/test_e2e_process_utils.py" - "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py" + - "tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py" upstream_refs: - "vLLM 0.26.0 serving and shutdown interfaces" - "lm-evaluation-harness GSM8K task and local-completions API" @@ -50,7 +51,7 @@ operator tests, benchmarks, and performance tests are outside this scope. ```text tests/e2e/ -├── models//test_.py # Test entry and case list +├── models//test_*.py # Test entries and case lists ├── accuracy/.py # Accuracy tool and result parsing ├── runner.py # Service startup, evaluation, and cleanup └── process_utils.py # Process-group termination and reaping @@ -72,7 +73,8 @@ cleanup. Production code does not depend on the E2E harness. incomplete results, failed validation, or failed cleanup. They **MUST NOT** use `skip`, `xfail`, or success-on-empty behavior. - `E2E-INV-005` — The harness **MUST** check service liveness before and after - evaluation, evaluator exit status, sample count, `NaN`, and accuracy. + evaluation. Accuracy cases **MUST** also check evaluator exit status, sample + count, `NaN`, and accuracy. - `E2E-INV-006` — Child processes **MUST** use owned process groups. Cancellation **MUST** send `SIGTERM`, use a bounded grace period, then reap every leader. A harness `SIGKILL` escalation **MUST** fail the case. @@ -97,6 +99,13 @@ for merge validation. Put slower coverage in a scheduled job. Prefer graph coverage. Keep one eager smoke test unless a feature cannot run in graph mode. +`afd-eager-async-cam` is a separate NPU-only smoke test. It uses four devices +for Attention DP1/TP2 and FFN DP2/TP1/EP2. It is not part of the PR gate above. + +`afd-graph-dbo-2a2f` is a separate GPU/NPU accuracy case. It uses four devices +for Attention DP2/TP1 and FFN DP2/TP1/EP2 and runs GSM8K-7. The default gate +selects only the four default cases by pytest node ID. + ## Accuracy gate | Setting | PR | Weekly | diff --git a/tests/e2e/README.md b/tests/e2e/README.md index 771984b6..a7fdd357 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -1,7 +1,7 @@ # End-to-End Tests These tests validate DeepSeek-V2-Lite on real GPU or Ascend NPU hardware. -They run four scenarios: +The default gate runs four scenarios: - `baseline-graph` - `afd-eager` @@ -47,11 +47,25 @@ Then run: ```bash python -m pytest -q -s \ - tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-eager]" \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph]" \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo]" \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[baseline-graph]" ``` Success means 4 passed and 0 skipped. +### Graph + DBO 2A2F + +This separate GPU/NPU case uses four devices: the first two for Attention +DP=2/TP=1 and the last two for FFN DP=2/TP=1/EP=2. It runs GSM8K-7. + +```bash +export AFD_E2E_DEVICES=0,1,2,3 +python -m pytest -q -s \ + "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo-2a2f]" +``` + For the weekly full GSM8K test, run only `afd-graph-dbo`: ```bash @@ -76,3 +90,21 @@ devices 0,1,2, and HF_HOME /data/huggingface. For either backend, provide the model path, three device IDs, and `HF_HOME`. The skill checks prerequisites, runs the same four tests, and reports failures and process cleanup. + +## NPU async CAM smoke test + +This separate test uses four NPUs: the first two for Attention TP=2, and the +last two for FFN DP=2/TP=1/EP=2. It sends one prompt and requests 32 tokens. It +does not run GSM8K. + +```bash +export AFD_E2E_BACKEND=npu +export AFD_E2E_DEVICES=0,1,2,3 +export AFD_NPU_E2E_MODEL=/path/to/DeepSeek-V2-Lite +python -m pytest -q -s \ + tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py +``` + +The CAM/CANN runtime and custom operators must already be installed. Missing +model configuration or a device list other than four unique IDs fails the +test. diff --git a/tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py b/tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py new file mode 100644 index 00000000..8c8a09c6 --- /dev/null +++ b/tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py @@ -0,0 +1,130 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the AFD plugin project +"""NPU smoke test for DeepSeek-V2-Lite with async CAM.""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path + +import pytest + +from tests.e2e.models.deepseek_v2_lite.test_deepseek_v2_lite import ( + _devices, + _required_env, + _run_runner, +) +from tests.e2e.runner import ( + ASYNC_CAM_ATTENTION_RANKS, + ASYNC_CAM_FFN_RANKS, + ASYNC_CAM_SCENARIO, +) + +CAM_VENDOR_PATH = Path("/usr/local/Ascend/cann-9.0.1/opp/vendors/CAM") +CAM_OP_API_PATH = CAM_VENDOR_PATH / "op_api" +CAM_OP_API_LIB_PATH = CAM_OP_API_PATH / "lib" +CAM_HCCL_BUFFER_SIZE = "4096" +CAM_MAX_NUM_SEQUENCES = "8" +CAM_MAX_BATCHED_TOKENS = "8000" +CAM_MEMORY_UTILIZATION = "0.75" + + +def _prepend_env_paths(env: dict[str, str], name: str, *paths: Path) -> None: + existing = [path for path in env.get(name, "").split(os.pathsep) if path] + env[name] = os.pathsep.join( + dict.fromkeys([*(str(path) for path in paths), *existing]), + ) + + +def _async_cam_env() -> dict[str, str]: + env = os.environ.copy() + env.setdefault("VLLM_USE_V1", "1") + env["HCCL_BUFFSIZE"] = CAM_HCCL_BUFFER_SIZE + env.setdefault("PYTORCH_NPU_ALLOC_CONF", "expandable_segments:True") + env.setdefault("ASCEND_LAUNCH_BLOCKING", "1") + env.setdefault("VLLM_ASCEND_ENABLE_CONTEXT_PARALLEL", "1") + env.setdefault("VLLM_ASCEND_ENABLE_FLASHCOMM1", "1") + _prepend_env_paths( + env, + "LD_LIBRARY_PATH", + CAM_OP_API_PATH, + CAM_OP_API_LIB_PATH, + ) + _prepend_env_paths(env, "ASCEND_CUSTOM_OPP_PATH", CAM_VENDOR_PATH) + return env + + +def _connector_extra_config(model: str) -> str: + dynamic_quant = int( + os.environ.get( + "AFD_NPU_ASYNC_CAM_E2E_DYNAMIC_QUANT", + "1" if (Path(model) / "quant_model_description.json").is_file() else "0", + ), + ) + return json.dumps({"dynamicQuant": dynamic_quant}, separators=(",", ":")) + + +def build_runner_command() -> list[str]: + backend = _required_env("AFD_E2E_BACKEND") + if backend != "npu": + raise RuntimeError("async CAM E2E requires AFD_E2E_BACKEND=npu") + + device_count = ASYNC_CAM_ATTENTION_RANKS + ASYNC_CAM_FFN_RANKS + devices = _devices("AFD_E2E_DEVICES", device_count) + model = _required_env("AFD_NPU_E2E_MODEL") + common_arguments = ( + "--trust-remote-code", + "--max-num-seqs", + CAM_MAX_NUM_SEQUENCES, + "--max-num-batched-tokens", + CAM_MAX_BATCHED_TOKENS, + "--gpu-memory-utilization", + CAM_MEMORY_UTILIZATION, + "--no-enable-prefix-caching", + ) + command = [ + sys.executable, + "-m", + "tests.e2e.runner", + "--model", + model, + "--vllm-bin", + os.environ.get("AFD_NPU_E2E_VLLM_BIN", "vllm"), + "--device-backend", + "npu", + "--attention-devices", + ",".join(devices[:ASYNC_CAM_ATTENTION_RANKS]), + "--ffn-devices", + ",".join(devices[ASYNC_CAM_ATTENTION_RANKS:]), + "--scenario", + ASYNC_CAM_SCENARIO, + "--served-model-name-prefix", + "cam-async", + "--afd-connector-extra-config", + _connector_extra_config(model), + "--api-port-base", + os.environ.get("AFD_NPU_ASYNC_CAM_E2E_API_PORT", "19080"), + "--afd-port", + os.environ.get("AFD_NPU_ASYNC_CAM_E2E_AFD_PORT", "6453"), + "--startup-timeout", + os.environ.get("AFD_NPU_E2E_STARTUP_TIMEOUT", "900"), + *(f"--common-vllm-arg={argument}" for argument in common_arguments), + ] + max_model_len = os.environ.get("AFD_NPU_ASYNC_CAM_E2E_MAX_MODEL_LEN") + if max_model_len: + command.extend( + [ + "--common-vllm-arg=--max-model-len", + f"--common-vllm-arg={max_model_len}", + ], + ) + return command + + +@pytest.mark.npu +@pytest.mark.e2e +@pytest.mark.slow +def test_deepseek_v2_lite_async_cam() -> None: + _run_runner(build_runner_command(), env=_async_cam_env()) diff --git a/tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py b/tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py index a262a272..3bc137d9 100644 --- a/tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py +++ b/tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py @@ -19,6 +19,7 @@ "afd-eager", "afd-graph", "afd-graph-dbo", + "afd-graph-dbo-2a2f", ) # Covers the roughly 64-second nested lm-eval/vLLM cleanup bound with buffer. RUNNER_CLEANUP_TIMEOUT_S = 90 @@ -51,7 +52,8 @@ def build_runner_command(scenario: str, gsm8k_output_path: Path) -> list[str]: else: raise RuntimeError("AFD_E2E_BACKEND must be 'gpu' or 'npu'") - devices = _devices("AFD_E2E_DEVICES", 3) + expected_device_count = 4 if scenario == "afd-graph-dbo-2a2f" else 3 + devices = _devices("AFD_E2E_DEVICES", expected_device_count) attention_devices = devices[:2] ffn_devices = devices[2:] if scenario == "baseline-graph": @@ -83,7 +85,7 @@ def build_runner_command(scenario: str, gsm8k_output_path: Path) -> list[str]: return command -def _run_runner(command: list[str]) -> None: +def _run_runner(command: list[str], env: dict[str, str] | None = None) -> None: handled_signals = (signal.SIGTERM, signal.SIGINT) previous_handlers = {signum: signal.getsignal(signum) for signum in handled_signals} process: subprocess.Popen | None = None @@ -113,6 +115,7 @@ def forward_cancellation(signum, _frame) -> None: process = subprocess.Popen( command, cwd=REPO_ROOT, + env=env, start_new_session=True, ) forward_received_signal() diff --git a/tests/e2e/runner.py b/tests/e2e/runner.py index 6139f507..6732c21e 100644 --- a/tests/e2e/runner.py +++ b/tests/e2e/runner.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the AFD plugin project -"""Run DeepSeekV2-Lite DeepEP and AFD GSM8K E2E scenarios.""" +"""Run DeepSeek-V2-Lite DeepEP baseline and AFD E2E scenarios.""" from __future__ import annotations @@ -27,6 +27,10 @@ REPO_ROOT = Path(__file__).resolve().parents[2] ASYNC_AFD_CONNECTOR = "CAMAsyncAFDConnector" +ASYNC_CAM_SCENARIO = "afd-eager-async-cam" +ASYNC_CAM_ATTENTION_RANKS = 2 +ASYNC_CAM_FFN_RANKS = 2 +ASYNC_CAM_ATTENTION_TP_SIZE = 2 PROCESS_TERMINATION_TIMEOUT_S = 20 PROCESS_POLL_INTERVAL_S = 0.2 PROCESS_REAP_TIMEOUT_S = 5 @@ -39,6 +43,31 @@ GSM8K_LIMIT_ENV = "AFD_GSM8K_LIMIT" GSM8K_THRESHOLD_ENV = "AFD_GSM8K_THRESHOLD" DEFAULT_GSM8K_THRESHOLD = 0.27 +COMPLETION_REQUEST_TIMEOUT_S = 120 +COMPLETION_MAX_TOKENS = 32 +COMPLETION_TEMPERATURE = 0 +ACCOUNTING_PROMPT = ( + "<|im_start|>system\n" + "You are a professional accountant. Answer questions using accounting " + "knowledge, output only the option letter (A/B/C/D).<|im_end|>\n" + "<|im_start|>user\n" + "Question: A company's balance sheet as of December 31, 2023 shows:\n" + " Current assets: Cash and equivalents 5 million yuan, Accounts " + "receivable 8 million yuan, Inventory 6 million yuan\n" + " Non-current assets: Net fixed assets 12 million yuan\n" + " Current liabilities: Short-term loans 4 million yuan, Accounts " + "payable 3 million yuan\n" + " Non-current liabilities: Long-term loans 9 million yuan\n" + " Owner's equity: Paid-in capital 10 million yuan, Retained earnings ?\n" + "Requirement: Calculate the company's Asset-Liability Ratio and Current " + "Ratio (round to two decimal places).\n" + "Options:\n" + "A. Asset-Liability Ratio=58.33%, Current Ratio=1.90\n" + "B. Asset-Liability Ratio=62.50%, Current Ratio=2.17\n" + "C. Asset-Liability Ratio=65.22%, Current Ratio=1.75\n" + "D. Asset-Liability Ratio=68.00%, Current Ratio=2.50<|im_end|>\n" + "<|im_start|>assistant\n" +) def main() -> int: @@ -108,7 +137,10 @@ def exit_after_cleanup(signum: int, _frame: Any) -> None: wait_for_openai_api(args, processes) ensure_processes_alive(processes) - run_gsm8k_evaluation(args) + if args.scenario == ASYNC_CAM_SCENARIO: + run_completion_evaluation(args) + else: + run_gsm8k_evaluation(args) ensure_processes_alive(processes) return 0 @@ -154,13 +186,19 @@ def parse_args() -> argparse.Namespace: ) parser.add_argument( "--scenario", - choices=["baseline-graph", "afd-eager", "afd-graph", "afd-graph-dbo"], + choices=[ + "baseline-graph", + "afd-eager", + "afd-graph", + "afd-graph-dbo", + "afd-graph-dbo-2a2f", + ASYNC_CAM_SCENARIO, + ], required=True, help="Fixed E2E scenario to run.", ) parser.add_argument( "--gsm8k-output-path", - required=True, help="Directory or file path where lm-eval writes GSM8K results.", ) parser.add_argument( @@ -173,7 +211,7 @@ def parse_args() -> argparse.Namespace: default="0", help=( "Comma-separated device IDs for the Attention serve process. " - "The number of devices must match Attention DP size." + "The number of devices must match Attention DP times TP." ), ) parser.add_argument( @@ -181,7 +219,7 @@ def parse_args() -> argparse.Namespace: default="", help=( "Comma-separated device IDs for the FFN serve process. " - "The number of devices must match FFN DP size." + "The number of devices must match FFN DP times TP." ), ) parser.add_argument("--api-host", default="127.0.0.1") @@ -255,11 +293,20 @@ def parse_args() -> argparse.Namespace: def configure_scenario(args: argparse.Namespace) -> None: """Set topology and features for the selected fixed scenario.""" + is_async_cam = args.scenario == ASYNC_CAM_SCENARIO scenario_settings = { "baseline-graph": (True, True, False, 1, 0), "afd-eager": (False, False, False, 2, 1), "afd-graph": (False, True, False, 2, 1), "afd-graph-dbo": (False, True, True, 2, 1), + "afd-graph-dbo-2a2f": (False, True, True, 2, 2), + ASYNC_CAM_SCENARIO: ( + False, + False, + False, + ASYNC_CAM_ATTENTION_RANKS, + ASYNC_CAM_FFN_RANKS, + ), } baseline, use_graph, enable_dbo, attention_ranks, ffn_ranks = scenario_settings[ args.scenario @@ -270,8 +317,21 @@ def configure_scenario(args: argparse.Namespace) -> None: args.num_attention_ranks = attention_ranks args.num_ffn_ranks = ffn_ranks args.tp_size = 1 - args.attention_tp_size = 1 + args.attention_tp_size = ASYNC_CAM_ATTENTION_TP_SIZE if is_async_cam else 1 args.ffn_tp_size = 1 + if not is_async_cam and args.gsm8k_output_path is None: + raise ValueError("--gsm8k-output-path is required for GSM8K scenarios") + if is_async_cam: + args.afd_connector = ASYNC_AFD_CONNECTOR + args.afd_async = True + args.compute_gate_on_attention = True + extra_config = parse_afd_connector_extra_config( + args.afd_connector_extra_config, + ) + extra_config["attn_ranks_per_dp"] = ASYNC_CAM_ATTENTION_TP_SIZE + args.afd_connector_extra_config = [ + json.dumps(extra_config, separators=(",", ":")), + ] if use_graph: args.cudagraph_capture_size = 8 if enable_dbo: @@ -311,8 +371,8 @@ def validate_topology( if role_tp_size(args, "attention") != 1: raise ValueError("baseline E2E requires Attention TP=1") return - if args.num_attention_ranks != 2 or args.num_ffn_ranks != 1: - raise ValueError("AFD E2E requires two Attention ranks and one FFN rank") + if args.scenario == ASYNC_CAM_SCENARIO and args.device_backend != "npu": + raise ValueError("async CAM E2E requires NPU") for role, rank_count in ( ("attention", args.num_attention_ranks), ("ffn", args.num_ffn_ranks), @@ -531,6 +591,8 @@ def ffn_api_port(args: argparse.Namespace) -> int: def run_gsm8k_evaluation(args: argparse.Namespace) -> None: """Run the configured GSM8K workload against the scenario's public API.""" + if args.gsm8k_output_path is None: + raise RuntimeError("--gsm8k-output-path is required for GSM8K scenarios") configured_limit = os.environ.get( GSM8K_LIMIT_ENV, str(DEFAULT_GSM8K_SAMPLE_LIMIT), @@ -568,6 +630,46 @@ def run_gsm8k_evaluation(args: argparse.Namespace) -> None: ) +def run_completion_evaluation(args: argparse.Namespace) -> None: + """Send the async CAM smoke request and require one returned choice.""" + payload = json.dumps( + { + "model": served_model_name(args, "attention"), + "prompt": ACCOUNTING_PROMPT, + "max_tokens": COMPLETION_MAX_TOKENS, + "temperature": COMPLETION_TEMPERATURE, + }, + ).encode() + request = urllib.request.Request( + f"http://{args.api_host}:{attention_api_port(args)}/v1/completions", + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen( + request, + timeout=COMPLETION_REQUEST_TIMEOUT_S, + ) as response: + result = json.loads(response.read()) + except urllib.error.HTTPError as exc: + body = exc.read().decode(errors="replace") + raise RuntimeError( + f"completion request failed with HTTP {exc.code}: {body}", + ) from exc + except (OSError, json.JSONDecodeError) as exc: + raise RuntimeError(f"completion request failed: {exc}") from exc + + choices = result.get("choices") + if not isinstance(choices, list) or not choices: + raise RuntimeError("completion response contains no choices") + choice = choices[0] + text = choice.get("text") if isinstance(choice, dict) else None + if not isinstance(text, str) or not text.strip(): + raise RuntimeError("completion response contains no text") + print(f"Completion response: {text}") + + def build_env( visible_devices: str, args: argparse.Namespace, diff --git a/tests/unit/test_e2e_runner.py b/tests/unit/test_e2e_runner.py index 970979a4..132e11a6 100644 --- a/tests/unit/test_e2e_runner.py +++ b/tests/unit/test_e2e_runner.py @@ -30,6 +30,12 @@ def _e2e_entrypoint(): ) +def _async_cam_entrypoint(): + return importlib.import_module( + "tests.e2e.models.deepseek_v2_lite.test_async_cam_npu", + ) + + def _set_e2e_entrypoint_env(monkeypatch, backend: str) -> None: for name in E2E_ENTRYPOINT_ENV_VARS: monkeypatch.delenv(name, raising=False) @@ -101,6 +107,25 @@ def test_e2e_entrypoint_builds_npu_runner_command(monkeypatch, tmp_path): ] +@pytest.mark.parametrize("backend", ["gpu", "npu"]) +def test_e2e_entrypoint_builds_four_device_2a2f_command( + monkeypatch, + tmp_path, + backend, +): + _set_e2e_entrypoint_env(monkeypatch, backend) + monkeypatch.setenv("AFD_E2E_DEVICES", "0,1,2,3") + + command = _e2e_entrypoint().build_runner_command( + "afd-graph-dbo-2a2f", + tmp_path / "afd-graph-dbo-2a2f", + ) + + assert command[command.index("--attention-devices") + 1] == "0,1" + assert command[command.index("--ffn-devices") + 1] == "2,3" + assert command[command.index("--scenario") + 1] == "afd-graph-dbo-2a2f" + + @pytest.mark.parametrize(("backend", "expected_device"), [("gpu", "2"), ("npu", "1")]) def test_e2e_entrypoint_baseline_uses_only_one_attention_device( monkeypatch, @@ -181,6 +206,61 @@ def test_e2e_entrypoint_rejects_reused_devices( _e2e_entrypoint().build_runner_command("afd-eager", tmp_path / "results") +def test_async_cam_entrypoint_builds_a_four_npu_runner_command( + monkeypatch, +): + monkeypatch.setenv("AFD_E2E_BACKEND", "npu") + monkeypatch.setenv("AFD_E2E_DEVICES", "0,2,4,6") + monkeypatch.setenv("AFD_NPU_E2E_MODEL", "/models/deepseek-v2-lite-npu") + monkeypatch.setenv("AFD_NPU_E2E_VLLM_BIN", "/opt/npu/bin/vllm") + monkeypatch.delenv("AFD_NPU_ASYNC_CAM_E2E_DYNAMIC_QUANT", raising=False) + + entrypoint = _async_cam_entrypoint() + command = entrypoint.build_runner_command() + env = entrypoint._async_cam_env() + + assert command[command.index("--attention-devices") + 1] == "0,2" + assert command[command.index("--ffn-devices") + 1] == "4,6" + assert command[command.index("--scenario") + 1] == "afd-eager-async-cam" + assert command[command.index("--model") + 1] == "/models/deepseek-v2-lite-npu" + assert command[command.index("--vllm-bin") + 1] == "/opt/npu/bin/vllm" + assert "--common-vllm-arg=--gpu-memory-utilization" in command + assert "--common-vllm-arg=0.75" in command + extra_config = json.loads( + command[command.index("--afd-connector-extra-config") + 1], + ) + assert extra_config == {"dynamicQuant": 0} + assert env["HCCL_BUFFSIZE"] == "4096" + assert str(entrypoint.CAM_OP_API_LIB_PATH) in env["LD_LIBRARY_PATH"] + assert str(entrypoint.CAM_VENDOR_PATH) in env["ASCEND_CUSTOM_OPP_PATH"] + + +@pytest.mark.parametrize( + ("backend", "devices", "model", "error"), + [ + ("gpu", "0,1,2,3", "model", "requires AFD_E2E_BACKEND=npu"), + ("npu", "0,1,2", "model", "exactly 4 devices"), + ("npu", "0,1,2,2", "model", "unique"), + ("npu", "0,1,2,3", None, "AFD_NPU_E2E_MODEL"), + ], +) +def test_async_cam_entrypoint_rejects_invalid_configuration( + monkeypatch, + backend, + devices, + model, + error, +): + monkeypatch.setenv("AFD_E2E_BACKEND", backend) + monkeypatch.setenv("AFD_E2E_DEVICES", devices) + monkeypatch.delenv("AFD_NPU_E2E_MODEL", raising=False) + if model: + monkeypatch.setenv("AFD_NPU_E2E_MODEL", model) + + with pytest.raises(RuntimeError, match=error): + _async_cam_entrypoint().build_runner_command() + + def test_e2e_entrypoint_forwards_cancellation_and_reaps_runner(monkeypatch): entrypoint = _e2e_entrypoint() monkeypatch.setattr(entrypoint, "os", runner.os, raising=False) @@ -248,6 +328,7 @@ def fake_signal(signum, handler): command, { "cwd": Path(__file__).resolve().parents[2], + "env": None, "start_new_session": True, }, ), @@ -402,10 +483,12 @@ def test_parse_args_rejects_legacy_fixed_scenario_options(monkeypatch, legacy_ar @pytest.mark.parametrize( ("scenario", "expected"), [ - ("baseline-graph", (True, True, False, 1, 0, 1)), - ("afd-eager", (False, False, False, 2, 1, 1)), - ("afd-graph", (False, True, False, 2, 1, 1)), - ("afd-graph-dbo", (False, True, True, 2, 1, 1)), + ("baseline-graph", (True, True, False, 1, 0, 1, 1)), + ("afd-eager", (False, False, False, 2, 1, 1, 1)), + ("afd-graph", (False, True, False, 2, 1, 1, 1)), + ("afd-graph-dbo", (False, True, True, 2, 1, 1, 1)), + ("afd-graph-dbo-2a2f", (False, True, True, 2, 2, 1, 1)), + ("afd-eager-async-cam", (False, False, False, 2, 2, 1, 2)), ], ) def test_configure_scenario_overwrites_fixed_topology_and_features( @@ -429,8 +512,8 @@ def test_configure_scenario_overwrites_fixed_topology_and_features( args.num_attention_ranks, args.num_ffn_ranks, args.tp_size, + args.attention_tp_size, ) == expected - assert args.attention_tp_size == 1 assert args.ffn_tp_size == 1 if args.cuda_graph_full_decode_only: assert args.cudagraph_capture_size == 8 @@ -439,6 +522,32 @@ def test_configure_scenario_overwrites_fixed_topology_and_features( assert args.dbo_prefill_token_threshold == 8 +def test_async_cam_scenario_builds_dp1tp2_attention_and_dp2tp1_ffn(): + args = _args() + args.scenario = "afd-eager-async-cam" + args.device_backend = "npu" + runner.configure_scenario(args) + runner.validate_topology(args, ["0", "1"], ["2", "3"]) + + attention_command = runner.build_vllm_command(args, role="attention") + ffn_command = runner.build_vllm_command(args, role="ffn") + + assert attention_command[attention_command.index("--data-parallel-size") + 1] == "1" + assert ( + attention_command[attention_command.index("--tensor-parallel-size") + 1] == "2" + ) + assert ffn_command[ffn_command.index("--data-parallel-size") + 1] == "2" + assert ffn_command[ffn_command.index("--tensor-parallel-size") + 1] == "1" + attention_config = json.loads( + attention_command[attention_command.index("--additional-config") + 1], + )["afd"] + assert attention_config["connector"] == runner.ASYNC_AFD_CONNECTOR + assert attention_config["async"] is True + assert attention_config["compute_gate_on_attention"] is True + assert attention_config["connector_extra_config"]["attn_ranks_per_dp"] == 2 + assert "--enable-expert-parallel" in ffn_command + + def test_build_baseline_command_uses_native_single_process_graph_server(): args = _args() args.scenario = "baseline-graph" @@ -1115,6 +1224,46 @@ def test_run_gsm8k_evaluation_rejects_accuracy_that_does_not_meet_gate( runner.run_gsm8k_evaluation(args) +@pytest.mark.parametrize( + ("body", "error"), + [ + ({"choices": [{"text": "B"}]}, None), + ({"choices": []}, "no choices"), + ({"choices": [{"text": ""}]}, "no text"), + ({"choices": [{"text": 32}]}, "no text"), + ], +) +def test_run_completion_evaluation_validates_one_32_token_request( + monkeypatch, + body, + error, +): + args = _args() + args.scenario = "afd-eager-async-cam" + runner.configure_scenario(args) + request = {} + + def fake_urlopen(http_request, timeout): + request.update(json.loads(http_request.data)) + request["timeout"] = timeout + return io.BytesIO(json.dumps(body).encode()) + + monkeypatch.setattr(runner.urllib.request, "urlopen", fake_urlopen) + + if error: + with pytest.raises(RuntimeError, match=error): + runner.run_completion_evaluation(args) + else: + runner.run_completion_evaluation(args) + assert request == { + "model": "deepseek-v2-lite-afd-attention", + "prompt": runner.ACCOUNTING_PROMPT, + "max_tokens": 32, + "temperature": 0, + "timeout": 120, + } + + def test_ensure_processes_alive_reports_exited_process_returncode(): process = argparse.Namespace(poll=lambda: 17)