[Feature] Add fault tolerance with retry and MC2 scale-down on Ascend NPU

[fangyuchu]

What this PR does / why we need it?

In DP+EP (Data Parallelism + Expert Parallelism) MoE deployments, a dead DP rank
makes the EP all2all on surviving ranks block indefinitely, hanging the whole
cluster. Upstream vLLM introduced a fault-tolerance framework
(vllm-project/vllm#44428) that detects the fault, aborts in-flight requests and
lets an external orchestrator trigger coordinated recovery, but Ascend NPU needs
its own backend.

This PR provides the complete Ascend NPU fault-tolerance backend in two layers:

1. Retry-based recovery for transient faults (foundation). Reuses the
   upstream GPUWorkerSentinel flow (clean worker state + DP-group
   re-initialization on retry) and adds the NPU-specific pieces upstream cannot
   cover: aborts hung NPU communication ops via torch.npu.npu.set_op_timeout_ms
   (new ft_communication_abort_timeout ascend config) and applies the CPU
   distributed timeout to DP-sync process groups, so peer failures surface as
   exceptions instead of hanging.

2. Scale-down recovery for MC2 (this PR's focus). When a rank is confirmed
   dead, retry alone cannot recover - survivors keep routing MoE tokens to the
   dead EP rank and the all2all keeps failing. This mirrors the upstream GPU
   scale_down flow while adapting the mask, routing and weight layers to
   Ascend's MC2 all2all backend:

   • npu_communicator: turns the no-op MC2 all2all manager into the owner of
     the dead-rank mask, encoded as the MC2 elastic_info tensor (flag, dense EP
     size, shared/physical expert counts plus orig-to-dense and dense-to-orig rank
     tables), rebuilt in place so captured graphs stay valid.
   • npu_worker_sentinel: adds scale_down with precondition validation (v2
     model runner, EPLB redundancy, no fused_mc2/hierarchy comm, dispatch_v2),
     dead EP-rank masking, expert redistribution with routing-table refresh and
     densification, selective on-disk expert weight reload, MC2 physical-expert
     width shrink, and a dummy-batch runnability check.
   • eplb_redistribute: NPU expert redistribution + weight reload mirroring
     process_weights_after_loading per slot (transpose, FRACTAL_NZ cast, quant
     scales) for unquantized and W8A8 dynamic schemes, plus
     densify_routing_table_physical_ids.
   • token_dispatcher / patch_distributed: pass elastic_info to MC2
     dispatch/combine and carry dead_dp_ranks so the DP allreduce neutralizes
     dead columns.
   • Clears a poisoned NPU stream on the first device error
     (fault_barrier_wrapper) so teardown cannot crash the worker before
     recovery runs.

Does this PR introduce any user-facing change?

Yes, when --enable-fault-tolerance is used on Ascend NPU:

• A new ascend config ft_communication_abort_timeout controls how long a hung
  NPU operator (e.g. a communication op) is allowed to run before being aborted
  with a timeout exception. When > 0, it derives HCCL_EVENT_TIMEOUT /
  HCCL_EXEC_TIMEOUT and calls torch.npu.set_op_timeout_ms (default 0,
  disabled).
• Scale-down recovery now supports MC2 deployments and requires EPLB with
  num_redundant_experts > 0 and the v2 model runner.
• No API/CLI interface changes beyond the upstream FT flags.

Example startup command for a 4-NPU DP=4 Qwen3-30B-A3B-W8A8 deployment (rank 0):

# Rank 0
ASCEND_RT_VISIBLE_DEVICES=0 \
HCCL_BUFFSIZE=2048 \
HCCL_EXEC_TIMEOUT=1 \
vllm serve /home/xxxxx/Qwen3-30B-A3B-W8A8/ \
  --tensor-parallel-size 1 \
  --data-parallel-size 4 \
  --trust-remote-code \
  --gpu-memory-utilization 0.9 \
  --max-model-len 1024 \
  --enable-expert-parallel \
  --quantization ascend \
  --max-num-batched-tokens 32 \
  --cpu-distributed-timeout-seconds 30 \
  --enable-fault-tolerance \
  --enable-eplb \
  --compilation-config '{"cudagraph_mode":"FULL_AND_PIECEWISE","cudagraph_capture_sizes":[4,8,12,16,20,24,28,32]}' \
  --fault-tolerance-config '{"engine_recovery_timeout_sec":32000}' \
  --eplb-config '{"num_redundant_experts":64, "use_async": false}' \
  --data-parallel-rank 0 \
  --port 8111 \
  --additional-config '{"ft_communication_abort_timeout": 30}' 

# Ranks 1/2/3 use the same command with:
#   ASCEND_RT_VISIBLE_DEVICES={1,2,3}
#   --data-parallel-rank {1,2,3}
#   --port {8112,8113,8114}

How was this patch tested?

Added DP=4 end-to-end coverage in tests/e2e/pull_request/four_card/fault_tolerance/test_fault_tolerance_e2e.py:

• test_injected_fault_retry_recovers_all_ranks — injected fault on rank 3 drives all 4 engines UNHEALTHY; retry restores all ranks to HEALTHY and factual-prompt serving resumes.
• test_scale_down_removes_dead_rank_and_recovers — SIGKILL of rank 3's worker marks survivors UNHEALTHY and the victim DEAD (DEAD rejects retry); scale_down re-hosts the dead rank's EPLB experts on the survivors, and all survivors return to HEALTHY and answer factual prompts correctly.

vLLM main: vllm-project/vllm@b2f6858

• vLLM main: vllm-project/vllm@ced6857

[github-actions]

👋 Hi! Thank you for contributing to the vLLM Ascend project. The following points will speed up your PR merge:‌‌

• A PR should do only one thing, smaller PRs enable faster reviews.
• Every PR should include unit tests and end-to-end tests ‌to ensure it works and is not broken by other future PRs.
• Write the commit message by fulfilling the PR description to help reviewer and future developers understand.

If CI fails, you can run linting and testing checks locally according Contributing and Testing.

---

Tip

💡 Consider Linking a Related Issue or RFC

Your PR title contains the [Feature] tag, indicating a bug fix or new feature.

Linking a related issue or RFC in the PR description is strongly encouraged — it gives reviewers helpful context and speeds up the review. You can use any of these keywords:

• Fixes #<issue_number>
• Closes #<issue_number>
• Resolves #<issue_number>
• Refs #<rfc_or_issue_number> (for RFCs)

🙏 Thanks for helping us keep the project well-organized!

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a robust fault-tolerance framework for Ascend NPU deployments, specifically addressing hanging issues in MoE models during rank failures. It implements a two-layered recovery mechanism: retry-based recovery for transient errors and scale-down recovery for permanent rank failures in MC2 deployments. The changes include NPU-specific communication timeouts, dynamic expert redistribution, and weight reloading, ensuring cluster stability and seamless recovery without manual intervention.

Highlights

• Retry-based Recovery: Implemented retry-based recovery for transient faults, utilizing NPU-specific operator timeouts to prevent cluster hangs.
• MC2 Scale-down Recovery: Added support for scale-down recovery in MC2 deployments, enabling dead-rank masking, expert redistribution, and weight reloading.
• Elastic Information Integration: Integrated an elastic_info tensor into MC2 dispatch and combine operators to dynamically handle rank failures.
• End-to-End Testing: Introduced a comprehensive test suite for fault tolerance on Ascend NPU, covering both retry and scale-down scenarios.

New Features

🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Ignored Files

• Ignored by pattern: .github/workflows/** (1)
  • .github/workflows/scripts/test_config.yaml

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

Suggested PR Title:

[Ops][Feature] Implement Fault Tolerance and Scale-Down Support for Ascend NPU

Suggested PR Summary:

### What this PR does / why we need it?
This pull request implements end-to-end fault tolerance and scale-down support for Ascend NPUs in vLLM. It introduces the `WorkerSentinel` for NPU, which manages device resets, collective process group re-initialization, and expert redistribution. It also adds an `All2All-manager` adapter (`_NpuAll2AllManager`) to handle elastic masking for MC2 dispatch/combine operators, and implements expert weight reloading from disk in `eplb_redistribute.py`. Additionally, it includes end-to-end tests to validate fault injection, retry recovery, and worker-kill scenarios.

Feedback on the implementation highlights several critical issues:
- Loading all expert weights into CPU memory during redistribution can cause CPU OOM; the loader should be restricted to only the required logical expert IDs.
- Resetting the device locally within the exception handler can cause deadlocks due to collective process group re-initialization; `reset_device` should support skipping this step during local quarantine.
- Wrapping caught exceptions in a generic `RuntimeError` in `synchronize_input_prep` hides the original exception type from recovery handlers.
- The host tensor `_elastic_info_host` should be allocated in pinned memory to ensure non-blocking asynchronous copies to the NPU.

### Does this PR introduce _any_ user-facing change?
Yes, it introduces fault-tolerance configuration options for Ascend NPU, such as `ft_communication_abort_timeout` in `AscendConfig`, and enables the `--enable-fault-tolerance` flag on Ascend platforms.

### How was this patch tested?
The patch was tested using the newly added end-to-end fault-tolerance tests on Ascend NPU (`test_fault_tolerance_e2e.py`), which cover injected fault retry recovery and worker process termination.

[gemini-code-assist]

Setting loader.local_expert_ids = None forces the DefaultModelLoader to load all experts of the model from disk. For large MoE models, loading all expert weights into CPU memory simultaneously is highly inefficient and can easily trigger a CPU Out-Of-Memory (OOM) error.

Since the required logical expert IDs for this rank are already known and stored in local_slots, we should restrict the loader to only load those specific experts.

Suggested change

	loader.local_expert_ids = None
	loader.local_expert_ids = {logical_id for _, logical_id in local_slots}

[gemini-code-assist]

Calling sentinel.reset_device() locally within the exception handler will trigger a collective process group re-initialization. Since other ranks are not participating yet, this rank will block indefinitely inside reinit_process_group, causing the worker to hang.

We should disable process group re-initialization during the local quarantine reset, and only perform it during the coordinated retry call.

Suggested change

	sentinel.reset_device()
	sentinel.reset_device(reinit_pg=False)

[205150940]

并未出现hang住情况，无需增加传参

[gemini-code-assist]

To prevent deadlocks during local quarantine resets, we should allow reset_device to skip the collective reinit_process_group call when requested. Re-initializing the process group is a collective operation that must only be performed when all ranks participate (i.e., during a coordinated retry).

Suggested change

	def reset_device(self) -> None:
	NPUPlatform.set_device(self.device)
	torch_npu.npu.stop_device(self.device.index)
	torch_npu.npu.restart_device(self.device.index)
	torch_npu.distributed.reinit_process_group(None, False)
	torch.npu.synchronize()
	def reset_device(self, reinit_pg: bool = True) -> None:
	NPUPlatform.set_device(self.device)
	torch_npu.npu.stop_device(self.device.index)
	torch_npu.npu.restart_device(self.device.index)
	if reinit_pg:
	torch_npu.distributed.reinit_process_group(None, False)
	torch.npu.synchronize()

[205150940]

同上，无需增加配置项

[gemini-code-assist]

Wrapping the caught exception in a generic RuntimeError hides the original exception type from upstream fault-tolerance and recovery handlers. This can interfere with decision-making logic that relies on specific exception types (e.g., distinguishing between transient communication timeouts and fatal model execution errors).

We should re-raise the original exception directly to preserve its type and traceback.

        except Exception:
            raise

[205150940]

此处需要手动抛出异常，避免进程hang住

[gemini-code-assist]

The host tensor self._elastic_info_host is copied to the NPU device tensor using non_blocking=True (at line 100). For non_blocking=True to be safe and actually asynchronous, the host tensor must be allocated in pinned memory (pin_memory()). Otherwise, the copy operation falls back to a synchronous copy, losing any asynchronous benefits.

Suggested change

	self._elastic_info_host = torch.zeros(size, dtype=torch.int32)
	self._elastic_info_host = torch.zeros(size, dtype=torch.int32, pin_memory=True)

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.